Agent 工程体系 · 第 42/98 篇。内容以 2026 年 9 月可验证的公开规范和稳定接口为基线;框架版本敏感能力会明确标注,不把实验行为写成通用保证。

OpenAI Agents SDK:Agent、Runner、Tool、Handoff、Guardrail 和 Trace

OpenAI Agents SDK 是一个以 Python 为中心的 Agent 运行时。它没有试图把所有工作流抽象成复杂的图模型,而是围绕少量核心原语组织一次 Agent 运行:Agent 描述“谁在解决问题”,Runner 驱动运行循环,Tool 让模型采取行动,Handoff 让一个 Agent 把当前回合交给另一个 Agent,Guardrail 对输入、输出或工具调用进行约束,Trace 则记录整个过程。SDK 默认使用 Responses API,但在其上增加了工具调度、交接、会话、守卫和追踪等运行时能力。(openai.github.io)

本文以 2026 年 9 月 Agent 工程基线为范围,重点解释这些对象如何共同实现一个完整的 Agent 运行循环:

输入观察决策行动反馈终止或继续\text{输入} \rightarrow \text{观察} \rightarrow \text{决策} \rightarrow \text{行动} \rightarrow \text{反馈} \rightarrow \text{终止或继续}

其中:

  • Agent 提供决策所需的身份、指令和能力边界;
  • Runner 执行状态转换;
  • Tool 把决策转化为外部行动;
  • Handoff 改变当前负责决策的 Agent;
  • Guardrail 限制状态转换和副作用;
  • Trace 记录每次状态变化,便于调试、评估和生产诊断。

一、先建立整体模型:Agent 不是一次模型调用

一个普通的模型调用通常可以表示为:

y=M(x,p)y = M(x, p)

其中:

  • xx 是用户输入;
  • pp 是提示词或系统指令;
  • MM 是模型;
  • yy 是模型输出。

但 Agent 运行通常不是一次调用,而是一个循环:

st+1=F(st,M(st),E)s_{t+1} = F(s_t, M(s_t), E)

其中:

  • sts_t 是第 tt 步的运行状态;
  • M(st)M(s_t) 是模型基于当前状态做出的决策;
  • EE 是外部环境,例如数据库、HTTP 服务、文件系统或另一个 Agent;
  • FF 是运行时对模型决策的解释和执行;
  • st+1s_{t+1} 是执行行动后得到的新状态。

模型可能产生三类结果:

  1. 最终答案:运行终止;
  2. 工具调用:执行工具,把工具结果反馈给模型;
  3. Handoff 调用:切换到另一个 Agent,继续当前任务。

因此,Agent SDK 的核心不是“创建一个有个性的聊天机器人”,而是维护下面这类状态:

当前 Agent
当前输入与历史
可用工具
可用 Handoff
Guardrail 结果
工具调用结果
最终输出
运行次数与资源使用
Trace 层级

Runner 正是负责推进这组状态的运行时。SDK 提供异步 Runner.run()、同步 Runner.run_sync() 和流式 Runner.run_streamed() 三种入口;后两者分别是同步封装和事件流封装。(openai.github.io)


二、Agent:对“决策者”的声明

2.1 Agent 的定义

在 Agents SDK 中,Agent 可以理解为:

A=(N,I,T,H,Gin,Gout,O,C)A = (N, I, T, H, G_{in}, G_{out}, O, C)

其中:

  • NN:名称 name
  • II:指令 instructions 或动态提示 prompt
  • TT:工具集合 tools
  • HH:可用的交接目标 handoffs
  • GinG_{in}:输入 Guardrail;
  • GoutG_{out}:输出 Guardrail;
  • OO:输出类型 output_type
  • CC:模型、模型参数和运行上下文。

官方将 Agent 描述为“带有指令和工具的 LLM”,并允许进一步配置 Guardrail、Handoff、模型和结构化输出。(openai.github.io)

一个最小 Agent 如下:

from agents import Agent

assistant = Agent(
    name="Assistant",
    instructions=(
        "你是一个技术支持助手。"
        "回答要基于已知事实;无法确认时明确说明不确定性。"
    ),
)

这里的 Agent 还没有执行任何工作。它只是声明:

  • 自己叫什么;
  • 应该遵循什么指令;
  • 当前没有额外工具;
  • 当前没有交接目标;
  • 默认使用 SDK 配置的模型。

因此,下面这段代码不会调用模型:

assistant = Agent(
    name="Assistant",
    instructions="回答问题",
)

它只创建了一个运行配置对象。真正推进状态的是 Runner

2.2 instructions 不是流程图

一个常见误解是,把复杂流程全部写进 instructions,然后期待模型严格执行:

Agent(
    name="OrderAssistant",
    instructions="""
    先查询订单,再判断是否已发货,
    如果未发货则取消订单,否则拒绝取消。
    """,
)

这段指令可以表达意图,但不能替代运行时约束。原因是“查询订单”和“取消订单”具有不同的副作用等级:

  • 查询订单通常是只读操作;
  • 取消订单可能改变业务状态;
  • 取消前还可能需要鉴权、二次确认或幂等检查。

更可靠的结构是把动作暴露为工具,并把不可违反的约束放在工具实现或 Guardrail 中:

from agents import Agent
from agents.decorators import tool

@tool
def get_order_status(order_id: str) -> str:
    """查询订单状态。"""
    # 实际项目中应访问订单服务
    return "未发货"

@tool
def cancel_order(order_id: str) -> str:
    """取消订单。仅用于已经确认且满足取消条件的订单。"""
    # 实际项目中应执行鉴权、幂等校验和状态更新
    return "订单已取消"

order_agent = Agent(
    name="Order Assistant",
    instructions=(
        "处理订单问题。取消订单前必须查询订单状态,"
        "只有未发货订单才允许取消。"
    ),
    tools=[get_order_status, cancel_order],
)

这里仍然不能把模型的判断当成安全边界。真正的 cancel_order() 必须再次校验订单状态,因为模型可能:

  • 直接选择取消工具;
  • 错误理解工具返回值;
  • 在上下文被截断后忘记前置条件;
  • 因提示注入而偏离原流程。

所以,Agent 的指令负责引导决策,工具实现负责执行约束

2.3 Agent 的输出类型

如果 Agent 只输出自然语言,可以直接使用:

text_agent = Agent(
    name="Text Agent",
    instructions="用简洁中文回答。",
)

如果下游程序需要稳定解析的结果,则应声明结构化输出类型。下面是概念示例:

from pydantic import BaseModel
from agents import Agent

class OrderDecision(BaseModel):
    order_id: str
    action: str
    reason: str

decision_agent = Agent(
    name="Order Decision Agent",
    instructions=(
        "根据订单信息给出决定。"
        "action 只能是 keep、cancel 或 escalate。"
    ),
    output_type=OrderDecision,
)

结构化输出解决的是“最终答案如何被程序读取”,不等于解决“模型是否有权执行动作”。即使 action 已经被严格解析,真正的写操作仍应经过权限、业务规则和审批控制。


三、Runner:Agent 运行循环的执行者

3.1 Runner 做什么

Runner 不是模型,也不是 Agent 的子类。它是运行时控制器,负责:

  1. 准备输入和运行上下文;
  2. 执行输入 Guardrail;
  3. 调用当前 Agent;
  4. 将模型产生的工具调用分派给工具;
  5. 将工具结果追加回运行状态;
  6. 处理 Handoff;
  7. 执行输出 Guardrail;
  8. 返回 RunResult 或流式结果。

可以将一次非流式运行抽象成如下伪代码:

state = initialize(agent, input)

check_input_guardrails(state)

while True:
    response = call_model(
        agent=state.current_agent,
        input=state.input_history,
        tools=state.available_tools,
        handoffs=state.available_handoffs,
    )

    if response.requests_tool:
        tool_result = execute_tool(response.tool_call)
        state.append(tool_result)
        continue

    if response.requests_handoff:
        state.current_agent = response.target_agent
        state.append_handoff(response)
        continue

    output = parse_final_output(response)
    check_output_guardrails(output)
    return RunResult(state=state, final_output=output)

实际 SDK 还处理流式事件、会话、取消、错误、追踪和模型提供商等问题,但这个循环足以解释核心因果关系。

3.2 三种运行入口

异步运行

import asyncio
from agents import Agent, Runner

agent = Agent(
    name="Assistant",
    instructions="简洁回答问题。",
)

async def main():
    result = await Runner.run(agent, "什么是幂等性?")
    print(result.final_output)

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

Runner.run() 返回 RunResult。结果不仅包含最终输出,还可以访问最后一个 Agent、输入、Guardrail 结果以及运行过程中产生的项目。由于运行期间可能发生 Handoff,最终输出的具体类型通常由最后执行的 Agent 决定。(openai.github.io)

同步运行

from agents import Agent, Runner

agent = Agent(
    name="Assistant",
    instructions="回答问题。",
)

result = Runner.run_sync(agent, "解释 TCP 三次握手")
print(result.final_output)

run_sync() 适合脚本、命令行工具和已经处于同步调用链的程序。它本质上是对异步运行的同步封装,不应在已经运行事件循环的环境中随意调用,例如某些异步 Web 框架处理函数中。

流式运行

import asyncio
from agents import Agent, Runner

agent = Agent(
    name="Assistant",
    instructions="回答问题,并在必要时使用工具。",
)

async def main():
    result = Runner.run_streamed(agent, "解释事件驱动架构")

    async for event in result.stream_events():
        print(event)

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

流式运行返回 RunResultStreaming。它不只传递文本增量,还可以传递模型响应、工具调用、Agent 作为工具时的嵌套事件等。流式输出不能被误认为“运行已经成功完成”:在最终事件到达之前,工具仍可能失败,Guardrail 仍可能触发,运行也可能被取消。(openai.github.io)

3.3 max_turns 是终止条件,不是质量控制

Agent 运行必须存在终止条件。最直接的条件是:

  • 模型返回最终输出;
  • 触发 Guardrail;
  • 抛出工具或模型错误;
  • 达到 max_turns
  • 被取消。

max_turns 限制的是运行循环可以推进的回合数。例如:

result = await Runner.run(
    agent,
    "完成一个需要多次工具调用的任务",
    max_turns=8,
)

它能防止以下错误形成无限循环:

模型调用工具 A
→ 工具 A 返回无法解决
→ 模型再次调用工具 A
→ 工具 A 再次返回无法解决
→ ...

max_turns 不能证明任务正确完成。若一个任务需要 5 回合,而设置为 3,结果可能只是被截断;若设置为 100,错误循环可能消耗大量令牌和外部资源。

因此,终止条件至少应区分:

终止=正常完成安全阻断资源耗尽异常失败\text{终止} = \text{正常完成} \lor \text{安全阻断} \lor \text{资源耗尽} \lor \text{异常失败}

生产代码不应只打印 final_output,还应记录最后 Agent、运行状态和异常类型。


四、Tool:让模型从“生成文字”进入“改变环境”

4.1 Tool 的本质

Tool 是模型可选择调用的动作接口。可以抽象为:

Ti:XiYiT_i: X_i \rightarrow Y_i

其中:

  • XiX_i 是工具参数的输入域;
  • YiY_i 是工具结果;
  • 工具内部可能访问外部环境 EE

例如:

get_weather(city: str) -> Weather
create_ticket(title: str, body: str) -> Ticket
cancel_order(order_id: str) -> CancelResult

对模型而言,工具通常以名称、描述和 JSON Schema 的形式出现。模型不是直接执行 Python 函数,而是先生成一个符合工具协议的调用请求;Runner 再解析参数、执行函数,并把结果反馈给模型。

4.2 使用 Python 函数创建工具

SDK 可以把 Python 函数包装成函数工具,并根据函数签名和文档字符串生成参数描述。官方快速入门使用了 agents.decorators.tool 装饰器。(openai.github.io)

from agents import Agent, Runner
from agents.decorators import tool

@tool
def lookup_user(user_id: str) -> str:
    """根据用户 ID 查询用户的基础信息。"""
    if user_id == "u-100":
        return "用户 u-100:杭州,普通会员"
    return "未找到用户"

agent = Agent(
    name="User Assistant",
    instructions=(
        "回答用户信息问题。"
        "如果问题涉及具体用户,先调用 lookup_user。"
    ),
    tools=[lookup_user],
)

执行时可能发生如下状态变化:

S0:
  用户输入:“u-100 是什么会员?”

S1:
  模型决定调用 lookup_user
  参数:{"user_id": "u-100"}

S2:
  Runner 执行 Python 函数
  结果:“用户 u-100:杭州,普通会员”

S3:
  Runner 把工具结果交给模型

S4:
  模型生成最终答案
  “u-100 是普通会员。”

工具结果必须是模型能够理解的反馈,而不是只对程序员有意义的调试对象。对于复杂结果,建议返回结构稳定、字段有限的 JSON 或 Pydantic 模型,而不是把整个数据库记录直接拼成字符串。

4.3 工具参数校验不等于业务校验

工具 Schema 主要解决格式问题:

{
  "order_id": "o-1001"
}

它可以阻止缺少 order_id 或类型错误,但不能判断:

  • 该订单是否属于当前用户;
  • 当前用户是否有权限取消;
  • 订单是否已发货;
  • 重复调用是否幂等;
  • 取消操作是否需要人工确认。

因此,工具应分成两层:

协议校验:
  参数存在、类型正确、枚举值有效

业务校验:
  身份、权限、状态、额度、幂等、审批

例如:

@tool
def cancel_order(order_id: str) -> str:
    """取消订单。"""
    order = order_service.get(order_id)

    if order is None:
        raise ValueError("订单不存在")

    if order.status != "PENDING":
        raise ValueError("只有待处理订单允许取消")

    if not authorization.can_cancel(order):
        raise PermissionError("当前用户无权取消该订单")

    order_service.cancel(order_id)
    return "cancelled"

如果工具失败,运行时可以把失败转换为模型可见的错误消息,也可以让异常继续向调用方抛出;具体行为受工具错误处理配置影响。SDK API 参考中提供了工具失败格式化函数和工具超时相关机制。(openai.github.io)

工程上要明确选择:

  • 模型可恢复的错误:例如参数不完整、资源暂时不可用,可以返回结构化错误,让模型改正或重试;
  • 调用方必须处理的错误:例如权限拒绝、数据库不可用、内部程序错误,应抛出异常并由服务层处理;
  • 不可重试的副作用错误:必须避免让模型盲目重试,否则可能重复扣款、重复发货或重复创建工单。

4.4 Hosted Tool、Runtime Tool 和 Agent as Tool

SDK 的 Tool 不只有 Python 函数。官方文档将工具分为 Hosted OpenAI tools、本地运行时工具、函数工具、Agents as tools,以及实验性的 Codex tool。(openai.github.io)

其中最容易混淆的是:

Agent
├── 直接拥有 function tool
├── 通过 handoff 转交控制权
└── 通过 Agent.as_tool() 调用另一个 Agent

Agent.as_tool() 的语义是:把另一个 Agent 暴露成一个工具。调用方 Agent 仍然拥有最终对话控制权。(openai.github.io)

例如:

from agents import Agent, Runner

researcher = Agent(
    name="Researcher",
    instructions="只负责收集事实,并返回带条理的研究摘要。",
)

writer = Agent(
    name="Writer",
    instructions=(
        "你负责给用户最终答复。"
        "必要时调用 researcher 获取研究摘要,"
        "但最终答案必须由你生成。"
    ),
    tools=[
        researcher.as_tool(
            tool_name="research_topic",
            tool_description="研究一个主题并返回事实摘要。",
        )
    ],
)

这里的调用关系是:

用户
  ↓
Writer
  ├── 直接回答
  └── 调用 research_topic
        ↓
      Researcher
        ↓
      返回摘要
  ↓
Writer 汇总并生成最终答案

如果 Researcher 直接接管用户对话,应该使用 Handoff,而不是 as_tool()


五、Handoff:把当前回合的控制权交给另一个 Agent

5.1 Handoff 的定义

Handoff 是 Agent 之间的控制权转移:

(Ai,Hij,st)(Aj,st+1)(A_i, H_{ij}, s_t) \rightarrow (A_j, s_{t+1})

其中:

  • AiA_i 是当前 Agent;
  • HijH_{ij} 是从 AiA_iAjA_j 的交接;
  • AjA_j 成为后续处理当前回合的 Agent。

SDK 把 Handoff 暴露成模型可调用的工具。例如目标 Agent 名为 Refund Agent 时,默认交接工具名类似于 transfer_to_refund_agent。模型先选择这个工具,Runner 再切换当前 Agent。(openai.github.io)

from agents import Agent, Runner

refund_agent = Agent(
    name="Refund Agent",
    handoff_description="处理退款、退货和退款进度问题。",
    instructions=(
        "你是退款专员。"
        "回答退款问题,必要时要求用户提供订单号。"
    ),
)

faq_agent = Agent(
    name="FAQ Agent",
    handoff_description="处理产品使用和常见问题。",
    instructions="回答产品常见问题。",
)

triage_agent = Agent(
    name="Triage Agent",
    instructions=(
        "判断用户问题属于退款还是产品常见问题,"
        "然后交给对应的专员。"
    ),
    handoffs=[refund_agent, faq_agent],
)

async def main():
    result = await Runner.run(
        triage_agent,
        "我的订单怎么申请退款?",
    )
    print(result.final_output)
    print(result.last_agent.name)

典型结果是:

Refund Agent

这里的 triage_agent 并不负责最终回答。它负责分类和路由;refund_agent 接管后续处理。

5.2 handoff_description 的作用

handoff_description 不是给用户看的欢迎语,而是帮助路由 Agent 判断什么时候应该交接:

refund_agent = Agent(
    name="Refund Agent",
    handoff_description="仅处理退款、退货和退款到账时间问题。",
    instructions="……",
)

描述越模糊,模型越可能在相邻领域之间错误路由。例如:

“处理订单相关问题”

可能覆盖查物流、改地址、取消订单、退款和发票。

更好的描述包含:

  • 处理范围;
  • 不处理的相邻范围;
  • 触发条件;
  • 需要的输入。
“处理已付款订单的退款和退货问题。
不处理物流查询、地址修改和发票问题。
当用户明确要求退款或退货时使用。”

5.3 handoff() 的定制

当需要自定义工具名称、描述、交接前回调或输入过滤时,可以显式使用 handoff()

from agents import Agent, handoff, RunContextWrapper

billing_agent = Agent(
    name="Billing Agent",
    instructions="处理账单和发票问题。",
)

def record_route(ctx: RunContextWrapper[None]):
    print("即将进入账单 Agent")

triage_agent = Agent(
    name="Triage Agent",
    instructions="将账单问题交给账单 Agent。",
    handoffs=[
        handoff(
            billing_agent,
            tool_name_override="route_to_billing",
            tool_description_override="将账单、发票和付款问题交给账单专员。",
            on_handoff=record_route,
        )
    ],
)

on_handoff 适合做记录、预取数据或初始化上下文,但不应被误解为动态路由器。handoff() 捕获的是传入的具体 Agent;如果存在多个候选目标,应分别注册多个 Handoff,让模型在多个工具之间选择。官方文档明确区分了这种静态目标交接与自定义 Handoff。(openai.github.io)

5.4 Handoff 输入与主输入不是一回事

可以为 Handoff 工具声明结构化参数:

from pydantic import BaseModel
from agents import Agent, handoff, RunContextWrapper

class EscalationData(BaseModel):
    reason: str

support_agent = Agent(
    name="Support Agent",
    instructions="处理升级后的客户支持问题。",
)

async def on_escalation(
    ctx: RunContextWrapper[None],
    data: EscalationData,
):
    print(f"升级原因:{data.reason}")

route = handoff(
    support_agent,
    input_type=EscalationData,
    on_handoff=on_escalation,
)

这里的 input_type 约束的是 Handoff 工具调用的参数,例如:

{
  "reason": "用户要求人工处理退款"
}

它不会自动改变下一个 Agent 的主输入。下一个 Agent 看到什么历史和输入,还受到 Handoff 输入过滤、运行配置和会话状态影响。这个区别非常重要:交接参数是路由元数据,不等于业务对话正文。(openai.github.io)

5.5 Handoff 与 Agent as Tool 的选择

需求 适合的机制
专员接管当前用户对话 Handoff
管理者调用专家获取一段结果 Agent.as_tool()
一个专家结果需要和多个专家结果汇总 Agent.as_tool()
路由本身就是业务流程的一部分 Handoff
希望始终由一个 Agent 生成最终答案 Agent.as_tool()

错误选择会导致对话所有权混乱。

例如,使用 Handoff 构建一个研究流程:

Triage → Researcher

之后 Researcher 可能直接面向用户输出,而不是返回研究结果给 Triage。

而使用 Agent.as_tool()

Manager → Researcher
       → Writer

Manager 保持最终输出权,但每次调用专家都会产生额外模型调用和上下文管理成本。


六、Guardrail:限制运行时的输入、输出和副作用

6.1 Guardrail 不是提示词

Guardrail 是一个可执行检查:

G(x){allowtripwireG(x) \rightarrow \begin{cases} \text{allow} \\ \text{tripwire} \end{cases}

与指令不同,Guardrail 可以直接改变运行结果:

  • 允许运行;
  • 阻止 Agent 启动;
  • 阻止最终输出返回;
  • 阻止工具执行;
  • 把错误转换为明确的业务响应。

SDK 支持输入 Guardrail、输出 Guardrail 和工具 Guardrail。输入 Guardrail 检查初始用户输入,输出 Guardrail 检查最终 Agent 输出,工具 Guardrail 则包围自定义函数工具调用。(openai.github.io)

6.2 输入 Guardrail

输入 Guardrail 适合检查:

  • 是否属于允许的业务范围;
  • 是否包含明显的越权请求;
  • 是否包含不应处理的敏感数据;
  • 是否需要转人工;
  • 是否满足最低输入条件。

SDK 的输入 Guardrail 可以同步或异步执行,并支持并行模式和阻塞模式。默认并行模式有更低延迟,但 Agent 可能已经开始消耗令牌甚至执行工具;阻塞模式则会等待 Guardrail 完成后再启动 Agent。(openai.github.io)

概念代码如下:

from agents import Agent, Runner, input_guardrail, GuardrailFunctionOutput

@input_guardrail(run_in_parallel=False)
async def reject_empty_input(ctx, agent, input):
    text = str(input).strip()

    return GuardrailFunctionOutput(
        output_info={"length": len(text)},
        tripwire_triggered=(len(text) == 0),
    )

agent = Agent(
    name="Support Agent",
    instructions="处理客户支持问题。",
    input_guardrails=[reject_empty_input],
)

run_in_parallel=False 的因果关系是:

用户输入
  ↓
输入 Guardrail
  ├── 触发 tripwire → Agent 不启动
  └── 通过 → Agent 开始模型调用

这对于带副作用工具的 Agent 很重要。例如一个 Agent 一启动就可能调用查询服务或创建任务,那么阻塞式 Guardrail 可以在模型和工具启动之前拒绝非法请求。

6.3 输出 Guardrail

输出 Guardrail 检查 Agent 的最终结果,而不是每一个中间模型片段:

模型输出
  ↓
结构化解析
  ↓
输出 Guardrail
  ├── 通过 → 返回给调用方
  └── 触发 tripwire → 抛出异常

它适合检查:

  • 结构化字段是否满足业务约束;
  • 是否泄露内部信息;
  • 是否给出不允许的承诺;
  • 是否缺失必需的免责声明;
  • 是否包含不应发送给用户的原始工具结果。

输出 Guardrail 不能替代工具级权限控制,因为它发生在动作之后。比如模型已经调用了退款工具,输出 Guardrail 才发现最终文本不合规,此时退款副作用可能已经发生。

6.4 工具 Guardrail

如果一个流程包含多个 Agent、Handoff 或 Agent as Tool,只在第一个 Agent 上配置输入 Guardrail,并不能自动检查每一次工具调用。官方文档指出:Agent 级输入 Guardrail 只运行在链路中的第一个 Agent,输出 Guardrail 只运行在产生最终输出的 Agent;工具 Guardrail 才会围绕每次自定义函数工具调用执行。(openai.github.io)

因此,对于写操作,更适合使用工具 Guardrail:

工具调用输入
  ↓
工具输入 Guardrail
  ↓
业务工具
  ↓
工具输出 Guardrail
  ↓
模型继续决策

可将三类 Guardrail 的责任划分为:

类型 主要问题 能否阻止模型启动 能否保护每次工具调用
输入 Guardrail 用户是否允许进入流程 可以,阻塞模式下可以 不能
输出 Guardrail 最终答案是否可返回 不能撤销已发生副作用 不能
工具 Guardrail 某次工具调用是否允许 不能阻止前面的模型调用 可以

6.5 Tripwire 与异常处理

Guardrail 返回 tripwire_triggered=True 时,SDK 会抛出对应异常,例如输入或输出 Guardrail 的 Tripwire 异常。(openai.github.io)

服务层应明确区分:

try:
    result = await Runner.run(agent, user_input)
except Exception as exc:
    # 实际项目中应分别捕获输入、输出、工具和模型异常
    return {
        "status": "blocked_or_failed",
        "message": "请求未完成",
    }

不要把所有异常都转换成“模型再试一次”。Guardrail 触发通常意味着需要:

  • 拒绝请求;
  • 引导用户修改输入;
  • 转人工;
  • 记录安全事件。

盲目重试会使同一请求反复触发安全策略,甚至让工具副作用被重复执行。


七、Trace:观察 Agent 运行,而不是只看最终答案

7.1 Trace 与日志的区别

最终答案只能告诉你:

用户看到了什么

Trace 需要回答:

为什么得到这个答案?
调用了哪个 Agent?
模型做了哪些回合?
调用了哪些工具?
工具花了多久?
是否发生了 Handoff?
哪个 Guardrail 被触发?
失败发生在模型、工具还是运行时?

Agents SDK 内置追踪,会记录模型生成、工具调用、Handoff、Guardrail 以及自定义事件。默认情况下,Trace 已启用,并可在 OpenAI Dashboard 的 Trace viewer 中查看。(openai.github.io)

7.2 Trace 和 Span 的关系

Trace 表示一次端到端工作流,Span 表示其中一个有开始时间和结束时间的操作。

可以抽象成树:

Trace: customer_support
└── task_span
    ├── turn_span
    │   ├── agent_span: Triage Agent
    │   ├── generation_span
    │   └── handoff_span
    ├── turn_span
    │   ├── agent_span: Refund Agent
    │   ├── generation_span
    │   ├── function_span: get_order_status
    │   └── guardrail_span
    └── output_guardrail_span

Trace 通常包含:

  • workflow_name
  • trace_id
  • 可选的 group_id
  • 元数据;
  • 是否禁用。

Span 通常包含:

  • 开始时间;
  • 结束时间;
  • 所属 Trace;
  • 父 Span;
  • 具体 Span 数据。

官方默认会为 Runner、任务、回合、Agent、模型生成、函数工具、Guardrail 和 Handoff 创建相应 Span。(openai.github.io)

7.3 自定义 Trace

如果一次业务请求包含多次独立的 Runner.run(),可以用 trace() 把它们归并为一个工作流:

from agents import Agent, Runner, trace

reviewer = Agent(
    name="Reviewer",
    instructions="审查输入文本并给出问题列表。",
)

writer = Agent(
    name="Writer",
    instructions="根据审查结果生成最终文本。",
)

async def main():
    with trace("document_review"):
        review = await Runner.run(
            reviewer,
            "审查这段文档:……",
        )

        final = await Runner.run(
            writer,
            f"根据审查结果生成修订稿:{review.final_output}",
        )

        print(final.final_output)

如果不显式创建外层 Trace,两次 Runner 调用可能各自形成独立运行记录;使用外层 trace() 后,它们被组织到同一业务工作流中。SDK 使用 Python contextvar 跟踪当前 Trace 和 Span,这使其可以在并发上下文中维护关联关系。(openai.github.io)

7.4 Trace 的敏感数据边界

Trace 很有价值,但它可能记录:

  • 用户输入;
  • 模型输入和输出;
  • 工具参数;
  • 工具结果;
  • 订单号、邮箱、内部业务字段。

因此,Trace 不是“免费日志”。生产环境需要决定:

是否记录原始输入?
是否脱敏工具参数?
是否截断大字段?
是否禁止记录凭证和令牌?
是否允许生产数据进入追踪平台?

SDK 支持全局禁用、代码禁用或按单次运行禁用追踪;文档还指出,使用 Zero Data Retention 策略的组织无法使用该追踪能力。(openai.github.io)

对于长时间运行的后台任务,如果进程即将退出,应在 Trace 关闭后调用 flush_traces(),等待缓冲中的 Trace 和 Span 导出完成,否则可能出现业务已经结束但追踪还未送达的情况。(openai.github.io)


八、把六个对象串起来:一个完整的支持流程

下面构造一个简化的客户支持流程:

  • Triage Agent 负责识别问题类型;
  • Order Agent 负责查询订单;
  • Refund Agent 负责退款;
  • get_order_status 是只读工具;
  • request_refund 是有副作用工具;
  • 输入 Guardrail 拒绝空请求;
  • 工具业务逻辑再次校验订单状态;
  • Trace 记录全过程。
import asyncio

from agents import (
    Agent,
    Runner,
    Runner,
    handoff,
    input_guardrail,
    GuardrailFunctionOutput,
    trace,
)
from agents.decorators import tool


@tool
def get_order_status(order_id: str) -> str:
    """查询订单状态。"""
    fake_orders = {
        "o-100": "PAID_NOT_SHIPPED",
        "o-200": "SHIPPED",
    }
    return fake_orders.get(order_id, "NOT_FOUND")


@tool
def request_refund(order_id: str) -> str:
    """申请退款。只有已付款且未发货的订单允许退款。"""
    status = get_order_status(order_id)

    if status != "PAID_NOT_SHIPPED":
        raise ValueError("当前订单状态不允许申请退款")

    # 实际系统中这里还应执行身份校验、幂等键检查和退款服务调用
    return f"退款申请已创建:{order_id}"


@input_guardrail(run_in_parallel=False)
async def non_empty_input(ctx, agent, input):
    text = str(input).strip()

    return GuardrailFunctionOutput(
        output_info={"input_length": len(text)},
        tripwire_triggered=(len(text) == 0),
    )


order_agent = Agent(
    name="Order Agent",
    handoff_description="处理订单状态、物流和订单号相关问题。",
    instructions=(
        "处理订单问题。"
        "需要订单状态时调用 get_order_status。"
        "不要自行编造订单状态。"
    ),
    tools=[get_order_status],
)

refund_agent = Agent(
    name="Refund Agent",
    handoff_description="处理退款和退货问题。",
    instructions=(
        "处理退款问题。"
        "申请退款前必须调用 get_order_status。"
        "只有 PAID_NOT_SHIPPED 状态才可以调用 request_refund。"
    ),
    tools=[get_order_status, request_refund],
)

triage_agent = Agent(
    name="Triage Agent",
    instructions=(
        "识别用户问题。"
        "订单查询交给 Order Agent;"
        "退款或退货请求交给 Refund Agent。"
    ),
    handoffs=[
        order_agent,
        refund_agent,
    ],
    input_guardrails=[non_empty_input],
)


async def main():
    with trace("customer_support"):
        result = await Runner.run(
            triage_agent,
            "请帮我申请订单 o-100 的退款",
            max_turns=8,
        )

        print("最后处理 Agent:", result.last_agent.name)
        print("最终结果:", result.final_output)


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

上例的关键路径是:

sequenceDiagram
    participant U as 用户
    participant R as Runner
    participant T as Triage Agent
    participant F as Refund Agent
    participant S as get_order_status
    participant X as request_refund
    participant G as Trace

    U->>R: 退款请求
    R->>G: 创建 Trace
    R->>T: 输入 Guardrail
    T-->>R: 通过
    R->>T: 模型决策
    T->>R: Handoff 到 Refund Agent
    R->>F: 切换当前 Agent
    F->>R: 调用 get_order_status
    R->>S: 查询 o-100
    S-->>R: PAID_NOT_SHIPPED
    R->>F: 返回工具结果
    F->>R: 调用 request_refund
    R->>X: 执行退款申请
    X-->>R: 退款申请已创建
    R->>F: 返回工具结果
    F-->>R: 最终输出
    R->>G: 完成 Trace
    R-->>U: 返回结果

其中有两个独立的安全边界:

  1. 模型指令要求先查询订单;
  2. request_refund() 自己再次检查订单状态。

第一个边界用于引导模型,第二个边界用于保护真实业务。这种重复不是冗余,而是因为模型决策和业务执行处于不同信任域。


九、常见误解与失败路径

9.1 “有了 Agent,就不需要自己写循环”

Runner 确实管理标准 Agent 循环,但它不会替你定义所有业务语义。你仍然需要决定:

  • 工具是否允许重试;
  • 工具异常是返回模型还是抛给服务层;
  • 哪些动作需要人工审批;
  • 哪些 Handoff 可以发生;
  • 会话历史如何保存;
  • 运行最多推进多少回合;
  • 取消后是否可以恢复。

SDK 管理的是通用执行机制,不会自动推导你的业务不变量。

9.2 “Handoff 就是调用另一个 Agent”

两者的结果所有权不同:

Agent.as_tool():
    当前 Agent 仍然负责最终输出

Handoff:
    目标 Agent 成为当前回合的处理者

如果把本应由 Manager 汇总的专家调用设计成 Handoff,最终输出可能由错误的 Agent 生成;如果把需要专员接管的对话设计成 as_tool(),Manager 可能在专家结果返回后重新改写用户意图。

9.3 “Guardrail 能撤销工具副作用”

输出 Guardrail 在最终输出阶段执行。若 Agent 已经调用了发货、退款、删除或写库工具,输出 Guardrail 触发并不能自动回滚这些动作。

对于不可逆操作,应在工具执行前建立保护:

输入 Guardrail
→ 工具输入 Guardrail
→ 权限检查
→ 幂等检查
→ 业务状态检查
→ 人工审批或确认
→ 执行动作

9.4 “Trace 里有完整记录,所以可以直接当审计系统”

Trace 主要服务于运行观察、调试和评估。它未必满足金融、医疗或高合规场景的审计要求,例如:

  • 不一定具备业务事件的不可抵赖性;
  • 不一定包含完整的数据库变更前后镜像;
  • 不一定能替代订单服务的操作日志;
  • 可能受采样、脱敏或禁用配置影响。

因此,业务审计事件应由业务系统单独记录,Trace 用于关联模型决策与运行过程。

9.5 “流式输出已经展示给用户,就代表结果可靠”

流式模式下,用户可能已经看到部分文字,但后续仍可能发生:

  • 工具调用失败;
  • Handoff;
  • 输出 Guardrail 失败;
  • 运行被取消;
  • 达到回合上限。

前端应区分:

增量事件:用于展示过程
完成事件:用于确认运行成功
错误事件:用于撤回、标记或补偿

不要在收到第一段文本时就把事务标记为成功。


十、会话、状态和 Trace 如何组合

Agent 的一次运行状态与跨回合会话状态并不相同。

运行状态

只在当前 Runner.run() 内部存在,例如:

  • 当前 Agent;
  • 当前回合;
  • 中间工具结果;
  • 当前 Guardrail 结果;
  • 当前输出。

会话状态

跨多个运行保留,例如:

  • 用户此前说过什么;
  • 上一次工具返回了什么;
  • 当前对话属于哪个会话;
  • 多轮交互的历史。

SDK 提供 Session 来自动维护多个 Agent 运行之间的对话历史,也支持手动传入 result.to_input_list(),或使用 conversation_idprevious_response_id 等服务端连续状态方案。一个运行中不能把 Session 与这些运行级连续状态选项叠加使用。(openai.github.io)

可以用下面的关系理解:

Trace = 一次或一组业务工作流的观察边界
Run   = 一次 Runner 执行
Session = 多次 Run 之间的会话记忆
Agent = 某一时刻负责决策的角色

例如:

Session: user-100-conversation-7
├── Run 1
│   └── Trace 1
├── Run 2
│   └── Trace 2
└── Run 3
    └── Trace 3

如果需要把同一会话的多次运行关联起来,可以使用 Trace 的 group_id 保存会话标识;但这只是观测层关联,不等于 Session 本身。(openai.github.io)


十一、生产中应如何划分责任

一个可维护的 Agent 系统,通常将责任划分为四层:

Agent 层:决策范围

负责:

  • 指令;
  • 工具和 Handoff 清单;
  • 输出类型;
  • 角色边界;
  • 模型选择。

Runner 层:运行控制

负责:

  • 回合推进;
  • 取消;
  • 流式;
  • 最大回合数;
  • 当前 Agent;
  • 结果和异常传播。

Tool 层:真实副作用

负责:

  • 参数和业务校验;
  • 权限;
  • 幂等;
  • 超时;
  • 重试策略;
  • 外部系统调用。

Guardrail 与 Trace 层:安全和观察

负责:

  • 输入拒绝;
  • 输出验证;
  • 工具调用前后检查;
  • 运行过程记录;
  • 失败定位;
  • 数据脱敏。

最危险的设计是把所有责任集中到 Prompt:

Prompt 负责路由
Prompt 负责权限
Prompt 负责业务状态
Prompt 负责幂等
Prompt 负责安全
Prompt 负责审计

Prompt 可以影响模型行为,但不能成为唯一的权限系统、事务系统或审计系统。


十二、结语:理解 SDK 的关键是理解控制权

AgentRunnerToolHandoffGuardrailTrace 不是六个孤立的类,而是围绕“控制权如何在运行中流动”形成的一套执行模型:

  • Agent 持有决策权;
  • Runner 推进状态;
  • Tool 暂时把执行权交给外部环境;
  • Handoff 把对话控制权交给另一个 Agent;
  • Guardrail 决定某个状态转换是否允许发生;
  • Trace 记录控制权转移和执行结果。

当任务只是一次回答时,直接使用 Responses API 往往更简单;当任务需要工具调度、多步运行、多个 Agent、会话、Guardrail 或完整可观测性时,Agents SDK 提供了更高层的运行时。官方文档也将这两种方式区分为:直接使用 Responses API 自己管理循环,或使用 Agents SDK 管理回合、工具、Guardrail、Handoff 和 Session。(openai.github.io)

真正成熟的 Agent 工程,不是让模型“看起来更自主”,而是明确每一步:

谁可以决策?
谁可以执行?
谁可以转交控制权?
什么条件下必须停止?
失败后由谁恢复?
每一步如何被观察和验证?

这正是 Agents SDK 中六个核心术语共同回答的问题。


系列导航与关联阅读

官方资料

本文依据 Agent、模型、协议与框架官方资料重新梳理;正文、示例与生产清单由 WR BLOG 编写。