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:

There are two main building blocks:

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 Algeriens, die größte Wüstenexpanse innerhalb eines Landes, 
fasziniert Besucher mit ihren imposanten Dünen, ruhigen Oasen, antiken 
Felskunstwerken und historischen Ksars und ist somit ein Schatz für den 
Wüstentourismus. Diese Region ist auch ein lebendiges kulturelles Zentrum, in 
dem Berbergemeinschaften Traditionen durch Musik und Festivals bewahren und 
kulturelle Enthusiasten ebenso anziehen wie Abenteuersuchende, die sich auf 
Kameltouren und 4x4-Expeditionen freuen. Obwohl die Tourismusinfrastruktur und 
die internationale Zugänglichkeit sich noch in der Entwicklung befinden, 
versprechen staatliche Initiativen, die auf nachhaltigen Tourismus ausgerichtet
sind, das volle Potenzial der Sahara zu erschließen, die lokalen Wirtschaften 
zu fördern und das einzigartige Wüstenerbe Algeriens mit der Welt zu teilen.

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 Algeriens, die größte Wüstenexpanse innerhalb eines Landes, 
fasziniert Besucher mit ihren imposanten Dünen, ruhigen Oasen, antiken 
Felskunstwerken und historischen Ksars und ist somit ein Schatz für den 
Wüstentourismus. Diese Region ist auch ein lebendiges kulturelles Zentrum, in 
dem Berbergemeinschaften Traditionen durch Musik und Festivals bewahren und 
kulturelle Enthusiasten ebenso anziehen wie Abenteuersuchende, die sich auf 
Kameltouren und 4x4-Expeditionen freuen. Obwohl die Tourismusinfrastruktur und 
die internationale Zugänglichkeit sich noch in der Entwicklung befinden, 
versprechen staatliche Initiativen, die auf nachhaltigen Tourismus ausgerichtet
sind, das volle Potenzial der Sahara zu erschließen, die lokalen Wirtschaften 
zu fördern und das einzigartige Wüstenerbe Algeriens mit der Welt zu teilen.

[[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, the largest portion of the vast desert in any country, 
offers unique landscapes including vast dunes, oases, ancient rock art, and 
historic ksars, making it a rich destination for desert tourism.  
2. The region holds significant cultural heritage with Berber communities, 
traditional music, and festivals that attract cultural tourism alongside 
adventure activities like camel trekking and 4x4 expeditions.  
3. Despite its immense potential, tourism infrastructure and international 
accessibility remain limited, but ongoing government efforts aim to develop 
sustainable tourism to boost local economies and showcase Algeria’s desert 
attractions.


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

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

[[1]]$trace[[2]]$input
1. Algeria's Sahara, the largest portion of the vast desert in any country, 
offers unique landscapes including vast dunes, oases, ancient rock art, and 
historic ksars, making it a rich destination for desert tourism.  
2. The region holds significant cultural heritage with Berber communities, 
traditional music, and festivals that attract cultural tourism alongside 
adventure activities like camel trekking and 4x4 expeditions.  
3. Despite its immense potential, tourism infrastructure and international 
accessibility remain limited, but ongoing government efforts aim to develop 
sustainable tourism to boost local economies and showcase Algeria’s desert 
attractions.

[[1]]$trace[[2]]$output
Algeria's Sahara, the largest expanse of the desert within any country, 
captivates visitors with its breathtaking landscapes of towering dunes, serene 
oases, ancient rock art, and historic ksars, making it a treasure trove for 
desert tourism. This region is also a vibrant cultural hub, where Berber 
communities keep traditions alive through music and festivals, drawing cultural
enthusiasts alongside adventure seekers eager for camel treks and 4x4 
expeditions. Although tourism infrastructure and international accessibility 
are still developing, government initiatives focused on sustainable tourism 
promise to unlock the Sahara’s full potential, aiming to boost local economies 
and share the unique allure of Algeria’s desert heritage with the world.


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

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

[[1]]$trace[[3]]$input
Algeria's Sahara, the largest expanse of the desert within any country, 
captivates visitors with its breathtaking landscapes of towering dunes, serene 
oases, ancient rock art, and historic ksars, making it a treasure trove for 
desert tourism. This region is also a vibrant cultural hub, where Berber 
communities keep traditions alive through music and festivals, drawing cultural
enthusiasts alongside adventure seekers eager for camel treks and 4x4 
expeditions. Although tourism infrastructure and international accessibility 
are still developing, government initiatives focused on sustainable tourism 
promise to unlock the Sahara’s full potential, aiming to boost local economies 
and share the unique allure of Algeria’s desert heritage with the world.

[[1]]$trace[[3]]$output
Algeria's Sahara, the largest desert expanse within any country, captivates 
visitors with its towering dunes, serene oases, ancient rock art, and historic 
ksars, making it a treasure trove for desert tourism. This region is also a 
vibrant cultural hub where Berber communities preserve traditions through music
and festivals, attracting cultural enthusiasts alongside adventure seekers 
eager for camel treks and 4x4 expeditions. Although tourism infrastructure and 
international accessibility are still developing, government initiatives 
focused on sustainable tourism promise to unlock the Sahara’s full potential, 
boost local economies, and share Algeria’s unique desert heritage with the 
world.


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

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

[[1]]$trace[[4]]$input
Algeria's Sahara, the largest desert expanse within any country, captivates 
visitors with its towering dunes, serene oases, ancient rock art, and historic 
ksars, making it a treasure trove for desert tourism. This region is also a 
vibrant cultural hub where Berber communities preserve traditions through music
and festivals, attracting cultural enthusiasts alongside adventure seekers 
eager for camel treks and 4x4 expeditions. Although tourism infrastructure and 
international accessibility are still developing, government initiatives 
focused on sustainable tourism promise to unlock the Sahara’s full potential, 
boost local economies, and share Algeria’s unique desert heritage with the 
world.

[[1]]$trace[[4]]$output
Die Sahara Algeriens, die größte Wüstenexpanse innerhalb eines Landes, 
fasziniert Besucher mit ihren imposanten Dünen, ruhigen Oasen, antiken 
Felskunstwerken und historischen Ksars und ist somit ein Schatz für den 
Wüstentourismus. Diese Region ist auch ein lebendiges kulturelles Zentrum, in 
dem Berbergemeinschaften Traditionen durch Musik und Festivals bewahren und 
kulturelle Enthusiasten ebenso anziehen wie Abenteuersuchende, die sich auf 
Kameltouren und 4x4-Expeditionen freuen. Obwohl die Tourismusinfrastruktur und 
die internationale Zugänglichkeit sich noch in der Entwicklung befinden, 
versprechen staatliche Initiativen, die auf nachhaltigen Tourismus ausgerichtet
sind, das volle Potenzial der Sahara zu erschließen, die lokalen Wirtschaften 
zu fördern und das einzigartige Wüstenerbe Algeriens mit der Welt zu teilen.

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 game will be against Argentina.  \n- This match marks the start of Algeria's World Cup campaign."

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 seas harbor rare and extraordinary fish species, many of which 
exist nowhere else in the Mediterranean. These exceptional fish not only enrich
the region’s vibrant biodiversity but also underscore the critical need for 
dedicated marine conservation efforts in Algeria.
end <- Sys.time()
print(end - start)
Time difference of 2.822783 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 seas harbor rare and extraordinary fish species, many of which 
exist nowhere else in the Mediterranean. These exceptional fish not only enrich
the region’s vibrant biodiversity but also underscore the critical need for 
dedicated marine conservation efforts in Algeria.
end <- Sys.time()
print(end - start)
Time difference of 0.05783606 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 home to several rare and vital fish species that play a key
role in maintaining the health of the marine ecosystem. Safeguarding these 
unique species is essential for preserving the rich biodiversity of the entire 
Mediterranean region.
end <- Sys.time()
print(end - start)
Time difference of 2.993069 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 your positive energy—it 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—please let me know how I can help 
improve your experience.

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 Desert Foxes, is the 
national football team of Algeria. Renowned for their skill and determination, 
they have a passionate fan base and have achieved significant success in 
African and international competitions.
# 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 crucial pollinators essential for plant reproduction and food 
security, but their populations are threatened by colony collapse disorder.

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")
Béjaïa is a Mediterranean port city in northern Algeria's Kabylie region, 
historically significant as a medieval trade and cultural center under the 
Hammadid dynasty, and today recognized for its diverse population and role as 
an economic hub for fishing, commerce, and tourism.
# 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")
Béjaïa ist eine lebendige Küstenstadt im Norden Algeriens, bekannt für ihr 
historisches Erbe, ihre wirtschaftliche Vitalität und die atemberaubenden 
mediterranen Landschaften.

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 a unique blend of traditional Arab customs and 
French colonial influences, creating a vibrant and diverse coffee scene. 
Popular beverages include both Arabic coffee and espresso-style drinks, enjoyed
widely in cafés and homes across the country. Despite this growing appreciation
for coffee, Algeria relies heavily on imports to meet demand due to limited 
local 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))

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