Handoff 交接机制

什么是 Handoff

状态机转换与上下文序列化的底层实现

Handoff 的本质是一次运行状态的原子转移。当源 Agent 决定交接时，Runner 会暂停当前的执行流，将目标 Agent 设置为活跃状态，并决定传递哪些上下文给新 Agent。这个过程涉及三个核心技术问题：

1. 状态捕获：Runner 需要捕获当前运行的完整状态，包括消息历史、上下文对象、已执行的工具结果等。
2. 上下文序列化：捕获到的状态必须被序列化为目标 Agent 可以理解的输入格式。默认情况下，SDK 会传递完整的对话历史，但通过 input_filter 可以进行裁剪和变换。
3. 状态恢复：目标 Agent 接管后，Runner 需要在其上下文中恢复执行，确保对话的连续性。

从状态机视角来看，整个运行可以建模为 (current_agent, message_history, context) 三元组。Handoff 触发了一次状态转移：

(A, H, C) --handoff--> (B, filter(H), C)

其中 filter(H) 是应用 input_filter 后的历史消息。值得注意的是，上下文对象 C 通常会被完整传递（因为它通常包含用户身份、会话状态等关键信息），但 input_filter 也可以对其进行修改。

nest_handoff_history 机制则是一种有损压缩策略：当发生多次 Handoff 时，之前的历史不再逐条传递，而是被压缩为一段摘要文本。这显著降低了上下文长度，但也可能导致信息丢失——特别是当之前的对话中包含精确的数据（如订单号、金额）时，摘要可能丢失这些细节。

与 AutoGen 的 GroupChat 相比，Agents SDK 的 Handoff 是确定性路由：交接目标在 Agent 配置时就已经确定，模型只能在预定义的选项中选择。AutoGen 的 GroupChat 则支持更动态的群体决策（如投票、轮询），但也因此更难以预测和调试。

Handoff 允许一个 Agent 将任务委托给另一个 Agent。这在不同 Agent 专精于不同领域的场景特别有用。

例如客服系统中：

• Triage Agent：负责分类用户问题
• Billing Agent：处理账单相关问题
• Refund Agent：处理退款请求

from agents import Agent, handoff

billing_agent = Agent(name="Billing agent")
refund_agent = Agent(name="Refund agent")

triage_agent = Agent(
    name="Triage agent",
    handoffs=[billing_agent, handoff(refund_agent)]
)

基本用法

Handoff 在底层被表示为工具。如果有一个名为 Refund Agent 的 handoff，对应的工具名就是 transfer_to_refund_agent。

直接使用 Agent

triage_agent = Agent(
    name="Triage agent",
    handoffs=[billing_agent, refund_agent],  # 直接传 Agent 实例
)

使用 handoff() 函数定制

from agents import handoff, RunContextWrapper

def on_handoff(ctx: RunContextWrapper[None]):
    print("Handoff called!")

handoff_obj = handoff(
    agent=refund_agent,
    on_handoff=on_handoff,
    tool_name_override="custom_transfer",
    tool_description_override="Transfer to refund specialist",
)

triage_agent = Agent(
    name="Triage agent",
    handoffs=[handoff_obj],
)

Handoff 参数详解

参数	说明
agent	交接目标 Agent
tool_name_override	覆盖默认工具名 transfer_to_<agent_name>
tool_description_override	覆盖默认工具描述
on_handoff	交接触发时的回调函数
input_type	Handoff 工具调用的参数 schema
input_filter	过滤传递给下一个 Agent 的输入
is_enabled	是否启用（支持动态判断）
nest_handoff_history	是否嵌套交接历史

带参数的 Handoff

让 LLM 在交接时提供额外数据：

from pydantic import BaseModel

class EscalationData(BaseModel):
    reason: str
    priority: Literal["low", "medium", "high"]

async def on_escalation(ctx, input_data: EscalationData):
    print(f"Escalated: {input_data.reason}, priority: {input_data.priority}")

handoff_obj = handoff(
    agent=escalation_agent,
    on_handoff=on_escalation,
    input_type=EscalationData,
)

输入过滤

控制下一个 Agent 能看到的历史：

from agents.handoffs import HandoffInputData

def filter_input(data: HandoffInputData) -> HandoffInputData:
    # 只保留最近 5 条消息
    data.input_history = data.input_history[-5:]
    return data

handoff_obj = handoff(
    agent=specialist_agent,
    input_filter=filter_input,
)

嵌套交接历史

默认关闭的 Beta 功能，将历史折叠为摘要：

from agents import RunConfig

result = await Runner.run(
    triage_agent,
    "I need a refund",
    run_config=RunConfig(nest_handoff_history=True),
)

完整客服系统示例

from agents import Agent, Runner, handoff

faq_agent = Agent(
    name="FAQ agent",
    instructions="Answer common customer questions.",
    handoff_description="For general questions about products and services.",
)

billing_agent = Agent(
    name="Billing agent",
    instructions="Help with billing, invoices, and payments.",
    handoff_description="For billing and payment related questions.",
)

refund_agent = Agent(
    name="Refund agent",
    instructions="Process refund requests.",
    handoff_description="For refund and return requests.",
)

triage_agent = Agent(
    name="Triage agent",
    instructions="Classify user requests and route to the right specialist.",
    handoffs=[faq_agent, billing_agent, refund_agent],
)

result = Runner.run_sync(triage_agent, "I want a refund for order #12345")
print(result.final_output)

下一步

学习 Agent 作为工具模式 和 多 Agent 编排策略。

完整实战示例：智能客服路由系统与交接审计

以下示例展示了如何在生产环境中构建一个支持审计追踪和动态启用的智能客服路由系统：

import asyncio
import time
from dataclasses import dataclass
from typing import Any
from agents import Agent, Runner, handoff, RunContextWrapper, RunConfig
from agents.extensions.handoff_prompt import RECOMMENDED_PROMPT_PREFIX


@dataclass
class AppContext:
    user_id: str
    user_tier: str
    session_start: float
    handoff_count: int = 0


class HandoffAuditor:
    """交接审计器，记录每次交接的详细上下文。"""
    def __init__(self):
        self.records: list[dict] = []

    async def on_handoff_callback(self, ctx: RunContextWrapper[AppContext], target_agent_name: str):
        ctx.context.handoff_count += 1
        self.records.append({
            "timestamp": time.time(),
            "run_id": ctx.run_id,
            "from_agent": getattr(ctx, "current_agent_name", "unknown"),
            "to_agent": target_agent_name,
            "user_tier": ctx.context.user_tier,
            "session_duration": time.time() - ctx.context.session_start,
            "handoff_number": ctx.context.handoff_count,
        })
        print(f"[AUDIT] Handoff #{ctx.context.handoff_count}: -> {target_agent_name}")


auditor = HandoffAuditor()


# 定义专家 Agent
billing_agent = Agent(
    name="BillingExpert",
    instructions=f"""{RECOMMENDED_PROMPT_PREFIX}
你是账单专家。处理发票、付款、退款和订阅相关问题。
如果用户问题超出账单范围，请求回退到总机。
""",
    model="gpt-5-nano"
)

常见问题与调试

问题一：交接后目标 Agent 缺乏关键上下文

默认情况下，Handoff 传递完整历史，但如果使用了 input_filter 或 nest_handoff_history，可能导致目标 Agent 看不到重要信息。排查方法：

1. 在开发阶段关闭 nest_handoff_history，确保完整历史可用。
2. 为关键信息（如用户身份、订单号）设计独立的上下文传递机制，绕过 input_filter。
3. 在目标 Agent 的 instructions 中明确告知"如果某些信息缺失，请礼貌地询问用户"。

问题二：循环交接（A -> B -> A）导致无限循环

当两个 Agent 的专长边界模糊时，可能出现来回交接的死循环。解决方案：

1. 在 AppContext 中记录 handoff_count，当超过阈值时强制终止并转人工。
2. 优化 Agent 的 instructions，明确各自的职责边界和"不处理"的范围。
3. 使用 is_enabled 动态禁用可能导致循环的交接路径。

问题三：交接工具名与模型理解不匹配

默认的交接工具名（如 transfer_to_billing_expert）对模型来说通常足够清晰，但如果 Agent 名称过长或包含特殊字符，生成的工具名可能不符合预期。建议：

1. 始终使用 tool_name_override 显式指定简洁、语义清晰的工具名。
2. 通过 tool_description_override 补充交接场景的描述，帮助模型做出正确决策。
3. 在 trace 中查看模型看到的工具定义，确认其理解是否准确。

与其他方案对比

维度	Agents SDK Handoff	AutoGen GroupChat	LangGraph 状态路由
路由方式	模型决策 + 预定义目标	群体协商/投票	图节点条件跳转
上下文传递	input_filter 可控	可选共享/隔离	全局状态池
循环检测	需自行实现	有限支持	图结构天然避免
可解释性	高（交接原因可追踪）	中（群体决策复杂）	高（节点流转清晰）

LangGraph 通过显式的状态图定义路由逻辑，提供了最强的可预测性和调试能力，适合需要严格流程控制的场景（如审批工作流）。AutoGen 的 GroupChat 更擅长模拟多人协作的开放式讨论，但代价是可控性的降低。Agents SDK 的 Handoff 在两者之间找到了平衡点：既保留了模型自主决策的灵活性，又通过预定义目标限制了失控范围。

Handoff 路由的责任链模式与防循环机制

当客服系统中的 Agent 数量超过三个时，简单的硬编码 Handoff 列表会变得难以维护。此时可以引入责任链模式（Chain of Responsibility Pattern）：将每个 Agent 包装为一个处理节点，请求沿着链条传递，直到找到能够处理的节点为止。这种模式天然支持动态增删节点，且将路由决策从集中式管理分散到各个节点自身，每个节点只关注自己能否处理请求以及应该传递给谁。
mermaid
stateDiagram-v2
[] --> TriageAgent: 用户输入
TriageAgent --> BillingAgent: 账单类问题
TriageAgent --> RefundAgent: 退款类问题
TriageAgent --> FAQAgent: 通用咨询
BillingAgent --> RefundAgent: 涉及退款
RefundAgent --> []: 处理完成
FAQAgent --> HumanAgent: 无法解答
HumanAgent --> [*]: 人工接管
note right of TriageAgent
每个 Agent 作为链上节点，
只负责一类问题或
继续向下传递
end note

责任链模式的优势在于**解耦发送者和接收者**。Triage Agent 不需要知道下游有哪些专家，只需要将请求发出；每个专家节点自主决定是否处理或继续传递。这种设计使得新增一个专业 Agent 时，只需在相关节点的 next_nodes 中注册即可，无需修改 Triage Agent 的配置。同时，责任链也支持条件路由：节点可以根据请求内容的特征（如关键词、用户等级、时间窗口）动态选择下一跳目标。

然而，责任链模式最大的风险是**循环路由**。例如 Billing Agent 遇到退款问题时交接给 Refund Agent，而 Refund Agent 又因缺少账单信息交接回 Billing Agent，形成死循环。解决这一问题需要引入**访问标记（Visitor Marker）**机制：在 `AppContext` 中记录已经访问过的 Agent 列表，当某条路径被重复触发时强制中断并转人工。同时，设置最大交接深度（max_depth）作为兜底保护，防止无限递归消耗上下文窗口。

以下是带防循环检测的责任链 Handoff 实现：

```python
from agents import Agent, handoff, RunContextWrapper
from dataclasses import dataclass, field

@dataclass
class ChainContext:
    visited_agents: list[str] = field(default_factory=list)
    handoff_depth: int = 0
    max_depth: int = 5

class HandoffNode:
    """责任链节点，封装 Agent 与下一跳判断逻辑。"""
    
    def __init__(self, agent: Agent, can_handle, next_nodes=None):
        self.agent = agent
        self.can_handle = can_handle
        self.next_nodes = next_nodes or []
    
    def get_handoffs(self, ctx: RunContextWrapper[ChainContext]):
        if ctx.context.handoff_depth >= ctx.context.max_depth:
            return []
        valid_next = []
        for node in self.next_nodes:
            if node.agent.name not in ctx.context.visited_agents:
                valid_next.append(node.agent)
        return valid_next

async def on_handoff_tracked(ctx: RunContextWrapper[ChainContext], target: str):
    ctx.context.visited_agents.append(target)
    ctx.context.handoff_depth += 1
    print(f"[CHAIN] 深度 \\{ctx.context.handoff_depth}: 交接至 \\{target}")

# 构建责任链
billing_node = HandoffNode(
    Agent(name="BillingAgent", instructions="处理账单"),
    can_handle=lambda q: "发票" in q or "付款" in q,
)
refund_node = HandoffNode(
    Agent(name="RefundAgent", instructions="处理退款"),
    can_handle=lambda q: "退款" in q,
)
billing_node.next_nodes = [refund_node]

除了责任链，对于复杂的客服系统还可以结合**状态机模式（State Machine Pattern）**显式建模整个流转过程。将每个 Agent 对应的状态、转移条件和终止状态以配置文件的形式维护，既能提高可解释性，也便于非技术团队（如运营）调整路由策略，而无需修改代码。状态机配置可以通过 YAML 文件热加载，实现动态路由规则的零停机更新。这种配置驱动的方式特别适合业务规则频繁变化的客服场景，运营团队可以自主调整路由逻辑，开发团队只需保证配置解析引擎的稳定性。

生产环境部署与性能优化

路由规则引擎的实践要点

将本章节的技术应用到生产环境时，首要考虑的是稳定性与可观测性。建议采用渐进式 rollout 策略：先在开发环境验证核心逻辑，再迁移到预发布环境进行压力测试，最后才全量上线。部署过程中应配置完善的日志收集和指标监控，确保任何问题都能被快速发现和定位。

具体来说，需要在基础设施层面做好以下准备：容器资源限制（CPU/内存）、网络策略配置（防火墙规则、服务网格）、持久化存储选型（SSD vs 标准盘）以及备份恢复方案。对于高可用要求严格的场景，建议部署多实例并配置负载均衡，避免单点故障导致服务中断。

交接成功率的关键指标

监控是生产系统的生命线。针对本章节涉及的功能，建议重点跟踪以下指标：请求延迟（P50/P95/P99）、错误率（4xx/5xx/超时）、吞吐量（QPS/TPS）以及资源利用率（CPU/内存/磁盘/网络）。这些指标应接入统一的监控大盘，并设置合理的告警阈值。

除了基础指标，还应关注业务层面的指标。例如功能成功率、用户满意度、成本消耗趋势等。通过将技术指标与业务指标关联分析，可以更准确地评估系统改进的实际价值，避免陷入"为了优化而优化"的陷阱。

动态 Agent 注册的架构考量

随着业务规模增长，单实例部署很快会成为瓶颈。扩展性设计应在项目初期就纳入考量，而非事后补救。水平扩展通常比垂直扩展更具成本效益，但也引入了分布式系统的复杂性（数据一致性、服务发现、负载均衡等）。

在扩展过程中，建议遵循"无状态优先"原则：将状态外置到独立的存储层（如 Redis、PostgreSQL），使计算层可以随时水平扩容。对于无法避免的状态（如会话、缓存），采用分布式一致性协议或最终一致性模型来管理。定期进行容量规划和压力测试，确保系统在流量峰值时仍能稳定运行。

运维团队的协作建议

技术方案的落地离不开高效的团队协作。建议建立清晰的运维手册（Runbook），涵盖常见故障的诊断步骤、应急处理流程和升级路径。同时，通过定期的复盘会议，将线上事故转化为团队的学习素材，持续完善系统的健壮性。

在工具链方面，推荐将本章节的配置和脚本纳入版本控制（Git），并使用 Infrastructure as Code（IaC）工具（如 Terraform、Ansible）管理基础设施变更。这不仅能提高部署效率，还能确保环境一致性，减少"在我机器上能跑"的问题。

Handoff 的上下文序列化与反序列化

Handoff 的核心技术挑战是如何在 Agent A 和 Agent B 之间可靠地传递上下文。上下文不仅包括对话历史，还包括工具执行结果、用户画像、会话状态等复杂数据结构。

序列化策略：

import json
from dataclasses import asdict

def serialize_context(ctx: RunContext) -> str:
    return json.dumps({
        "messages": [asdict(m) for m in ctx.messages],
        "metadata": ctx.metadata,
        "completed_tools": ctx.completed_tools,
    }, ensure_ascii=False, default=str)

def deserialize_context(data: str) -> dict:
    return json.loads(data)

序列化时需要注意以下问题：

1. 循环引用：如果上下文中包含互相引用的对象，JSON 序列化会失败。需要在序列化前打破循环引用，或者使用支持循环引用的序列化库（如 pickle，但 pickle 有安全风险）。
2. 大数据对象：工具返回的二进制文件、大型表格等不应全部序列化。应该只传递摘要信息或对象引用（如文件路径、数据库 ID）。
3. 版本兼容性：当 SDK 升级后，上下文数据结构可能发生变化。反序列化时应使用宽松的策略，缺失字段使用默认值，未知字段忽略而非报错。

反序列化后的状态恢复：

目标 Agent 接收到上下文后，不应直接替换自己的上下文，而是做合并操作。目标 Agent 的 system instructions 和默认工具应该保留，只有用户对话历史和特定状态变量被覆盖。

Handoff 的审计日志对于排查问题和优化路由策略至关重要。每次 Handoff 应记录：源 Agent 名称、目标 Agent 名称、Handoff 原因（由 LLM 生成）、上下文大小、Handoff 耗时。通过分析这些数据，可以识别出频繁 Handoff 的场景，优化路由规则或增强源 Agent 的能力。

在多租户环境中，Handoff 还需要考虑租户隔离。确保 Agent A 不会将租户 X 的上下文 Handoff 给属于租户 Y 的 Agent B。这通常通过在 Agent 名称或配置中嵌入租户 ID 来实现。