主题
在**交接(handoffs)**架构中,行为会根据状态动态变化。其核心机制是:工具更新一个在多个对话轮次中持久存在的状态变量(例如 current_step 或 active_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_step或active_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_id 的 ToolMessage 完成了这个请求-响应周期——没有它,对话历史就会变得格式错误。每当你的交接工具更新消息时,这都是必需的。
关于完整实现,请参阅下面的教程。
教程:使用交接模式构建客户支持系统
学习如何使用交接模式构建一个客户支持智能体,其中单个智能体在不同配置之间转换。
了解更多 →
实现方法
有两种实现交接的方式:使用中间件的单智能体(一个具有动态配置的智能体)或**多智能体子图**(作为图节点的不同智能体)。
使用中间件的单智能体
单个智能体根据状态改变其行为。中间件拦截每个模型调用,并动态调整系统提示词和可用工具。工具更新状态变量以触发转换:
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