Skip to content

在**交接(handoffs)**架构中,行为会根据状态动态变化。其核心机制是:工具更新一个在多个对话轮次中持久存在的状态变量(例如 current_stepactive_agent),系统读取此变量以调整行为——要么应用不同的配置(系统提示词、工具),要么路由到不同的智能体(agent)。此模式既支持不同智能体之间的交接,也支持单个智能体内部的动态配置变更。

交接(handoffs) 一词由 OpenAI 创造,用于描述使用工具调用(例如 transfer_to_sales_agent)在智能体或状态之间转移控制权。

mermaid
sequenceDiagram
    participant User
    participant Agent
    participant Workflow State

    User->>Agent: "My phone is broken"
    Note over Agent,Workflow State: Step: Get warranty status<br/>Tools: record_warranty_status
    Agent-->>User: "Is your device under warranty?"

    User->>Agent: "Yes, it's still under warranty"
    Agent->>Workflow State: record_warranty_status("in_warranty")
    Note over Agent,Workflow State: Step: Classify issue<br/>Tools: record_issue_type
    Agent-->>User: "Can you describe the issue?"

    User->>Agent: "The screen is cracked"
    Agent->>Workflow State: record_issue_type("hardware")
    Note over Agent,Workflow State: Step: Provide resolution<br/>Tools: provide_solution, escalate_to_human
    Agent-->>User: "Here's the warranty repair process..."

关键特性

  • 状态驱动行为:行为基于状态变量(例如 current_stepactive_agent)变化。
  • 基于工具的转换:工具更新状态变量以在不同状态间移动。
  • 直接用户交互:每个状态的配置直接处理用户消息。
  • 持久化状态:状态在对话轮次间持续存在。

何时使用

当你需要强制执行顺序约束(仅在满足前提条件后解锁功能)、智能体需要在不同状态下直接与用户对话,或者你正在构建多阶段对话流程时,请使用交接模式。此模式对于需要按特定顺序收集信息的客户支持场景特别有价值——例如,在处理退款前收集保修 ID。

基础实现

核心机制是一个返回 Command 来更新状态的工具,从而触发向新步骤或新智能体的转换:

python
from langchain.tools import tool
from langchain.messages import ToolMessage
from langgraph.types import Command

@tool
def transfer_to_specialist(runtime) -> Command:
    """Transfer to the specialist agent."""
    return Command(
        update={
            "messages": [
                ToolMessage(  
                    content="Transferred to specialist",
                    tool_call_id=runtime.tool_call_id  
                )
            ],
            "current_step": "specialist"  # Triggers behavior change
        }
    )

为什么包含 ToolMessage 当 LLM 调用一个工具时,它期望得到一个响应。带有匹配 tool_call_idToolMessage 完成了这个请求-响应周期——没有它,对话历史就会变得格式错误。每当你的交接工具更新消息时,这都是必需的。

关于完整实现,请参阅下面的教程。

教程:使用交接模式构建客户支持系统

学习如何使用交接模式构建一个客户支持智能体,其中单个智能体在不同配置之间转换。

了解更多

实现方法

有两种实现交接的方式:使用中间件的单智能体(一个具有动态配置的智能体)或**多智能体子图**(作为图节点的不同智能体)。

使用中间件的单智能体

单个智能体根据状态改变其行为。中间件拦截每个模型调用,并动态调整系统提示词和可用工具。工具更新状态变量以触发转换:

python
from langchain.tools import ToolRuntime, tool
from langchain.messages import ToolMessage
from langgraph.types import Command

@tool
def record_warranty_status(
    status: str,
    runtime: ToolRuntime[None, SupportState]
) -> Command:
    """Record warranty status and transition to next step."""
    return Command(
        update={
            "messages": [
                ToolMessage(
                    content=f"Warranty status recorded: {status}",
                    tool_call_id=runtime.tool_call_id
                )
            ],
            "warranty_status": status,
            "current_step": "specialist"  # Update state to trigger transition
        }
    )
完整示例:使用中间件的客户支持
python
from langchain.agents import AgentState, create_agent
from langchain.agents.middleware import wrap_model_call, ModelRequest, ModelResponse
from langchain.tools import tool, ToolRuntime
from langchain.messages import ToolMessage
from langgraph.types import Command
from typing import Callable

# 1. Define state with current_step tracker
class SupportState(AgentState):  
    """Track which step is currently active."""
    current_step: str = "triage"
    warranty_status: str | None = None

# 2. Tools update current_step via Command
@tool
def record_warranty_status(
    status: str,
    runtime: ToolRuntime[None, SupportState]
) -> Command:  
    """Record warranty status and transition to next step."""
    return Command(update={  
        "messages": [  
            ToolMessage(
                content=f"Warranty status recorded: {status}",
                tool_call_id=runtime.tool_call_id
            )
        ],
        "warranty_status": status,
        # Transition to next step
        "current_step": "specialist"
    })

# 3. Middleware applies dynamic configuration based on current_step
@wrap_model_call
def apply_step_config(
    request: ModelRequest,
    handler: Callable[[ModelRequest], ModelResponse]
) -> ModelResponse:
    """Configure agent behavior based on current_step."""
    step = request.state.get("current_step", "triage")  

    # Map steps to their configurations
    configs = {
        "triage": {
            "prompt": "Collect warranty information...",
            "tools": [record_warranty_status]
        },
        "specialist": {
            "prompt": "Provide solutions based on warranty: {warranty_status}",
            "tools": [provide_solution, escalate]
        }
    }

    config = configs[step]
    request = request.override(  
        system_prompt=config["prompt"].format(**request.state),  
        tools=config["tools"]  
    )
    return handler(request)

# 4. Create agent with middleware
agent = create_agent(
    model,
    tools=[record_warranty_status, provide_solution, escalate],
    state_schema=SupportState,
    middleware=[apply_step_config],  
    checkpointer=InMemorySaver()  # Persist state across turns  #
)

多智能体子图

多个不同的智能体作为单独的节点存在于图中。交接工具使用 Command.PARENT 在智能体节点之间导航,以指定接下来执行哪个节点。

子图交接需要仔细的上下文工程(context engineering)。与单智能体中间件(消息历史自然流动)不同,你必须明确决定哪些消息在智能体之间传递。如果处理不当,智能体将收到格式错误的对话历史或臃肿的上下文。请参阅下面的上下文工程

python
from langchain.messages import AIMessage, ToolMessage
from langchain.tools import tool, ToolRuntime
from langgraph.types import Command

@tool
def transfer_to_sales(
    runtime: ToolRuntime,
) -> Command:
    """Transfer to the sales agent."""
    last_ai_message = next(  
        msg for msg in reversed(runtime.state["messages"]) if isinstance(msg, AIMessage)  
    )  
    transfer_message = ToolMessage(  
        content="Transferred to sales agent",  
        tool_call_id=runtime.tool_call_id,  
    )  
    return Command(
        goto="sales_agent",
        update={
            "active_agent": "sales_agent",
            "messages": [last_ai_message, transfer_message],  
        },
        graph=Command.PARENT
    )

此示例展示了一个包含独立销售和支持智能体的多智能体系统。每个智能体都是一个独立的图节点,交接工具允许智能体将会话转移给对方。

:::python

python
from typing import Literal

from langchain.agents import AgentState, create_agent
from langchain.messages import AIMessage, ToolMessage
from langchain.tools import tool, ToolRuntime
from langgraph.graph import StateGraph, START, END
from langgraph.types import Command
from typing_extensions import NotRequired

# 1. Define state with active_agent tracker
class MultiAgentState(AgentState):
active_agent: NotRequired[str]

# 2. Create handoff tools
@tool
def transfer_to_sales(
runtime: ToolRuntime,
) -> Command:
"""Transfer to the sales agent."""
last_ai_message = next(  
msg for msg in reversed(runtime.state["messages"]) if isinstance(msg, AIMessage)  
)  
transfer_message = ToolMessage(  
content="Transferred to sales agent from support agent",  
tool_call_id=runtime.tool_call_id,  
)  
return Command(
goto="sales_agent",
update={
"active_agent": "sales_agent",
"messages": [last_ai_message, transfer_message],  
},
graph=Command.PARENT,
)

@tool
def transfer_to_support(
runtime: ToolRuntime,
) -> Command:
"""Transfer to the support agent."""
last_ai_message = next(  
msg for msg in reversed(runtime.state["messages"]) if isinstance(msg, AIMessage)  
)  
transfer_message = ToolMessage(  
content="Transferred to support agent from sales agent",  
tool_call_id=runtime.tool_call_id,  
)  
return Command(
goto="support_agent",
update={
"active_agent": "support_agent",
"messages": [last_ai_message, transfer_message],  
},
graph=Command.PARENT,
)

# 3. Create agents with handoff tools
sales_agent = create_agent(
model="anthropic:claude-sonnet-4-20250514",
tools=[transfer_to_support],
system_prompt="You are a sales agent. Help with sales inquiries. If asked about technical issues or support, transfer to the support agent.",
)

support_agent = create_agent(
model="anthropic:claude-sonnet-4-20250514",
tools=[transfer_to_sales],
system_prompt="You are a support agent. Help with technical issues. If asked about pricing or purchasing, transfer to the sales agent.",
)

# 4. Create agent nodes that invoke the agents
def call_sales_agent(state: MultiAgentState) -> Command:
"""Node that calls the sales agent."""
response = sales_agent.invoke(state)
return response

def call_support_agent(state: MultiAgentState) -> Command:
"""Node that calls the support agent."""
response = support_agent.invoke(state)
return response

# 5. Create router that checks if we should end or continue
def route_after_agent(
state: MultiAgentState,
) -> Literal["sales_agent", "support_agent", "__end

LangChain 中文文档