OpenAI Agents SDK源码深度拆解

从一个@dataclass开始，如何构建出一个完整的多Agent编排框架？

一、项目概览

OpenAI Agents SDK^[1]（原名Swarm，后正式更名）是OpenAI官方开源的多Agent编排框架，在GitHub上已积累约28k Star。它的设计哲学可以用一句话概括：用最少的概念，编排最复杂的Agent工作流。

整个框架的核心概念只有五个：Agent（智能体）、Runner（执行器）、Handoff（移交）、Guardrail（护栏）、Tool（工具）。但围绕这五个概念，SDK构建了一套极其精密的执行引擎——支持异步流式输出、并发护栏检测、MCP协议集成、会话持久化、沙箱执行、语音交互等企业级能力。

与市面上其他Agent框架（如LangChain、CrewAI）相比，OpenAI Agents SDK的最大特点是极简主义。它没有引入复杂的图编排、状态机或角色系统，而是将一切抽象为"Agent调用工具"这一统一范式。Handoff本质上是工具调用，Agent as_tool也是工具调用，甚至审批请求也通过工具机制处理。这种统一性使得执行循环的设计异常简洁，同时也带来了极高的可组合性。

本文将从源码层面逐模块拆解这个框架的每一个核心组件，揭示其设计精妙之处。

二、完整目录结构解析

SDK的源码位于src/agents/目录下，模块组织层次分明。整个代码库大约有两万行Python代码，分布在数十个模块中。下面我们逐一列出并解释每个模块的职责和设计意图。

src/agents/├── agent.py              # Agent核心类定义，约1000行├── run.py                # Runner执行器，约1900行├── run_config.py         # RunConfig运行配置，约390行├── run_context.py        # RunContextWrapper运行上下文├── run_state.py          # RunState可序列化状态（用于断点续传）├── run_error_handlers.py # 错误处理器注册表├── result.py             # RunResult/RunResultStreaming结果类├── items.py              # 运行时条目类型（消息、工具调用、推理项等）├── tool.py               # Tool工具体系，约2400行，最大的单文件├── function_schema.py    # 函数签名内省，从type hints自动生成JSON Schema├── tool_context.py       # ToolContext工具执行上下文├── tool_guardrails.py    # ToolInputGuardrail/ToolOutputGuardrail工具护栏├── guardrail.py          # InputGuardrail/OutputGuardrail输入输出护栏├── strict_schema.py      # JSON Schema严格模式转换工具├── agent_output.py       # AgentOutputSchema结构化输出Schema├── agent_tool_input.py   # Agent as_tool的输入处理├── agent_tool_state.py   # Agent工具状态管理├── lifecycle.py          # RunHooks/AgentHooks生命周期钩子定义├── exceptions.py         # 自定义异常层次结构├── logger.py             # 日志配置├── prompts.py            # Prompt对象及动态Prompt函数├── usage.py              # Usage使用量统计├── stream_events.py      # 流式事件类型定义├── _debug.py             # 调试工具├── _tool_identity.py     # 工具标识与名称空间管理├── util/                 # 工具函数集合│   ├── _transforms.py    # 字符串转换工具│   ├── _json.py          # JSON验证工具│   ├── _error_tracing.py # 错误追踪工具│   └── _types.py         # 类型辅助定义│├── handoffs/             # Handoff移交机制模块│   ├── __init__.py       # Handoff dataclass + handoff()工厂函数，约350行│   └── history.py        # nest_handoff_history历史嵌套逻辑│├── models/               # 模型抽象层│   ├── interface.py      # 抽象Model类和ModelProvider接口，约155行│   ├── multi_provider.py # MultiProvider多模型提供商路由│   ├── default_models.py # 默认模型配置与设置│   └── _response_terminal.py # 响应终止条件判断│├── tracing/              # 自定义追踪系统（注意：非OpenTelemetry）│   ├── __init__.py       # 导出所有追踪API，约130行│   ├── traces.py         # Trace抽象基类及三种实现，约530行│   ├── spans.py          # Span抽象基类及SpanImpl实现，约400行│   ├── span_data.py      # 13种SpanData类型定义，约450行│   ├── processor_interface.py # TracingProcessor处理器接口│   ├── processors.py     # BatchTraceProcessor批量导出处理器│   ├── provider.py       # TraceProvider追踪提供者管理│   ├── scope.py          # 基于contextvars的作用域管理│   ├── context.py        # TraceCtxManager上下文管理器│   ├── config.py         # TracingConfig追踪配置│   ├── create.py         # 创建Span/Trace的工厂函数│   ├── model_tracing.py  # 模型调用追踪辅助│   └── setup.py          # 全局追踪设置│├── run_internal/         # 运行时内部引擎（非公开API，不保证稳定性）│   ├── __init__.py│   ├── run_loop.py       # 核心执行循环，约1950行，整个SDK最大的文件│   ├── turn_resolution.py # 模型响应处理与NextStep决策，约2140行│   ├── tool_execution.py # 工具执行引擎，约2440行，第二大的文件│   ├── streaming.py      # 流式事件队列管理│   ├── guardrails.py     # 护栏并发执行逻辑│   ├── tool_planning.py  # 工具调用规划与审批决策│   ├── approvals.py      # 审批流程管理│   ├── items.py          # 输入项规范化与去重│   ├── tool_actions.py   # 工具动作类定义│   ├── tool_use_tracker.py # 工具使用追踪器│   ├── turn_preparation.py # 单轮执行前的准备工作│   ├── agent_bindings.py # Agent绑定与执行Agent分离│   ├── agent_runner_helpers.py # Runner辅助函数集合│   ├── error_handlers.py # 运行时错误处理逻辑│   ├── model_retry.py    # 模型调用重试机制│   ├── session_persistence.py # Session持久化逻辑│   ├── oai_conversation.py # OpenAI服务端对话管理│   ├── prompt_cache_key.py # Prompt缓存键解析│   └── run_steps.py      # NextStep枚举与SingleStepResult定义│├── mcp/                  # MCP（Model Context Protocol）协议集成│   ├── __init__.py       # MCPUtil工具转换 + MCPServer抽象接口│   └── manager.py        # MCPServerManager生命周期管理器│├── memory/               # 会话记忆管理│   ├── __init__.py       # Session接口导出│   └── session.py        # Session抽象基类定义│├── realtime/             # 实时Agent模块│   ├── __init__.py       # RealtimeAgent定义│   └── ...               # WebSocket实时通信相关│├── voice/                # 语音管线模块│   ├── __init__.py       # VoicePipeline语音管线│   ├── stt.py            # 语音转文字（Speech-to-Text）│   ├── tts.py            # 文字转语音（Text-to-Speech）│   └── ...               # 音频处理相关│└── sandbox/              # 沙箱执行环境    ├── __init__.py       # 沙箱API导出    ├── runtime.py        # SandboxRuntime运行时管理    ├── manifest.py       # Manifest文件清单声明式配置    ├── snapshot.py       # 沙箱快照管理    └── session/          # 多种沙箱后端实现        ├── base_sandbox_session.py # BaseSandboxSession抽象基类        ├── sandbox_client.py      # BaseSandboxClient客户端接口        ├── docker.py     # Docker容器沙箱后端        ├── e2b.py        # E2B云端代码执行沙箱        ├── modal.py      # Modal无服务器计算沙箱        └── cloudflare.py # Cloudflare Workers边缘沙箱

这个目录结构体现了SDK的清晰分层设计。最外层是公开API层（agent.py、run.py、tool.py等），这些是开发者直接使用的接口，API稳定性有保障。中间层是内部执行层（run_internal/），包含执行循环、工具规划、审批流程等核心运行时逻辑，这些是SDK的"发动机"，不对外暴露。再往下是扩展层（mcp/、memory/、sandbox/、voice/、realtime/），提供各种扩展能力。最后是观测层（tracing/），提供全链路追踪能力。

值得注意的是，run_internal/目录下的文件总行数超过一万行，占据了整个代码库的一半以上。这说明SDK的复杂度主要体现在执行引擎的工程实现上，而非概念模型上。概念层面只有五个核心概念，但要将它们正确地组合起来，需要处理大量的边界情况、并发协调和错误恢复。

三、Agent类深度剖析

Agent是整个SDK的核心数据结构，定义在agent.py中，文件约1000行。它使用@dataclass装饰器实现，并通过Generic[TContext]支持泛型上下文。

3.1 继承体系与设计选择

@dataclassclassAgentBase(Generic[TContext]):"""Base class for Agent and RealtimeAgent."""    name: str    handoff_description: str | None = None    tools: list[Tool] = field(default_factory=list)    mcp_servers: list[MCPServer] = field(default_factory=list)    mcp_config: MCPConfig = field(default_factory=lambda: MCPConfig())@dataclassclassAgent(AgentBase, Generic[TContext]):"""An agent is an AI model configured with instructions, tools,    guardrails, handoffs and more."""    instructions: str | Callable[..., MaybeAwaitable[str]] | None = None    prompt: Prompt | DynamicPromptFunction | None = None    handoffs: list[Agent[Any] | Handoff[TContext, Any]] = field(default_factory=list)    model: str | Model | None = None# ... 更多字段

SDK选择@dataclass而非传统的类继承来定义Agent，这是一个深思熟虑的决定。Dataclass天然支持dataclasses.replace()浅拷贝，使得Agent.clone()方法可以一行代码实现。同时，Dataclass的字段定义语法非常清晰，每个字段的名称、类型、默认值一目了然，配合docstring形成了极好的自文档化效果。

AgentBase提取了与RealtimeAgent共享的基础字段（name、tools、mcp_servers、mcp_config），Agent则在此基础上扩展了所有标准Agent特有的配置。这种拆分避免了RealtimeAgent继承不需要的字段（如instructions、handoffs等），体现了接口隔离原则。

3.2 全部字段逐一解析

Agent类包含十余个核心字段，每一个都经过精心设计。下面我们逐一深入分析：

instructions字段——最灵活的系统提示

instructions: (str    | Callable[        [RunContextWrapper[TContext], Agent[TContext]],        MaybeAwaitable[str],    ]    | None) = None

instructions可以是静态字符串，也可以是一个动态函数。这个函数接收运行上下文RunContextWrapper和Agent实例，返回系统提示词字符串。函数可以是同步的也可以是异步的（通过MaybeAwaitable类型别名支持）。这意味着同一个Agent可以根据不同的运行时上下文生成完全不同的指令。举个实际例子：一个客服Agent可以根据用户是VIP还是普通用户来切换不同的服务策略和语气，或者根据当前对话的语言自动选择对应语言的系统提示。

prompt字段——与OpenAI平台深度集成

prompt: Prompt | DynamicPromptFunction | None = None

prompt字段允许通过OpenAI平台的Prompt管理功能动态配置Agent的指令和工具。Prompt对象有一个get()方法，可以从OpenAI平台拉取最新的Prompt配置。这实现了"代码外的Agent行为管理"——产品运营人员可以在OpenAI平台上调整Agent的行为，而无需修改代码和重新部署。

handoffs字段——多Agent协作的核心

handoffs: list[Agent[Any] | Handoff[TContext, Any]] = field(default_factory=list)

handoffs支持两种形式：直接传入Agent对象（SDK会自动创建默认的Handoff，工具名为transfer_to_{agent_name}），或传入自定义的Handoff对象以精细控制移交行为（包括工具名称、描述、输入过滤器、历史嵌套等）。这种双模式设计既照顾了简单场景的便利性，又保留了复杂场景的灵活性。

model字段——灵活的模型指定

model: str | Model | None = None

model字段支持三种指定方式。传入字符串时，由RunConfig中的ModelProvider解析为具体的Model实例。传入Model实例时，直接使用。传入None时，使用SDK的默认模型配置。这种设计使得模型的选择可以在不同层级灵活覆盖。

output_type字段——结构化输出

output_type: type[Any] | AgentOutputSchemaBase | None = None

支持任意Python类型（dataclass、Pydantic模型、TypedDict等），SDK会自动从类型生成JSON Schema并约束LLM的输出格式。如果传入None，默认输出为纯字符串。还支持传入AgentOutputSchemaBase的子类来自定义Schema生成逻辑，比如使用非严格模式的JSON Schema。

model_settings字段——模型参数配置

model_settings: ModelSettings = field(default_factory=get_default_model_settings)

ModelSettings是一个包含20多个可配置参数的dataclass，包括temperature、top_p、frequency_penalty、presence_penalty、tool_choice、max_tokens、reasoning等。SDK实现了层级覆盖机制：RunConfig的model_settings覆盖Agent的model_settings，Agent的model_settings覆盖默认配置。

input_guardrails和output_guardrails字段——安全护栏

input_guardrails: list[InputGuardrail[TContext]] = field(default_factory=list)output_guardrails: list[OutputGuardrail[TContext]] = field(default_factory=list)

输入护栏在Agent首次运行前并行执行，用于检查输入是否合规（如是否偏离主题、是否包含有害内容）。输出护栏在Agent产出最终结果后执行，用于验证输出质量。护栏可以设置为与Agent并发执行（默认）或串行执行。

tool_use_behavior字段——工具调用策略

tool_use_behavior: (Literal["run_llm_again", "stop_on_first_tool"]    | StopAtTools    | ToolsToFinalOutputFunction) = "run_llm_again"

这是控制Agent工具调用行为的关键配置，支持四种模式：

• "run_llm_again"（默认）：工具执行后将结果反馈给LLM，LLM决定下一步
• "stop_on_first_tool"：第一个工具调用的结果直接作为最终输出
• StopAtTools字典：指定哪些工具名触发停止
• 自定义函数：完全自定义的决策逻辑

其他重要字段

hooks字段允许注册Agent级的生命周期钩子（如on_start、on_end、on_tool_start等），用于在Agent执行的关键节点插入自定义逻辑。reset_tool_choice字段默认为True，在每次工具调用后重置tool_choice为auto，防止LLM陷入无限工具调用循环。

3.3 post_init()——运行时验证

Agent的__post_init__()方法包含了大量运行时类型检查：

def__post_init__(self):ifnotisinstance(self.name, str):raise TypeError(f"Agent name must be a string, got {type(self.name).__name__}")ifself.instructions isnotNoneandnotisinstance(self.instructions, str) andnotcallable(self.instructions):raise TypeError(f"Agent instructions must be a string, callable, or None, ...")# ... 更多检查

每个字段都有严格的类型校验，确保在构造阶段就能发现配置错误，而不是等到运行时才暴露问题。这种"快速失败"的设计哲学极大地提升了开发体验。

3.4 as_tool()方法——Agent即工具

Agent类提供了一个非常精妙的as_tool()方法，可以将一个Agent转化为一个FunctionTool供其他Agent调用：

defas_tool(    self,    tool_name: str | None,    tool_description: str | None,    custom_output_extractor: Callable[[RunResult], Awaitable[str]] | None = None,    is_enabled: bool | Callable[..., MaybeAwaitable[bool]] = True,    on_stream: Callable[[AgentToolStreamEvent], MaybeAwaitable[None]] | None = None,    run_config: RunConfig | None = None,    max_turns: int | None = None,    ...) -> FunctionTool:

与Handoff模式的根本区别在于：Handoff是"委托"——新Agent接管整个对话；而as_tool是"调用"——子Agent作为工具被父Agent调用，接收结构化输入，返回结构化结果，父Agent继续控制对话。这实现了Agent之间的组合模式，允许构建任意深度的Agent调用树。

3.5 clone()方法

defclone(self, **kwargs: Any) -> Agent[TContext]:if"model"in kwargs and"model_settings"notin kwargs:if _model_settings_match_implicit_model_defaults(self.model, self.model_settings):            kwargs["model_settings"] = _initial_model_settings_for_model(kwargs["model"])return dataclasses.replace(self, **kwargs)

clone()使用dataclasses.replace()实现浅拷贝。一个贴心的细节是：当只修改model时，它会自动检测当前model_settings是否是默认值，如果是则自动更新为新模型的默认配置。这避免了用户切换模型后还需要手动重置model_settings的麻烦。

四、Runner与执行循环

Runner定义在run.py中，文件约1900行，是SDK的执行引擎入口。

4.1 Runner类——优雅的门面模式

classRunner:    @classmethodasyncdefrun(        cls,        starting_agent: Agent[TContext],input: str | list[TResponseInputItem] | RunState[TContext],        *,        context: TContext | None = None,        max_turns: int | None = DEFAULT_MAX_TURNS,        hooks: RunHooks[TContext] | None = None,        run_config: RunConfig | None = None,        error_handlers: RunErrorHandlers[TContext] | None = None,        previous_response_id: str | None = None,        auto_previous_response_id: bool = False,        conversation_id: str | None = None,        session: Session | None = None,) -> RunResult:

Runner.run()是一个类方法门面，内部将调用委托给AgentRunner实例。这种设计允许用户通过set_default_agent_runner()替换默认的Runner实现，实现自定义的执行策略。

input参数支持三种形式：纯文本字符串（最常见的用法）、OpenAI Responses格式的输入项列表（用于精细控制输入格式）、或可序列化的RunState（用于断点续传——这是企业级应用的关键特性）。

默认max_turns=10是一个精心选择的安全阀值。大多数Agent任务在10轮以内就能完成，如果超过这个限制，很可能是Agent进入了死循环。当然，对于复杂任务可以设置为None来禁用限制。

4.2 RunConfig——全局运行配置

@dataclassclassRunConfig:    model: str | Model | None = None    model_provider: ModelProvider = field(default_factory=MultiProvider)    model_settings: ModelSettings | None = None    handoff_input_filter: HandoffInputFilter | None = None    nest_handoff_history: bool = False    input_guardrails: list[InputGuardrail[Any]] | None = None    output_guardrails: list[OutputGuardrail[Any]] | None = None    tracing_disabled: bool = False    tracing: TracingConfig | None = None    workflow_name: str = "Agent workflow"    trace_id: str | None = None    group_id: str | None = None    call_model_input_filter: CallModelInputFilter | None = None    tool_error_formatter: ToolErrorFormatter | None = None    sandbox: SandboxRunConfig | None = None    tool_execution: ToolExecutionConfig | None = None    tool_not_found_behavior: ToolNotFoundBehavior = "raise_error"# ... 更多配置项

RunConfig是整个Agent运行的全局配置。它控制了模型选择、追踪设置、护栏配置、沙箱配置、工具执行策略等方方面面。与Agent级配置不同，RunConfig的设置会覆盖所有Agent的对应配置，适合用于统一管理整个工作流的行为。

4.3 执行循环核心——run_single_turn()

真正的执行循环在run_internal/run_loop.py中，文件约1950行，是SDK中最大的单文件。这个文件的复杂度主要来自以下方面：并发工具执行的协调、审批流程的处理、流式输出的管理、会话持久化的逻辑、以及大量的错误处理和恢复代码。

核心的单轮执行流程可以抽象为以下步骤：

1. 准备阶段：收集当前Agent的所有可用工具（包括MCP工具）、可用的Handoff列表、输出Schema
2. 模型调用：通过Model接口调用LLM获取响应（支持重试机制）
3. 响应处理：解析模型输出，分类为不同的操作类型（消息、工具调用、Handoff请求等）
4. 工具执行：执行所有工具调用（支持并发和审批流程）
5. 决策阶段：根据执行结果决定下一步（继续循环、产出结果、切换Agent、或暂停等待审批）

4.4 NextStep枚举——执行状态机

处理模型响应后，SDK通过NextStep枚举决定下一步行动。这个枚举是整个执行循环的状态转移函数：

classNextStepHandoff:"""模型请求移交到另一个Agent"""    agent: Agent[Any]    handoff: Handoff[Any, Any]classNextStepFinalOutput:"""模型产出了最终结果，循环应该终止"""    output: AnyclassNextStepRunAgain:"""需要再次运行模型，将工具结果反馈给LLM"""    new_items: list[RunItem]classNextStepInterruption:"""执行被中断，需要等待外部审批后才能继续"""    interruptions: list[ToolApprovalItem]

这四种状态构成了执行循环的完整状态机。Runner的主循环本质上就是一个while True循环，不断调用run_single_turn()获取SingleStepResult，然后根据其中的next_step字段决定是继续循环（RunAgain）、切换Agent（Handoff）、退出循环（FinalOutput）还是暂停（Interruption）。

4.5 流式执行——start_streaming()

除了同步执行，SDK还支持流式输出。start_streaming()函数实现了与同步版本相同的逻辑，但通过asyncio.Queue将中间事件实时推送给调用者：

asyncdefstart_streaming(    starting_input, streamed_result, starting_agent,    max_turns, hooks, context_wrapper, run_config, ...):whileTrue:# 执行单轮...# 将产生的事件推入队列        streamed_result._event_queue.put_nowait(event)

流式模式下，调用者可以通过RunResultStreaming对象实时获取Agent更新事件、工具调用事件、消息输出事件等，实现类似ChatGPT的实时交互体验。

4.6 会话持久化与断点续传

SDK支持将执行状态序列化保存，并在之后从断点恢复执行。这是通过RunState类实现的，它记录了完整的执行状态：当前Agent、已生成的条目、模型响应历史、工具使用追踪等。当执行因审批而中断时，Runner会将RunState保存下来；审批通过后，可以从保存的RunState恢复执行，而无需重新运行之前的所有步骤。

五、工具系统

工具系统定义在tool.py（约2400行）和function_schema.py（约500行）中。这是SDK中代码量最大的子系统，反映了工具调用在Agent执行中的核心地位。

5.1 Tool基类与类型层次

classTool(abc.ABC):    @property    @abc.abstractmethoddefname(self) -> str: ...

SDK定义了一个极简的Tool抽象基类，只包含一个name属性。具体的工具类型通过子类实现：

• FunctionTool：Python函数工具，最常用
• HostedMCPTool：远程MCP服务器提供的工具
• ComputerTool：计算机使用工具（模拟鼠标键盘操作）
• ShellTool：Shell命令执行工具
• LocalShellTool：本地Shell调用
• ApplyPatchTool：代码补丁应用工具
• CustomTool：自定义工具类型

每种工具类型都有独立的执行逻辑和审批支持。FunctionTool是最核心的类型，大多数用户自定义工具都属于此类。

5.2 FunctionTool——函数即工具

@dataclassclassFunctionTool(Tool):    name: str    description: str    params_json_schema: dict[str, Any]    on_invoke_tool: Callable[[RunContextWrapper[Any], str], Awaitable[Any]]    strict_json_schema: bool = True    failure_error_function: ToolErrorFunction | None = default_tool_error_function    needs_approval: bool | Callable[..., Awaitable[bool]] = False    is_enabled: bool | Callable[..., MaybeAwaitable[bool]] = True

FunctionTool的设计非常精练。on_invoke_tool是核心——它接收运行上下文和JSON字符串参数，返回工具执行结果。JSON字符串的格式与OpenAI API的tool_call参数格式完全一致，确保了与LLM输出的无缝对接。

needs_approval字段控制是否需要人工审批才能执行此工具。它可以是一个布尔值，也可以是一个动态决策函数，根据运行时上下文和工具参数决定是否需要审批。这对于敏感操作（如发送邮件、修改数据库）至关重要。

failure_error_function控制工具执行失败时的错误处理策略。默认行为是将错误转换为LLM可见的错误消息，让LLM自行决定如何处理。如果设置为None，则直接抛出异常终止执行。

5.3 function_schema.py——类型自省的魔法

这是SDK中最精巧的模块之一。它通过Python的类型注解系统和docstring自动为任意函数生成符合OpenAI API规范的工具Schema：

@dataclassclassFuncSchema:    name: str    description: str | None    params_pydantic_model: type[BaseModel]    params_json_schema: dict[str, Any]    signature: inspect.Signature    takes_context: bool = False

核心流程分为五步：

1. 使用inspect.signature()获取Python函数的签名对象
2. 使用typing.get_type_hints()解析所有类型注解（包括from __future__ import annotations的情况）
3. 使用第三方库griffe解析docstring中的Args描述，将其映射到对应的参数
4. 使用Pydantic的create_model()动态创建一个参数模型类
5. 从Pydantic模型生成严格模式的JSON Schema

# 举例：给定如下Python函数defsearch_database(query: str, limit: int = 10, filters: dict[str, Any] | None = None) -> list[dict]:"""Search the database for matching records.    Args:        query: The search query string.        limit: Maximum number of results to return.        filters: Optional filters to apply to the search.    """    ...# function_schema会自动生成如下JSON Schema：# {#   "type": "object",#   "properties": {#     "query": {"type": "string", "description": "The search query string."},#     "limit": {"type": "integer", "description": "Maximum number of results to return."},#     "filters": {"type": "object", "description": "Optional filters to apply to the search."}#   },#   "required": ["query"],#   "additionalProperties": false# }

takes_context字段是一个关键设计：如果函数的第一个参数名为context且类型为RunContextWrapper或ToolContext，SDK会自动识别并在调用时注入运行上下文，而不会将其暴露给LLM作为可调用参数。这使得工具函数可以访问运行时状态，而不需要在Schema中暴露实现细节。

六、Handoff移交机制

Handoff定义在handoffs/__init__.py中，文件约350行。它是SDK实现多Agent协作的核心机制，设计精巧而优雅。

6.1 设计哲学——Handoff即工具

Handoff的核心设计理念是：Agent之间的移交本质上是一次工具调用。当一个客服Agent判断用户的问题属于技术问题时，它不是通过某种特殊的"切换"机制来移交控制权，而是"调用"一个名为transfer_to_technical_support的"工具"。SDK在执行层识别出这是一个Handoff工具调用后，会执行Agent切换逻辑而非普通工具执行逻辑。

这种设计的美妙之处在于，对于LLM来说，移交和普通工具调用的格式完全一致，不需要任何特殊的训练或提示就能正确使用。

6.2 Handoff数据类

@dataclassclassHandoff(Generic[TContext, TAgent]):    tool_name: str    tool_description: str    input_json_schema: dict[str, Any]    on_invoke_handoff: Callable[[RunContextWrapper[Any], str], Awaitable[TAgent]]    agent_name: str    input_filter: HandoffInputFilter | None = None    nest_handoff_history: bool | None = None    strict_json_schema: bool = True    is_enabled: bool | Callable[..., MaybeAwaitable[bool]] = True    _agent_ref: weakref.ReferenceType[AgentBase[Any]] | None = field(        default=None, init=False, repr=False    )

on_invoke_handoff是核心回调——它接收运行上下文和LLM生成的JSON参数字符串，返回目标Agent实例。这个设计允许Handoff的目标是动态确定的，而非静态绑定。

_agent_ref使用弱引用持有目标Agent，避免循环引用导致的内存泄漏。这是一个看似微小但非常重要的工程细节。

6.3 handoff()工厂函数

defhandoff(    agent: Agent[TContext],    tool_name_override: str | None = None,    tool_description_override: str | None = None,    on_handoff: OnHandoffWithInput[THandoffInput] | OnHandoffWithoutInput | None = None,    input_type: type[THandoffInput] | None = None,    input_filter: Callable[[HandoffInputData], HandoffInputData] | None = None,    nest_handoff_history: bool | None = None,    is_enabled: bool | Callable[..., MaybeAwaitable[bool]] = True,) -> Handoff[TContext, Agent[TContext]]:

工厂函数通过Python的@overload装饰器支持三种调用形式：无回调（最简形式）、带输入类型和回调、不带输入类型的回调。input_type参数允许定义Handoff工具接受的结构化参数，LLM可以通过JSON传递额外信息给目标Agent。

6.4 HandoffInputFilter——对话历史过滤

HandoffInputFilter: TypeAlias = Callable[[HandoffInputData], MaybeAwaitable[HandoffInputData]]

当Agent A移交给Agent B时，默认情况下B会看到完整的对话历史。但在很多场景下这是不理想的——比如Agent A已经处理了很多细节，Agent B只需要知道最终结论。input_filter允许开发者过滤或修改传递给新Agent的对话历史，包括输入历史、移交前的条目和移交时的新条目。

6.5 Handoff执行流程

在turn_resolution.py的execute_handoffs()函数中实现了Handoff的完整执行逻辑：

asyncdefexecute_handoffs(*, public_agent, run_handoffs, ...) -> SingleStepResult:# 多个Handoff只取第一个，其余生成错误消息if multiple_handoffs:        output_message = "Multiple handoffs detected, ignoring this one."# 为其余handoff生成错误tool output    actual_handoff = run_handoffs[0]with handoff_span(from_agent=public_agent.name) as span_handoff:        new_agent = await handoff.on_invoke_handoff(            context_wrapper, actual_handoff.tool_call.arguments        )        span_handoff.span_data.to_agent = new_agent.name

一个有趣的细节是，当模型在一个turn中请求了多个Handoff时，SDK只执行第一个，其余的会收到"Multiple handoffs detected, ignoring this one"的错误消息。这是一个合理的降级策略——同时移交给多个Agent在语义上是矛盾的。

七、Guardrail护栏系统

护栏系统是SDK的安全机制，定义在guardrail.py中。它实现了"绊线"（tripwire）模式——当检查不通过时，立即停止Agent执行。

7.1 核心数据结构

@dataclassclassGuardrailFunctionOutput:    output_info: Any# 检查结果的详细信息    tripwire_triggered: bool# 是否触发了绊线@dataclassclassInputGuardrail(Generic[TContext]):    guardrail_function: Callable[        [RunContextWrapper[TContext], Agent[Any], str | list[TResponseInputItem]],        MaybeAwaitable[GuardrailFunctionOutput],    ]    name: str | None = None    run_in_parallel: bool = True# 默认与Agent并发执行

护栏函数接收三个参数：运行上下文、当前Agent实例、以及输入文本或输入项列表。它返回GuardrailFunctionOutput，其中output_info可以包含任意检查详情（用于日志和调试），tripwire_triggered为True时触发异常终止。

7.2 并发执行——关键的性能优化

这是护栏系统最精妙的设计。默认的run_in_parallel=True意味着输入护栏检查与Agent的模型调用是并发执行的，而非串行等待：

asyncdefrun_input_guardrails(agent, guardrails, input, context):"""Run input guardrails concurrently and raise on tripwires."""    guardrail_tasks = [        asyncio.create_task(run_single_input_guardrail(agent, g, input, context))for g in guardrails    ]try:for done in asyncio.as_completed(guardrail_tasks):            result = await doneif result.output.tripwire_triggered:# 第一个触发的护栏立即取消所有其他任务for t in guardrail_tasks:                    t.cancel()await asyncio.gather(*guardrail_tasks, return_exceptions=True)raise InputGuardrailTripwireTriggered(result)            guardrail_results.append(result)except BaseException:for t in guardrail_tasks:ifnot t.done():                t.cancel()await asyncio.gather(*guardrail_tasks, return_exceptions=True)raise

使用asyncio.as_completed()确保第一个触发的护栏能立即终止所有其他正在运行的检查。这不仅节省了计算资源，更重要的是缩短了安全响应时间——在安全场景中，每一毫秒都很重要。

流式模式下的护栏检查更加复杂。run_input_guardrails_with_queue()函数需要将护栏结果推入事件队列，让调用者能够实时获知护栏状态：

asyncdefrun_input_guardrails_with_queue(agent, guardrails, input, context, streamed_result, parent_span):    queue = streamed_result._input_guardrail_queue    guardrail_tasks = [...]for done in asyncio.as_completed(guardrail_tasks):        result = await doneif result.output.tripwire_triggered:            streamed_result._triggered_input_guardrail_result = result            queue.put_nowait(result)  # 推入队列通知调用者# 取消其他任务...break        queue.put_nowait(result)

7.3 装饰器语法糖

SDK提供了@input_guardrail和@output_guardrail装饰器，支持直接使用（无括号）和带参数使用两种形式：

@input_guardraildefcheck_topic(ctx, agent, input):"""直接使用，函数名自动作为护栏名称"""    ...@input_guardrail(name="my_guard", run_in_parallel=False)asyncdefdetailed_check(ctx, agent, input):"""带参数：自定义名称，串行执行"""    ...

7.4 四种护栏类型

SDK支持四种护栏类型，覆盖Agent执行的各个阶段：

1. InputGuardrail：在Agent首次运行前检查输入
2. OutputGuardrail：在Agent产出最终结果后检查输出
3. ToolInputGuardrail：在工具调用前检查工具参数
4. ToolOutputGuardrail：在工具执行后检查工具结果

InputGuardrail和OutputGuardrail是全局性质的，通过Agent或RunConfig配置。ToolInputGuardrail和ToolOutputGuardrail则更加细粒度，可以针对单个工具进行安全检查。

八、Model接口——模型抽象层

模型抽象定义在models/interface.py中，文件约155行。它是SDK支持多模型提供商的基础。

8.1 抽象Model接口

classModel(abc.ABC):asyncdef_cleanup_on_run_end(self, owner: object) -> None:returnNoneasyncdefclose(self) -> None:returnNonedefget_retry_advice(self, request: ModelRetryAdviceRequest) -> ModelRetryAdvice | None:returnNone    @abc.abstractmethodasyncdefget_response(        self,        system_instructions: str | None,input: str | list[TResponseInputItem],        model_settings: ModelSettings,        tools: list[Tool],        output_schema: AgentOutputSchemaBase | None,        handoffs: list[Handoff],        tracing: ModelTracing,        *,        previous_response_id: str | None,        conversation_id: str | None,        prompt: ResponsePromptParam | None,) -> ModelResponse:pass    @abc.abstractmethoddefstream_response(self, ...) -> AsyncIterator[TResponseStreamEvent]:pass

Model接口定义了两个核心抽象方法和三个可选的生命周期方法。get_response()一次性获取完整模型响应，stream_response()以异步迭代器的形式流式获取响应事件。两个方法接收完全相同的参数，确保行为一致性。

_cleanup_on_run_end()在Runner执行结束后被调用，用于释放运行期间持有的资源（如持久连接）。close()用于释放Model实例持有的所有资源。get_retry_advice()允许Model实现提供提供商特定的重试建议（如重试延迟、是否安全重试等）。

8.2 ModelTracing——追踪级别控制

classModelTracing(enum.Enum):    DISABLED = 0# 完全禁用追踪    ENABLED = 1# 启用追踪，包含所有数据    ENABLED_WITHOUT_DATA = 2# 启用追踪，但不包含输入输出数据

三级追踪控制让用户可以精确平衡可观测性和隐私保护。在处理敏感数据时，ENABLED_WITHOUT_DATA模式可以保留调用结构信息而隐藏具体内容。

8.3 ModelProvider与MultiProvider

classModelProvider(abc.ABC):    @abc.abstractmethoddefget_model(self, model_name: str | None) -> Model:pass

ModelProvider是一个简单的工厂接口。SDK默认使用MultiProvider，它支持同时注册多个ModelProvider，根据模型名称前缀路由到对应的提供商。这意味着在同一个工作流中可以混合使用OpenAI、Anthropic、Google等不同提供商的模型——前端Agent使用GPT-4o，后端Agent使用Claude，完全透明。

九、Tracing追踪系统

SDK实现了一套完全自定义的追踪系统，而非业界标准的OpenTelemetry。这个选择是有意为之的——自定义方案更轻量，且可以与OpenAI的追踪后端深度集成，提供更精确的语义建模。

9.1 三层抽象架构

追踪系统的核心是三层抽象：

Trace（工作流级别）  └── Span（操作级别）       └── SpanData（操作特定数据）

一个Trace代表一个完整的端到端工作流（如"处理客户投诉"），其中包含多个Span（如"理解意图"、"查询数据库"、"生成回复"），每个Span携带特定类型的SpanData（如LLM调用的输入输出、工具调用的参数和结果等）。

9.2 Trace——工作流追踪

classTrace(abc.ABC):    @abc.abstractmethoddefstart(self, mark_as_current: bool = False): ...    @abc.abstractmethoddeffinish(self, reset_current: bool = False): ...    @property    @abc.abstractmethoddeftrace_id(self) -> str: ...    @property    @abc.abstractmethoddefname(self) -> str: ...    @abc.abstractmethoddefexport(self) -> dict[str, Any] | None: ...

Trace支持Python上下文管理器协议（with trace("name") as t:），通过contextvars实现线程安全的当前Trace追踪。即使在复杂的异步并发场景中，每个协程也能正确地关联到自己的Trace。

SDK提供了三种Trace实现：

• TraceImpl：正常工作的追踪实现，将数据发送给TracingProcessor
• NoOpTrace：追踪禁用时的空操作实现，所有方法都是no-op但不报错
• ReattachedTrace：从持久化状态恢复的Trace，用于断点续传场景

9.3 13种SpanData类型

这是追踪系统最丰富的部分。SDK定义了13种精细化的Span类型，覆盖了Agent执行的所有关键操作：

classAgentSpanData(SpanData):       # type = "agent" —— Agent运行classTaskSpanData(SpanData):        # type = "task" —— 顶层Runner运行classTurnSpanData(SpanData):        # type = "turn" —— 单轮执行循环classFunctionSpanData(SpanData):    # type = "function" —— 函数工具调用classGenerationSpanData(SpanData):  # type = "generation" —— LLM文本生成classResponseSpanData(SpanData):    # type = "response" —— API响应classHandoffSpanData(SpanData):     # type = "handoff" —— Agent移交classCustomSpanData(SpanData):      # type = "custom" —— 自定义操作classGuardrailSpanData(SpanData):   # type = "guardrail" —— 护栏检查classTranscriptionSpanData(SpanData): # type = "transcription" —— 语音转文字classSpeechSpanData(SpanData):      # type = "speech" —— 文字转语音classSpeechGroupSpanData(SpanData): # type = "speech_group" —— 语音组classMCPListToolsSpanData(SpanData): # type = "mcp_tools" —— MCP工具列表

每种SpanData精确记录了对应操作的关键信息。例如GenerationSpanData记录了input（输入消息列表）、output（输出消息列表）、model（模型名称）、model_config（模型配置）、usage（token使用量）；HandoffSpanData记录了from_agent（源Agent名称）和to_agent（目标Agent名称）。

最后四种Span类型（Transcription、Speech、SpeechGroup、MCPListTools）反映了SDK对语音交互和MCP协议的深度支持，这些在传统Agent框架中是看不到的。

9.4 TracingProcessor接口

classTracingProcessor(abc.ABC):    @abc.abstractmethoddefon_trace_start(self, trace: Trace) -> None: ...    @abc.abstractmethoddefon_trace_end(self, trace: Trace) -> None: ...    @abc.abstractmethoddefon_span_start(self, span: Span[Any]) -> None: ...    @abc.abstractmethoddefon_span_end(self, span: Span[Any]) -> None: ...    @abc.abstractmethoddefshutdown(self) -> None: ...    @abc.abstractmethoddefforce_flush(self) -> None: ...

用户可以通过实现TracingProcessor接口将追踪数据导出到任意后端（如DataDog、Jaeger、自建系统等）。SDK默认提供BatchTraceProcessor，在后台线程中批量异步导出追踪数据到OpenAI追踪后端，最小化对Agent执行性能的影响。

十、扩展生态系统

SDK的核心之外还有丰富的扩展模块，提供了从MCP协议集成到沙箱执行的全方位能力。

10.1 MCP（Model Context Protocol）集成

MCP是Anthropic提出的一种标准化模型上下文协议，允许LLM通过统一接口访问外部工具和数据源。SDK原生支持MCP，Agent可以直接连接MCP服务器并自动发现可用工具：

asyncdefget_mcp_tools(self, run_context):    convert_schemas_to_strict = self.mcp_config.get("convert_schemas_to_strict", False)returnawait MCPUtil.get_all_function_tools(self.mcp_servers,        convert_schemas_to_strict,        run_context,self,        failure_error_function=failure_error_function,        include_server_in_tool_names=include_server_in_tool_names,        reserved_tool_names=reserved_tool_names,    )

mcp_config支持三个配置项：convert_schemas_to_strict控制是否将MCP工具的JSON Schema转换为严格模式（提高LLM输出正确率）；failure_error_function定义MCP工具调用失败时的错误处理策略；include_server_in_tool_names控制是否在工具名中包含服务器前缀以避免多服务器间的名称冲突。

MCPServerManager提供了MCP服务器的生命周期管理，确保connect和cleanup在同一任务上下文中配对执行，避免资源泄漏。

10.2 Memory/Session会话管理

Session抽象基类定义了会话持久化的接口，支持将对话历史存储到外部系统（如Redis、数据库等）。SDK自动管理Session的读写：

# Runner.run()中的session参数asyncdefrun(cls, starting_agent, input, *, session: Session | None = None, ...):# 自动从session加载历史    prepared_input, session_items = await prepare_input_with_session(input, session, ...    )# ... 执行 ...# 自动保存结果到sessionawait save_result_to_session(session, ...)

开发者只需要实现Session接口并传入Runner，就可以获得完整的会话管理能力，无需手动处理历史的加载、保存和去重。

10.3 Sandbox沙箱执行

沙箱系统是SDK最具野心的扩展，允许Agent在安全的隔离环境中执行代码。SDK支持四种沙箱后端：

Docker容器沙箱：在本地Docker容器中执行Agent生成的代码，适合需要完全控制执行环境的场景。

E2B云端沙箱：使用E2B提供的云端代码执行环境，无需本地Docker，适合SaaS产品。

Modal无服务器沙箱：利用Modal的无服务器计算平台，按需启动沙箱实例，按使用量计费。

Cloudflare Workers沙箱：在Cloudflare的边缘网络上执行代码，延迟极低，适合全球分布式场景。

沙箱配置通过RunConfig.sandbox统一管理：

@dataclassclassSandboxRunConfig:    client: BaseSandboxClient[Any] | None = None# 沙箱客户端    session: BaseSandboxSession | None = None# 已有会话    manifest: Manifest | None = None# 文件清单    snapshot: SnapshotSpec | SnapshotBase | None = None# 快照    concurrency_limits: SandboxConcurrencyLimits = ...   # 并发限制    archive_limits: SandboxArchiveLimits | None = None# 归档限制

Manifest系统允许声明式地定义沙箱中的文件结构，SDK负责自动物化（materialize）这些文件到沙箱中。支持从本地目录、归档文件或远程URL等多种来源获取文件。

10.4 Realtime Agent

realtime/模块提供了基于WebSocket的实时Agent能力。与标准Agent的请求-响应模式不同，RealtimeAgent支持持续的流式输入输出，适合语音对话、实时协作等场景。

10.5 Voice Pipeline

voice/模块提供完整的语音交互管线，包括：

• STT（Speech-to-Text）：将用户语音转换为文本
• TTS（Text-to-Speech）：将Agent回复转换为语音

语音管线与追踪系统深度集成，每个语音操作都有对应的SpanData类型（TranscriptionSpanData、SpeechSpanData、SpeechGroupSpanData），确保全链路可观测。

十一、独特设计决策总结

通过数千行源码的深入分析，以下是OpenAI Agents SDK最值得关注的设计决策：

泛型上下文TContext——整个框架围绕泛型上下文构建。RunContextWrapper包装TContext并在整个执行链中传递，工具函数、护栏、钩子都能访问到类型安全的上下文对象。这让SDK在保持极高灵活性的同时不失类型安全。

Dataclass优于继承——Agent、Handoff、Tool等核心类都使用@dataclass而非传统OOP继承。这使得它们天然支持浅拷贝（clone）、序列化（RunState）、以及清晰的字段定义。

内部/公开API严格分离——run_internal/目录包含所有非公开API，通过__all__列表精确控制导出。这给了SDK团队重构内部实现的极大自由度。

异步优先，同步为辅——所有核心路径都是异步的，同步调用只是在内部创建事件循环。这确保了在高并发场景下的最佳性能。

工具即一等公民——Handoff是工具，Agent as_tool也是工具，审批请求也通过工具机制处理。这种统一性是SDK最核心的设计哲学。

可中断的执行流——NextStepInterruption和ToolApprovalItem实现了执行流的优雅暂停与恢复，配合RunState的序列化能力，支持企业级的审批工作流。

自定义追踪优于OTel——虽然牺牲了OpenTelemetry生态的兼容性，但换来了更精确的语义建模（13种SpanData类型）和更紧密的OpenAI平台集成。

十二、RunContextWrapper与运行时上下文传递

运行时上下文是SDK中一个非常重要但容易被忽略的概念。RunContextWrapper定义在run_context.py中，它是整个Agent执行过程中上下文信息的载体和传递机制。

12.1 RunContextWrapper的核心结构

RunContextWrapper的核心职责有三个。第一，它包装了用户提供的泛型上下文对象TContext，这个对象可以是任何Python类型——一个数据库连接、一个用户会话对象、或者一个包含业务状态的dataclass。SDK在整个执行链中传递这个wrapper，确保所有组件都能访问到相同的上下文实例。

第二，它追踪整个运行过程中的token使用量。Usage对象记录了输入token数、输出token数和累计调用次数，每次模型调用后都会自动更新。这对于成本控制和性能监控至关重要，开发者可以通过context_wrapper.usage随时获取当前的资源消耗情况。

第三，它管理工具审批状态。内部的ApprovalRecord记录了哪些工具调用已被批准、哪些被拒绝、以及拒绝的原因消息。这个设计使得审批决策可以在执行过程中持久化，并在断点续传时恢复，实现了企业级的审批工作流支持。

12.2 上下文传递机制

在SDK的执行链中，RunContextWrapper像一条隐形的线贯穿始终。从Runner.run()开始创建wrapper，到run_single_turn()传递给模型调用，到工具执行时注入ToolContext，到护栏检查时传入guardrail函数——每一步都能访问到同一个wrapper实例。

这种设计的一个重要好处是状态共享。不同的工具函数可以通过wrapper共享状态，而不需要通过全局变量或复杂的参数传递。例如，一个搜索工具可以将搜索结果缓存在context中，后续的分析工具可以直接使用缓存的结果，避免重复调用外部API。

13.3 AgentHookContext专用上下文

除了RunContextWrapper，SDK还定义了AgentHookContext专门用于生命周期钩子。它包含了当前的上下文、使用量、审批状态和当前轮次的输入，为钩子函数提供了完整的执行快照。钩子函数可以通过这个上下文对象做出智能决策，例如根据当前的token使用量决定是否需要压缩对话历史。

十三、Result结果类与流式输出

result.py定义了Agent执行的结果类型，包括同步的RunResult和流式的RunResultStreaming。这两个类的设计体现了SDK对开发者体验的极致追求。

13.1 RunResult的丰富信息

RunResult包含了执行的所有信息：原始输入、生成的条目列表、模型的原始响应、最终输出、所有护栏检查结果、以及运行上下文。final_output是最常用的字段，它直接包含了Agent的最终输出结果。如果设置了output_type，这个字段会是对应的Python类型实例（如dataclass对象或Pydantic模型实例）；如果没有设置，就是纯字符串。

new_items列表包含了执行过程中生成的所有条目——消息输出、工具调用、工具结果、推理项等。这对于调试和审计非常有用，开发者可以回溯Agent执行的每一步操作。raw_responses保存了模型的原始API响应，包含了完整的token使用量、模型名称等元信息。

13.2 流式事件体系

RunResultStreaming通过asyncio.Queue实现了事件驱动的流式输出。开发者可以通过stream_events()异步迭代器实时获取执行过程中的所有事件。流式事件包括多种类型：RawResponsesStreamEvent是模型的原始输出流，用于实现打字机效果；RunItemStreamEvent是运行条目事件，如工具调用、消息输出等；AgentUpdatedStreamEvent是Agent切换事件，当Handoff发生时触发。

RunItemStreamEvent是最丰富的事件类型，通过name字段区分不同的事件类型。message_output_created表示Agent生成了文本消息，tool_called表示Agent调用了工具，tool_output表示工具返回了结果，handoff_requested表示Agent请求移交给另一个Agent，reasoning_item_created表示推理模型产生了推理步骤。这些事件类型涵盖了Agent执行的所有关键节点，开发者可以根据需要过滤和处理特定类型的事件。

十四、异常层次结构与错误恢复

SDK定义了一套完善的异常层次结构。AgentsException是所有SDK异常的基类，其下有MaxTurnsExceeded（超过最大轮次限制）、ModelBehaviorError（模型输出格式错误等行为异常）、UserError（用户配置错误）等子类。

特别值得关注的是四类护栏触发异常：InputGuardrailTripwireTriggered、OutputGuardrailTripwireTriggered、ToolInputGuardrailTripwireTriggered和ToolOutputGuardrailTripwireTriggered。每个护栏异常都携带了对应的检查结果（guardrail_result），包含触发的护栏名称、检查详情等信息，方便开发者实现自定义的安全响应策略。

SDK还支持通过RunErrorHandlers注册自定义错误处理器。开发者可以针对不同类型的错误注册不同的处理策略，例如在模型输出格式错误时自动重试、在护栏触发时返回友好的错误消息、在超过轮次限制时返回当前的中间结果等。这种灵活的错误处理机制使得Agent应用能够在各种异常情况下优雅降级，而不是简单地崩溃。

十五、lifecycle生命周期钩子的精确时机

SDK提供了两层生命周期钩子：RunHooks是运行级别钩子，在整个运行期间有效，不管当前活跃的是哪个Agent；AgentHooks是Agent级别钩子，只在其绑定的Agent活跃时触发。这种双层设计使得开发者既可以在全局层面监控执行（如统一的日志记录和性能追踪），也可以在单个Agent层面插入特定逻辑。

钩子函数的执行时机经过精心设计。on_agent_start在Agent的第一轮模型调用之前触发，确保钩子函数可以修改Agent的初始状态。on_agent_end在Agent产出最终结果之后、输出护栏执行之前触发，使得钩子函数可以对最终结果做最后的检查或转换。on_tool_start在工具函数被调用之前触发，允许钩子函数记录工具调用的开始时间或做参数预处理。on_handoff在旧Agent的on_agent_end之后、新Agent的on_agent_start之前触发，确保移交过程中的状态转换是有序的。

on_handoff钩子特别有用，它可以用于记录Agent之间的移交日志、在移交时清理临时状态、或者根据业务逻辑动态修改目标Agent的配置。在客服场景中，这个钩子可以用来记录客户从哪个部门转到了哪个部门，方便后续的服务质量分析。

十六、Prompts系统与外部配置

SDK的prompts.py模块实现了一套与OpenAI平台深度集成的Prompt管理系统。Prompt对象可以从OpenAI平台动态拉取最新的Prompt配置，实现代码外的Agent行为管理。

PromptUtil负责将Prompt配置解析为SDK可使用的格式，包括提取系统提示词、工具列表、模型设置等。DynamicPromptFunction则支持在运行时动态生成Prompt配置，根据当前的上下文和Agent状态返回不同的配置。

这种设计的核心价值在于解耦。Agent的行为定义可以在OpenAI平台上独立管理，产品经理可以直接在平台上调整Agent的指令和工具，而不需要修改代码和重新部署。同时，代码中仍然可以覆盖平台配置，实现环境特定的定制。这在企业级AI应用的迭代中非常实用，大大加速了Agent行为的调优周期。

十七、工具审批机制深度解析

工具审批是SDK中一个非常重要的安全特性，它允许在执行敏感操作之前暂停Agent的执行并等待人工确认。这个机制的实现贯穿了tool.py、run_internal/tool_execution.py和run_internal/approvals.py三个核心模块。

17.1 审批声明与决策

FunctionTool的needs_approval字段支持两种形式：布尔值和异步决策函数。当设置为True时，该工具的每次调用都需要人工审批。当设置为决策函数时，SDK会在每次工具调用时调用该函数，根据运行上下文和工具参数动态决定是否需要审批。

决策函数的签名接收三个参数：运行上下文、工具参数字典和工具名称字符串，返回一个布尔值。这使得审批逻辑可以非常精细——例如，只有当邮件发送的目标地址是外部域名时才需要审批，内部邮件则自动放行。

17.2 审批暂停与恢复

当工具调用需要审批时，Runner会暂停当前的执行循环，将所有待审批的工具调用收集到NextStepInterruption中返回给调用者。调用者（通常是前端应用或审批系统）在获得用户确认后，通过RunState恢复执行。

这个过程中，SDK需要处理大量的边界情况。例如，当多个工具调用同时需要审批时，它们应该被一起展示给用户。当部分工具被批准、部分被拒绝时，被拒绝的工具需要生成合适的错误消息反馈给LLM。当执行恢复时，已经被批准的工具不应该再次执行，而是直接使用之前缓存的结果。

17.3 审批与流式模式

在流式模式下，审批机制更加复杂。Runner需要在检测到审批需求时立即暂停流式输出，将审批项推入事件队列通知前端，然后等待前端的审批决策。审批通过后，Runner需要从断点恢复流式输出。SDK通过RunResultStreaming的interruptions字段和事件队列实现了这一复杂的交互流程。

17.4 审批状态的持久化

SDK的审批状态管理支持跨会话持久化。ApprovalRecord中的approved和rejected字段既可以是布尔值（表示永久批准或拒绝），也可以是特定调用ID的列表（表示仅对特定的工具调用生效）。这种细粒度的审批状态管理确保了在断点续传场景下，审批决策能够被准确地恢复。

当审批被拒绝时，SDK会调用tool_error_formatter来生成合适的错误消息。开发者可以通过RunConfig.tool_error_formatter自定义错误消息的格式，使其对LLM更加友好，引导LLM尝试其他方式完成任务而不是简单地放弃。

stickey_rejection_message字段记录了最近一次拒绝的详细原因，当同一工具被多次拒绝时，后续的错误消息会引用之前的原因，帮助LLM理解为什么被拒绝以及应该如何调整策略。这种智能化的错误反馈机制大大提升了Agent在遇到审批拒绝时的自我纠错能力。

总结

OpenAI Agents SDK的源码展现了一个精心设计的多Agent框架应有的样子：核心概念极少但表达力极强，内部实现精密复杂但对外API简洁易用。从@dataclass装饰的Agent到近两千行的执行循环，从十三种追踪Span类型到四种沙箱后端，每一个设计决策都在灵活性和复杂度之间找到了精妙的平衡点。

对于想要构建多Agent应用的开发者来说，理解这个框架的源码不仅能帮助更好地使用它，更能启发如何在自己的项目中设计Agent编排系统。在本系列的后续文章中，我们将继续深入剖析Google ADK和Anthropic Claude Agent SDK的源码，对比三大框架在架构设计、执行模型和扩展能力上的异同。

引用链接