Pattern Handoffs

Capsule overview

In the previous capsule you learned the Supervisor pattern: a central agent that coordinates the others, decides who works, and collects results. It works well when you don't know the execution order ahead of time. But many flows are predictable: research → analyze → write always follows the same order. In those cases, why route through a supervisor if the researcher always hands off to the analyst, and the analyst always hands off to the writer?

A handoff is when one agent transfers control directly to another agent. There's no supervisor in the middle. Agent A finishes its part, decides Agent B is next, and passes control along with the context it needs. Agent B runs, and can pass to Agent C or finish. It's a chain of responsibility where each agent knows who to delegate to.

The difference with the supervisor is architectural: with a supervisor, every decision goes through a central node. With handoffs, the agents coordinate among themselves. This cuts latency (fewer roundtrips to the supervisor), simplifies the flow when it's predictable, and makes each agent more autonomous.

When to use handoffs vs supervisor

The decision isn't "one is better than the other" — they're tools for different problems:

Practical rule:

• ✅ Use a supervisor when you don't know ahead of time which agent you need or in what order
• ✅ Use handoffs when the flow is sequential and predictable (research → analyze → write)
• ✅ Use a hybrid when you have predictable phases with dynamic decisions inside each phase

The handoff mechanism in LangGraph

LangGraph implements handoffs in two ways:

1. At the node level: The node returns a Command with goto pointing at the next node. The graph follows that instruction directly.

2. At the tool level: An agent (inside a node) calls a handoff tool that returns a Command. The Command navigates to the next agent in the parent graph using graph=Command.PARENT.

Both use the same Command object — the difference is where the decision originates. At the node level, the node function decides. At the tool level, the LLM decides (by calling the tool).

Basic handoff: Agent A → Agent B

The simplest handoff: a researcher investigates and transfers directly to the analyst.

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langgraph.types import Command

class State(TypedDict):
    task: str
    research: str
    analysis: str
    log: Annotated[list[str], operator.add]

def researcher(state: State) -> Command:
    result = f"5 sources found on '{state['task']}': papers, blogs, official docs"
    return Command(
        goto="analyst",
        update={
            "research": result,
            "log": ["researcher_done → handoff to analyst"],
        },
    )

def analyst(state: State) -> dict:
    analysis = f"Analysis based on: {state['research'][:50]}... → 3 trends identified"
    return {
        "analysis": analysis,
        "log": ["analyst_done"],
    }

builder = StateGraph(State)
builder.add_node("researcher", researcher)
builder.add_node("analyst", analyst)

builder.add_edge(START, "researcher")
builder.add_edge("analyst", END)

graph = builder.compile()

result = graph.invoke({
    "task": "State of the AI market 2025",
    "research": "",
    "analysis": "",
    "log": [],
})

print(f"Research: {result['research']}")
print(f"Analysis: {result['analysis']}")
print(f"Log: {result['log']}")
# Expected output:
# Research: 5 sources found on 'State of the AI market 2025': papers, blogs, official docs
# Analysis: Analysis based on: 5 sources found on 'State of the AI market 2025': ... → 3 trends identified
# Log: ['researcher_done → handoff to analyst', 'analyst_done']

The researcher returns Command(goto="analyst", update={...}). That does two things:

1. Updates the state with the research results
2. Transfers control directly to the analyst — without going through a supervisor

The analyst receives the already-updated state and works with it. There's no node in between.

The full flow of a handoff

A real handoff isn't just "the researcher passes to the analyst." It's a complete sequence where each agent prepares the context for the next one:

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langgraph.types import Command

class PipelineState(TypedDict):
    topic: str
    sources: list[str]
    research_summary: str
    analysis: str
    report: str
    log: Annotated[list[str], operator.add]

def researcher(state: PipelineState) -> Command:
    sources = [
        f"Paper: '{state['topic']}' trends 2025",
        f"Blog: Industry analysis of {state['topic']}",
        f"Docs: Official {state['topic']} documentation",
    ]
    summary = f"Summary of {len(sources)} sources on '{state['topic']}'"
    return Command(
        goto="analyst",
        update={
            "sources": sources,
            "research_summary": summary,
            "log": [f"researcher: {len(sources)} sources → handoff to analyst"],
        },
    )

def analyst(state: PipelineState) -> Command:
    analysis = (
        f"Analysis of {len(state['sources'])} sources: "
        f"upward trend in '{state['topic']}', "
        f"2 risks identified, 1 clear opportunity"
    )
    return Command(
        goto="writer",
        update={
            "analysis": analysis,
            "log": ["analyst: analysis complete → handoff to writer"],
        },
    )

def writer(state: PipelineState) -> dict:
    report = (
        f"EXECUTIVE REPORT\n"
        f"Topic: {state['topic']}\n"
        f"Sources: {len(state['sources'])}\n"
        f"Main finding: {state['analysis'][:60]}...\n"
        f"Recommendation: Invest in this area"
    )
    return {
        "report": report,
        "log": ["writer: report generated"],
    }

builder = StateGraph(PipelineState)
builder.add_node("researcher", researcher)
builder.add_node("analyst", analyst)
builder.add_node("writer", writer)

builder.add_edge(START, "researcher")
builder.add_edge("writer", END)

graph = builder.compile()

result = graph.invoke({
    "topic": "LLMs in production",
    "sources": [],
    "research_summary": "",
    "analysis": "",
    "report": "",
    "log": [],
})

print("=== Pipeline result ===\n")
print(result["report"])
print(f"\n=== Handoff flow ===")
for entry in result["log"]:
    print(f"  → {entry}")
# Expected output:
# === Pipeline result ===
#
# EXECUTIVE REPORT
# Topic: LLMs in production
# Sources: 3
# Main finding: Analysis of 3 sources: upward trend in 'LLMs in production',...
# Recommendation: Invest in this area
#
# === Handoff flow ===
#   → researcher: 3 sources → handoff to analyst
#   → analyst: analysis complete → handoff to writer
#   → writer: report generated

The full flow: researcher investigates → prepares context (sources + summary) → transfers to the analyst → the analyst analyzes → prepares context (analysis + findings) → transfers to the writer → the writer generates the final report. Each agent receives exactly what it needs from the previous one.

Handoff tools: transferring via tools

In the previous example, the node functions decide directly who to transfer to. But in a real system, agents are LLMs making dynamic decisions. LangGraph's pattern for that: handoff tools — tools the agent calls to transfer control.

When an agent (created with create_agent) decides another agent should take over, it calls a handoff tool. That tool returns a Command that navigates to the destination agent in the parent graph.

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated, Literal
import operator
from langchain.tools import tool, ToolRuntime
from langchain.messages import AIMessage, ToolMessage
from langchain.agents import create_agent, AgentState
from langgraph.graph import StateGraph, START, END
from langgraph.types import Command

class TeamState(AgentState):
    active_agent: str

@tool
def transfer_to_analyst(runtime: ToolRuntime) -> Command:
    """Transfer control to the analyst agent for data analysis."""
    last_ai = next(
        m for m in reversed(runtime.state["messages"])
        if isinstance(m, AIMessage)
    )
    return Command(
        goto="analyst_node",
        update={
            "active_agent": "analyst",
            "messages": [
                last_ai,
                ToolMessage(
                    content="Transferred to the analyst",
                    tool_call_id=runtime.tool_call_id,
                ),
            ],
        },
        graph=Command.PARENT,
    )

@tool
def transfer_to_writer(runtime: ToolRuntime) -> Command:
    """Transfer control to the writer agent to generate reports."""
    last_ai = next(
        m for m in reversed(runtime.state["messages"])
        if isinstance(m, AIMessage)
    )
    return Command(
        goto="writer_node",
        update={
            "active_agent": "writer",
            "messages": [
                last_ai,
                ToolMessage(
                    content="Transferred to the writer",
                    tool_call_id=runtime.tool_call_id,
                ),
            ],
        },
        graph=Command.PARENT,
    )

researcher_agent = create_agent(
    "openai:gpt-4.1-mini",
    tools=[transfer_to_analyst],
    prompt=(
        "You are a researcher. When the user asks you to research something, "
        "summarize the main sources and then transfer to the analyst."
    ),
)

analyst_agent = create_agent(
    "openai:gpt-4.1-mini",
    tools=[transfer_to_writer],
    prompt=(
        "You are an analyst. Analyze the researcher's information, "
        "identify trends and risks, and transfer to the writer."
    ),
)

writer_agent = create_agent(
    "openai:gpt-4.1-mini",
    tools=[],
    prompt="You are a writer. Generate an executive report with the information available.",
)

def researcher_node(state: TeamState):
    return researcher_agent.invoke(state)

def analyst_node(state: TeamState):
    return analyst_agent.invoke(state)

def writer_node(state: TeamState):
    return writer_agent.invoke(state)

def route_after_agent(state: TeamState) -> Literal["researcher_node", "analyst_node", "writer_node", "__end__"]:
    messages = state.get("messages", [])
    if messages:
        last = messages[-1]
        if isinstance(last, AIMessage) and not last.tool_calls:
            return "__end__"
    active = state.get("active_agent", "researcher")
    return f"{active}_node"

builder = StateGraph(TeamState)
builder.add_node("researcher_node", researcher_node)
builder.add_node("analyst_node", analyst_node)
builder.add_node("writer_node", writer_node)

builder.add_conditional_edges(START, lambda _: "researcher_node")
builder.add_conditional_edges("researcher_node", route_after_agent, ["analyst_node", "writer_node", END])
builder.add_conditional_edges("analyst_node", route_after_agent, ["researcher_node", "writer_node", END])
builder.add_conditional_edges("writer_node", route_after_agent, ["researcher_node", "analyst_node", END])

graph = builder.compile()

result = graph.invoke({
    "messages": [{"role": "user", "content": "Research the state of AI agents in 2025 and generate an executive report"}],
    "active_agent": "researcher",
})

for msg in result["messages"]:
    msg.pretty_print()
# Expected output (varies by model):
# ================================ Human Message =================================
# Research the state of AI agents in 2025 and generate an executive report
# ================================== Ai Message ==================================
# I've researched the main sources on AI agents...
# ================================== Tool Message ================================
# Transferred to the analyst
# ================================== Ai Message ==================================
# Based on the research, I identify 3 key trends...
# ================================== Tool Message ================================
# Transferred to the writer
# ================================== Ai Message ==================================
# EXECUTIVE REPORT: AI Agents in 2025
# ...

The mechanism:

1. Researcher receives the task from the user
2. It researches and decides to transfer → calls transfer_to_analyst
3. The tool returns Command(goto="analyst_node", graph=Command.PARENT) → the parent graph navigates to the analyst
4. Analyst receives the context, analyzes, and calls transfer_to_writer
5. Writer receives everything and generates the final report with no handoff tools (the flow ends)

The ToolMessage is critical: when an LLM calls a tool, it expects a response. Without the ToolMessage carrying the right tool_call_id, the conversation history ends up malformed.

State transfer: full vs filtered

When Agent A passes control to Agent B, what does Agent B see? There are two strategies:

Full handoff: Agent B sees the entire history — every message, every tool result, every intermediate decision. Simple to implement, but the context grows with each handoff.

Filtered handoff: Agent B only sees what's relevant — a summary of what came before, or just the data it needs. More implementation work, but the context stays focused.

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langgraph.types import Command

class State(TypedDict):
    task: str
    raw_data: str
    cleaned_summary: str
    internal_notes: str
    analysis: str
    log: Annotated[list[str], operator.add]

def data_collector(state: State) -> Command:
    raw = (
        "Raw data: 500 sales records, "
        "200 with formatting errors, "
        "50 duplicates, "
        "fields: date, product, amount, region"
    )
    notes = "INTERNAL NOTE: the data API responded in 2.3s, rate limit close to 80%"
    summary = "Clean data: 250 valid Q4 sales records, 4 regions, 12 products"
    return Command(
        goto="analyst",
        update={
            "raw_data": raw,
            "cleaned_summary": summary,
            "internal_notes": notes,
            "log": ["collector: data collected → handoff to analyst"],
        },
    )

def analyst_full_context(state: State) -> dict:
    """Analyst that sees the ENTIRE state (full handoff)."""
    sees_raw = len(state["raw_data"]) > 0
    sees_notes = len(state["internal_notes"]) > 0
    sees_summary = len(state["cleaned_summary"]) > 0
    analysis = (
        f"Analyst (full context) sees: "
        f"raw_data={'yes' if sees_raw else 'no'}, "
        f"internal_notes={'yes' if sees_notes else 'no'}, "
        f"cleaned_summary={'yes' if sees_summary else 'no'}"
    )
    return {"analysis": analysis, "log": ["analyst_full: analysis with the full context"]}

def analyst_filtered_context(state: State) -> dict:
    """Analyst that only sees the clean summary (filtered handoff)."""
    analysis = f"Analyst (filtered context) works with: '{state['cleaned_summary']}'"
    return {"analysis": analysis, "log": ["analyst_filtered: analysis with focused context"]}


print("=== FULL HANDOFF: the analyst sees everything ===\n")

builder_full = StateGraph(State)
builder_full.add_node("collector", data_collector)
builder_full.add_node("analyst", analyst_full_context)
builder_full.add_edge(START, "collector")
builder_full.add_edge("analyst", END)

graph_full = builder_full.compile()
result_full = graph_full.invoke({
    "task": "Q4 analysis", "raw_data": "", "cleaned_summary": "",
    "internal_notes": "", "analysis": "", "log": [],
})
print(f"Analysis: {result_full['analysis']}")


print("\n=== FILTERED HANDOFF: the analyst only sees the summary ===\n")

def data_collector_filtered(state: State) -> Command:
    raw = "Raw data: 500 records..."
    notes = "INTERNAL NOTE: rate limit 80%"
    summary = "250 valid Q4 sales records"
    return Command(
        goto="analyst",
        update={
            "raw_data": "",
            "cleaned_summary": summary,
            "internal_notes": "",
            "log": ["collector: filtered data → handoff to analyst"],
        },
    )

builder_filtered = StateGraph(State)
builder_filtered.add_node("collector", data_collector_filtered)
builder_filtered.add_node("analyst", analyst_filtered_context)
builder_filtered.add_edge(START, "collector")
builder_filtered.add_edge("analyst", END)

graph_filtered = builder_filtered.compile()
result_filtered = graph_filtered.invoke({
    "task": "Q4 analysis", "raw_data": "", "cleaned_summary": "",
    "internal_notes": "", "analysis": "", "log": [],
})
print(f"Analysis: {result_filtered['analysis']}")
# Expected output:
# === FULL HANDOFF: the analyst sees everything ===
#
# Analysis: Analyst (full context) sees: raw_data=yes, internal_notes=yes, cleaned_summary=yes
#
# === FILTERED HANDOFF: the analyst only sees the summary ===
#
# Analysis: Analyst (filtered context) works with: '250 valid Q4 sales records'

In the filtered handoff, the collector doesn't pass raw_data or internal_notes to the analyst. It only passes cleaned_summary. That keeps the analyst's context focused on what it actually needs.

Rule: Filter the context when the destination agent doesn't need the intermediate details. Pass everything when the destination agent needs the full history to make decisions.

Handoff chains: A → B → C → answer

Handoffs chain naturally. Each agent processes, transforms, and passes to the next:

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langgraph.types import Command

class ChainState(TypedDict):
    query: str
    search_results: str
    fact_check: str
    draft: str
    edited: str
    log: Annotated[list[str], operator.add]

def searcher(state: ChainState) -> Command:
    results = f"3 results for '{state['query']}': [scientific paper, technical blog, documentation]"
    return Command(
        goto="fact_checker",
        update={"search_results": results, "log": ["searcher → fact_checker"]},
    )

def fact_checker(state: ChainState) -> Command:
    check = f"Verified: 2 of 3 sources are trustworthy. Source discarded: blog with no references"
    return Command(
        goto="drafter",
        update={"fact_check": check, "log": ["fact_checker → drafter"]},
    )

def drafter(state: ChainState) -> Command:
    draft = (
        f"Draft on '{state['query']}':\n"
        f"Based on {state['fact_check'][:40]}...\n"
        f"Content: detailed analysis with 3 sections"
    )
    return Command(
        goto="editor",
        update={"draft": draft, "log": ["drafter → editor"]},
    )

def editor(state: ChainState) -> dict:
    edited = f"FINAL VERSION (edited): {state['draft'][:60]}... [Style and grammar fixed]"
    return {"edited": edited, "log": ["editor: ready to publish"]}

builder = StateGraph(ChainState)
builder.add_node("searcher", searcher)
builder.add_node("fact_checker", fact_checker)
builder.add_node("drafter", drafter)
builder.add_node("editor", editor)

builder.add_edge(START, "searcher")
builder.add_edge("editor", END)

graph = builder.compile()

result = graph.invoke({
    "query": "Best practices for RAG in production",
    "search_results": "", "fact_check": "",
    "draft": "", "edited": "", "log": [],
})

print(f"Final result: {result['edited']}")
print(f"\nHandoff chain:")
for step in result["log"]:
    print(f"  {step}")
# Expected output:
# Final result: FINAL VERSION (edited): Draft on 'Best practices for RAG in production':
# Based on Ve... [Style and grammar fixed]
#
# Handoff chain:
#   searcher → fact_checker
#   fact_checker → drafter
#   drafter → editor
#   editor: ready to publish

Four agents, zero supervisors. Each agent knows exactly who to pass the work to. The flow is predictable and each handoff transfers exactly the context the next one needs.

Handoff with return: A → B → back to A

Sometimes an agent needs to delegate a subtask and get control back. The researcher needs a specialist to verify a data point, and then continues the investigation:

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langgraph.types import Command

class RoundTripState(TypedDict):
    task: str
    research: str
    verification: str
    final_report: str
    phase: str
    log: Annotated[list[str], operator.add]

def researcher(state: RoundTripState) -> Command:
    if state["phase"] == "initial":
        research = f"Initial research on '{state['task']}': key data point found but needs verification"
        return Command(
            goto="specialist",
            update={
                "research": research,
                "phase": "verifying",
                "log": ["researcher: initial research → handoff to specialist"],
            },
        )
    else:
        report = (
            f"FINAL REPORT\n"
            f"Research: {state['research'][:50]}...\n"
            f"Verification: {state['verification'][:50]}...\n"
            f"Conclusion: data confirmed, positive recommendation"
        )
        return Command(
            goto="__end__",
            update={
                "final_report": report,
                "phase": "complete",
                "log": ["researcher: final report generated"],
            },
        )

def specialist(state: RoundTripState) -> Command:
    verification = f"Data point verified against 3 independent sources: CONFIRMED with 95% confidence"
    return Command(
        goto="researcher",
        update={
            "verification": verification,
            "phase": "verified",
            "log": ["specialist: verification complete → handoff back to researcher"],
        },
    )

builder = StateGraph(RoundTripState)
builder.add_node("researcher", researcher)
builder.add_node("specialist", specialist)

builder.add_edge(START, "researcher")

graph = builder.compile()

result = graph.invoke({
    "task": "Impact of AI on enterprise productivity",
    "research": "", "verification": "", "final_report": "",
    "phase": "initial", "log": [],
})

print(result["final_report"])
print(f"\nRound-trip flow:")
for step in result["log"]:
    print(f"  {step}")
# Expected output:
# FINAL REPORT
# Research: Initial research on 'Impact of AI on enterprise pr...
# Verification: Data point verified against 3 independent sources:...
# Conclusion: data confirmed, positive recommendation
#
# Round-trip flow:
#   researcher: initial research → handoff to specialist
#   specialist: verification complete → handoff back to researcher
#   researcher: final report generated

The researcher uses phase to know where in the flow it is. On the first invocation it researches and delegates to the specialist. When the specialist hands control back (return handoff), the researcher is in phase "verified" and generates the final report.

This pattern is useful when an agent needs external validation before continuing. The researcher doesn't lose its context — everything persists in the shared state.

When handoffs get complicated

Handoffs are elegant when the chain is short (2-4 agents) and the flow is predictable. They get messy when:

• ⚠️ Too many handoffs: A → B → C → D → E → F. With 6+ agents in a chain, debugging an error means tracing the whole chain. Who corrupted the state? Who received the wrong context?
• ⚠️ Circular handoffs with no exit condition: A → B → A → B → A... With no clear termination condition, the flow never ends (or it hits recursion_limit)
• ⚠️ State that grows unchecked: Every agent adds data to the state. If you don't filter, the last agent in the chain receives everything the previous ones piled up
• ⚠️ Complex conditional handoffs: If Agent A can transfer to B, C, or D depending on 5 conditions, you're reinventing a supervisor with more complexity

Rule: If your handoff chain needs a "map" to understand it, you probably need a supervisor.

Comparison table: supervisor vs handoffs vs hybrid

The hybrid pattern is the most common one in production: a high-level supervisor that coordinates phases (research phase → analysis phase → writing phase), and inside each phase, the agents use handoffs for the internal sequence.

Troubleshooting

Problem 1: "The handoff doesn't transfer to the right agent"

Symptom: Command(goto="analyst") doesn't navigate to the analyst, or the graph ends prematurely.

Cause: The name in goto doesn't match the name registered in add_node().

Fix: Check that the name is identical:

builder.add_node("analyst", analyst_fn)

Command(goto="analyst")

Problem 2: "Missing ToolMessage in a tool-based handoff"

Symptom: The destination agent receives a malformed message history and produces errors or incoherent responses.

Cause: You didn't include the ToolMessage when doing the handoff. When an LLM calls a tool, it expects a response carrying the same tool_call_id.

Fix: Always include the AIMessage + ToolMessage pair in the update:

@tool
def transfer_to_analyst(runtime: ToolRuntime) -> Command:
    last_ai = next(
        m for m in reversed(runtime.state["messages"])
        if isinstance(m, AIMessage)
    )
    return Command(
        goto="analyst_node",
        update={
            "messages": [
                last_ai,
                ToolMessage(content="Transferred", tool_call_id=runtime.tool_call_id),
            ],
        },
        graph=Command.PARENT,
    )

Problem 3: "Infinite circular handoff (A → B → A → B...)"

Symptom: The graph runs indefinitely until it hits recursion_limit.

Cause: There's no termination condition in the handoff cycle.

Fix: Use a state field (phase, iteration_count) to control when the cycle should end:

def agent_a(state: State) -> Command:
    if state["iteration"] >= 3:
        return Command(goto="__end__", update={"log": ["stopping"]})
    return Command(
        goto="agent_b",
        update={"iteration": state["iteration"] + 1},
    )

Problem 4: "The destination agent's context has data it doesn't need"

Symptom: The analyst receives the researcher's internal data (notes, failed attempts, raw data) that it doesn't need.

Cause: You're using a full handoff when you should be filtering.

Fix: In Command.update, only include the fields relevant to the destination agent. Clear the internal fields:

return Command(
    goto="analyst",
    update={
        "cleaned_summary": summary,
        "internal_notes": "",
        "raw_data": "",
    },
)

Exercises

Exercise 1: Basic handoff between two agents (Easy)

Build a system with two agents: translator and reviewer. The translator translates a text from English to French, and hands off to the reviewer, who checks the translation. Use Command for the handoff.

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langgraph.types import Command

class State(TypedDict):
    original_text: str
    translation: str
    review: str
    log: Annotated[list[str], operator.add]

def translator(state: State) -> Command:
    translated = f"Translation of '{state['original_text'][:30]}...': [French version here]"
    return Command(
        goto="reviewer",
        update={"translation": translated, "log": ["translator → reviewer"]},
    )

def reviewer(state: State) -> dict:
    review = f"Review: '{state['translation'][:40]}...' — Accuracy: 95%, fluency: good"
    return {"review": review, "log": ["reviewer: review complete"]}

builder = StateGraph(State)
builder.add_node("translator", translator)
builder.add_node("reviewer", reviewer)

builder.add_edge(START, "translator")
builder.add_edge("reviewer", END)

graph = builder.compile()
result = graph.invoke({
    "original_text": "AI agents are transforming the industry",
    "translation": "", "review": "", "log": [],
})

print(f"Translation: {result['translation']}")
print(f"Review: {result['review']}")
print(f"Log: {result['log']}")
# Expected output:
# Translation: Translation of 'AI agents are transforming the...': [French version here]
# Review: Review: 'Translation of 'AI agents are transformi...' — Accuracy: 95%, fluency: good
# Log: ['translator → reviewer', 'reviewer: review complete']

Exercise 2: Chain of 3 agents with context filtering (Medium)

Build a chain: extractor → classifier → formatter. The extractor pulls raw data (name, email, phone, internal notes). The classifier should only receive name, email, phone (no internal notes). The formatter receives the classification and produces clean output. Implement context filtering in the handoffs.

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langgraph.types import Command

class State(TypedDict):
    input_text: str
    name: str
    email: str
    phone: str
    internal_notes: str
    category: str
    formatted_output: str
    log: Annotated[list[str], operator.add]

def extractor(state: State) -> Command:
    return Command(
        goto="classifier",
        update={
            "name": "María García",
            "email": "maria@empresa.com",
            "phone": "+52 555 1234567",
            "internal_notes": "Hot lead, contact before Friday. Internal score: 85/100",
            "log": ["extractor: data extracted → filtered handoff to classifier"],
        },
    )

def classifier(state: State) -> Command:
    has_notes = len(state.get("internal_notes", "")) > 0
    category = "enterprise" if "@empresa" in state["email"] else "individual"
    return Command(
        goto="formatter",
        update={
            "category": category,
            "internal_notes": "",
            "log": [f"classifier: category={category}, sees internal notes={has_notes} → handoff to formatter"],
        },
    )

def formatter(state: State) -> dict:
    output = (
        f"CONTACT [{state['category'].upper()}]\n"
        f"  Name: {state['name']}\n"
        f"  Email: {state['email']}\n"
        f"  Phone: {state['phone']}"
    )
    return {"formatted_output": output, "log": ["formatter: output generated"]}

builder = StateGraph(State)
builder.add_node("extractor", extractor)
builder.add_node("classifier", classifier)
builder.add_node("formatter", formatter)

builder.add_edge(START, "extractor")
builder.add_edge("formatter", END)

graph = builder.compile()
result = graph.invoke({
    "input_text": "Contact data...", "name": "", "email": "",
    "phone": "", "internal_notes": "", "category": "",
    "formatted_output": "", "log": [],
})

print(result["formatted_output"])
print(f"\nInternal notes at the end: '{result['internal_notes']}'")
print(f"\nLog:")
for step in result["log"]:
    print(f"  {step}")
# Expected output:
# CONTACT [ENTERPRISE]
#   Name: María García
#   Email: maria@empresa.com
#   Phone: +52 555 1234567
#
# Internal notes at the end: ''
#
# Log:
#   extractor: data extracted → filtered handoff to classifier
#   classifier: category=enterprise, sees internal notes=True → handoff to formatter
#   formatter: output generated

Exercise 3: Handoff with return and verification (Medium)

Build a system where a planner generates a plan, hands off to a validator, the validator checks the plan and hands control back to the planner. If validation fails, the planner regenerates. If it passes, the planner produces the final result. Use phase to control the flow.

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langgraph.types import Command

class State(TypedDict):
    goal: str
    plan: str
    validation: str
    is_valid: bool
    result: str
    phase: str
    attempt: int
    log: Annotated[list[str], operator.add]

def planner(state: State) -> Command:
    attempt = state.get("attempt", 0)

    if state["phase"] == "initial" or state["phase"] == "retry":
        attempt += 1
        if attempt == 1:
            plan = f"Plan v1 for '{state['goal']}': basic approach"
        else:
            plan = f"Plan v{attempt} for '{state['goal']}': improved approach (fix: {state['validation'][:30]}...)"
        return Command(
            goto="validator",
            update={
                "plan": plan,
                "attempt": attempt,
                "phase": "validating",
                "log": [f"planner: plan v{attempt} → handoff to validator"],
            },
        )
    else:
        result = f"EXECUTED: {state['plan']} — Validation: {state['validation']}"
        return Command(
            goto="__end__",
            update={
                "result": result,
                "phase": "complete",
                "log": ["planner: plan validated, result generated"],
            },
        )

def validator(state: State) -> Command:
    is_valid = state["attempt"] >= 2
    if is_valid:
        validation = "APPROVED: the plan meets every criterion"
        next_phase = "approved"
    else:
        validation = "REJECTED: missing error handling and logging"
        next_phase = "retry"
    return Command(
        goto="planner",
        update={
            "validation": validation,
            "is_valid": is_valid,
            "phase": next_phase,
            "log": [f"validator: {validation[:30]}... → handoff back to planner"],
        },
    )

builder = StateGraph(State)
builder.add_node("planner", planner)
builder.add_node("validator", validator)
builder.add_edge(START, "planner")

graph = builder.compile()
result = graph.invoke({
    "goal": "Migrate the database to PostgreSQL",
    "plan": "", "validation": "", "is_valid": False,
    "result": "", "phase": "initial", "attempt": 0, "log": [],
})

print(f"Attempts: {result['attempt']}")
print(f"Result: {result['result'][:80]}...")
print(f"\nFlow:")
for step in result["log"]:
    print(f"  {step}")
# Expected output:
# Attempts: 2
# Result: EXECUTED: Plan v2 for 'Migrate the database to PostgreSQL': improved approach (f...
#
# Flow:
#   planner: plan v1 → handoff to validator
#   validator: REJECTED: missing error handli... → handoff back to planner
#   planner: plan v2 → handoff to validator
#   validator: APPROVED: the plan meets every... → handoff back to planner
#   planner: plan validated, result generated

Exercise 4: Dynamic handoff based on task type (Medium)

Build a router_agent that receives a task and decides who to hand off to: code_agent (for coding tasks), writing_agent (for writing tasks), or data_agent (for data tasks). The handoff is decided with keywords in the task. Use Command with dynamic routing.

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langgraph.types import Command

class State(TypedDict):
    task: str
    routed_to: str
    result: str
    log: Annotated[list[str], operator.add]

CODE_KEYWORDS = ["code", "function", "bug", "implement", "python", "api"]
DATA_KEYWORDS = ["data", "analysis", "metrics", "dashboard", "sql"]

def router_agent(state: State) -> Command:
    task_lower = state["task"].lower()
    if any(kw in task_lower for kw in CODE_KEYWORDS):
        target = "code_agent"
    elif any(kw in task_lower for kw in DATA_KEYWORDS):
        target = "data_agent"
    else:
        target = "writing_agent"
    return Command(
        goto=target,
        update={
            "routed_to": target,
            "log": [f"router → {target}"],
        },
    )

def code_agent(state: State) -> dict:
    return {
        "result": f"[CODE] Implementation completed for: {state['task'][:40]}...",
        "log": ["code_agent: task complete"],
    }

def writing_agent(state: State) -> dict:
    return {
        "result": f"[WRITING] Content generated for: {state['task'][:40]}...",
        "log": ["writing_agent: task complete"],
    }

def data_agent(state: State) -> dict:
    return {
        "result": f"[DATA] Analysis completed for: {state['task'][:40]}...",
        "log": ["data_agent: task complete"],
    }

builder = StateGraph(State)
builder.add_node("router_agent", router_agent)
builder.add_node("code_agent", code_agent)
builder.add_node("writing_agent", writing_agent)
builder.add_node("data_agent", data_agent)

builder.add_edge(START, "router_agent")
builder.add_edge("code_agent", END)
builder.add_edge("writing_agent", END)
builder.add_edge("data_agent", END)

graph = builder.compile()

tasks = [
    "Implement a Python function to validate emails",
    "Write an article about trends in AI",
    "Analysis of Q4 sales metrics",
]

for task in tasks:
    result = graph.invoke({"task": task, "routed_to": "", "result": "", "log": []})
    print(f"Task: {task[:50]}...")
    print(f"  Result: {result['result']}")
    print(f"  Log: {result['log']}\n")
# Expected output:
# Task: Implement a Python function to validate emails...
#   Result: [CODE] Implementation completed for: Implement a Python function to validate ...
#   Log: ['router → code_agent', 'code_agent: task complete']
#
# Task: Write an article about trends in AI...
#   Result: [WRITING] Content generated for: Write an article about trends in AI...
#   Log: ['router → writing_agent', 'writing_agent: task complete']
#
# Task: Analysis of Q4 sales metrics...
#   Result: [DATA] Analysis completed for: Analysis of Q4 sales metrics...
#   Log: ['router → data_agent', 'data_agent: task complete']

Exercise 5: Full pipeline with handoffs and cumulative state (Advanced)

Build a research pipeline with 4 chained agents: sourcer (finds sources) → reader (reads and extracts key info) → synthesizer (synthesizes into findings) → presenter (generates an executive presentation). Each agent must accumulate its result in the state AND pass a filtered summary to the next one. At the end, print the full state showing what each agent produced.

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langgraph.types import Command

class ResearchState(TypedDict):
    topic: str
    sources: list[str]
    key_extracts: list[str]
    synthesis: str
    presentation: str
    handoff_context: str
    log: Annotated[list[str], operator.add]

def sourcer(state: ResearchState) -> Command:
    sources = [
        "arxiv.org/paper-llm-agents-2025",
        "blog.langchain.dev/multi-agent-patterns",
        "docs.anthropic.com/agent-design",
    ]
    context = f"{len(sources)} academic and technical sources identified"
    return Command(
        goto="reader",
        update={
            "sources": sources,
            "handoff_context": context,
            "log": [f"sourcer: {len(sources)} sources → reader"],
        },
    )

def reader(state: ResearchState) -> Command:
    extracts = [
        "LLM agents improve productivity by 40%",
        "Multi-agent beats single-agent on complex tasks",
        "Context window management is the main challenge",
    ]
    context = f"{len(extracts)} key findings extracted from {len(state['sources'])} sources"
    return Command(
        goto="synthesizer",
        update={
            "key_extracts": extracts,
            "handoff_context": context,
            "log": [f"reader: {len(extracts)} extracts → synthesizer"],
        },
    )

def synthesizer(state: ResearchState) -> Command:
    synthesis = (
        f"SYNTHESIS: Out of {len(state['key_extracts'])} findings, "
        f"the main trend is that AI agents are maturing fast. "
        f"Key finding: '{state['key_extracts'][0]}'"
    )
    context = "Synthesis ready: 1 main trend, 1 key finding, 1 challenge"
    return Command(
        goto="presenter",
        update={
            "synthesis": synthesis,
            "handoff_context": context,
            "log": ["synthesizer: synthesis complete → presenter"],
        },
    )

def presenter(state: ResearchState) -> dict:
    presentation = (
        f"EXECUTIVE PRESENTATION\n"
        f"{'=' * 40}\n"
        f"Topic: {state['topic']}\n"
        f"Sources consulted: {len(state['sources'])}\n"
        f"Key findings: {len(state['key_extracts'])}\n"
        f"\n{state['synthesis']}\n"
        f"{'=' * 40}\n"
        f"Recommendation: Invest in multi-agent capabilities"
    )
    return {
        "presentation": presentation,
        "log": ["presenter: presentation generated"],
    }

builder = StateGraph(ResearchState)
builder.add_node("sourcer", sourcer)
builder.add_node("reader", reader)
builder.add_node("synthesizer", synthesizer)
builder.add_node("presenter", presenter)

builder.add_edge(START, "sourcer")
builder.add_edge("presenter", END)

graph = builder.compile()

result = graph.invoke({
    "topic": "AI Agents in enterprise production",
    "sources": [], "key_extracts": [],
    "synthesis": "", "presentation": "",
    "handoff_context": "", "log": [],
})

print(result["presentation"])
print(f"\nFull pipeline:")
for step in result["log"]:
    print(f"  → {step}")
# Expected output:
# EXECUTIVE PRESENTATION
# ========================================
# Topic: AI Agents in enterprise production
# Sources consulted: 3
# Key findings: 3
#
# SYNTHESIS: Out of 3 findings, the main trend is that AI agents are maturing fast. Key finding: 'LLM agents improve productivity by 40%'
# ========================================
# Recommendation: Invest in multi-agent capabilities
#
# Full pipeline:
#   → sourcer: 3 sources → reader
#   → reader: 3 extracts → synthesizer
#   → synthesizer: synthesis complete → presenter
#   → presenter: presentation generated

Summary

In this capsule you learned:

• A handoff is a direct transfer of control between agents, without going through a supervisor. Agent A decides Agent B should continue and hands it control along with the context it needs
• Command(goto="target", update={...}) is LangGraph's mechanism for implementing handoffs at the node level. The node returns a Command that tells the graph where to go and what to update
• Handoff tools let LLM-driven agents (created with create_agent) dynamically decide who to transfer to. The tool returns a Command with graph=Command.PARENT to navigate the parent graph
• The ToolMessage is mandatory in tool-based handoffs: it completes the request-response cycle the LLM expects when it calls a tool
• Context can be filtered during the handoff — you don't always need to pass the whole state. Pass only what the destination agent needs
• Handoffs chain (A → B → C) for predictable pipelines, and they support returns (A → B → A) for delegation with control coming back
• Handoffs vs supervisor: use handoffs when the flow is predictable and sequential; use a supervisor when you need dynamic coordination
• Watch out for long chains: more than 4-5 handoffs makes debugging hard. If you need a "map" to understand the flow, consider a supervisor

Next capsule: Pattern Subagents — agents that run in an isolated context. Unlike handoffs, where state is shared, subagents work in their own space and only return the result. That prevents context bloat when you have many subtasks.

Further resources

1. LangChain — Handoffs — Official documentation of the handoffs pattern
2. LangGraph — Command — Reference for the Command object used to navigate graphs
3. create_handoff_tool API Reference — Convenience API for creating handoff tools
4. LangGraph — Multi-Agent Systems — Overview of multi-agent patterns
5. Context Engineering for Agents — How to design the flow of context between agents

Module 10 — LangChain & LangGraph: From Chains to Agents