Swarm and Handoff Patterns

Agent A ← → Agent B
  ↓         ↓
Agent C ← → Agent D
(decentralized, peer-to-peer)

from langgraph.prebuilt import create_react_agent
from langgraph_swarm import create_handoff_tool, create_swarm
from langchain_openai import ChatOpenAI

gpt4o = ChatOpenAI(model="gpt-4o")
gpt4o_mini = ChatOpenAI(model="gpt-4o-mini")

# --- Agents with handoff tools ---

billing_agent = create_react_agent(
    model=gpt4o,
    tools=[
        get_invoice,
        process_refund,
        create_handoff_tool(
            agent_name="support_agent",
            description="Transfer to technical support when the question is about "
                        "product functionality rather than billing."
        ),
    ],
    name="billing_agent",
    prompt="""You handle billing and payment questions: invoices, charges, refunds,
    subscription plans. When a question is technical rather than financial,
    transfer to support_agent."""
)

support_agent = create_react_agent(
    model=gpt4o,
    tools=[
        search_docs,
        get_error_logs,
        create_handoff_tool(
            agent_name="billing_agent",
            description="Transfer to billing when the question involves charges, "
                        "invoices, or payment processing."
        ),
        create_handoff_tool(
            agent_name="escalation_agent",
            description="Transfer to escalation when the issue cannot be resolved "
                        "with documentation or standard troubleshooting."
        ),
    ],
    name="support_agent",
    prompt="""You handle technical support: troubleshooting, documentation, error diagnosis.
    Transfer to billing_agent for financial questions.
    Transfer to escalation_agent for unresolvable issues."""
)

escalation_agent = create_react_agent(
    model=gpt4o,
    tools=[
        create_ticket,
        notify_engineer,
    ],
    name="escalation_agent",
    prompt="""You handle escalations that other agents cannot resolve.
    Create tickets and notify the on-call engineer. Do not transfer back."""
)

# --- Assemble the swarm ---

swarm = create_swarm(
    agents=[billing_agent, support_agent, escalation_agent],
    default_active_agent="billing_agent"  # entry point
)

app = swarm.compile()

# Usage
result = app.invoke({
    "messages": [{"role": "user", "content": "I'm getting a 500 error on the API and I was charged twice this month"}]
})

# The handoff tool uses LangGraph's Command primitive internally:
# Command(goto="target_agent", update={"active_agent": "target_agent"})
# The state is updated but messages are preserved in full.

# You can verify handoff history in the state:
for msg in result["messages"]:
    if hasattr(msg, "name"):  # tool message from handoff tool
        print(f"Handoff: {msg.name}")
    else:
        print(f"{msg.type}: {msg.content[:100]}")

from langgraph.checkpoint.memory import MemorySaver

checkpointer = MemorySaver()
app = swarm.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": "user-session-42"}}

# First turn
result = app.invoke(
    {"messages": [{"role": "user", "content": "I need help with my invoice"}]},
    config=config
)

# User returns hours later — full state is restored from checkpoint
result = app.invoke(
    {"messages": [{"role": "user", "content": "The issue is still not resolved"}]},
    config=config
)

from openai.agents import Agent, Runner

billing_agent = Agent(
    name="BillingAgent",
    instructions="""You handle billing questions: invoices, charges, refunds, subscriptions.
    
    Hand off to TechSupportAgent when:
    - The question is about product functionality, not payment
    - The user reports a technical error unrelated to billing
    
    Do not attempt to diagnose technical issues yourself.""",
    tools=[get_invoice, process_refund, list_subscriptions],
    handoffs=["TechSupportAgent"]  # declared by name, resolved at runtime
)

support_agent = Agent(
    name="TechSupportAgent",
    instructions="""You handle technical support: troubleshooting, documentation, error logs.
    
    Hand off to BillingAgent when:
    - The question involves a charge, invoice, or payment
    
    Hand off to EscalationAgent when:
    - The issue requires engineering intervention
    - Standard troubleshooting has not resolved the problem""",
    tools=[search_docs, get_error_logs, run_diagnostic],
    handoffs=["BillingAgent", "EscalationAgent"]
)

escalation_agent = Agent(
    name="EscalationAgent",
    instructions="""You handle escalations. Create a ticket and notify on-call.
    Do not hand off further — you are the final escalation point.""",
    tools=[create_incident_ticket, page_on_call],
    handoffs=[]  # terminal agent
)

# Entry point for the swarm
result = await Runner.run(
    billing_agent,
    "I've been charged three times and now the API is returning 500 errors"
)
print(result.final_output)

from openai.agents import Agent, handoff, RunContextWrapper

def log_handoff(ctx: RunContextWrapper, input_data: str | None):
    print(f"Handoff initiated. Input: {input_data}")

billing_agent = Agent(
    name="BillingAgent",
    instructions="...",
    handoffs=[
        handoff(
            agent=support_agent,
            on_handoff=log_handoff,
            input_type=str  # typed input passed during handoff
        )
    ]
)

from openai.agents import Agent, handoff
from openai.agents.extensions import remove_all_tools

escalation_agent = Agent(
    name="EscalationAgent",
    instructions="...",
)

support_agent = Agent(
    name="TechSupportAgent",
    instructions="...",
    handoffs=[
        handoff(
            agent=escalation_agent,
            # Remove tool calls from history — escalation agent sees only messages
            input_filter=remove_all_tools
        )
    ]
)

# In LangGraph, track via state
class SwarmState(TypedDict):
    messages: list
    active_agent: str
    visited_agents: list[str]  # agents that have already executed

# In each agent node, check before accepting a handoff
def check_cycle(state: SwarmState, target: str) -> bool:
    if target in state["visited_agents"]:
        # Hand off to escalation instead of cycling
        return False
    return True