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

Algerien, Heimat des größten Teils der Sahara, bietet weite und vielfältige 
Landschaften – von ausgedehnten Sanddünen und uralten Oasen bis hin zu 
bemerkenswerten Felsformationen – und ist somit ein erstklassiges Ziel für 
Abenteuertourismus. Sein reiches kulturelles Erbe wird durch historische 
Stätten wie die zum UNESCO-Weltkulturerbe gehörenden Felskunst von Tassili 
n'Ajjer und gut erhaltene römische Ruinen, die verstreut in der Wüste liegen, 
präsentiert, was es für Kulturreisende besonders attraktiv macht. Obwohl die 
Tourismusinfrastruktur Algeriens noch unterentwickelt ist, versprechen laufende
Sicherheitsverbesserungen und verstärkte Investitionen großes Potenzial, um den
vollen Schatz des Sahara-Tourismus zu heben und das Land als faszinierendes 
Ziel für Entdecker und Geschichtsinteressierte gleichermaßen zu positionieren.

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
Algerien, Heimat des größten Teils der Sahara, bietet weite und vielfältige 
Landschaften – von ausgedehnten Sanddünen und uralten Oasen bis hin zu 
bemerkenswerten Felsformationen – und ist somit ein erstklassiges Ziel für 
Abenteuertourismus. Sein reiches kulturelles Erbe wird durch historische 
Stätten wie die zum UNESCO-Weltkulturerbe gehörenden Felskunst von Tassili 
n'Ajjer und gut erhaltene römische Ruinen, die verstreut in der Wüste liegen, 
präsentiert, was es für Kulturreisende besonders attraktiv macht. Obwohl die 
Tourismusinfrastruktur Algeriens noch unterentwickelt ist, versprechen laufende
Sicherheitsverbesserungen und verstärkte Investitionen großes Potenzial, um den
vollen Schatz des Sahara-Tourismus zu heben und das Land als faszinierendes 
Ziel für Entdecker und Geschichtsinteressierte gleichermaßen zu positionieren.

[[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 has the largest portion of the Sahara Desert, offering vast and 
diverse landscapes including sand dunes, ancient oases, and unique rock 
formations ideal for adventure tourism.

2. Historical sites like the UNESCO World Heritage-listed Tassili n'Ajjer rock 
art and Roman ruins in the desert enhance its cultural tourism appeal.

3. Despite its potential, Algeria's tourism infrastructure is underdeveloped, 
but improving security and investment could significantly boost Sahara tourism 
prospects.


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

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

[[1]]$trace[[2]]$input
1. Algeria has the largest portion of the Sahara Desert, offering vast and 
diverse landscapes including sand dunes, ancient oases, and unique rock 
formations ideal for adventure tourism.

2. Historical sites like the UNESCO World Heritage-listed Tassili n'Ajjer rock 
art and Roman ruins in the desert enhance its cultural tourism appeal.

3. Despite its potential, Algeria's tourism infrastructure is underdeveloped, 
but improving security and investment could significantly boost Sahara tourism 
prospects.

[[1]]$trace[[2]]$output
Algeria, home to the largest portion of the Sahara Desert, boasts vast and 
diverse landscapes ranging from expansive sand dunes and ancient oases to 
remarkable rock formations, making it a prime destination for adventure 
tourism. The country's rich cultural heritage is exemplified by historical 
sites such as the UNESCO World Heritage-listed Tassili n'Ajjer rock art and 
well-preserved Roman ruins scattered throughout the desert, adding a compelling
dimension for cultural travelers. Although Algeria’s tourism infrastructure 
remains underdeveloped, ongoing improvements in security and increased 
investment hold great promise for unlocking the full potential of Sahara 
tourism, positioning the country as a captivating destination for explorers and
history enthusiasts alike.


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

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

[[1]]$trace[[3]]$input
Algeria, home to the largest portion of the Sahara Desert, boasts vast and 
diverse landscapes ranging from expansive sand dunes and ancient oases to 
remarkable rock formations, making it a prime destination for adventure 
tourism. The country's rich cultural heritage is exemplified by historical 
sites such as the UNESCO World Heritage-listed Tassili n'Ajjer rock art and 
well-preserved Roman ruins scattered throughout the desert, adding a compelling
dimension for cultural travelers. Although Algeria’s tourism infrastructure 
remains underdeveloped, ongoing improvements in security and increased 
investment hold great promise for unlocking the full potential of Sahara 
tourism, positioning the country as a captivating destination for explorers and
history enthusiasts alike.

[[1]]$trace[[3]]$output
Algeria, home to the largest part of the Sahara Desert, offers vast and diverse
landscapes—from expansive sand dunes and ancient oases to remarkable rock 
formations—making it a prime destination for adventure tourism. Its rich 
cultural heritage is showcased by historical sites such as the UNESCO World 
Heritage-listed Tassili n'Ajjer rock art and well-preserved Roman ruins 
scattered throughout the desert, adding appeal for cultural travelers. Although
Algeria’s tourism infrastructure remains underdeveloped, ongoing security 
improvements and increased investment hold great promise for unlocking the full
potential of Sahara tourism, positioning the country as a captivating 
destination for explorers and history enthusiasts alike.


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

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

[[1]]$trace[[4]]$input
Algeria, home to the largest part of the Sahara Desert, offers vast and diverse
landscapes—from expansive sand dunes and ancient oases to remarkable rock 
formations—making it a prime destination for adventure tourism. Its rich 
cultural heritage is showcased by historical sites such as the UNESCO World 
Heritage-listed Tassili n'Ajjer rock art and well-preserved Roman ruins 
scattered throughout the desert, adding appeal for cultural travelers. Although
Algeria’s tourism infrastructure remains underdeveloped, ongoing security 
improvements and increased investment hold great promise for unlocking the full
potential of Sahara tourism, positioning the country as a captivating 
destination for explorers and history enthusiasts alike.

[[1]]$trace[[4]]$output
Algerien, Heimat des größten Teils der Sahara, bietet weite und vielfältige 
Landschaften – von ausgedehnten Sanddünen und uralten Oasen bis hin zu 
bemerkenswerten Felsformationen – und ist somit ein erstklassiges Ziel für 
Abenteuertourismus. Sein reiches kulturelles Erbe wird durch historische 
Stätten wie die zum UNESCO-Weltkulturerbe gehörenden Felskunst von Tassili 
n'Ajjer und gut erhaltene römische Ruinen, die verstreut in der Wüste liegen, 
präsentiert, was es für Kulturreisende besonders attraktiv macht. Obwohl die 
Tourismusinfrastruktur Algeriens noch unterentwickelt ist, versprechen laufende
Sicherheitsverbesserungen und verstärkte Investitionen großes Potenzial, um den
vollen Schatz des Sahara-Tourismus zu heben und das Land als faszinierendes 
Ziel für Entdecker und Geschichtsinteressierte gleichermaßen zu positionieren.

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- This marks an important upcoming fixture for Algeria."

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 fish species, notably 
including the endangered Mediterranean sand tiger shark. These extraordinary 
marine inhabitants enrich the region’s vibrant biodiversity and underscore the 
urgent need for dedicated conservation initiatives to safeguard their delicate 
and diminishing habitats.

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

Time difference of 4.072386 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 fish species, notably 
including the endangered Mediterranean sand tiger shark. These extraordinary 
marine inhabitants enrich the region’s vibrant biodiversity and underscore the 
urgent need for dedicated conservation initiatives to safeguard their delicate 
and diminishing habitats.

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

Time difference of 0.05695391 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 vibrant sanctuary for an array of rare fish species, 
including the brilliantly colored Mediterranean rainbow wrasse and the striking
ornate wrasse, both celebrated for their vivid hues and limited geographic 
ranges. Safeguarding these extraordinary fish is essential to preserving the 
ecological balance and sustaining the rich biodiversity of this unique marine 
ecosystem.

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

Time difference of 2.939737 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—your positivity truly brightens my day!

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

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

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, also known as the Algerian national team, is 
highly regarded in African and international football for their skillful play 
and passionate style. Representing Algeria, the team has achieved significant 
success in regional and global competitions, earning a devoted fan base known 
as the "Desert Foxes."

# 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 are vital pollinators essential for plant reproduction and food 
production, but their populations are threatened by colony collapse disorder, 
risking food security and ecosystem health.

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, a historic northeastern Algerian port city situated between the 
Mediterranean Sea and the Kabylie Mountains, played a key role in medieval 
Mediterranean trade and today thrives on a diverse economy of fishing, 
agriculture, tourism, and Berber cultural heritage.

# 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 ist eine historische Küstenstadt im Nordosten Algeriens, die für ihren 
malerischen Mittelmeerhafen, ihre lebendige Berberkultur und ihre Lage in der 
Nähe der Kabylei-Berge bekannt ist.

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")

In Algeria, coffee is a popular social beverage, commonly prepared strong or 
spiced with cardamom. Cafés across the country serve as important community 
hubs where people gather to enjoy their coffee. Despite its cultural 
significance, Algeria relies on imported coffee beans because the local climate
and soil conditions are unsuitable for coffee cultivation.

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), mode = "interactive")

Non-blocking HITL: pause and resume (Not yet on CRAN)

The HITL above with mode="interactive" (the default) only works in an interactive console. Pass mode = "pause" to $set_hitl() and $run() returns immediately instead of blocking, handing you back a mini007_pending object that describes exactly where execution stopped. This is what lets HITL work from a Shiny app, a Plumber API, or any batch job: persist the pending object, resume it whenever the human actually responds, even in a different R session.

Let’s route a teaser about Tassili n’Ajjer, the Algerian Sahara’s prehistoric rock-art plateau through a drafter and a translator, pausing after the draft for review:

sahara_pipeline <- Workflow$new("sahara pipeline", use_cache = FALSE)

drafter <- Agent$new(
  name = "sahara_drafter",
  instruction = "Write a punchy two-sentence teaser about the Saharan destination provided.",
  llm_object = openai_4_1_mini
)

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

sahara_pipeline$add_station(name = "draft", handler = drafter)
sahara_pipeline$add_station(name = "translate", handler = translator)
sahara_pipeline$add_route(from = "draft", to = "translate")

# mode = "pause" instead of the default "console"
sahara_pipeline$set_hitl(steps = 1, mode = "pause")

pending <- sahara_pipeline$run("Tassili n'Ajjer, Algeria")
pending

── Pending HITL request - step 1 ──
Request id: "b3f1c2a0-..."
Station/Agent: draft
ℹ Input:
Tassili n'Ajjer, Algeria
ℹ Proposed output:
Deep in Algeria's Sahara, Tassili n'Ajjer hides 15,000 years of rock art
carved into sandstone cathedrals — a UNESCO World Heritage plateau most
travellers never set foot on.
! Call `$resume("b3f1c2a0-...", action = "continue"|"edit"|"abort")` to proceed.

is_pending(pending) returns TRUE for this object (and FALSE for a normal completed result). Once the human has weighed in, call $resume() with one of three actions:

# 1. Accept the draft as-is — the translator receives it next
sahara_pipeline$resume(pending$request_id, action = "continue")

# 2. Or edit it first. The edited text is what the translator receives
sahara_pipeline$resume(
  pending$request_id,
  action = "edit",
  value  = "Tassili n'Ajjer: 15,000 years of rock art etched into Saharan sandstone, in Algeria's far southeast."
)

# 3. Or abort altogether — raises an error, nothing downstream runs
sahara_pipeline$resume(pending$request_id, action = "abort")

If the workflow has more HITL steps ahead, $resume() returns another mini007_pending object instead of the final result — so a caller typically loops while (is_pending(result)), persisting result$request_id between iterations, until a plain character string comes back.

Note that a Workflow wrapped as a WorkflowAgent (via $as_agent()) must keep mode = "console" (the default): a WorkflowAgent is invoked synchronously from inside a Workflow Station or a LeadAgent step, and neither knows how to hand a paused run back to you — $invoke() raises an error if the wrapped workflow pauses.

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.

---

Per-station retry (Not yet on CRAN)

Real-world pipelines talk to external services — LLM providers, APIs, databases — that can fail transiently: rate-limit errors, network blips, brief provider outages. Rather than letting a single transient failure abort the entire run, you can attach a retry logic to any Station via two parameters of $add_station():

Parameter	Default	Meaning
max_retries	0	Extra attempts after the first failure
retry_delay	1	Seconds to sleep between attempts

The station is tried up to max_retries + 1 times in total. If it succeeds on any attempt, execution continues normally. If all attempts fail, the error is either forwarded to a fallback handler (see below) or re-raised.

Simulating a flaky service

# Factory that returns a handler which fails the first `fail_times` calls
make_flaky_service <- function(fail_times = 2L) {
  attempts <- 0
  function(text) {
    attempts <<- attempts + 1L
    if (attempts <= fail_times) {
      stop(sprintf("HTTP 429: Rate limit (attempt %d of %d)", attempts, fail_times + 1L))
    }
    paste("Service result:", toupper(text))
  }
}

retry_wf <- Workflow$new("retry pipeline", use_cache = FALSE)

retry_wf$add_station(
  name        = "call_service",
  handler     = make_flaky_service(fail_times = 2L),
  max_retries = 3,
  retry_delay = 0     # set to a positive number (e.g. 2) in production
)

retry_wf$run("summarise this document")

[1] "Service result: SUMMARISE THIS DOCUMENT"

The station fails on attempts 1 and 2, then succeeds on attempt 3. The workflow returns the successful result transparently.

Attaching retries to an Agent station

max_retries works identically for Agent stations — the LLM call is retried the same way.

fragile_pipeline <- Workflow$new("fragile pipeline")

fragile_pipeline$add_station(
  name        = "analyse",
  handler     = some_agent,      # any Agent or WorkflowAgent
  max_retries = 2,
  retry_delay = 3                # wait 3 s between retries
)

Behaviour when all retries are exhausted

When every attempt fails and no fallback is defined, the last error is re-raised and propagates up from $run().

always_fails_wf <- Workflow$new("always-fails pipeline", use_cache = FALSE)

always_fails_wf$add_station(
  name        = "broken_step",
  handler     = function(x) stop("Service permanently unavailable"),
  max_retries = 1,
  retry_delay = 0
)

tryCatch(
  always_fails_wf$run("input"),
  error = function(e) paste("Caught:", conditionMessage(e))
)

[1] "Caught: Service permanently unavailable"

---

Fallback handlers {#fallback-handlers} (not yet on CRAN)

A fallback is a function(input, error) attached to a specific Station. It is invoked only after all retry attempts for that Station have been exhausted, giving you a controlled degradation path instead of a hard failure.

wf$add_station(
  name     = "step",
  handler  = primary_handler,
  fallback = function(input, error) {
    # `input`  — the string the station received
    # `error`  — the condition object from the last failed attempt
    "default output"
  }
)

The fallback’s return value is used as the station’s output and the pipeline continues normally from there.

Simple graceful degradation

safe_wf <- Workflow$new("safe pipeline", use_cache = FALSE)

safe_wf$add_station(
  name     = "fetch_summary",
  handler  = function(x) stop("Connection refused"),
  fallback = function(input, err) {
    paste("(Summary unavailable — service error:", conditionMessage(err), ")\nOriginal:", input)
  }
)

safe_wf$run("Analyse this quarterly report")

[1] "(Summary unavailable — service error: Connection refused )\nOriginal: Analyse this quarterly report"

Fallback fires only after all retries

The fallback is a last resort. With max_retries > 0 the primary handler is tried repeatedly; the fallback only runs if every attempt fails.

attempts <- 0L

resilient_wf <- Workflow$new("resilient pipeline", use_cache = FALSE)

resilient_wf$add_station(
  name        = "external_call",
  handler     = function(x) {
    attempts <<- attempts + 1L
    stop(sprintf("Attempt %d failed", attempts))
  },
  max_retries = 2,
  retry_delay = 0,
  fallback    = function(input, err) {
    paste0("[Fallback after ", attempts, " attempt(s)] ", input)
  }
)

resilient_wf$run("Process this document")

[1] "[Fallback after 3 attempt(s)] Process this document"

Routing continues after a fallback

The fallback’s output flows into the next Station via the normal routing mechanism. The rest of the pipeline is unaware that a fallback fired.

routed_fallback_wf <- Workflow$new("routed fallback", use_cache = FALSE)

routed_fallback_wf$add_station(
  name     = "fetch",
  handler  = function(x) stop("API down"),
  fallback = function(input, err) paste("CACHED VERSION OF:", input)
)

routed_fallback_wf$add_station(
  "format",
  function(x) paste0("**", trimws(x), "**")
)

routed_fallback_wf$add_route("fetch", "format")

routed_fallback_wf$run("monthly sales figures")

[1] "**CACHED VERSION OF: monthly sales figures**"

Using a backup LLM as the fallback

A common production pattern is to call a premium model as the primary handler and fall back to a cheaper or local model when the primary is unavailable.

primary_agent  <- Agent$new("primary",  "You are a senior analyst.", premium_llm)
fallback_agent <- Agent$new("fallback", "You are a concise analyst.", cheap_llm)

pipeline <- Workflow$new("model-with-backup")

pipeline$add_station(
  name     = "analyse",
  handler  = primary_agent,
  max_retries = 1,
  retry_delay = 2,
  fallback = function(input, err) {
    fallback_agent$invoke(input)   # delegate to the backup model
  }
)

---

Parallel Stations (not yet on CRAN)

Multiple independent Stations can run concurrently using $set_daemons() and $add_parallel_group().

Multi-perspective parallel analysis

# Create three specialized agents for different perspectives
technical_agent <- Agent$new(
  name = "technical_analyst",
  instruction = paste(
    "You are a technical expert in AI and software engineering.",
    "Analyze topics from a technical/engineering perspective.",
    "Focus on: implementation details, technology stack, technical challenges.",
    "Keep responses to 2-3 sentences."
  ),
  llm_object = openai_4_1_mini
)

business_agent <- Agent$new(
  name = "business_strategist",
  instruction = paste(
    "You are a business strategist and market analyst.",
    "Analyze topics from a business/market perspective.",
    "Focus on: market opportunities, ROI, competitive advantage, business models.",
    "Keep responses to 2-3 sentences."
  ),
  llm_object = openai_4_1_mini
)

ux_agent <- Agent$new(
  name = "ux_specialist",
  instruction = paste(
    "You are a UX and product design specialist.",
    "Analyze topics from a user experience perspective.",
    "Focus on: user needs, design considerations, accessibility, user satisfaction.",
    "Keep responses to 2-3 sentences."
  ),
  llm_object = openai_4_1_mini
)


# Create workflow with parallel execution
wf <- Workflow$new(
  name = "parallel_multi_agent_analysis",
  description = "Multi-perspective analysis using parallel Agent invocations"
)

# Set up 3 parallel workers for concurrent API calls
wf$set_daemons(3)

# Add input station (passthrough)
wf$add_station(
  "input",
  handler = function(x) x,
  description = "Input topic"
)

# Add three parallel Agent stations
wf$add_station(
  "technical_analysis",
  handler = technical_agent,
  description = "Technical perspective"
)

wf$add_station(
  "business_analysis",
  handler = business_agent,
  description = "Business perspective"
)

wf$add_station(
  "ux_analysis",
  handler = ux_agent,
  description = "UX perspective"
)

# Add synthesis station
wf$add_station(
  "synthesis",
  handler = function(x) {
    paste0(
      "═══════════════════════════════════════════\n",
      "MULTI-PERSPECTIVE ANALYSIS REPORT\n",
      "═══════════════════════════════════════════\n\n",
      x
    )
  },
  description = "Synthesis"
)

# Define parallel group: input -> [technical, business, ux] -> synthesis
wf$add_parallel_group(
  from = "input",
  stations = c("technical_analysis", "business_analysis", "ux_analysis"),
  to = "synthesis",
  merge_fn = function(results) {
    paste0(
      "📱 TECHNICAL PERSPECTIVE:\n",
      results$technical_analysis, "\n\n",
      "💼 BUSINESS PERSPECTIVE:\n",
      results$business_analysis, "\n\n",
      "👤 UX PERSPECTIVE:\n",
      results$ux_analysis
    )
  }
)

wf$visualize()

# Execute workflow with parallel execution
topic <- "Artificial Intelligence in Healthcare"

result <- wf$run(topic)

# Clean up any existing daemons from previous runs
mirai::daemons(0)

[1] 0

result

[1] "═══════════════════════════════════════════\nMULTI-PERSPECTIVE ANALYSIS REPORT\n═══════════════════════════════════════════\n\n📱 TECHNICAL PERSPECTIVE:\nAI in healthcare commonly leverages deep learning frameworks like TensorFlow and PyTorch for tasks such as medical image analysis, natural language processing for clinical notes, and predictive analytics. Key technical challenges include integrating heterogeneous healthcare data sources (EHRs, imaging, genomics), ensuring data privacy through techniques like federated learning, and meeting stringent regulatory compliance like HIPAA. Robust model interpretability and real-time processing capabilities are critical for clinical adoption.\n\n💼 BUSINESS PERSPECTIVE:\nArtificial Intelligence in healthcare offers significant market opportunities through predictive diagnostics, personalized treatment, and operational efficiency, driving substantial ROI by reducing costs and improving patient outcomes. Companies leveraging proprietary AI algorithms and extensive healthcare data gain a competitive advantage, enabling scalable business models like SaaS platforms for hospitals, telemedicine integration, and AI-powered medical devices. Early adoption and regulatory compliance are critical for capturing market share in this rapidly evolving space.\n\n👤 UX PERSPECTIVE:\nAI in healthcare addresses user needs for faster diagnostics, personalized treatment, and efficient data management. Design considerations include intuitive interfaces for diverse users (patients, doctors), data privacy, and seamless integration with existing systems. Ensuring accessibility involves accommodating varying tech literacy and disabilities, while user satisfaction hinges on trust, accuracy, and transparency."

How it works

1. $set_daemons(n) Creates n parallel worker processes.

2. $add_parallel_group(from, stations, to, merge_fn) Defines which stations run in parallel:

   • from: predecessor station (all parallel stations receive its output)
   • stations: vector of station names to run concurrently
   • to: successor station (receives merged results)
   • merge_fn: function to combine results from all parallel stations

3. Execution All parallel stations run concurrently in separate daemons

4. Visualization $visualize() shows parallel routes in red with ∥ symbol to distinguish from sequential routes