数据截至 (上游 commit 2c5560339cd7)
OpenAI Agents SDK (Python) — 架构与原理
30 秒导读: 这是 OpenAI 官方的 Python 框架,用来搭建「多个 AI agent 协作完成任务」的工作流。它本质上就是一个循环:把对话喂给 LLM → 看 LLM 想干嘛(说话?调工具?换个 agent 接手?)→ 执行 → 再喂回去,直到产出最终答案。它不绑定 OpenAI 模型,100+ 模型都能接。
1. 这是什么(零基础也能懂)
一句话定义: OpenAI Agents SDK 是一个 Python 库,帮你把「一个或多个由 LLM 驱动、能调用工具的 agent」编排成一个能自动运行到底的工作流。
解决什么问题 / 给谁用: 假设你想做一个客服机器人——用户问问题,AI 要么直接答,要么查数据库(调工具),要么转给「退款专员 agent」处理。你不想自己手写「调一次模型、解析它要调哪个函数、跑函数、把结果塞回去、再调模型」这一长串胶水代码。这个 SDK 就是那层胶水,而且把它做得很薄、很清晰。
它的几个核心概念(对外暴露的积木):
| 概念 | 白话 | 在哪 |
|---|---|---|
Agent | 一个配好「指令 + 工具 + 交接对象」的 LLM | agent.py:269 |
Runner | 把 agent 跑起来的引擎(Runner.run(...)) | run.py:234 |
工具(function_tool) | 把普通 Python 函数变成 LLM 能调的工具 | tool.py:2509 |
交接(handoff) | 一个 agent 把任务转交给另一个 agent | handoffs/__init__.py:260 |
护栏(guardrail) | 在输入/输出上跑的安全校验,触发就中断 | guardrail.py:71 |
会话(Session) | 自动管理多轮对话历史 | memory/session.py:16 |
追踪(tracing) | 自动记录整个 run 的每一步,可调试可观测 | tracing/__init__.py |
用起来什么样: 一个最小例子——定义一个带工具的 agent,然后 Runner.run_sync 跑它:
# 示意,基于 README 的真实 API 形态
from agents import Agent, Runner, function_tool
@function_tool # 把这个函数变成 LLM 可调用的工具
def get_weather(city: str) -> str: # 类型注解会被自动转成 JSON schema
"""查某城市天气。""" # docstring 变成给 LLM 看的工具描述
return f"{city} 今天晴。"
agent = Agent(
name="助手",
instructions="你是一个乐于助人的助手。", # 这就是 system prompt
tools=[get_weather],
)
result = Runner.run_sync(agent, "北京天气怎么样?")
print(result.final_output) # → 模型先调 get_weather,拿到结果后再总结回答
一句话直觉/类比: 把 Runner.run() 想成一个 「带工具箱的对话回合制游戏」:每个回合,LLM 看完当前局面后只能做三件事之一——说出最终答案(游戏结束)、调用工具(执行后进入下回合)、把话筒交给另一个 agent(换人接着玩)。SDK 就是那个忠实执行规则、记分、记录回放的「裁判」。
2. 顶层全景(它大概怎么转)
整个库的价值集中在一个循环里。下面这张图是控制流主线(从左到右一个回合,循环回到顶部直到出最终输出):
┌──────────────────────────── 一次 Runner.run() ────────────────────────────┐
│ │
输入 ─┤ [准备] 拼 system prompt + 历史 + 工具列表 (run_single_turn) │
│ │ │
│ ▼ │
│ [调模型] get_new_response → ModelResponse │
│ │ │
│ ▼ │
│ [解析输出] process_model_response │
│ 分拣模型想干的事: 文本 / 函数调用 / 交接 / 计算机操作 / MCP 审批… │
│ │ │
│ ▼ │
│ [执行+决定下一步] execute_tools_and_side_effects │
│ ┌─────────────┬──────────────┬───────────────┬─────────────────┐ │
│ ▼ ▼ ▼ ▼ ▼ │
│ 有交接? 工具结果即 纯文本/结构 要审批? 否则 │
│ →换 agent 最终输出? 化输出 → →再跑 │
│ (Handoff) (FinalOutput) (FinalOutput) (Interruption) (RunAgain) │
│ │ │ │ │ │ │
└─────┴─────────────┴──────────────┴───────────────┴──────┬──────────┘ │
│ FinalOutput/Interruption: 跑输出护栏, 返回 RunResult │
└── RunAgain/Handoff: 回到顶部, 进入下一回合 ──────────────────────────┘
怎么读这张图: 中间那行五个分支,就是 SDK 对「模型这一回合到底产出了什么」的五种归类,对应代码里五个 NextStep* 类型(run_steps.py:155-175)。每回合最多走一个分支;走到 FinalOutput 或 Interruption 就结束 run,走到 RunAgain 或 Handoff 就再循环。
部件一句话职责:
| 部件 | 干什么 | 在哪个文件 |
|---|---|---|
Runner / AgentRunner | 对外入口 + 那个大 while 循环 | run.py:234 / run.py:523 |
run_single_turn | 一个回合:准备输入→调模型→把响应转成「下一步」 | run_internal/run_loop.py:2221 |
process_model_response | 把模型原始输出分拣成 函数/交接/计算机操作… | run_internal/turn_resolution.py:2684 |
execute_tools_and_side_effects | 真正跑工具,并算出 NextStep* | run_internal/turn_resolution.py:784 |
Model 接口 | 抽象掉「具体哪个 LLM」 | models/interface.py:37 |
RunResult / RunState | 结果与可序列化的运行状态(支持中断恢复) | result.py:334 / run_state.py |
主线走一遍(高层): Runner.run(agent, "...") → 进入 AgentRunner.run 的 while 循环(run.py:938)→ 每圈调 run_single_turn 拿到一个 SingleStepResult → 根据它的 next_step 字段决定:NextStepFinalOutput 就跑输出护栏后返回 RunResult;NextStepHandoff 就换 current_agent 再循环;NextStepRunAgain 直接再循环;NextStepInterruption 则打包成「待审批」的 RunResult 提前返回。
3. 阅读地图(按这个顺序往下读)
这个项目很大(光 run_internal/ 就上万行),但价值高度集中。建议顺序:
- 01-agent-loop.md — 先读这个。核心 while 循环 + 五种「下一步」决策。读懂它,整个库就懂了一半。
- 02-tools.md — 工具系统。
@function_tool如何用反射把函数签名变成 JSON schema,执行时如何校验、调用、容错。 - 03-handoffs-guardrails.md — 多 agent 协作(handoff 本质是「返回新 agent 的特殊工具」)与安全护栏(并行/前置两种时序)。
- 04-sessions-tracing-hitl.md — 三个横切机制:会话记忆、追踪、人在环路(工具审批的中断/恢复)。
4. 边界与本文范围
本文聚焦核心编排引擎:agent 循环、工具、交接、护栏、会话、追踪、人在环路。SDK 还有几大块本文未深入(只在相关处点到):
- Sandbox Agents(
sandbox/):让 agent 在受控容器/文件系统里长时间干活,是 0.14+ 的新主打,但工程量巨大,自成体系。 - Realtime / Voice(
realtime/、voice/):语音实时 agent(WebSocket)。 - MCP(
mcp/):Model Context Protocol 服务器集成,只讲它如何并入工具列表。 - 模型适配层细节(
models/openai_responses.py、chatcmpl_*):Responses API 与 Chat Completions 的双向转换、重试,本文只讲Model接口契约。
5. 代码地图(导航索引)
| 主题 | 文件路径 | 符号名 |
|---|---|---|
| 对外入口 | src/agents/run.py | Runner.run、Runner.run_streamed、Runner.run_sync |
| 核心 while 循环 | src/agents/run.py | AgentRunner.run(while 在 run.py:938) |
| 一个回合 | src/agents/run_internal/run_loop.py | run_single_turn、get_new_response |
| 响应分拣 | src/agents/run_internal/turn_resolution.py | process_model_response |
| 执行+下一步 | src/agents/run_internal/turn_resolution.py | execute_tools_and_side_effects、check_for_final_output_from_tools |
| 「下一步」类型 | src/agents/run_internal/run_steps.py | NextStepHandoff、NextStepFinalOutput、NextStepRunAgain、NextStepInterruption |
| Agent 定义 | src/agents/agent.py | Agent、AgentBase、Agent.as_tool |
| 工具 | src/agents/tool.py | function_tool、FunctionTool |
| 工具 schema | src/agents/function_schema.py | function_schema、FuncSchema |
| 交接 | src/agents/handoffs/__init__.py | handoff、Handoff |
| 护栏 | src/agents/guardrail.py | InputGuardrail、OutputGuardrail |
| 会话 | src/agents/memory/session.py | Session、SessionABC |
| 模型接口 | src/agents/models/interface.py | Model、ModelProvider |
| 追踪 | src/agents/tracing/__init__.py | agent_span、generation_span、trace |
| 运行状态 | src/agents/run_state.py | RunState、RunState.approve、RunState.reject |