Multi-agent workflows in LlamaIndex explained

If you’re searching for a llamaindex multi-agent workflow tutorial that goes beyond hello-world examples, you’re likely hitting the same wall most engineers do: the documentation shows you how to spawn agents, but not how to make them collaborate reliably in production. This guide walks through the actual patterns that work — AgentWorkflow orchestration, handoff mechanics, shared state management, and the tradeoffs you’ll face when scaling past a demo.

Understanding the agentworkflow abstraction

LlamaIndex’s AgentWorkflow is the orchestration layer that sits above individual agents. It manages the event loop, handles tool calls, routes messages between agents, and maintains conversation state. Unlike the older FunctionCallingAgent or ReActAgent patterns where you manually drove the loop, AgentWorkflow gives you a declarative graph of agents with defined handoff rules.

from llama_index.core.agent.workflow import AgentWorkflow
from llama_index.core.agent.workflow import FunctionAgent
from llama_index.core.tools import FunctionTool

def get_weather(location: str) -> str:
    """Get current weather for a location."""
    # Real implementation would call an API
    return f"Weather in {location}: 72°F, sunny"

def calculate_mortgage(principal: float, rate: float, years: int) -> dict:
    """Calculate monthly mortgage payment."""
    monthly_rate = rate / 12 / 100
    months = years * 12
    payment = principal * (monthly_rate * (1 + monthly_rate) ** months) / ((1 + monthly_rate) ** months - 1)
    return {"monthly_payment": round(payment, 2), "total_paid": round(payment * months, 2)}

weather_agent = FunctionAgent(
    name="weather_agent",
    description="Handles weather queries",
    tools=[FunctionTool.from_defaults(fn=get_weather)],
    system_prompt="You are a weather specialist. Use the get_weather tool for all location queries."
)

mortgage_agent = FunctionAgent(
    name="mortgage_agent",
    description="Handles mortgage calculations",
    tools=[FunctionTool.from_defaults(fn=calculate_mortgage)],
    system_prompt="You are a mortgage calculator. Use the calculate_mortgage tool for all loan queries."
)

workflow = AgentWorkflow(
    agents=[weather_agent, mortgage_agent],
    root_agent="weather_agent",  # Entry point
    initial_state={"user_id": "user_123", "session_id": "sess_456"}
)

The root_agent receives the initial user message. From there, agents can hand off to each other using the built-in handoff tool that AgentWorkflow automatically injects.

Handoff patterns that actually work

The handoff mechanism is where most multi-agent systems break. You have two primary patterns: explicit handoffs (agent decides to transfer) and implicit handoffs (workflow routes based on intent classification).

Explicit handoffs with context passing

from llama_index.core.agent.workflow import AgentWorkflow, FunctionAgent
from llama_index.core.workflow import Context

# Agent with explicit handoff capability
research_agent = FunctionAgent(
    name="research_agent",
    description="Researches topics and hands off to writer",
    tools=[FunctionTool.from_defaults(fn=search_web)],
    system_prompt=(
        "You research topics thoroughly. When you have enough information, "
        "hand off to the writer_agent with a summary of findings. "
        "Use the handoff tool: handoff(agent_name='writer_agent', reason='...', context={...})"
    )
)

writer_agent = FunctionAgent(
    name="writer_agent",
    description="Writes content based on research",
    tools=[],
    system_prompt="You write clear, well-structured content from research summaries."
)

workflow = AgentWorkflow(
    agents=[research_agent, writer_agent],
    root_agent="research_agent"
)

# Running with context preservation
ctx = Context(workflow)
result = await workflow.run(user_msg="Write an article about quantum computing", ctx=ctx)

The context parameter in handoff is critical — it’s how you pass structured data between agents without stuffing everything into the conversation history. The receiving agent gets this via ctx.get("handoff_context") in its system prompt or tools.

Implicit routing with a supervisor agent

For more complex routing, a supervisor agent classifies intent and delegates:

supervisor_agent = FunctionAgent(
    name="supervisor",
    description="Routes requests to specialized agents",
    tools=[],  # No tools — pure routing
    system_prompt=(
        "You are a router. Analyze the user's request and hand off to exactly one agent:\n"
        "- weather_agent: weather, temperature, forecast queries\n"
        "- mortgage_agent: loan, mortgage, payment, interest rate calculations\n"
        "- research_agent: general knowledge, fact-finding, explanations\n"
        "Always use the handoff tool with a brief reason."
    )
)

workflow = AgentWorkflow(
    agents=[supervisor_agent, weather_agent, mortgage_agent, research_agent],
    root_agent="supervisor"
)

Tradeoff: Explicit handoffs give agents autonomy but can create circular delegations. Supervisor routing is more predictable but creates a bottleneck — every request hits the supervisor first. For latency-sensitive paths, consider hybrid: supervisor for ambiguous requests, direct handoffs for known workflows.

Shared state and memory management

AgentWorkflow maintains a global Context object that persists across the entire workflow execution. This is your source of truth for shared state — user preferences, accumulated facts, intermediate results.

from llama_index.core.workflow import Context

async def run_with_persistent_state(user_id: str, message: str):
    ctx = Context(workflow)
    
    # Load existing state from your database
    existing_state = await db.get_workflow_state(user_id)
    if existing_state:
        ctx.data.update(existing_state)
    
    result = await workflow.run(user_msg=message, ctx=ctx)
    
    # Persist updated state
    await db.save_workflow_state(user_id, dict(ctx.data))
    return result

Pitfall: The context grows unbounded. Long-running workflows accumulate every tool result, handoff payload, and agent response in ctx.data. Implement a compaction strategy — summarize old interactions, drop intermediate tool outputs, keep only decision-relevant state.

def compact_context(ctx: Context, max_tokens: int = 8000):
    """Summarize and truncate context to stay within token limits."""
    # Count tokens in conversation history
    history = ctx.data.get("conversation_history", [])
    if estimate_tokens(history) > max_tokens:
        # Keep system prompts, last N turns, and structured state
        summary = summarize_old_turns(history[:-10])
        ctx.data["conversation_history"] = [
            {"role": "system", "content": summary},
            *history[-10:]
        ]

Parallel execution and fan-out patterns

AgentWorkflow executes agents sequentially by default. For true parallelism — say, querying multiple APIs simultaneously — you need to structure tools as async and use asyncio.gather within a single agent, or spawn sub-workflows.

async def parallel_research(topics: list[str]) -> dict:
    """Research multiple topics in parallel."""
    async def research_one(topic: str):
        agent = FunctionAgent(
            name=f"researcher_{topic}",
            tools=[search_tool],
            system_prompt=f"Research {topic} thoroughly."
        )
        return await agent.run(f"Research {topic}")
    
    results = await asyncio.gather(*[research_one(t) for t in topics])
    return dict(zip(topics, results))

parallel_agent = FunctionAgent(
    name="parallel_researcher",
    description="Researches multiple topics simultaneously",
    tools=[FunctionTool.from_defaults(fn=parallel_research)],
    system_prompt="Use parallel_research for multi-topic queries."
)

Warning: Each sub-agent invocation creates its own LLM calls. Parallel research with 5 topics = 5x the token cost and latency of the slowest call. Batch where possible, and set timeouts.

Structured output and validation

Agents returning unstructured text forces downstream agents to parse fragile strings. Use Pydantic models for tool outputs and handoff context:

from pydantic import BaseModel, Field
from typing import Literal

class WeatherReport(BaseModel):
    location: str
    temperature_f: float
    conditions: str
    humidity_pct: int
    source: Literal["api", "cache"]

class MortgageQuote(BaseModel):
    monthly_payment: float
    total_interest: float
    loan_amount: float
    apr: float
    term_years: int

def get_weather_structured(location: str) -> WeatherReport:
    # ... API call ...
    return WeatherReport(
        location=location,
        temperature_f=72.0,
        conditions="sunny",
        humidity_pct=45,
        source="api"
    )

# Agent receives typed output automatically
weather_agent = FunctionAgent(
    name="weather_agent",
    tools=[FunctionTool.from_defaults(fn=get_weather_structured)],
    system_prompt="Return structured weather data. The tool handles formatting."
)

When handing off, pass the Pydantic model directly in context:

# In sending agent's tool or logic
await ctx.set("weather_data", weather_report.model_dump())

# In receiving agent's system prompt or tool
weather_data = WeatherReport(**ctx.data.get("weather_data", {}))

Common pitfalls and how to avoid them

1. Infinite handoff loops

Agents hand off to each other recursively until token limits explode.

# Prevention: track handoff count in context
MAX_HANDOFFS = 5

async def safe_handoff(agent_name: str, ctx: Context, reason: str):
    count = ctx.data.get("handoff_count", 0)
    if count >= MAX_HANDOFFS:
        raise ValueError(f"Max handoffs ({MAX_HANDOFFS}) exceeded")
    ctx.data["handoff_count"] = count + 1
    return handoff(agent_name=agent_name, reason=reason)

2. Tool hallucination

Agents invent tool parameters or call non-existent tools. Mitigate with strict schemas and few-shot examples in system prompts:

system_prompt = (
    "You have access to these tools ONLY:\n"
    "1. get_weather(location: str) -> WeatherReport\n"
    "2. calculate_mortgage(principal: float, rate: float, years: int) -> MortgageQuote\n\n"
    "Examples:\n"
    "User: 'Weather in NYC'\n"
    "Assistant: [calls get_weather(location='New York, NY')]\n\n"
    "NEVER invent tools or parameters. If unsure, ask for clarification."
)

3. Context window overflow

Long conversations + multiple agents + tool outputs = context explosion. Solutions:

• Summarization agent: Periodically condense history
• Sliding window: Keep only last N turns + structured state
• RAG over history: Embed and retrieve relevant past turns

class HistoryManager:
    def __init__(self, max_turns: int = 20, embed_model=None):
        self.max_turns = max_turns
        self.embed_model = embed_model
    
    async def compact(self, ctx: Context):
        history = ctx.data.get("messages", [])
        if len(history) <= self.max_turns:
            return
        
        # Keep system + recent, summarize old
        old_turns = history[1:-self.max_turns]  # Skip system prompt
        recent = history[-self.max_turns:]
        
        summary = await self.summarize(old_turns)
        ctx.data["messages"] = [
            history[0],  # System prompt
            {"role": "system", "content": f"Previous context: {summary}"},
            *recent
        ]

4. Silent failures in tool execution

Tools that raise exceptions crash the workflow unless wrapped. Always wrap external calls:

from llama_index.core.tools import FunctionTool

def safe_api_call(func):
    async def wrapper(*args, **kwargs):
        try:
            return await func(*args, **kwargs)
        except Exception as e:
            return {"error": str(e), "type": type(e).__name__}
    return wrapper

weather_tool = FunctionTool.from_defaults(
    fn=safe_api_call(get_weather_structured),
    name="get_weather",
    description="Get weather. Returns error object on failure."
)

The agent then sees the error in the tool result and can retry, handoff, or apologize gracefully.

Production deployment considerations

Observability

Instrument every handoff, tool call, and agent transition. You need to answer: “Why did this request take 45 seconds?” and “Which agent caused the error?”

from llama_index.core.workflow import Event

class AgentHandoffEvent(Event):
    from_agent: str
    to_agent: str
    reason: str
    timestamp: float

class ToolCallEvent(Event):
    agent: str
    tool: str
    args: dict
    duration_ms: float
    success: bool

# Emit in your agents/tools
ctx.write_event_to_stream(AgentHandoffEvent(
    from_agent=current_agent,
    to_agent=target_agent,
    reason=reason,
    timestamp=time.time()
))

Rate limiting and fallbacks

When using multiple models across agents (e.g., GPT-4o for reasoning, Haiku for classification), implement per-model rate limiting with fallback:

# This is where a gateway like n4n.ai helps — single endpoint, 
# automatic fallback when a provider is rate-limited or degraded,
# per-token usage metering across all models.

Testing multi-agent flows

Unit test individual agents with mocked tools. Integration test the workflow with controlled inputs:

import pytest
from llama_index.core.workflow import Context

@pytest.mark.asyncio
async def test_mortgage_handoff_flow():
    ctx = Context(workflow)
    result = await workflow.run(
        user_msg="What's my monthly payment on $400k at 6.5% for 30 years?",
        ctx=ctx
    )
    
    # Verify supervisor routed correctly
    assert "mortgage_agent" in ctx.data.get("agents_visited", [])
    
    # Verify structured output
    assert "monthly_payment" in str(result)
    assert float(result.split("$")[1].split(",")[0]) > 2000  # Sanity check

When to use multi-agent vs single-agent

Multi-agent adds latency, complexity, and failure modes. Use it when:

• Distinct expertise domains require different system prompts, tools, or models (legal + medical + financial)
• Parallelizable subtasks benefit from concurrent execution
• Human-in-the-loop checkpoints need different reviewers per domain
• Compliance requires audit trails per agent type

Stick with a single well-prompted agent when:

• Tasks are sequential and interdependent
• Context sharing is dense (every step needs full history)
• Latency budget is tight (<2s p95)
• Team lacks bandwidth to maintain orchestration logic

Next steps

Start with a two-agent supervisor pattern — one router, one specialist. Add agents only when you hit a genuine domain boundary. Instrument everything from day one. And remember: the workflow graph is code, not configuration. Version it, test it, and deploy it like any other service.