返回博客列表

OpenAI Agents SDK 完整教程:从 Agent 到多 Agent、护栏、会话、MCP 与语音

2026-10-05T22:00:00+08:00
OpenAIAgents SDKAgent多 AgentMCPGuardrailsTutorial

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 怎么看,最后到语音与实时。

本文提纲

  1. 安装与第一个 Agent
  2. Agent 的配置项:instructions、模型、结构化输出、动态提示词
  3. 运行 Agent:agent loop 与 RunConfig
  4. 读懂运行结果:8 个结果面与中断恢复
  5. 流式输出与事件
  6. 工具体系:hosted、local runtime、函数工具、Agent as tool
  7. Handoffs:把对话交给专家
  8. 多 Agent 的两种编排方式
  9. 会话与记忆:8 种内置实现
  10. 上下文、护栏与错误处理
  11. MCP、模型提供商、追踪与成本
  12. 测试、语音与实时

安装与第一个 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 跑一个循环,官方把规则讲得很清楚:

  1. 用当前输入调用当前 Agent 的 LLM;
  2. LLM 产出输出;
  3. 如果被判定为最终输出,循环结束并返回结果;
  4. 如果 LLM 请求交接,更新当前 Agent 与输入,重新循环;
  5. 如果 LLM 产生工具调用,执行这些工具、把结果追加进上下文,重新循环;
  6. 超过 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 处理窄任务。

通过代码编排更适合要确定性(速度、成本、可预测性)的场景,常见四种:

  1. 用结构化输出做分类,再根据分类结果选择下一个 Agent;
  2. 链式:研究 → 大纲 → 写稿 → 评审 → 改写,上一步输出是下一步输入;
  3. 评估循环:while 循环里跑任务 Agent 产出结果、跑评估 Agent 打分,直到通过;
  4. 并行:用 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 不可用——如果你要做浏览器直连,得在架构上另外安排。

参考链接

你现在的 Agent 是"一个 while 循环 + 一堆 if",还是已经拆成基元了?评论区聊聊你的架构,觉得这份教程有用就点个赞收藏。


作者: itech001 来源: 公众号:AI人工智能时代(the-ai-era) 网站: https://www.theaiera.top/ 关注每日最新AI新闻和技术博客,主页有更多的文章的AI 技术参考:https://www.theaiera.top

本文首发于 AI人工智能时代,转载请注明出处。

分享给朋友