Workflow

library(mini007)

retrieve_open_ai_credential <- function() {
  Sys.getenv("OPENAI_API_KEY")
}

openai_4_1_mini <- ellmer::chat(
  name = "openai/gpt-4.1-mini",
  credentials = retrieve_open_ai_credential,
  echo = "none"
)

A Workflow is a predefined, sequential pipeline of processing units called Stations. Each Station receives the output of the previous one as its input, transforms it, and passes the result forward. Routes define which Station comes next and can be gated by optional condition functions that inspect the current output to decide the execution path.

Workflows complement Agent and LeadAgent:

• An Agent is LLM-driven and decides its own steps dynamically.
• A LeadAgent decomposes a prompt and delegates subtasks to agents automatically.
• A Workflow gives you explicit control over every step of the pipeline — you decide the order, the branching logic, and which handler runs where.

There are two main building blocks:

• add_station(name, handler) — registers a named processing unit. The handler can be an Agent, a WorkflowAgent, or a plain R function.
• add_route(from, to) — connects two Stations in sequence.

A minimal linear pipeline

The simplest Workflow is a straight chain of Stations executed one after the other. Here we wire three specialised agents: a researcher that gathers facts, a writer that shapes them into a paragraph, and an editor that polishes the result.

researcher <- Agent$new(
  name = "researcher",
  instruction = "You are a research assistant. Provide concise, accurate facts on the given topic in 3 max",
  llm_object  = openai_4_1_mini
)

writer <- Agent$new(
  name = "writer",
  instruction = "You are a skilled writer. Turn the research notes you receive into a single, engaging paragraph.",
  llm_object  = openai_4_1_mini
)

editor <- Agent$new(
  name = "editor",
  instruction = "You are a copy editor. Polish the paragraph you receive for grammar, clarity, and conciseness. Return only the final text.",
  llm_object  = openai_4_1_mini
)

translator <- Agent$new(
  name = "translator", 
  instruction = "You're a English-German Translater, translate text from English to German", 
  llm_object = openai_4_1_mini
)

article_pipeline <- Workflow$new(
  name        = "article pipeline",
  description = "Research → Write → Edit -> Transalte", 
  use_cache = TRUE
)

article_pipeline$add_station(name = "research", handler = researcher, )
article_pipeline$add_station(name = "write", handler = writer)
article_pipeline$add_station(name = "edit", handler = editor)
article_pipeline$add_station(name = "translate", handler = translator)

article_pipeline$add_route(from = "research", to = "write")
article_pipeline$add_route(from = "write", to = "edit")
article_pipeline$add_route(from = "edit", to = "translate")

article_pipeline$set_entry(station_name = "research")

Calling $run() executes every Station in order. Each Station’s output becomes the next Station’s input automatically.

result <- article_pipeline$run("The potential of Algeria when it comes to tourism in the Sahara")
result

Die Sahara, die etwa 80 % Algeriens bedeckt, beeindruckt mit atemberaubenden 
Landschaften und einem reichen kulturellen Erbe, einschließlich der 
Felsformationen von Tassili n'Ajjer und der antiken Berberzivilisationen. Diese
majestätische Region bietet Reisenden eine Vielzahl von Attraktionen, von 
ruhigen Wüstenoasen und hohen Sanddünen bis hin zu prähistorischer Höhlenkunst 
und lebendigen Tuareg-Gemeinschaften, was sie zu einem faszinierenden Reiseziel
für Abenteuerlustige und Kulturinteressierte macht. Obwohl infrastrukturelle 
Herausforderungen und Sicherheitsbedenken historisch das Tourismuswachstum 
behinderten, positionieren Entwicklungsfortschritte und eine zunehmende 
Stabilität die algerische Sahara als eine vielversprechende und weitgehend 
unerschlossene Region für einzigartige Wüstentourismus-Erlebnisse.

The run_history attribute records every execution, including the full trace of inputs and outputs for each Station:

article_pipeline$run_history

[[1]]
[[1]]$input
[1] "The potential of Algeria when it comes to tourism in the Sahara"

[[1]]$output
Die Sahara, die etwa 80 % Algeriens bedeckt, beeindruckt mit atemberaubenden 
Landschaften und einem reichen kulturellen Erbe, einschließlich der 
Felsformationen von Tassili n'Ajjer und der antiken Berberzivilisationen. Diese
majestätische Region bietet Reisenden eine Vielzahl von Attraktionen, von 
ruhigen Wüstenoasen und hohen Sanddünen bis hin zu prähistorischer Höhlenkunst 
und lebendigen Tuareg-Gemeinschaften, was sie zu einem faszinierenden Reiseziel
für Abenteuerlustige und Kulturinteressierte macht. Obwohl infrastrukturelle 
Herausforderungen und Sicherheitsbedenken historisch das Tourismuswachstum 
behinderten, positionieren Entwicklungsfortschritte und eine zunehmende 
Stabilität die algerische Sahara als eine vielversprechende und weitgehend 
unerschlossene Region für einzigartige Wüstentourismus-Erlebnisse.

[[1]]$steps
[1] 4

[[1]]$trace
[[1]]$trace[[1]]
[[1]]$trace[[1]]$step
[1] 1

[[1]]$trace[[1]]$station
[1] "research"

[[1]]$trace[[1]]$input
[1] "The potential of Algeria when it comes to tourism in the Sahara"

[[1]]$trace[[1]]$output
1. Algeria's Sahara covers about 80% of the country, featuring vast deserts, 
unique landscapes like the Tassili n'Ajjer rock formations, and rich cultural 
heritage from ancient Berber civilizations.  
2. It offers attractions such as desert oases, prehistoric cave art, sand 
dunes, and traditional Tuareg communities, appealing to adventure and cultural 
tourism.  
3. Infrastructure and security concerns have limited tourism growth, but with 
development and stability, Algeria's Sahara has significant potential as a 
unique and untapped desert tourism destination.


[[1]]$trace[[2]]
[[1]]$trace[[2]]$step
[1] 2

[[1]]$trace[[2]]$station
[1] "write"

[[1]]$trace[[2]]$input
1. Algeria's Sahara covers about 80% of the country, featuring vast deserts, 
unique landscapes like the Tassili n'Ajjer rock formations, and rich cultural 
heritage from ancient Berber civilizations.  
2. It offers attractions such as desert oases, prehistoric cave art, sand 
dunes, and traditional Tuareg communities, appealing to adventure and cultural 
tourism.  
3. Infrastructure and security concerns have limited tourism growth, but with 
development and stability, Algeria's Sahara has significant potential as a 
unique and untapped desert tourism destination.

[[1]]$trace[[2]]$output
Covering approximately 80% of Algeria, the vast Sahara Desert is a land of 
breathtaking landscapes and rich cultural heritage, home to stunning features 
like the Tassili n'Ajjer rock formations and ancient Berber civilizations. This
majestic region offers travelers an array of attractions, from tranquil desert 
oases and towering sand dunes to prehistoric cave art and vibrant Tuareg 
communities, making it a captivating destination for both adventure seekers and
cultural enthusiasts. Although infrastructure challenges and security concerns 
have historically limited tourism growth, ongoing development and improving 
stability position Algeria’s Sahara as a promising and largely untapped 
frontier for unique desert tourism experiences.


[[1]]$trace[[3]]
[[1]]$trace[[3]]$step
[1] 3

[[1]]$trace[[3]]$station
[1] "edit"

[[1]]$trace[[3]]$input
Covering approximately 80% of Algeria, the vast Sahara Desert is a land of 
breathtaking landscapes and rich cultural heritage, home to stunning features 
like the Tassili n'Ajjer rock formations and ancient Berber civilizations. This
majestic region offers travelers an array of attractions, from tranquil desert 
oases and towering sand dunes to prehistoric cave art and vibrant Tuareg 
communities, making it a captivating destination for both adventure seekers and
cultural enthusiasts. Although infrastructure challenges and security concerns 
have historically limited tourism growth, ongoing development and improving 
stability position Algeria’s Sahara as a promising and largely untapped 
frontier for unique desert tourism experiences.

[[1]]$trace[[3]]$output
Covering approximately 80% of Algeria, the vast Sahara Desert boasts 
breathtaking landscapes and a rich cultural heritage, including the Tassili 
n'Ajjer rock formations and ancient Berber civilizations. This majestic region 
offers travelers a range of attractions, from tranquil desert oases and 
towering sand dunes to prehistoric cave art and vibrant Tuareg communities, 
making it a captivating destination for both adventure seekers and cultural 
enthusiasts. Although infrastructure challenges and security concerns have 
historically hindered tourism growth, ongoing development and improving 
stability position Algeria’s Sahara as a promising and largely untapped 
frontier for unique desert tourism experiences.


[[1]]$trace[[4]]
[[1]]$trace[[4]]$step
[1] 4

[[1]]$trace[[4]]$station
[1] "translate"

[[1]]$trace[[4]]$input
Covering approximately 80% of Algeria, the vast Sahara Desert boasts 
breathtaking landscapes and a rich cultural heritage, including the Tassili 
n'Ajjer rock formations and ancient Berber civilizations. This majestic region 
offers travelers a range of attractions, from tranquil desert oases and 
towering sand dunes to prehistoric cave art and vibrant Tuareg communities, 
making it a captivating destination for both adventure seekers and cultural 
enthusiasts. Although infrastructure challenges and security concerns have 
historically hindered tourism growth, ongoing development and improving 
stability position Algeria’s Sahara as a promising and largely untapped 
frontier for unique desert tourism experiences.

[[1]]$trace[[4]]$output
Die Sahara, die etwa 80 % Algeriens bedeckt, beeindruckt mit atemberaubenden 
Landschaften und einem reichen kulturellen Erbe, einschließlich der 
Felsformationen von Tassili n'Ajjer und der antiken Berberzivilisationen. Diese
majestätische Region bietet Reisenden eine Vielzahl von Attraktionen, von 
ruhigen Wüstenoasen und hohen Sanddünen bis hin zu prähistorischer Höhlenkunst 
und lebendigen Tuareg-Gemeinschaften, was sie zu einem faszinierenden Reiseziel
für Abenteuerlustige und Kulturinteressierte macht. Obwohl infrastrukturelle 
Herausforderungen und Sicherheitsbedenken historisch das Tourismuswachstum 
behinderten, positionieren Entwicklungsfortschritte und eine zunehmende 
Stabilität die algerische Sahara als eine vielversprechende und weitgehend 
unerschlossene Region für einzigartige Wüstentourismus-Erlebnisse.

Mixing agents with plain R functions

Stations do not have to be Agent objects. Any R function that accepts a single character argument and returns a character value is a valid handler. This makes it easy to add pre-processing or post-processing steps without an LLM call.

# Plain R function stations, no LLM involved
normalise <- function(text) {
  trimws(gsub("\\s+", " ", text))
}

add_markdown_header <- function(text) {
  paste0("## Summary\n\n", text)
}

summariser <- Agent$new(
  name        = "summariser",
  instruction = "Summarise the text you receive into exactly three concise bullet points.",
  llm_object  = openai_4_1_mini
)

format_pipeline <- Workflow$new("format and summarise")

format_pipeline$add_station(
  name = "normalise",
  handler = normalise, 
  description = "Collapse whitespace"
)
format_pipeline$add_station(
  name = "summarise", 
  handler = summariser, 
  description = "LLM bullet summary"
)
format_pipeline$add_station(
  name = "add_header", 
  handler = add_markdown_header, 
  description = "Prepend markdown header"
)

format_pipeline$add_route(from = "normalise",  to = "summarise")
format_pipeline$add_route(from = "summarise",  to = "add_header")

messy_text <- "  The   2026    worldcup is    approachine.   Algeria's first game   is against     Argentina          "

format_pipeline$run(messy_text)

[1] "## Summary\n\n- The 2026 World Cup is approaching.  \n- Algeria's first match will be against Argentina.  \n- The opening game is significant for both teams."

Caching station results

When use_cache = TRUE (the default), every Station’s output is stored keyed by its name and the input it received. A subsequent $run() call with the same input skips re-invoking any handler and returns the cached result immediately — saving both time and LLM costs when you are iterating on later Stations.

cached_pipeline <- Workflow$new(
  name      = "cached-pipeline",
  use_cache = TRUE
)

drafter <- Agent$new(
  name        = "drafter",
  instruction = "Write a two-sentence draft on the topic provided.",
  llm_object  = openai_4_1_mini
)

refiner <- Agent$new(
  name        = "refiner",
  instruction = "Improve the draft you receive. Make it more vivid and precise.",
  llm_object  = openai_4_1_mini
)

cached_pipeline$add_station(name = "draft", handler = drafter)
cached_pipeline$add_station(name = "refine", handler = refiner)
cached_pipeline$add_route(from = "draft", to = "refine")

# First run — both stations are executed
start <- Sys.time()
first_run <- cached_pipeline$run("Rare fish in the Algerian see")
first_run

The Algerian Sea harbors a remarkable array of rare marine life, including the 
endangered Mediterranean sand tiger shark and the distinctive Algerian grouper.
These extraordinary species play a crucial role in maintaining the delicate 
balance of the marine ecosystem. To safeguard their survival, intensified 
conservation efforts are urgently needed to shield their fragile habitats from 
the escalating threats of pollution and overfishing.

end <- Sys.time()
print(end - start)

Time difference of 20.84 secs

# Second run with the same input — cache is served, no LLM calls are made
start <- Sys.time()
second_run <- cached_pipeline$run("Rare fish in the Algerian see")
second_run

The Algerian Sea harbors a remarkable array of rare marine life, including the 
endangered Mediterranean sand tiger shark and the distinctive Algerian grouper.
These extraordinary species play a crucial role in maintaining the delicate 
balance of the marine ecosystem. To safeguard their survival, intensified 
conservation efforts are urgently needed to shield their fragile habitats from 
the escalating threats of pollution and overfishing.

end <- Sys.time()
print(end - start)

Time difference of 0.05784392 secs

To force a fresh execution, call $clear_cache():

cached_pipeline$clear_cache()

# Fresh run after clearing the cache
start <- Sys.time()
fresh_run <- cached_pipeline$run("Rare fish in the Algerian see")
fresh_run

The Algerian Sea is a sanctuary for several rare fish species, including the 
elusive Mediterranean spearfish and the critically endangered dusky grouper. 
Preserving these extraordinary species is essential for sustaining the region’s
rich marine biodiversity and ensuring the health and resilience of its 
underwater ecosystems.

end <- Sys.time()
print(end - start)

Time difference of 3.006597 secs

Conditional routing

Routes can carry a condition, a function that inspects the current Station’s output and returns TRUE or FALSE. When a Station has multiple outgoing Routes, conditional ones are evaluated first (in the order they were added). The first condition that returns TRUE determines the next Station. If none match, the first unconditional Route is used as the default fallback.

This makes it possible to build branching pipelines where the path taken depends on what an earlier Station produces.

classifier <- Agent$new(
  name        = "classifier",
  instruction = 'Classify the sentiment of the text. Reply with exactly one word: "positive" or "negative".',
  llm_object  = openai_4_1_mini
)

positive_handler <- Agent$new(
  name        = "positive_handler",
  instruction = "Write a warm, enthusiastic one-sentence reply to the following positive message.",
  llm_object  = openai_4_1_mini
)

negative_handler <- Agent$new(
  name        = "negative_handler",
  instruction = "Write an empathetic, solution-focused one-sentence reply to the following negative message.",
  llm_object  = openai_4_1_mini
)

sentiment_router <- Workflow$new("sentiment router")

sentiment_router$add_station(name = "classify",  handler = classifier)
sentiment_router$add_station(name = "reply_pos", handler = positive_handler)
sentiment_router$add_station(name = "reply_neg", handler = negative_handler)

# Conditional routes — evaluated against the classifier's output
sentiment_router$add_route(
  from      = "classify",
  to        = "reply_pos",
  condition = function(out) grepl("positive", tolower(out))
)
sentiment_router$add_route(
  from      = "classify",
  to        = "reply_neg",
  condition = function(out) grepl("negative", tolower(out))
)

sentiment_router$set_entry(station_name = "classify")

sentiment_router$run("I absolutely love this product, it changed my life!")

Thank you so much for the positive vibes – they truly made my day brighter!

sentiment_router$run("Very disappointed. The quality is nothing like advertised.")

I’m sorry to hear you’re feeling this way—how can I help make things better for
you?

A more structured example uses a plain function as the router so the branching logic is deterministic and free of LLM calls.

Important pattern: the router station must be a passthrough — it returns the original text unchanged so that downstream agents receive the real content. The routing classification belongs in the route conditions, which inspect the station output and decide where it goes next.

brief_expander <- Agent$new(
  name        = "brief_expander",
  instruction = "The input is short. Expand it into a well-rounded paragraph of 2 sentences.",
  llm_object  = openai_4_1_mini
)

long_summariser <- Agent$new(
  name        = "long_summariser",
  instruction = "The input is long. Distill it into a single crisp sentence.",
  llm_object  = openai_4_1_mini
)

length_router <- Workflow$new("length router")

# The router station is a passthrough: it returns the text as-is.
# The length check lives in the route conditions below, not here.
length_router$add_station(name = "router", handler = function(text) text)
length_router$add_station(name = "short-to-long",  handler = brief_expander)
length_router$add_station(name = "long-to-short",   handler = long_summariser)

# Conditions receive the router's output (the original text) and decide the branch.
length_router$add_route(
  from      = "router",
  to        = "short-to-long",
  condition = function(x) nchar(x) < 80
)
length_router$add_route(
  from      = "router",
  to        = "long-to-short",
  condition = function(x) nchar(x) >= 80
)

length_router$set_entry(station_name = "router")

# Short input — routed to the expander
length_router$run("Fennec Algeria football team")

The Fennec Algeria football team, often referred to simply as "Les Fennecs," is
the national football team of Algeria known for its passionate style of play 
and rich football history. Representing the country in international 
competitions, they have earned recognition for their skillful players and 
significant achievements, including winning the Africa Cup of Nations multiple 
times.

# Long input — routed to the summariser
long_input <- paste(
  "Bees are among the most important pollinators on the planet.",
  "They transfer pollen from one flower to another, enabling plants to reproduce.",
  "Without bees, many of the fruits and vegetables we eat would disappear.",
  "Colony collapse disorder has threatened bee populations worldwide,",
  "raising serious concerns about food security and ecosystem stability."
)
length_router$run(long_input)

Bees play a crucial role in pollination essential for plant reproduction and 
food production, but their populations are threatened by colony collapse 
disorder, posing risks to food security and ecosystems.

Wrapping a workflow as an agent

$as_agent() converts any Workflow into a WorkflowAgent. The result exposes the same $invoke() interface as a regular Agent, which means it can be:

• registered with a LeadAgent as one of its sub-agents, or
• used as a Station handler inside another Workflow.

This lets you compose complex pipelines from simpler ones without rewriting any logic.

Using a WorkflowAgent inside a LeadAgent

# Inner workflow: research + summarise
research_wf <- Workflow$new("research workflow")

gatherer <- Agent$new(
  name = "gatherer",
  instruction = "Gather three key facts about the topic provided. Be concise.",
  llm_object  = openai_4_1_mini
)

condonser <- Agent$new(
  name = "condenser",
  instruction = "Condense the facts you receive into a single sentence.",
  llm_object  = openai_4_1_mini
)

research_wf$add_station(name = "gather", handler = gatherer)
research_wf$add_station(name = "condense", handler = condonser)
research_wf$add_route(from = "gather", to = "condense")

research_wf$set_entry(station_name = "gather")
# Wrap as a WorkflowAgent
research_agent <- research_wf$as_agent(
  name = "research_agent",
  instruction = "An agent that gathers and condenses facts on any topic into one sentence."
)

# Use the WorkflowAgent directly
research_agent$invoke("The city of Bejaia in Algeria")

Bejaia is a Mediterranean port city in northeastern Algeria, historically 
significant as a medieval commercial and cultural center during the Hammadid 
dynasty, and today recognized for its beautiful coastline, oil and gas 
industries, and role as a regional economic hub.

# Register with a LeadAgent alongside regular agents
translator <- Agent$new(
  name        = "translator",
  instruction = "Translate the text you receive from English into German",
  llm_object  = openai_4_1_mini
)

lead <- LeadAgent$new(
  name       = "Lead",
  llm_object = openai_4_1_mini
)

lead$register_agents(c(research_agent, translator))

lead$invoke("Tell me about the city of Bejaia in one sentence, then translate it into German")

Bejaia is a historic Mediterranean port city in northeastern Algeria’s Kabylie 
region, known for its medieval commerce, vibrant Berber culture, scenic 
coastline, and significant oil and gas industry today.  
Bejaia ist eine historische Mittelmeer-Hafenstadt in der Kabylien-Region im 
Nordosten Algeriens, bekannt für ihren mittelalterlichen Handel, die lebendige 
Berberkultur, die malerische Küstenlinie und die heute bedeutende Öl- und 
Gasindustrie.

Nesting a WorkflowAgent inside another Workflow

# Outer workflow: use the wrapped inner workflow as a Station
presentation_wf <- Workflow$new("Presentation workflow")

presentation_wf$add_station(
  name    = "research",
  handler = research_agent  # WorkflowAgent used as a station handler
)

presenter <- Agent$new(
    name = "presenter",
    instruction = "Turn the one-sentence summary you receive into a polished three-sentence report.",
    llm_object  = openai_4_1_mini
  )

presentation_wf$add_station(
  name = "present",
  handler = presenter
)

presentation_wf$add_route(from = "research", to = "present")
presentation_wf$set_entry(station_name = "research")

presentation_wf$run("Algeria and Coffee")

Algeria’s coffee culture is deeply influenced by Mediterranean and French 
traditions, with cafés serving as important social hubs in communities. The 
culture primarily depends on imported coffee and features popular beverages 
such as espresso, Turkish coffee, and café au lait. These drinks are often 
enjoyed alongside traditional pastries, highlighting the rich blend of 
historical and cultural flavors in Algerian coffee enjoyment.

Human In The Loop (HITL)

$set_hitl(steps) lets you pause the workflow at specific steps so a human can inspect the Station’s output before execution continues. Steps are numbered from 1 in execution order — the same counter shown in the run() console output.

When execution reaches a HITL step, it prints the Station name, the input it received, and the output it produced, then offers three choices:

1. Continue — use the output as-is and move to the next Station.
2. Edit — type a replacement output; that value is forwarded to the next Station (and cached, if caching is on).
3. Stop — abort the workflow immediately.

HITL only fires on fresh executions; cache hits are skipped since the result has already been reviewed in a prior run.

hitl_pipeline <- Workflow$new("hitl pipeline", use_cache = FALSE)

drafter <- Agent$new(
  name = "drafter",
  instruction = "Write a short two-sentence paragraph on the topic provided.",
  llm_object = openai_4_1_mini
)

translator <- Agent$new(
  name = "translator",
  instruction = "Translate the text you receive from English into Spanish.",
  llm_object = openai_4_1_mini
)

hitl_pipeline$add_station(name = "draft", handler = drafter)
hitl_pipeline$add_station(name = "translate", handler = translator)

hitl_pipeline$add_route(from = "draft", to = "translate")

# Pause after step 1 (the drafter) so a human can review the draft
# before it is sent to the translator
hitl_pipeline$set_hitl(steps = 1)

hitl_pipeline$run("The discovery of penicillin")

You can set HITL at multiple steps at once:

hitl_pipeline$set_hitl(steps = c(1, 2))

Visualising a workflow

Call $visualize() on any Workflow to render an interactive directed graph. Stations appear as rounded blue boxes, conditional Routes as dashed arrows labelled “cond”, and the entry Station is reached via a green START node.

article_pipeline$visualize()

sentiment_router$visualize()

The graph updates automatically to reflect whatever Stations and Routes are currently registered, making it a convenient tool for checking that the pipeline is wired correctly before running it.