LlamaIndex AgentWorkflow for multi-step tasks

If you’ve built anything with LlamaIndex agents before the 0.10 release, you know the pain: stitching together ReActAgent, OpenAIAgent, and custom state machines by hand. The llamaindex agentworkflow tutorial you’re reading exists because AgentWorkflow replaces all that with a single, typed, checkpointable abstraction. We’ll build a research assistant that plans, searches, extracts, and synthesizes — four distinct steps with explicit state passing and retry logic.

Prerequisites

You need Python 3.10+ and a virtual environment. Install the current LlamaIndex core and the agent workflow extras:

python -m venv .venv && source .venv/bin/activate
pip install -U "llama-index-core>=0.10.0" "llama-index-llms-openai>=0.1.0" "llama-index-tools-tavily-research>=0.1.0" python-dotenv

Set your API keys in a .env file:

OPENAI_API_KEY=sk-...
TAVILY_API_KEY=tvly-...

We’ll use OpenAI’s gpt-4o-mini for planning and synthesis, and Tavily for web search. If you prefer a different provider, swap the LLM class — the workflow code doesn’t change.

The mental model

AgentWorkflow is a directed acyclic graph of FunctionAgent nodes. Each node receives a typed Context object that carries state across steps. You define the graph declaratively, then call run() with an initial message. The framework handles:

• Serialization of context to JSON (for checkpointing or human-in-the-loop)
• Automatic retries with exponential backoff per step
• Streaming intermediate outputs if you need a UI

No global singletons, no hidden thread locals. It’s just functions and data.

Step 1: Define the state schema

Create workflow.py and start with a TypedDict for the shared context. Explicit typing catches bugs at edit time, not runtime.

# workflow.py
from typing import TypedDict, List, Optional
from llama_index.core.workflow import Context

class ResearchState(TypedDict):
    topic: str
    plan: Optional[List[str]]
    search_results: Optional[List[dict]]
    extracted_facts: Optional[List[str]]
    final_report: Optional[str]

The Context object at runtime will hold an instance of this dict under ctx.data. Every agent step reads and writes the fields it owns.

Step 2: Build the planner agent

The planner takes the user’s topic and returns a list of search queries. We’ll use a FunctionAgent with a strict output schema.

# workflow.py (continued)
from llama_index.core.agent.workflow import FunctionAgent
from llama_index.core.tools import FunctionTool
from llama_index.llms.openai import OpenAI
from pydantic import BaseModel, Field

llm = OpenAI(model="gpt-4o-mini", temperature=0)

class PlanOutput(BaseModel):
    queries: List[str] = Field(description="3-5 specific search queries to research the topic")

planner = FunctionAgent(
    name="planner",
    description="Breaks a research topic into concrete search queries",
    llm=llm,
    tools=[],
    output_cls=PlanOutput,
    system_prompt=(
        "You are a research planner. Given a topic, produce 3-5 specific, "
        "non-overlapping search queries that together cover the topic thoroughly. "
        "Return only the structured output."
    ),
)

Note: output_cls forces structured output via OpenAI’s function calling. The agent will not emit free text.

Step 3: Build the search agent

This agent receives the plan, executes searches in parallel, and writes raw results to state.

# workflow.py (continued)
from llama_index.tools.tavily_research import TavilyToolSpec

tavily = TavilyToolSpec(api_key="env:TAVILY_API_KEY")
search_tool = tavily.to_tool_list()[0]  # single tool: tavily_search

search_agent = FunctionAgent(
    name="searcher",
    description="Executes search queries and returns raw results",
    llm=llm,
    tools=[search_tool],
    system_prompt=(
        "You receive a list of search queries. Call the search tool for each query. "
        "Return the aggregated results as a JSON array of objects with keys: "
        "query, title, url, content, score."
    ),
)

The Tavily tool returns a list of dicts. We’ll normalize them in the next step.

Step 4: Build the extractor agent

The extractor reads raw search results, pulls out verifiable claims with citations, and writes a flat list of facts.

# workflow.py (continued)
class Fact(BaseModel):
    claim: str
    source_url: str
    source_title: str

class ExtractorOutput(BaseModel):
    facts: List[Fact]

extractor = FunctionAgent(
    name="extractor",
    description="Extracts cited facts from search results",
    llm=llm,
    tools=[],
    output_cls=ExtractorOutput,
    system_prompt=(
        "You receive raw search results. Extract every verifiable claim as a separate fact. "
        "Each fact must include the exact claim text, the source URL, and the source title. "
        "Discard opinions, speculation, and duplicate claims. Return only the structured output."
    ),
)

Step 5: Build the synthesizer agent

The synthesizer takes the fact list and writes a coherent report with inline citations.

# workflow.py (continued)
class ReportOutput(BaseModel):
    markdown: str

synthesizer = FunctionAgent(
    name="synthesizer",
    description="Writes a cited research report from extracted facts",
    llm=llm,
    tools=[],
    output_cls=ReportOutput,
    system_prompt=(
        "You receive a list of cited facts. Write a well-structured markdown report "
        "with sections, bullet points, and inline citations using the format "
        "[source_title](source_url). Be concise but comprehensive. Return only the structured output."
    ),
)

Step 6: Wire the workflow

Now connect the agents into a graph. Each step is a function that receives Context, reads state, calls its agent, and writes results back.

# workflow.py (continued)
from llama_index.core.workflow import (
    Workflow, StartEvent, StopEvent, step, Context
)

class ResearchWorkflow(Workflow):
    @step
    async def plan(self, ctx: Context, ev: StartEvent) -> StopEvent | None:
        topic = ev.get("topic")
        if not topic:
            return StopEvent(result={"error": "No topic provided"})
        await ctx.set("topic", topic)
        result = await planner.run(f"Research topic: {topic}")
        await ctx.set("plan", result.queries)
        return None  # continue to next step

    @step
    async def search(self, ctx: Context, ev: StartEvent) -> StopEvent | None:
        plan = await ctx.get("plan")
        if not plan:
            return StopEvent(result={"error": "No plan found"})
        queries_str = "\n".join(f"- {q}" for q in plan)
        result = await search_agent.run(f"Execute these searches:\n{queries_str}")
        # The search agent returns raw text; parse JSON from its response
        import json
        try:
            results = json.loads(result.response)
        except json.JSONDecodeError:
            results = [{"query": q, "error": "parse failed"} for q in plan]
        await ctx.set("search_results", results)
        return None

    @step
    async def extract(self, ctx: Context, ev: StartEvent) -> StopEvent | None:
        results = await ctx.get("search_results")
        if not results:
            return StopEvent(result={"error": "No search results"})
        result = await extractor.run(f"Extract facts from: {results}")
        await ctx.set("extracted_facts", [f.model_dump() for f in result.facts])
        return None

    @step
    async def synthesize(self, ctx: Context, ev: StartEvent) -> StopEvent:
        facts = await ctx.get("extracted_facts")
        if not facts:
            return StopEvent(result={"error": "No facts extracted"})
        result = await synthesizer.run(f"Write report from facts: {facts}")
        await ctx.set("final_report", result.markdown)
        return StopEvent(result={"report": result.markdown})

The @step decorator registers each method as a node. Returning None means “continue to the next step in declaration order.” Returning a StopEvent ends the run and surfaces the payload.

Step 7: Run it

Create main.py to execute the workflow end-to-end.

# main.py
import asyncio
from workflow import ResearchWorkflow

async def main():
    workflow = ResearchWorkflow(timeout=300, verbose=True)
    result = await workflow.run(topic="Impact of RAG on hallucination rates in production LLMs")
    print(result["report"])

if __name__ == "__main__":
    asyncio.run(main())

Run it:

python main.py

Expected output (truncated):

# Impact of RAG on Hallucination Rates in Production LLMs

## Executive Summary
Retrieval-Augmented Generation (RAG) reduces hallucination rates by 60-80% 
in benchmark evaluations [Lewis et al., 2020](https://arxiv.org/abs/2005.11401) 
and production case studies [Databricks, 2023](https://www.databricks.com/blog/...).

## Key Findings
- **Grounding effect**: Models with access to retrieved context hallucinate 
  less on factual QA [Shuster et al., 2021](https://arxiv.org/abs/2104.07567).
- **Retrieval quality matters**: Poor retrieval increases hallucinations 
  versus no retrieval [Liu et al., 2023](https://arxiv.org/abs/2309.09378).
- **Citation discipline**: Forcing inline citations cuts unsupported claims 
  by 40% [Menick et al., 2022](https://arxiv.org/abs/2203.11171).

## Caveats
- RAG does not eliminate hallucinations in reasoning-heavy tasks.
- Latency and cost increase with retrieval depth.

The verbose=True flag prints each step’s input/output to stdout — useful for debugging.

Checkpointing and resume

Context serializes to JSON. Save it mid-run, resume later, or hand it to a human reviewer.

# Save checkpoint after search step
ctx_dict = ctx.to_dict()
with open("checkpoint.json", "w") as f:
    json.dump(ctx_dict, f)

# Resume
with open("checkpoint.json") as f:
    ctx_dict = json.load(f)
ctx = Context.from_dict(workflow, ctx_dict)
result = await workflow.run(ctx=ctx, topic="...")  # topic ignored; state restored

This is how you build human-in-the-loop review: pause after extraction, present facts to a domain expert, approve or edit, then resume synthesis.

Retry and error handling

Each step can declare its own retry policy. Add retry_policy to the @step decorator:

from llama_index.core.workflow import RetryPolicy

@step(retry_policy=RetryPolicy(max_attempts=3, base_delay=2.0))
async def search(self, ctx: Context, ev: StartEvent) -> StopEvent | None:
    ...

The workflow will retry the entire step on any exception, with exponential backoff. For provider-level fallback (e.g., OpenAI rate limit → Anthropic), you’d swap the llm instance inside the agent at runtime — something we handle at the gateway layer in n4n.ai by honoring client routing directives and forwarding provider cache-control hints.

Streaming intermediate output

If you’re building a UI, stream each step’s output as it arrives:

async for event in workflow.stream(topic="..."):
    if event.name == "search":
        print(f"Search completed: {len(event.result)} results")
    elif event.name == "extract":
        print(f"Extracted {len(event.result)} facts")
    elif event.name == "synthesize":
        print("Report ready")

The stream() method yields WorkflowEvent objects with name and result fields.

Testing individual steps

Because each step is a pure async function of Context, you can unit-test them in isolation:

# test_workflow.py
import pytest
from workflow import ResearchWorkflow
from llama_index.core.workflow import Context

@pytest.mark.asyncio
async def test_plan_step():
    wf = ResearchWorkflow()
    ctx = Context(wf)
    await ctx.set("topic", "test topic")
    result = await wf.plan(ctx, StartEvent(topic="test topic"))
    plan = await ctx.get("plan")
    assert isinstance(plan, list)
    assert len(plan) >= 3

No mocking the LLM required if you use a deterministic test model. For integration tests, point at a local Ollama instance.

Common pitfalls

1. Forgetting await ctx.set() — state changes are not automatic. Every write needs an explicit set.

2. Mixing StartEvent and custom events — the workflow engine routes by event type. If you emit a custom event, add a @step that accepts it.

3. Large context blowing the token window — the entire Context serializes to JSON and can be passed to agents as context. If your facts list grows large, summarize before synthesis or use a separate summarization step.

4. Silent failures in structured output — if the LLM returns invalid JSON for output_cls, the agent raises. Wrap agent calls in try/except and write errors to state for visibility.

Scaling patterns

For production workloads, consider:

• Parallel search: Fan out queries to multiple search agents, then merge. Use asyncio.gather inside a single step.
• Caching: Hash the query list; if unchanged, load cached results from Redis instead of calling Tavily.
• Observability: Emit OpenTelemetry spans from each step. The workflow’s verbose flag is a starting point; replace with structured logging.

What’s next

You now have a typed, checkpointable, retryable multi-step agent. From here you can:

• Add a critic agent that scores the report and triggers a rewrite loop
• Plug in a code interpreter tool for quantitative analysis
• Expose the workflow as a FastAPI endpoint with Server-Sent Events for streaming UI

The llamaindex agentworkflow tutorial pattern scales because the graph is data, not code. Modify the topology without rewriting agents. Swap models per step. Serialize, inspect, resume. That’s the architecture that survives contact with production.