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 运行循环:
其中:
Agent提供决策所需的身份、指令和能力边界;Runner执行状态转换;Tool把决策转化为外部行动;Handoff改变当前负责决策的 Agent;Guardrail限制状态转换和副作用;Trace记录每次状态变化,便于调试、评估和生产诊断。
一、先建立整体模型:Agent 不是一次模型调用
一个普通的模型调用通常可以表示为:
其中:
- 是用户输入;
- 是提示词或系统指令;
- 是模型;
- 是模型输出。
但 Agent 运行通常不是一次调用,而是一个循环:
其中:
- 是第 步的运行状态;
- 是模型基于当前状态做出的决策;
- 是外部环境,例如数据库、HTTP 服务、文件系统或另一个 Agent;
- 是运行时对模型决策的解释和执行;
- 是执行行动后得到的新状态。
模型可能产生三类结果:
- 最终答案:运行终止;
- 工具调用:执行工具,把工具结果反馈给模型;
- 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 可以理解为:
其中:
- :名称
name; - :指令
instructions或动态提示prompt; - :工具集合
tools; - :可用的交接目标
handoffs; - :输入 Guardrail;
- :输出 Guardrail;
- :输出类型
output_type; - :模型、模型参数和运行上下文。
官方将 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 的子类。它是运行时控制器,负责:
- 准备输入和运行上下文;
- 执行输入 Guardrail;
- 调用当前 Agent;
- 将模型产生的工具调用分派给工具;
- 将工具结果追加回运行状态;
- 处理 Handoff;
- 执行输出 Guardrail;
- 返回
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,错误循环可能消耗大量令牌和外部资源。
因此,终止条件至少应区分:
生产代码不应只打印 final_output,还应记录最后 Agent、运行状态和异常类型。
四、Tool:让模型从“生成文字”进入“改变环境”
4.1 Tool 的本质
Tool 是模型可选择调用的动作接口。可以抽象为:
其中:
- 是工具参数的输入域;
- 是工具结果;
- 工具内部可能访问外部环境 。
例如:
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 之间的控制权转移:
其中:
- 是当前 Agent;
- 是从 到 的交接;
- 成为后续处理当前回合的 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 是一个可执行检查:
与指令不同,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: 返回结果
其中有两个独立的安全边界:
- 模型指令要求先查询订单;
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_id、previous_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 的关键是理解控制权
Agent、Runner、Tool、Handoff、Guardrail 和 Trace 不是六个孤立的类,而是围绕“控制权如何在运行中流动”形成的一套执行模型:
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 工程完整路线:从运行循环、记忆与协议到安全、评测和生产交付
- 上一篇:AG-UI 协议:Agent 事件、前端状态、流式交互和人工确认
- 下一篇:OpenAI Agents SDK 会话工程:历史、Session、流式、取消和恢复
- 延伸:Agent 运行循环:观察、决策、行动、反馈、状态与终止
官方资料
本文依据 Agent、模型、协议与框架官方资料重新梳理;正文、示例与生产清单由 WR BLOG 编写。

评论
0 条讨论