Agents in workflows

How agent()-derived tasks integrate with Flux's durability model, what gets checkpointed, how replay works, and what idempotency requirements follow for tools.

When you await an agent task inside a @workflow, every LLM call and every tool call is recorded to the workflow’s event log as a separate checkpoint. If the worker crashes or is restarted, Flux replays the event log forward and resumes execution after the last completed step, without re-calling the LLM or re-running any tool that already returned a result.

What gets checkpointed

The agent loop in agent_loop.py wraps each LLM call as a named Flux task (llm_0, llm_1, …). Each tool invocation is an await of a Flux @task function. Both are recorded to the event log before Flux moves to the next step.

For a single agent invocation with one tool call:

workflow.started
  task.started   (agent_anthropic_claude_sonnet_4_20250514)
    task.started   (llm_0)
    task.completed (llm_0)          ← LLM response recorded
    task.started   (get_weather)    ← tool call
    task.completed (get_weather)    ← tool result recorded
    task.started   (llm_1)
    task.completed (llm_1)          ← final LLM response recorded
  task.completed (agent_anthropic_...)
workflow.completed

Each task.completed event stores the output. On replay, Flux reads that stored output directly and skips re-execution of the corresponding function body.

Replay in practice

The replay_demo workflow demonstrates the full replay cycle: an agent calls a tool, the workflow pauses for human confirmation, and on resume Flux replays the completed steps before continuing with a fresh second call.

from flux import ExecutionContext, task, workflow
from flux.tasks import pause
from flux.tasks.ai import agent


@task
async def get_weather(city: str) -> str:
    """Get the current weather for a city."""
    return f"Sunny, 22C in {city}"


@workflow
async def replay_demo(ctx: ExecutionContext[dict]):
    city = (ctx.input or {}).get("city", "London")

    assistant = await agent(
        "You are a weather assistant. Always use the get_weather tool.",
        model="ollama/llama3.2",
        name="weather_bot",
        tools=[get_weather],
        stream=False,
    )

    first_answer = await assistant(f"What is the weather in {city}?")

    await pause(
        "confirm_replay",
        output={"message": "Resume to trigger replay.", "first_answer": first_answer},
    )

    second_answer = await assistant(f"Tell me the weather forecast for {city}")

    return {"first_answer": first_answer, "second_answer": second_answer}

Run it and then resume:

flux workflow run replay_demo '{"city": "London"}'
# copy the execution ID from the output

flux workflow resume replay_demo <execution_id> '{"confirmed": true}'

On resume, Flux re-enters the workflow body from the top. When it reaches await assistant(...) for the first time, it finds llm_0, get_weather, and llm_1 already in the event log. It returns the recorded results without making any API call or running get_weather again. Only the second_answer call after the pause creates fresh checkpoints.

Conversation state and replay

Working memory is serialized into the event log as part of the tool-call iteration. When working_memory is attached to an agent, each memorize() call is checkpointed along with the rest of the step. On replay, the memory is rebuilt from the recorded events, so the conversation history the LLM sees on the second invocation is identical to the original run.

from flux.tasks.ai.memory import working_memory

@workflow
async def multi_turn(ctx: ExecutionContext):
    mem = working_memory()
    assistant = await agent(
        "You are a helpful assistant.",
        model="anthropic/claude-sonnet-4-20250514",
        working_memory=mem,
    )

    first = await assistant("What is 6 × 7?")
    second = await assistant("What did I just ask you?")  # memory carries context
    return {"first": first, "second": second}

If the worker crashes between the two calls, replay delivers the first response from the event log, restores the conversation history, and runs the second call fresh with the same context the LLM had before the crash.

Tool idempotency

A tool that completed before a crash will not run again on resume. The risk is a tool that was mid-execution: Flux does not record a task.completed event until the tool’s return statement runs, so a partially-written side effect (a half-sent email, a partially-written file) leaves no checkpoint and the tool runs again on resume.

Design agent tools so that running them twice with the same arguments produces the same result:

@task
async def send_notification(user_id: str, message: str) -> str:
    if await notification_already_sent(user_id, message):
        return "already_sent"
    await deliver_notification(user_id, message)
    return "sent"

See Reliability — idempotency for the full pattern, including how Flux’s caching mechanism can serve as an idempotency guard.

Crash inside a tool loop

Flux checkpoints each LLM call and each tool call individually. A crash mid-loop resumes after the last completed checkpoint:

task.completed (llm_0)        ← recorded
task.completed (search_web)   ← tool 1, recorded
task.started   (read_file)    ← tool 2 — CRASH HERE

On resume, Flux replays llm_0 and search_web from the event log. read_file starts fresh. The LLM receives exactly the same tool results it received before the crash, up to the last recorded one.

The name parameter and stable task IDs

Flux generates stable task IDs from the task name and its arguments. By default, an agent task’s name is derived from the provider and model string (e.g. agent_anthropic_claude_sonnet_4_20250514). If you have two agents in the same workflow using the same provider and model, assign explicit names to keep task IDs distinct:

researcher = await agent(
    "You research topics thoroughly.",
    model="anthropic/claude-sonnet-4-20250514",
    name="researcher",
)
writer = await agent(
    "You turn research into clear prose.",
    model="anthropic/claude-sonnet-4-20250514",
    name="writer",
)

Without distinct names, the two agents share an ID namespace and replay may return the wrong cached result.

What’s next