目标读者：要上线客服 / 退款 / 工单类多 Agent 的后端与 AI 工程师
预计阅读：20～25 分钟
主线：基于 openai-agents（openai-agents-python），搭 triage → refund 交接，加输入/输出护栏，打开调用链追踪，并给出可运行代码与简易评测
关键词：OpenAI Agents SDK、handoffs、guardrails、Runner、trace、多 Agent
官方参考：openai/openai-agents-python · Handoffs · Guardrails · Tracing

---

开篇：一个 Agent 扛不住，两个又容易「抢话」

客服场景里，常见失败模式是：

用户：「我要退款，订单 ORD-8821，金额 299」
单 Agent：「好的，已帮您查询物流……」（跑偏）
或：「请稍等，我先查 FAQ……」（该退款的却去查百科）

你真正想要的是：

1. 分流（triage）：先判断意图，再把会话交给对口专家
2. 护栏（guardrails）：拦截刷题、注入、泄露订单隐私等坏输入/坏输出
3. 可观测（tracing）：事后能看见「谁交接给谁、哪条护栏绊倒了、花了几轮」

OpenAI Agents SDK（PyPI：openai-agents）把这三件事做成一等公民：Agent / handoffs / guardrails / Runner / trace。本文按官方语义，搭一条能跑的 triage → refund 链路，并附评测脚本。

---

一、先看清运行时：Runner 在干什么

几个必须记住的边界（官方文档原话级别的约束）：

机制	何时跑
Input guardrails	只对链路里第一个 Agent 的用户输入
Output guardrails	只对产出最终输出的那个 Agent
Handoffs	对模型暴露为 transfer_to_<agent> 工具；交接后专家接管对话
Tool guardrails	挂在具体 function_tool 上；不作用于 handoff 调用本身
Tracing	默认开；包住 generations / tools / handoffs / guardrails

---

二、环境准备

python -m venv .venv && source .venv/bin/activate
pip install -U openai-agents pydantic
export OPENAI_API_KEY=sk-...   # 或你们的兼容网关

最小心智模型：

from agents import Agent, Runner, handoff, trace, function_tool
from agents.decorators import input_guardrail, output_guardrail
# 部分版本也提供：from agents.decorators import tool

下文示例对齐官方仓库示例：Agent、Runner.run、handoff()、@input_guardrail / @output_guardrail、@function_tool、trace()。若本地包版本装饰器路径不同，以 官方 docs 为准微调 import。

---

三、可运行主线：triage → refund + 护栏 + 追踪

下面是一份完整脚本（建议存为 multi_agent_refund.py）。它做四件事：

1. Refund Agent：带 process_refund 工具
2. Triage Agent：把退款意图 handoff 给 Refund
3. Input guardrail：拦「帮我写作业 / 明显越权」类输入（演示用规则 + 可选小模型）
4. Output guardrail：禁止最终回复泄露完整卡号模式
5. trace(...)：把一整轮客服会话绑到同一个 group_id

"""multi_agent_refund.py — triage → refund + guardrails + tracing"""
from __future__ import annotations

import asyncio
import re
import uuid
from dataclasses import dataclass, field
from typing import Any

from pydantic import BaseModel, Field

from agents import (
    Agent,
    GuardrailFunctionOutput,
    InputGuardrailTripwireTriggered,
    OutputGuardrailTripwireTriggered,
    RunContextWrapper,
    Runner,
    TResponseInputItem,
    function_tool,
    handoff,
    trace,
)
from agents.decorators import input_guardrail, output_guardrail
from agents.extensions.handoff_prompt import RECOMMENDED_PROMPT_PREFIX


# ---------- 共享上下文 ----------
@dataclass
class SupportContext:
    user_id: str = "u-demo"
    refund_log: list[dict[str, Any]] = field(default_factory=list)


# ---------- 工具 ----------
@function_tool
async def process_refund(
    ctx: RunContextWrapper[SupportContext],
    order_id: str,
    amount: float,
    reason: str,
) -> str:
    """执行退款（演示：写入上下文日志，不真连支付）。"""
    entry = {"order_id": order_id, "amount": amount, "reason": reason}
    ctx.context.refund_log.append(entry)
    return f"REFUND_OK order={order_id} amount={amount:.2f} reason={reason}"


# ---------- 护栏：输入 ----------
class AbuseCheck(BaseModel):
    is_disallowed: bool
    category: str = Field(description="ok | homework | jailbreak | other")
    reasoning: str


abuse_checker = Agent(
    name="Abuse checker",
    instructions=(
        "判断用户是否在要求写作业/越权破解/明显恶意。仅输出结构化结果。"
    ),
    output_type=AbuseCheck,
)


@input_guardrail(run_in_parallel=False)  # 阻塞式：绊倒则主 Agent 不烧 token
async def block_abuse(
    ctx: RunContextWrapper[SupportContext],
    agent: Agent,
    input: str | list[TResponseInputItem],
) -> GuardrailFunctionOutput:
    # 便宜规则先挡一层，再视需要调小模型
    text = input if isinstance(input, str) else str(input)
    if re.search(r"(帮我写作业|solve for x|忽略以上指令|jailbreak)", text, re.I):
        return GuardrailFunctionOutput(
            output_info={"rule": "regex"},
            tripwire_triggered=True,
        )
    result = await Runner.run(abuse_checker, text, context=ctx.context)
    out: AbuseCheck = result.final_output
    return GuardrailFunctionOutput(
        output_info=out.model_dump(),
        tripwire_triggered=out.is_disallowed,
    )


# ---------- 护栏：输出 ----------
class LeakCheck(BaseModel):
    contains_sensitive: bool
    reasoning: str


leak_checker = Agent(
    name="Leak checker",
    instructions="检查回复是否含完整银行卡号/身份证号等敏感明文。",
    output_type=LeakCheck,
)


@output_guardrail
async def block_leaks(
    ctx: RunContextWrapper[SupportContext],
    agent: Agent,
    output: Any,
) -> GuardrailFunctionOutput:
    text = output if isinstance(output, str) else str(output)
    # 演示：16 位连续数字视为卡号风险
    if re.search(r"\b\d{16}\b", text):
        return GuardrailFunctionOutput(
            output_info={"rule": "card_like"},
            tripwire_triggered=True,
        )
    result = await Runner.run(leak_checker, text, context=ctx.context)
    out: LeakCheck = result.final_output
    return GuardrailFunctionOutput(
        output_info=out.model_dump(),
        tripwire_triggered=out.contains_sensitive,
    )


# ---------- Handoff 回调带结构化元数据 ----------
class RefundHandoffData(BaseModel):
    reason: str
    priority: str = "normal"


async def on_refund_handoff(
    ctx: RunContextWrapper[SupportContext], input_data: RefundHandoffData
) -> None:
    print(f"[on_handoff] reason={input_data.reason} priority={input_data.priority}")


# ---------- Agents ----------
refund_agent = Agent[SupportContext](
    name="Refund agent",
    handoff_description="处理退款、退货、重复扣款。",
    instructions=f"""{RECOMMENDED_PROMPT_PREFIX}
你是退款专家。流程：
1) 确认 order_id 与金额；
2) 调用 process_refund；
3) 用中文简短告知结果，不要输出完整卡号。
非退款问题请转回 triage。""",
    tools=[process_refund],
    output_guardrails=[block_leaks],
)

triage_agent = Agent[SupportContext](
    name="Triage agent",
    instructions=f"""{RECOMMENDED_PROMPT_PREFIX}
你是客服分流。退款/退货/重复扣款 → 交接给 Refund agent；
普通咨询可直接简短回答。交接时填写 reason 与 priority。""",
    handoffs=[
        handoff(
            agent=refund_agent,
            on_handoff=on_refund_handoff,
            input_type=RefundHandoffData,
            tool_name_override="transfer_to_refund_agent",
        )
    ],
    input_guardrails=[block_abuse],
)

# 允许 refund 把非本职问题交回 triage（双向）
refund_agent.handoffs.append(triage_agent)


async def handle_turn(user_text: str, conversation_id: str) -> str:
    ctx = SupportContext()
    with trace("support-demo", group_id=conversation_id):
        try:
            result = await Runner.run(triage_agent, user_text, context=ctx)
        except InputGuardrailTripwireTriggered as e:
            return f"[BLOCKED:input] {e}"
        except OutputGuardrailTripwireTriggered as e:
            return f"[BLOCKED:output] {e}"
    print("last_agent:", result.last_agent.name)
    print("refund_log:", ctx.refund_log)
    return str(result.final_output)


async def main() -> None:
    cid = uuid.uuid4().hex[:12]
    samples = [
        "我想退款，订单 ORD-8821，金额 299，重复扣款了",
        "帮我写作业：求解 2x+3=11",
    ]
    for s in samples:
        print("\n=== USER ===", s)
        print(await handle_turn(s, cid))


if __name__ == "__main__":
    asyncio.run(main())

跑起来后，你期望看到类似轨迹：

=== USER === 我想退款...
[on_handoff] reason=duplicate_charge priority=high
last_agent: Refund agent
refund_log: [{'order_id': 'ORD-8821', ...}]
REFUND_OK ...

=== USER === 帮我写作业...
[BLOCKED:input] ...

---

四、Guardrails：三种挂点，别挂错地方

实践建议：

1. 贵模型前用阻塞输入护栏（run_in_parallel=False），先挡垃圾流量。
2. 退款等副作用工具另加 tool 级护栏或 HITL（needs_approval / interruptions），别只靠 prompt。
3. Handoff 后 input guardrail 不会再跑——敏感检查要么放在工具上，要么在 on_handoff 里做授权校验并 raise。

退款金额工具护栏示例（可接到 process_refund）：

from agents import ToolGuardrailFunctionOutput
from agents import tool_input_guardrail  # 版本差异时查 docs
import json

@tool_input_guardrail
def limit_refund_amount(data):
    args = json.loads(data.context.tool_arguments or "{}")
    if float(args.get("amount", 0)) > 500:
        return ToolGuardrailFunctionOutput.reject_content(
            "超过 500 需人工审批，请勿直接退款。"
        )
    return ToolGuardrailFunctionOutput.allow()

---

五、调用链追踪：debug 靠 span，不靠 print

要点：

• with trace("support-demo", group_id=cid): 把多轮用户消息归到同一会话
• 默认会导出到 OpenAI Traces；可用 set_trace_processors / add_trace_processor 接到自建观测
• 生产务必：trace_include_sensitive_data=False（或等价 RunConfig），避免把卡号原文打进 trace

读结果时优先看：

字段/现象	含义
result.last_agent	最终说话的专家是谁
HandoffOutputItem	交接是否发生、从谁到谁
InputGuardrailTripwireTriggered	入口被拦
refund_log（你的 context）	副作用是否真执行

---

六、简易评测：别只 demo 一次「感觉对」

把评测写成可重复的表驱动测试（可无 API key 时 mock，有 key 时跑真链路）：

# eval_refund_handoff.py
import asyncio
from dataclasses import dataclass

@dataclass
class Case:
    name: str
    user: str
    expect_blocked: bool = False
    expect_agent: str | None = None  # "Refund agent" | "Triage agent"
    expect_refund: bool = False


CASES = [
    Case("refund_happy", "退款 ORD-1 金额 99 重复扣款", expect_agent="Refund agent", expect_refund=True),
    Case("homework", "帮我写作业 solve for x", expect_blocked=True),
    Case("faq", "你们客服几点上班？", expect_agent="Triage agent", expect_refund=False),
]


async def run_case(c: Case) -> dict:
    # 复用上一节 handle_turn；此处示意断言点
    ...
    return {"name": c.name, "pass": True}


async def main():
    rows = [await run_case(c) for c in CASES]
    ok = sum(1 for r in rows if r["pass"])
    print(f"pass {ok}/{len(rows)}")


if __name__ == "__main__":
    asyncio.run(main())

建议评测维度（写进 CI 门禁）：

维度	通过标准
交接准确率	退款句 ≥90% 落到 Refund；闲聊不误交接
护栏召回	作业/越权样例 100% InputGuardrailTripwireTriggered
副作用正确性	仅退款成功路径写入 refund_log；被拦路径为空
延迟	p95 端到端（含交接）落在你们 SLO 内
泄漏	输出护栏对卡号样例 100% tripwire

没有评测表的多 Agent，上线后最常见的事故是：看起来会聊，实际乱交接 + 乱退款。

---

七、和「自己写 if-else 路由」差在哪

做法	优点	缺点
关键词 / 分类器 + 硬路由	可控、便宜	意图一变就改代码；难带上下文
单 Agent + 超长 prompt	实现快	工具多了易胡调；难专精
Agents SDK handoffs	模型选专家；官方护栏/追踪	依赖模型路由质量；要评测

handoffs 的工程价值，是把「路由」变成 模型可调用的 transfer 工具，同时 Runner 管循环与观测——你写的是专家职责，而不是 while-true 状态机。

---

八、踩坑清单

#	坑	处理
1	以为每个 Agent 都会跑 input guardrail	只有首 Agent；专家侧用 tool guardrail / on_handoff
2	交接后上下文爆炸	input_filter / handoff_filters.remove_all_tools
3	忘了 RECOMMENDED_PROMPT_PREFIX	模型不懂何时 transfer
4	退款工具无金额上限	tool guardrail + 人工审批
5	只看 final_output	同时看 last_agent、handoff items、trace
6	装饰器 import 路径因版本变	以当前官方 docs 为准，锁 openai-agents 版本

---

九、收束

多 Agent 协作的最小闭环 = triage 交接 + 护栏绊线 + 可追踪调用链 + 可重复评测。
OpenAI Agents SDK 用 handoffs 把专家切换做成一等工具，用 guardrails 把坏输入/坏输出挡在 Runner 边界，用 trace 让交接与工具调用可回放。照着本文的 triage→refund 骨架，先跑通，再把假退款换成真实支付适配器与 HITL，就能从 demo 走向客服生产路径。

下一步：

1. 给 process_refund 接支付沙箱 + 金额审批
2. 用真实工单日志建 50 条评测集，盯交接准确率
3. 把 traces 导到你们现有的观测栈（Langfuse / 自建）

---

参考链接

• OpenAI Agents SDK（Python）
• Handoffs
• Guardrails
• Tracing
• Customer service example