Agent Communication & Handoffs

The Handoff Problem

Handoffs are where multi-agent systems most commonly fail. An agent builds up 15 steps of context, understands the user’s real intent, has found the key relevant facts, and then hands off to another agent with a single line: “Process this billing issue.”

The receiving agent has lost everything: the full conversation context, what was tried and failed, the specific nuances of the user’s situation, and any established facts from prior research.

Good handoff design is the difference between multi-agent systems that feel like a coherent team and ones that feel like a broken phone chain.

Structured Handoff Schemas

Typed Pydantic models make handoff payloads explicit and validated, if a required field is missing, the error surfaces at handoff time, not when the receiving agent tries to use it.

# src/agents/handoffs.py
from pydantic import BaseModel, Field
from enum import Enum
from datetime import datetime

class UrgencyLevel(str, Enum):
    low = "low"
    medium = "medium"
    high = "high"
    critical = "critical"

class BaseHandoff(BaseModel):
    """Base model for all agent handoffs."""
    from_agent: str       # tracks provenance, useful for debugging and audit logs
    to_agent: str
    timestamp: str = Field(default_factory=lambda: datetime.utcnow().isoformat())
    reason: str = Field(description="Why this handoff is happening")
    user_goal: str = Field(description="What the user ultimately wants to achieve")
    urgency: UrgencyLevel = UrgencyLevel.medium

class SupportHandoff(BaseHandoff):
    """Handoff from a general support agent to a specialist."""
    customer_id: str
    issue_category: str
    issue_summary: str
    # list of what was tried: prevents the specialist from repeating failed solutions
    attempted_solutions: list[str] = Field(default_factory=list)
    established_facts: dict = Field(
        default_factory=dict,
        description="Facts confirmed during the support session"
    )
    relevant_account_data: dict = Field(default_factory=dict)

class ResearchHandoff(BaseHandoff):
    """Handoff from a researcher to an analyst or writer."""
    research_query: str
    sources_searched: list[str] = Field(default_factory=list)
    key_findings: list[str] = Field(default_factory=list)
    data_points: dict = Field(default_factory=dict)
    gaps_identified: list[str] = Field(
        default_factory=list,
        description="Areas where information was insufficient"  # tells the next agent where not to over-promise
    )
    # ge=0, le=1 enforces 0.0–1.0 range at validation time: not just documentation
    confidence_level: float = Field(ge=0, le=1, default=0.8)

Implementing Handoffs with LangGraph Command

Command lets a node simultaneously update shared state and specify the next node to run, keeping the handoff logic co-located rather than split across nodes and routing functions.

from langgraph.types import Command

class TriageState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]
    user_input: str
    handoff: dict | None  # serialized handoff payload, dict so it's JSON-serializable in state
    route_to: str | None

def triage_agent(state: TriageState) -> Command:
    """Classify the request and route to the right specialist."""
    
    classification_prompt = f"""
Classify this support request and route to the appropriate team:
Request: {state['user_input']}

Teams available:
- billing: Payment issues, invoices, subscriptions
- technical: Bugs, errors, integration problems
- account: Password, access, profile updates
- general: Questions, information requests

Respond with JSON:
{{
  "team": "billing|technical|account|general",
  "issue_summary": "one sentence description",
  "urgency": "low|medium|high|critical",
  "established_facts": {{}}
}}"""
    
    result = llm.invoke([HumanMessage(content=classification_prompt)])
    import json
    classification = json.loads(result.content)
    
    # Build typed handoff payload: Pydantic validates all required fields are present
    handoff = SupportHandoff(
        from_agent="triage",
        to_agent=classification["team"],
        reason=f"Classified as {classification['team']} issue",
        user_goal=state["user_input"],
        urgency=classification["urgency"],
        customer_id="",  # would come from auth context in a real system
        issue_category=classification["team"],
        issue_summary=classification["issue_summary"],
        established_facts=classification.get("established_facts", {}),
    )
    
    # Command(update=..., goto=...) both updates state AND routes to a new node in one operation
    return Command(
        update={
            "handoff": handoff.model_dump(),  # serialize to dict for state storage
            "route_to": classification["team"],
        },
        goto=classification["team"],  # jump directly to the specialist, no conditional edge needed
    )

def billing_specialist(state: TriageState) -> dict:
    """Handle billing issues with full handoff context."""
    # Deserialize back to typed model: catches schema drift between triage and specialist
    handoff = SupportHandoff(**state["handoff"]) if state["handoff"] else None
    
    context = ""
    if handoff:
        context = f"""
HANDOFF CONTEXT:
From: {handoff.from_agent}
Issue: {handoff.issue_summary}
Urgency: {handoff.urgency}
Previously tried: {', '.join(handoff.attempted_solutions) or 'Nothing yet'}
Established facts: {handoff.established_facts}
"""
    
    system = f"""You are a billing specialist. Handle the customer's billing issue.
{context}
Access billing records, process refunds, update subscriptions as needed."""
    
    response = billing_llm.invoke([
        SystemMessage(content=system),
        HumanMessage(content=state["user_input"]),
    ])
    
    return {"messages": [response]}

Building the Full Triage System

# Build the multi-agent routing graph
triage_builder = StateGraph(TriageState)

triage_builder.add_node("triage", triage_agent)
triage_builder.add_node("billing", billing_specialist)
triage_builder.add_node("technical", technical_specialist)
triage_builder.add_node("account", account_specialist)
triage_builder.add_node("general", general_handler)

triage_builder.set_entry_point("triage")

# Specialists all route to END after handling: single interaction per handoff
for team in ["billing", "technical", "account", "general"]:
    triage_builder.add_edge(team, END)

# triage uses Command(goto=...) so no explicit routing edge needed
# LangGraph handles Command-based routing automatically
triage_system = triage_builder.compile(checkpointer=MemorySaver())  # MemorySaver persists state across turns

Multi-hop Handoffs

Sometimes an agent needs to pass through multiple specialists:

def analyst_with_handoff(state: AnalysisState) -> Command:
    """Analyst hands off to writer after completing analysis."""
    analysis_result = run_analysis(state["research_findings"])
    
    # Build a rich handoff to the writer: include everything the writer needs to produce the final output
    research_handoff = ResearchHandoff(
        from_agent="analyst",
        to_agent="writer",
        reason="Analysis complete, ready for writing",
        user_goal=state["original_goal"],
        research_query=state["query"],
        key_findings=analysis_result.key_findings,   # structured findings, not raw text
        data_points=analysis_result.data_points,
        gaps_identified=analysis_result.gaps,        # tells writer what to hedge or omit
        confidence_level=analysis_result.confidence,
    )
    
    # Command(update=..., goto=...) both updates state AND routes to a new node in one operation
    return Command(
        update={"handoff": research_handoff.model_dump()},
        goto="writer",
    )

Handoff Failure Handling

When classification fails, always route to a safe fallback rather than crashing, include the error context in the handoff so the receiving agent knows the situation.

def robust_triage(state: TriageState) -> Command:
    """Triage with fallback if classification fails."""
    try:
        result = llm.invoke(...)
        classification = json.loads(result.content)
        team = classification["team"]
        
        if team not in ["billing", "technical", "account", "general"]:
            raise ValueError(f"Unknown team: {team}")  # explicit validation before using the value
        
        handoff = SupportHandoff(...)
        return Command(update={"handoff": handoff.model_dump()}, goto=team)
    
    except Exception as e:
        # Fallback: route to general with error context so the agent knows classification failed
        fallback_handoff = SupportHandoff(
            from_agent="triage",
            to_agent="general",
            reason=f"Classification failed ({e}), routing to general support",
            user_goal=state["user_input"],
            issue_category="unclassified",
            issue_summary="Classification failed, handle manually",
            urgency=UrgencyLevel.medium,
            customer_id="unknown",
        )
        return Command(
            update={"handoff": fallback_handoff.model_dump()},
            goto="general",  # general handler is the safest catch-all destination
        )

Robust handoffs are the connective tissue of multi-agent systems. Get them right and agents cooperate seamlessly. Get them wrong and you’ll spend most of your debugging time on information-loss bugs between agents.