OpenAI Agents SDK 完整教程:从 Agent 到多 Agent、护栏、会话、MCP 与语音
OpenAI Agents SDK 完整教程:从 Agent 到多 Agent、护栏、会话、MCP 与语音
这个 SDK 的野心不是"帮你调 API",而是把 Agent 该有的零件全部定义成基元。
如果你写过一阵子 Agent,大概会经历同一个过程:先用一个 while 循环把工具调用包起来,然后加记忆,然后加护栏,然后发现自己造了一套没法维护的框架。OpenAI Agents SDK 就是把这个过程标准化——它只定义一组精简的基元(Agent、Handoff、Guardrail、Session、Tracing),把编排逻辑露在 Python 里,而不是藏进一层又一层的抽象。
官方文档在 openai.github.io/openai-agents-python。这篇是完整教程,按"从单个 Agent 到生产过程"的顺序,把它文档里的知识点全部走一遍:Agent 怎么配、Runner 怎么跑、结果怎么读、工具怎么分、交接怎么做、会话怎么存、上下文怎么传、护栏怎么写、MCP 怎么接、模型怎么换、Trace 怎么看,最后到语音与实时。
本文提纲
- 安装与第一个 Agent
- Agent 的配置项:instructions、模型、结构化输出、动态提示词
- 运行 Agent:agent loop 与 RunConfig
- 读懂运行结果:8 个结果面与中断恢复
- 流式输出与事件
- 工具体系:hosted、local runtime、函数工具、Agent as tool
- Handoffs:把对话交给专家
- 多 Agent 的两种编排方式
- 会话与记忆:8 种内置实现
- 上下文、护栏与错误处理
- MCP、模型提供商、追踪与成本
- 测试、语音与实时
安装与第一个 Agent
pip install openai-agents
export OPENAI_API_KEY=sk-...最小的 Agent 只有三行:
from agents import Agent, Runner
agent = Agent(name="Assistant", instructions="You are a helpful assistant")
result = Runner.run_sync(agent, "Write a haiku about recursion in programming.")
print(result.final_output)把它拆开看,SDK 的核心概念就这几个:Agent 是一份配置(指令、工具、交接、模型),Runner 负责跑循环,result.final_output 是最终输出。Agent 本身不做任何事——它没有状态、不持有连接,只是一个描述"这个 Agent 是谁、能用什么"的数据对象。这一点很关键:同一个 Agent 实例可以被并发跑很多次,状态在 Runner 和 Session 里。
生产项目里更常用异步版本:
import asyncio
from agents import Agent, Runner
agent = Agent(
name="History Tutor",
instructions="You answer history questions clearly and concisely.",
)
async def main():
result = await Runner.run(agent, "When did the Roman Empire fall?")
print(result.final_output)
if __name__ == "__main__":
asyncio.run(main())Agent 的配置项:instructions、模型、结构化输出、动态提示词
一个完整的 Agent 长这样:
from agents import Agent
from agents.decorators import tool
@tool
def get_weather(city: str) -> str:
"""returns weather info for the specified city."""
return f"The weather in {city} is sunny"
agent = Agent(
name="Haiku agent",
instructions="Always respond in haiku form",
model="gpt-5-nano",
tools=[get_weather],
)几个容易忽略的点:
handoff_description:当这个 Agent 会被别的 Agent 交接时,这段描述会进到对方的工具列表里,帮助对方判断"什么情况下该把活交出去"。
结构化输出(output_type):给一个 Pydantic 模型,final_output 就会是它的实例,而不是字符串。
from pydantic import BaseModel
from agents import Agent
class CalendarEvent(BaseModel):
name: str
date: str
participants: list[str]
agent = Agent(
name="Calendar extractor",
instructions="Extract calendar events from text",
output_type=CalendarEvent,
)动态提示词模板(prompt):可以直接引用平台上的提示词模板并传入变量。
agent = Agent(
name="Prompted assistant",
prompt={
"id": "pmpt_123",
"version": "1",
"variables": {"poem_style": "haiku"},
},
)模板还能根据运行时上下文动态生成——传一个函数进去,它会在每次运行前被调用:
from dataclasses import dataclass
from agents import Agent, GenerateDynamicPromptData, Runner
@dataclass
class PromptContext:
prompt_id: str
poem_style: str
async def build_prompt(data: GenerateDynamicPromptData):
ctx: PromptContext = data.context.context
return {
"id": ctx.prompt_id,
"version": "1",
"variables": {"poem_style": ctx.poem_style},
}
agent = Agent(name="Prompted assistant", prompt=build_prompt)
result = await Runner.run(
agent, "Say hello",
context=PromptContext(prompt_id="pmpt_123", poem_style="limerick"),
)工具使用行为(tool_use_behavior):默认是"工具跑完继续让模型说",但你可以改成"第一个工具的结果就是最终答案",或者指定某些工具一旦被调用就停:
agent = Agent(..., tool_use_behavior="stop_on_first_tool")
from agents.agent import StopAtTools
agent = Agent(..., tool_use_behavior=StopAtTools(stop_at_tool_names=["get_weather"]))也能传一个自定义处理函数,接收 List[FunctionToolResult],返回 ToolsToFinalOutputResult——适合"工具结果本身就是结构化数据、不需要模型再润色一遍"的场景。
运行 Agent:agent loop 与 RunConfig
Runner 有三个入口:Runner.run()(异步)、Runner.run_sync()(同步)、Runner.run_streamed()(流式)。传入的输入可以是字符串(当作一条用户消息)、Responses API 格式的输入项列表,或者一个 RunState(恢复被暂停/取消的运行)。
然后 Runner 跑一个循环,官方把规则讲得很清楚:
- 用当前输入调用当前 Agent 的 LLM;
- LLM 产出输出;
- 如果被判定为最终输出,循环结束并返回结果;
- 如果 LLM 请求交接,更新当前 Agent 与输入,重新循环;
- 如果 LLM 产生工具调用,执行这些工具、把结果追加进上下文,重新循环;
- 超过
max_turns抛MaxTurnsExceeded——传max_turns=None可以关掉这个上限。
"最终输出"的判定标准值得记住:产出了期望类型的文本输出,且没有任何工具调用。
RunConfig 是单次运行的全局覆盖层,不改 Agent 定义。它的选项很多,按用途分四类是最好记的:
| 类别 | 关键项 |
|---|---|
| 模型与会话 | model、model_provider、model_settings、session_settings、session_input_callback |
| 护栏与交接 | input_guardrails、output_guardrails、handoff_input_filter、nest_handoff_history、handoff_history_mapper |
| 模型输入整形 | call_model_input_filter(模型调用前最后一刻改输入,比如裁历史或注入系统提示)、reasoning_item_id_policy |
| 追踪与可观测 | tracing_disabled、tracing、trace_include_sensitive_data、workflow_name、trace_id、group_id、trace_metadata |
| 工具执行 | tool_execution(如限制本地函数工具并发)、tool_not_found_behavior、tool_name_collision_policy、tool_error_formatter |
其中两个值得单独说。call_model_input_filter 是"裁剪历史"最干净的落点:不用改 Agent、不用改 Session,直接在模型调用前拿到完整输入再决定留什么。tool_not_found_behavior 处理的是模型幻觉出一个不存在的工具名——默认抛 ModelBehaviorError,你可以选择把它变成一条模型可见的错误输出,让模型自己纠正。
读懂运行结果:8 个结果面与中断恢复
RunResult 上有很多字段,第一次看容易晕。官方给了张"按需求查表"的对照,我按同样思路整理:
| 你需要什么 | 用哪个 |
|---|---|
| 给用户看的最终答案 | final_output |
| 可直接作为下一轮输入的完整对话列表 | to_input_list() |
| 带 Agent、工具、交接、审批元数据的运行项 | new_items |
| 下一轮通常该由哪个 Agent 处理 | last_agent |
用 Responses API 的 previous_response_id 续接 |
last_response_id |
| 待审批项 + 可恢复快照 | interruptions 和 to_state() |
当前嵌套 Agent.as_tool() 调用的元数据 |
agent_tool_invocation |
| 原始模型响应与护栏诊断 | raw_responses 与护栏结果数组 |
final_output 的类型取决于最后一个 Agent:没定义 output_type 就是 str,定义了就是那个类型的实例,如果运行因审批中断而暂停则是 None。注意交接会改变"最后一个 Agent"是谁,所以 SDK 静态上无法知道最终类型——这就是它被标注为 Any 的原因。
中断与恢复是这套 SDK 里设计得最完整的部分。当某个工具需要人工审批时,待审批项出现在 result.interruptions(流式下是 RunResultStreaming.interruptions),你可以把结果转成可恢复状态、批准或拒绝,再继续跑:
from agents import Agent, Runner
agent = Agent(name="Assistant", instructions="Use tools when needed.")
result = await Runner.run(agent, "Delete temp files that are no longer needed.")
if result.interruptions:
state = result.to_state()
for interruption in result.interruptions:
state.approve(interruption) # 或者 state.reject(interruption)
result = await Runner.run(agent, state)RunState 可以序列化(to_json() / to_string()),所以能把"等审批"这件事跨进程、跨重启地接住。恢复期间如果用户又补了一句话,用 state.add_input(...) 暂存,它会在下一次模型调用前被送进去:
state = result.to_state()
state.add_input("Also keep the generated report in the project folder.")
for interruption in state.get_interruptions():
state.approve(interruption)
result = await Runner.run(agent, state)这里有一个工程边界值得记住:跑到"最终输出已被接受、输出护栏与收尾 hook 都已完成、只是最后没能持久化"这一步之后再恢复,SDK 会把该状态标记为不可恢复,后续用这个状态重跑会直接抛 UserError。因为重放可能重复触发终态副作用。这类细节平时用不到,但在做断点续跑的持久化工作流时,会决定你的重试策略怎么写。
流式输出与事件
流式用 Runner.run_streamed(),它返回 RunResultStreaming,用 stream_events() 消费事件:
from openai.types.responses import ResponseTextDeltaEvent
from agents import Agent, Runner
result = Runner.run_streamed(agent, input="Please tell me 5 jokes.")
async for event in result.stream_events():
if event.type == "raw_response_event" and isinstance(event.data, ResponseTextDeltaEvent):
print(event.data.delta, end="", flush=True)事件分三个层次,选对层次很重要:
RawResponsesStreamEvent:包着模型原始事件(response.created、response.output_text.delta等)。适合"一个字一个字打给用户看"。RunItemStreamEvent:更高层,在一个完整条目生成后触发,name是一组固定的语义事件名,比如message_output_created、handoff_requested、handoff_occured、tool_called、tool_output。适合"工具跑完了,更新一下进度条"。AgentUpdatedStreamEvent:当前 Agent 变化时触发(比如发生交接)。流式模式下current_agent会实时更新,你能在流结束前就看到交接发生了。
两个坑:
流式运行在 stream_events() 结束前都不算完成。 最后一个可见 token 之后,SDK 可能还在持久化 session、落定审批状态、压缩历史。
要中途停止用 result.cancel(),默认立即停;result.cancel(mode="after_turn") 会等当前轮干净地跑完。如果你是自己用 to_input_list(mode="normalized") 续接,注意 after_turn 停在工具轮时,应该用 result.last_agent 加这个 normalized 输入继续"未完成的用户轮",而不是马上追加一条新的用户消息。
工具体系:hosted、local runtime、函数工具、Agent as tool
SDK 的工具分成三大类,官方文档里区分得很清楚。
一类:Hosted tools(在服务端跑)
这些工具由 OpenAI 的 Responses API 直接执行,你不提供实现:
from agents import Agent, FileSearchTool, Runner, WebSearchTool
agent = Agent(
name="Assistant",
tools=[
WebSearchTool(
search_content_types=["image", "text"],
image_settings={"max_results": 3, "caption": True},
),
FileSearchTool(
max_num_results=3,
vector_store_ids=["VECTOR_STORE_ID"],
),
],
)这一类还包括几个更"重"的能力:
ShellTool:托管容器里的 shell,可以挂载 skill 引用,并配置网络策略(例如完全禁网);ToolSearchTool+tool_namespace+@tool(defer_loading=True):宿主侧的工具搜索。工具多了以后不把全部定义塞进上下文,而是让模型按需"搜索并延迟加载";ProgrammaticToolCallingTool:让模型写代码调用工具,而不是逐个发起工具调用。用法上给工具加allowed_callers=["programmatic"],并在ModelSettings里设tool_choice="programmatic_tool_calling"。
二类:Local runtime tools(在你机器上执行)
ComputerTool(配合实现 AsyncComputer 接口:screenshot / click / double_click / scroll / type / wait / move / keypress)和 ApplyPatchTool(配一个实现 ApplyPatchEditor 的编辑器)。这两个是"让 Agent 操作你的电脑"的接口,也意味着权限与审批的门必须设在这里(文档专门有一节讲 local shell 与文件编辑的审批)。
三类:Function tools(你自己写的函数)
import json
from typing_extensions import TypedDict, Any
from agents import Agent, FunctionTool, RunContextWrapper
from agents.decorators import tool
class Location(TypedDict):
lat: float
long: float
@tool
async def fetch_weather(location: Location) -> str:
...关键知识点:
- 参数与 docstring 自动解析成 JSON Schema,所以类型标注和 docstring 就是模型的说明书,不是可选注释;
- 可以用 Pydantic
Field约束和描述参数(比如给数值加范围、给字符串加枚举),比在 docstring 里口头约定可靠; - 函数工具超时可配,默认值之外要显式设;
- 错误处理:工具抛异常时的行为可以定制,避免一个工具挂掉打断整个 run;
- 返回图片或文件也是支持的(工具不只能返回文本);
- 需要完全控制 schema 时,可以用
FunctionTool手工构造(custom function tools)。
第四类:Agents as tools
Agent.as_tool() 把一个 Agent 包装成工具,交给管理者 Agent 调用。它有几个进阶用法:定制工具名与描述、给工具定义结构化输入、给工具加审批门、流式查看嵌套 Agent 的运行,以及按条件启用工具。
这里有个交叉知识点:Agent 作为工具被调用时,被调用的 Agent 能通过 RunContextWrapper.tool_input 拿到这次调用的结构化输入,返回时也能通过 agent_tool_invocation 元数据被追溯。
Handoffs:把对话交给专家
交接和"Agent as tool"的区别在于控制权:交接之后,接手的是那个专家 Agent,它直接面对用户,成为本轮剩下的活跃 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)])直接放 Agent 是简写,handoff() 才能定制:
from agents import Agent, handoff, RunContextWrapper
def on_handoff(ctx: RunContextWrapper[None]):
print("Handoff called")
handoff_obj = handoff(
agent=agent,
on_handoff=on_handoff,
tool_name_override="custom_handoff_tool",
tool_description_override="Custom description",
)带输入参数的交接很实用:交接时可以要求模型填一份结构化数据,比如升级原因。
from pydantic import BaseModel
class EscalationData(BaseModel):
reason: str
async def on_handoff(ctx: RunContextWrapper[None], input_data: EscalationData):
print(f"Escalation agent called with reason: {input_data.reason}")
handoff_obj = handoff(agent=agent, on_handoff=on_handoff, input_type=EscalationData)另外两个知识点:输入过滤器(input_filter)可以改写交给下一个 Agent 的内容,官方内置了 handoff_filters.remove_all_tools 这类过滤器;推荐提示词前缀 RECOMMENDED_PROMPT_PREFIX 应当加在会接收交接的 Agent 指令前,它包含让模型正确处理"我接手了"的说明——不用它,交接后的第一句话经常很奇怪。
多 Agent 的两种编排方式
官方把它分成"通过 LLM 编排"和"通过代码编排",这个划分很关键。
通过 LLM 编排:Agent 本身就是"LLM + 指令 + 工具 + 交接",所以开放式任务可以让模型自己规划。这里有两种核心模式,选哪个取决于你要谁持有最终答案:
| 模式 | 工作方式 | 什么时候用 |
|---|---|---|
| Agents as tools | 管理者 Agent 保持对话控制权,通过 Agent.as_tool() 调用专家 |
你想让一个 Agent 拥有最终答案、汇总多个专家的输出,或者在一处统一施加护栏 |
| Handoffs | 分诊 Agent 把对话路由给专家,专家成为本轮活跃 Agent | 你想让专家直接回复、保持提示词聚焦,或者让路由本身成为工作流的一部分 |
两者可以混用:分诊 Agent 交接给专家,专家再用 Agent as tools 调别的 Agent 处理窄任务。
通过代码编排更适合要确定性(速度、成本、可预测性)的场景,常见四种:
- 用结构化输出做分类,再根据分类结果选择下一个 Agent;
- 链式:研究 → 大纲 → 写稿 → 评审 → 改写,上一步输出是下一步输入;
- 评估循环:while 循环里跑任务 Agent 产出结果、跑评估 Agent 打分,直到通过;
- 并行:用
asyncio.gather同时跑互不依赖的任务。
会话与记忆:8 种内置实现
会话解决的是"下一轮怎么带上历史"。用法极简——传 session 就行:
from agents import Agent, Runner, SQLiteSession
agent = Agent(name="Assistant", instructions="Reply very concisely.")
session = SQLiteSession("conversation_123")
result = await Runner.run(agent, "What city is the Golden Gate Bridge in?", session=session)
result = await Runner.run(agent, "What state is it in?", session=session) # 自动记得上文SQLiteSession("user_123") 是内存库(进程结束即丢),SQLiteSession("user_123", "conversations.db") 才落盘。中断恢复要用同一个 session(同一 ID + 同一存储后端),否则恢复的那一轮接不上原来的历史。
内置实现覆盖面很广,按场景挑:
| 实现 | 适合 |
|---|---|
SQLiteSession / AsyncSQLiteSession |
默认轻量方案、单机、原型 |
RedisSession |
多实例、需要共享会话 |
SQLAlchemySession |
已有关系型数据库、要接 ORM |
MongoDBSession |
文档型存储 |
DaprSession |
用 Dapr 做分布式运行时 |
| 加密会话 | 历史含敏感数据,需要落盘加密 |
OpenAIConversationsSession |
让服务端托管对话 |
OpenAIResponsesCompactionSession |
历史太长,需要自动压缩 |
会话层还有三个"细节但重要"的控制点:历史与新输入的合并方式(用 RunConfig.session_input_callback 定制)、限制取回的历史条数(SessionSettings(limit=...))、以及记忆操作与纠错——除了基础增删查,pop_item 可以撤掉某一条历史,这在"模型基于一条错误记忆跑偏"时很好用。
上下文、护栏与错误处理
上下文:把你自己的状态传进去
RunContextWrapper 是个包装器,日常只用到几项:wrapper.context(你自己的对象)、wrapper.usage(本次运行的 token 用量)、wrapper.tool_input(在 Agent.as_tool() 内的结构化输入)、wrapper.approve_tool(...) / reject_tool(...)(程序化地改审批状态)。
from dataclasses import dataclass
from agents import Agent, RunContextWrapper, Runner
from agents.decorators import tool
@dataclass
class UserInfo:
name: str
uid: int
@tool
async def fetch_user_age(wrapper: RunContextWrapper[UserInfo]) -> str:
"""Fetch the age of the user."""
return f"The user {wrapper.context.name} is 47 years old"
agent = Agent[UserInfo](name="Assistant", tools=[fetch_user_age])
result = await Runner.run(agent, "What is the age of the user?", context=UserInfo(name="John", uid=123))两个提醒。Agent[UserInfo] 的泛型标注是给类型检查器用的:工具声明了 RunContextWrapper[UserInfo],如果 Agent 的上下文类型不匹配,静态检查就能提前发现。不要把密钥放进 context:如果你打算序列化 RunState(做人工审批或持久化工作流),上下文里的运行元数据会被一起写进去。
护栏:三类,都靠 tripwire 中断
护栏的实现方式是"一个接收输入、返回 GuardrailFunctionOutput 的函数",官方示例里干脆用另一个 Agent 来判断:
from pydantic import BaseModel
from agents import (
Agent, GuardrailFunctionOutput, InputGuardrailTripwireTriggered,
RunContextWrapper, Runner, TResponseInputItem,
)
from agents.decorators import input_guardrail
class MathHomeworkOutput(BaseModel):
is_math_homework: bool
reasoning: str
guardrail_agent = Agent(
name="Guardrail check",
instructions="Check if the user is asking you to do their math homework.",
output_type=MathHomeworkOutput,
)
@input_guardrail
async def math_guardrail(
ctx: RunContextWrapper[None], agent: Agent, input: str | list[TResponseInputItem]
) -> GuardrailFunctionOutput:
result = await Runner.run(guardrail_agent, input, context=ctx.context)
return GuardrailFunctionOutput(
output_info=result.final_output,
tripwire_triggered=result.final_output.is_math_homework,
)三类护栏与对应的异常要把名字记准,因为中断后你靠异常类型决定怎么处理:
| 护栏 | 装饰器 | 触发时的异常 |
|---|---|---|
| 输入护栏 | @input_guardrail |
InputGuardrailTripwireTriggered |
| 输出护栏 | @output_guardrail |
OutputGuardrailTripwireTriggered |
| 工具护栏 | @tool_guardrail |
ToolInputGuardrailTripwireTriggered / ToolOutputGuardrailTripwireTriggered |
异常里带着诊断信息:Agent 级 tripwire 的 guardrail_result 指出是哪条护栏触发;输入 tripwire 的 run_data.input_guardrail_results 保存了在中断前完成的所有输入护栏结果,输出同理。工具级的异常则直接暴露触发护栏与输出,run_data.tool_input_guardrail_results / tool_output_guardrail_results 保留此前累积的结果——而且像 MaxTurnsExceeded 这类运行器管理的失败也会保留这些结果,所以"任务失败了,但护栏检查记录还在",便于事后审计。
护栏还有执行模式(input guardrails 可选并行或阻塞),这是延迟与确定性的取舍:并行更快,但可能在护栏判定出来之前就已经调用了模型。
MCP、模型提供商、追踪与成本
MCP:四种接法
| 你的需求 | 用哪个 |
|---|---|
| 让 Responses API 代替模型调用公网可达的 MCP 服务 | HostedMCPTool |
| 连接你自建或远程的 Streamable HTTP 服务 | MCPServerStreamableHttp |
| 连接实现了 HTTP + SSE 的服务 | MCPServerSse |
| 启动本地进程、走 stdin/stdout | MCPServerStdio |
stdio 的例子:
from pathlib import Path
from agents import Agent, Runner
from agents.mcp import MCPServerStdio
async with MCPServerStdio(
name="Filesystem Server via npx",
params={
"command": "npx",
"args": ["-y", "@modelcontextprotocol/server-filesystem", str(samples_dir)],
},
) as server:
agent = Agent(
name="Assistant",
instructions="Use the files in the sample directory to answer questions.",
mcp_servers=[server],
)
result = await Runner.run(agent, "List the files available to you.")还有几个实战要点:多个服务器用 MCPServerManager 统一连接,并把连上的子集暴露给 Agent;工具过滤支持静态和动态两种;MCP 工具同样能加工具护栏;MCP 的 prompts 也能被取用;托管 MCP 支持审批流和连接器支持的服务端。
依赖上有个容易踩的点:SDK 支持 mcp>=1.19.0,<3 两个大版本,会自动探测已安装的包版本并适配 stdio / SSE / Streamable HTTP 连接(v2 下会先用 server/discover 探测,老服务器则回落到 initialize 握手)。但如果你自定义 HTTP 栈——params["auth"] 或 params["httpx_client_factory"] 的返回类型——必须匹配已安装的 mcp 包主版本(v1 用 httpx、v2 用 httpx2),否则会直接报错。
模型与提供商
默认模型由 SDK 决定,覆盖方式有三种:Agent 上写 model="gpt-5-nano"、运行时在 RunConfig(model=...) 里全局覆盖、或者换整个 provider。密钥与客户端可以程序化设置:
from agents import set_default_openai_key, set_default_openai_client
from openai import AsyncOpenAI
set_default_openai_key("sk-...")
custom_client = AsyncOpenAI(base_url="...", api_key="...")
set_default_openai_client(custom_client)注意 SDK 的客户端归属规则:如果给 OpenAIProvider 传了显式 openai_client,就不能再传 api_key / base_url / organization / project,否则抛 UserError,而不是悄悄忽略——这类"要么全给我、要么全别给"的设计能省掉很多诡异问题。非 OpenAI 模型可以通过第三方适配器接入(官方列出 Any-LLM 等路径),也可以在一个工作流里混用不同厂商的模型(比如便宜模型做分类、强模型做生成)。
其余生产相关项:ModelSettings 里的高级 Responses 参数、extra_args 透传、模型调用超时、Runner 托管的自动重试、Responses 的 WebSocket 传输(可复用连接,注意它是 Responses over websocket,不是 Realtime API)、Chat Completions 兼容选项,以及实验性的托管多 Agent。
追踪与成本
Tracing 默认开启,每个 Agent 步骤都会产生 span。手动建 trace 用上下文管理器,并发安全(当前 trace 存在 contextvar 里):
from agents import trace
with trace("My workflow") as my_trace:
... # 这中间所有的模型调用、工具调用都会挂在这条 trace 下custom_span()、generation_span()、function_span() 用于自定义 span;注意 generation_span 存 LLM 输入输出、function_span 存函数输入输出,都可能含敏感数据,用 RunConfig.trace_include_sensitive_data 控制,或者用环境变量 OPENAI_AGENTS_TRACE_INCLUDE_SENSITIVE_DATA 设默认值。还可以注册自定义追踪处理器,在导出前做脱敏。
成本统计在 result.context_wrapper.usage 上:
usage = result.context_wrapper.usage
print(usage.requests, usage.input_tokens, usage.output_tokens, usage.total_tokens)它聚合了本次运行里所有模型调用(包括产生工具调用和交接的那些),细分字段包含 input_tokens_details.cached_tokens、cache_write_tokens、output_tokens_details.reasoning_tokens,还有 request_usage_entries 给出逐请求明细。会话与 RunState 检查点里也能读到用量。
测试、语音与实时
测试的符号刻意不放在顶层导入里,而是紧贴它们所替代的运行时边界:agents.testing(Agent 模型与 Sandbox 工作流)、agents.realtime.testing(实时传输)、agents.voice.testing(语音 STT/TTS)。官方给的是"配方式"文档:返回固定响应、测试工具工作流、从请求推导响应、检查模型调用、测试流式、注入模型故障、检测工作流漂移。最后两条在生产里最有用——在无网络、无 provider 请求的情况下确定性地复现失败。
语音的核心是 VoicePipeline,三步:语音转文字 → 跑你的代码(通常是 Agent 工作流)→ 文本转语音。
实时(Realtime) 是另一套组件:
from agents.realtime import RealtimeAgent, RealtimeRunner
agent = RealtimeAgent(
name="Assistant",
instructions="You are a helpful voice assistant. Keep responses short and conversational.",
)
runner = RealtimeRunner(
starting_agent=agent,
config={
"model_settings": {
"model_name": "gpt-realtime-2.1",
"audio": {
"input": {
"format": "pcm16",
"transcription": {"model": "gpt-4o-mini-transcribe"},
"turn_detection": {"type": "semantic_vad", "interrupt_response": True},
},
"output": {"format": "pcm16", "voice": "ash"},
},
}
},
)runner.run() 返回一个 RealtimeSession,连接在你进入 session 上下文时建立。要记住一个平台边界:Python SDK 只支持 WebSocket 传输,WebRTC 不可用——如果你要做浏览器直连,得在架构上另外安排。
参考链接
- OpenAI Agents SDK 官方文档 — 本文主要来源
- Quickstart — 安装、首个 Agent、工具、交接、追踪
- Agents — 指令、提示词模板、上下文泛型、输出类型、工具使用行为
- Running agents — agent loop、RunConfig、错误处理、异常
- Results — 结果面、
to_state()、中断恢复与RunState - Streaming — 三类流式事件、审批兼容、取消
- Tools — hosted / local runtime / 函数工具 / Agent as tool
- Handoffs — 交接定制、输入类型、过滤器与推荐前缀
- Multi-agent patterns — LLM 编排 vs 代码编排
- Sessions — 八种内置会话实现与记忆操作
- Context —
RunContextWrapper与ToolContext - Guardrails — 输入/输出/工具护栏与 tripwire
- MCP — 四种集成方式、服务器管理与工具过滤
- Models — 模型选择、非 OpenAI 提供商、混合模型与重试
- Usage and pricing — 用量字段与计费口径
- Configuration — 密钥、客户端归属、日志与敏感数据
- Tracing — trace/span、自定义处理器与脱敏
- Testing — 确定性测试配方(含故障注入)
- Voice pipeline / Realtime — 语音与实时 Agent
- GitHub: openai/openai-agents-python — 源码、示例与 issue
你现在的 Agent 是"一个 while 循环 + 一堆 if",还是已经拆成基元了?评论区聊聊你的架构,觉得这份教程有用就点个赞收藏。
作者: itech001 来源: 公众号:AI人工智能时代(the-ai-era) 网站: https://www.theaiera.top/ 关注每日最新AI新闻和技术博客,主页有更多的文章的AI 技术参考:https://www.theaiera.top
本文首发于 AI人工智能时代,转载请注明出处。