多 Agent 协作（Handoff）

Handoff 机制详解

Handoff（移交）是指一个 Agent 将当前对话的控制权转交给另一个 Agent 的机制。被移交的 Agent 会接收到完整的对话历史，就像它从一开始就参与了这个对话。

基础 Handoff 实现

from agents import Agent, Runner, handoff
import asyncio

# ── 专家 Agent（Workers）──────────────────────────────────
tech_agent = Agent(
    name="技术支持专员",
    instructions="""你是技术支持专员，专注于解决产品技术问题。
    遇到非技术问题（账单、退款、账号）时，明确说明需要转交给其他专员。
    """,
    model="gpt-4o-mini"
)

billing_agent = Agent(
    name="账单专员",
    instructions="""你是账单处理专员，专注于发票、付款、退款问题。
    退款超过 1000 元时，需要获取用户同意后才能操作。
    """,
    model="gpt-4o-mini"
)

account_agent = Agent(
    name="账号安全专员",
    instructions="""你是账号安全专员，专注于密码重置、账号验证、安全设置。
    操作前必须通过安全验证（手机号或邮箱验证码）。
    """,
    model="gpt-4o-mini"
)

# ── Orchestrator Agent ────────────────────────────────────
# handoffs 参数声明该 Agent 可以移交给哪些 Agent
triage_agent = Agent(
    name="客服分诊",
    instructions="""你是客服分诊助手，负责接待用户并将问题转交给正确的专员。

转交规则（严格遵守）：
- 产品使用、bug、错误信息 → 技术支持专员
- 发票、扣费、退款、付款 → 账单专员
- 密码、账号、登录、安全 → 账号安全专员
- 问题涉及多个领域时，优先处理安全问题，再处理账单，最后技术

你只负责分诊，不直接解决具体问题。
""",
    model="gpt-4o-mini",
    handoffs=[tech_agent, billing_agent, account_agent]  # 声明可移交对象
)

async def main():
    result = await Runner.run(
        triage_agent,
        "你好，我的账单里有一笔重复扣款，能帮我退款吗？"
    )
    print(f"最终回复（来自 Agent: {result.last_agent.name}）：")
    print(result.final_output)

asyncio.run(main())

自定义 Handoff 配置

使用 handoff() 函数可以精细控制移交行为：

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

@dataclass
class SupportContext:
    user_id: str
    handoff_count: int = 0         # 记录转交次数
    handoff_reasons: list[str] = field(default_factory=list)

# on_handoff 回调：每次发生移交时触发
def on_tech_handoff(ctx: RunContextWrapper[SupportContext]):
    ctx.context.handoff_count += 1
    ctx.context.handoff_reasons.append("技术问题")
    print(f"[监控] 用户 {ctx.context.user_id} 被转接到技术支持")

def on_billing_handoff(ctx: RunContextWrapper[SupportContext]):
    ctx.context.handoff_count += 1
    ctx.context.handoff_reasons.append("账单问题")
    print(f"[监控] 用户 {ctx.context.user_id} 被转接到账单专员")

# 使用 handoff() 函数创建带配置的移交
triage_agent_v2 = Agent(
    name="高级客服分诊",
    instructions="将用户问题转交给正确的专员处理。",
    model="gpt-4o-mini",
    handoffs=[
        handoff(
            tech_agent,
            tool_name_override="transfer_to_tech_support",  # 自定义工具名
            tool_description_override="转接到技术支持，处理产品使用和技术故障问题",
            on_handoff=on_tech_handoff,  # 移交时的回调函数
        ),
        handoff(
            billing_agent,
            tool_name_override="transfer_to_billing",
            tool_description_override="转接到账单专员，处理付款、退款和发票问题",
            on_handoff=on_billing_handoff,
        ),
    ]
)

完整客服系统实战

from agents import Agent, Runner, handoff, function_tool, RunContextWrapper
from dataclasses import dataclass, field
from datetime import datetime
import asyncio

@dataclass
class SupportContext:
    user_id: str
    session_id: str
    start_time: datetime = field(default_factory=datetime.now)
    resolved: bool = False
    satisfaction_score: int | None = None

# ── 账单工具 ──────────────────────────────────────────────
@function_tool
async def lookup_invoice(
    ctx: RunContextWrapper[SupportContext],
    invoice_id: str
) -> str:
    """查询发票详情。"""
    user_id = ctx.context.user_id
    # 实际查询数据库...
    return f"用户 {user_id} 的发票 {invoice_id}：金额 ¥299，状态：已付款，时间：2025-03-15"

@function_tool
async def process_refund(
    ctx: RunContextWrapper[SupportContext],
    invoice_id: str,
    amount: float,
    reason: str
) -> str:
    """处理退款申请。需要先查询发票确认金额。"""
    if amount > 1000:
        return "退款金额超过 1000 元，需要人工审核，预计 1-3 个工作日处理。"
    # 实际退款操作...
    ctx.context.resolved = True
    return f"退款申请已提交：¥{amount}，退款原因：{reason}，预计 3-5 个工作日到账。"

# ── 技术支持工具 ──────────────────────────────────────────
@function_tool
async def search_knowledge_base(query: str) -> str:
    """在技术文档库中搜索解决方案。"""
    # 实际搜索知识库...
    return f"搜索结果（{query}）：找到 3 篇相关文章..."

# ── 专家 Agent 定义 ───────────────────────────────────────
billing = Agent(
    name="账单专员",
    instructions="""你是账单专员。
    处理退款前，必须先用 lookup_invoice 工具查询发票确认金额。
    退款超过 1000 元时，明确告知需要人工审核。
    """,
    model="gpt-4o-mini",
    tools=[lookup_invoice, process_refund]
)

tech_support = Agent(
    name="技术支持",
    instructions="""你是技术支持专员。
    遇到技术问题时，先搜索知识库，找到方案后给用户详细指引。
    """,
    model="gpt-4o-mini",
    tools=[search_knowledge_base]
)

# ── 分诊 Agent（入口）────────────────────────────────────
def log_handoff(target_name: str):
    def callback(ctx: RunContextWrapper[SupportContext]):
        elapsed = (datetime.now() - ctx.context.start_time).seconds
        print(f"[{elapsed}s] 转接到: {target_name}")
    return callback

triage = Agent(
    name="客服分诊",
    instructions="""你是客服总入口，礼貌接待用户并快速判断问题类型。
    账单/退款/发票 → 账单专员
    技术/bug/使用 → 技术支持
    不确定时先询问用户具体情况再转接。
    """,
    model="gpt-4o-mini",
    handoffs=[
        handoff(billing, on_handoff=log_handoff("账单专员")),
        handoff(tech_support, on_handoff=log_handoff("技术支持")),
    ]
)

async def handle_customer(user_id: str, message: str):
    ctx = SupportContext(user_id=user_id, session_id=f"sess_{user_id[:8]}")
    result = await Runner.run(triage, message, context=ctx)
    print(f"\n最终回复（{result.last_agent.name}）：\n{result.final_output}")
    print(f"问题已解决: {ctx.resolved}")

asyncio.run(handle_customer("user_abc123", "我的订单被重复扣款了，发票号是 INV-2025-001"))