跳到主要内容

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一个配好「指令 + 工具 + 交接对象」的 LLMagent.py:269
Runner把 agent 跑起来的引擎(Runner.run(...))run.py:197
工具(function_tool)把普通 Python 函数变成 LLM 能调的工具tool.py:1899
交接(handoff)一个 agent 把任务转交给另一个 agenthandoffs/__init__.py:225
护栏(guardrail)在输入/输出上跑的安全校验,触发就中断guardrail.py:71
会话(Session)自动管理多轮对话历史memory/session.py:14
追踪(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:154-174)。每回合最多走一个分支;走到 FinalOutput 或 Interruption 就结束 run,走到 RunAgain 或 Handoff 就再循环。

部件一句话职责:

部件干什么在哪个文件
Runner / AgentRunner对外入口 + 那个大 while 循环run.py:197 / run.py:444
run_single_turn一个回合:准备输入→调模型→把响应转成「下一步」run_internal/run_loop.py:1708
process_model_response把模型原始输出分拣成 函数/交接/计算机操作…run_internal/turn_resolution.py:1551
execute_tools_and_side_effects真正跑工具,并算出 NextStep*run_internal/turn_resolution.py:629
Model 接口抽象掉「具体哪个 LLM」models/interface.py:37
RunResult / RunState结果与可序列化的运行状态(支持中断恢复)result.py:334 / run_state.py

主线走一遍(高层): Runner.run(agent, "...") → 进入 AgentRunner.run 的 while 循环(run.py:767)→ 每圈调 run_single_turn 拿到一个 SingleStepResult → 根据它的 next_step 字段决定:NextStepFinalOutput 就跑输出护栏后返回 RunResult;NextStepHandoff 就换 current_agent 再循环;NextStepRunAgain 直接再循环;NextStepInterruption 则打包成「待审批」的 RunResult 提前返回。

3. 阅读地图(按这个顺序往下读)​

这个项目很大(光 run_internal/ 就上万行),但价值高度集中。建议顺序:

  1. 01-agent-loop.md — 先读这个。核心 while 循环 + 五种「下一步」决策。读懂它,整个库就懂了一半。
  2. 02-tools.md — 工具系统。@function_tool 如何用反射把函数签名变成 JSON schema,执行时如何校验、调用、容错。
  3. 03-handoffs-guardrails.md — 多 agent 协作(handoff 本质是「返回新 agent 的特殊工具」)与安全护栏(并行/前置两种时序)。
  4. 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.pyRunner.run、Runner.run_streamed、Runner.run_sync
核心 while 循环src/agents/run.pyAgentRunner.run(while 在 run.py:767)
一个回合src/agents/run_internal/run_loop.pyrun_single_turn、get_new_response
响应分拣src/agents/run_internal/turn_resolution.pyprocess_model_response
执行+下一步src/agents/run_internal/turn_resolution.pyexecute_tools_and_side_effects、check_for_final_output_from_tools
「下一步」类型src/agents/run_internal/run_steps.pyNextStepHandoff、NextStepFinalOutput、NextStepRunAgain、NextStepInterruption
Agent 定义src/agents/agent.pyAgent、AgentBase、Agent.as_tool
工具src/agents/tool.pyfunction_tool、FunctionTool
工具 schemasrc/agents/function_schema.pyfunction_schema、FuncSchema
交接src/agents/handoffs/__init__.pyhandoff、Handoff
护栏src/agents/guardrail.pyInputGuardrail、OutputGuardrail
会话src/agents/memory/session.pySession、SessionABC
模型接口src/agents/models/interface.pyModel、ModelProvider
追踪src/agents/tracing/__init__.pyagent_span、generation_span、trace
运行状态src/agents/run_state.pyRunState、RunState.approve、RunState.reject