OpenAI Agents SDK: Lightweight Multi-Agent Workflow Framework

code openai-openai-agents-python
agentmulti-agenttool-callingMCPhandoffsguardrails

OpenAI Agents SDK (Python) — 代码解读报告 #

Date: 2026-04-16

URL: https://github.com/openai/openai-agents-python

Version: v0.14.1

Domain: agent | Language: Python 99.7% | LOC: ~87K (src) / ~245K (total)

Stars: 21K | Forks: 3.4K | License: MIT


What It Does #

一句话总结: 一个轻量但功能强大的多 Agent 工作流编排框架,让开发者用声明式方式定义 AI Agent 的指令、工具、护栏和协作关系,然后自动执行。

类比: 如果把 LLM 比作一个聪明但被动的大脑,Agents SDK 就是给这个大脑装上了"手脚"(Tools)、"同事"(Handoffs)、"安全绳"(Guardrails)、和"手术室"(Sandbox)。你只需要描述每个 Agent 该干什么,框架负责编排执行循环、错误重试、中断恢复和追踪记录。

目标用户: 需要构建 multi-agent 系统的 ML 工程师和平台开发者。

生态关系:


Architecture Overview #


┌─────────────────────────────────────────────────────────────────┐
│                         User Code                               │
│  agent = Agent(instructions=..., tools=[...], handoffs=[...])   │
│  result = Runner.run(agent, input="...")                        │
└────────────────────────────┬────────────────────────────────────┘
                             │
                    ┌────────▼────────┐
                    │     Runner      │   run.py — 核心入口
                    │  (run / stream) │   orchestrate turn loop
                    └────────┬────────┘
                             │
         ┌───────────────────┼───────────────────┐
         │                   │                   │
   ┌─────▼─────┐   ┌────────▼────────┐   ┌──────▼──────┐
   │  Guardrails│   │   Run Loop      │   │   Session   │
   │  (input/   │   │  run_internal/  │   │  (memory)   │
   │   output)  │   │  run_loop.py    │   │  persist    │
   └────────────┘   └────────┬────────┘   └─────────────┘
                             │
              ┌──────────────┼──────────────┐
              │              │              │
        ┌─────▼─────┐ ┌─────▼─────┐ ┌──────▼──────┐
        │   Model    │ │   Tools   │ │  Handoffs   │
        │  Provider  │ │ execution │ │  (delegate) │
        │ (OpenAI/   │ │ function/ │ │  to other   │
        │  LiteLLM)  │ │ MCP/hosted│ │  agents     │
        └─────┬──────┘ └───────────┘ └─────────────┘
              │
    ┌─────────┼─────────┐
    │         │         │
┌───▼──┐ ┌───▼──┐ ┌────▼────┐
│OpenAI│ │Chat  │ │LiteLLM/ │
│Resp. │ │Compl.│ │any-llm  │
│ API  │ │ API  │ │         │
└──────┘ └──────┘ └─────────┘

        ┌──────────────────┐
        │    Tracing        │  全链路追踪
        │  (spans, traces)  │  OpenAI Dashboard
        └──────────────────┘

        ┌──────────────────┐
        │    Sandbox        │  v0.14 新增
        │  (Docker/Local/   │  Agent 在容器中执行
        │   E2B/Modal/...)  │  长任务工作空间
        └──────────────────┘

        ┌──────────────────┐
        │  Realtime / Voice │  实时语音 Agent
        │  (WebSocket)      │  gpt-realtime-1.5
        └──────────────────┘

Directory Structure #


openai-agents-python/
├── src/agents/                    # 核心库 (~87K LOC)
│   ├── __init__.py                # 公共 API 导出 (538 行, ~180 个导出符号)
│   ├── agent.py                   # Agent 类定义 (941 行)
│   ├── run.py                     # Runner 主入口 (1859 行)
│   ├── run_state.py               # 运行状态序列化/恢复 (3304 行)
│   ├── tool.py                    # Tool 抽象层 (1938 行)
│   ├── guardrail.py               # 输入/输出护栏 (344 行)
│   ├── result.py                  # RunResult (896 行)
│   ├── items.py                   # 消息/工具调用项 
│   ├── lifecycle.py               # RunHooks / AgentHooks 回调
│   ├── run_internal/              # Runner 内部实现
│   │   ├── run_loop.py            # 单轮执行循环 (1894 行)
│   │   ├── turn_resolution.py     # 模型输出解析 (1911 行)
│   │   ├── tool_execution.py      # 工具并发执行 (2329 行)
│   │   ├── tool_planning.py       # 工具规划
│   │   ├── streaming.py           # 流式输出处理
│   │   └── session_persistence.py # 会话持久化
│   ├── models/                    # 模型后端
│   │   ├── interface.py           # Model / ModelProvider 抽象 
│   │   ├── openai_responses.py    # OpenAI Responses API (2043 行)
│   │   ├── openai_chatcompletions.py # Chat Completions 适配
│   │   └── multi_provider.py      # 多 Provider 路由
│   ├── handoffs/                  # Agent 间委托
│   ├── mcp/                       # Model Context Protocol 集成
│   │   ├── server.py              # MCP 服务端 (1620 行)
│   │   └── manager.py             # 生命周期管理
│   ├── memory/                    # 会话记忆
│   │   ├── session.py             # Session 抽象
│   │   ├── sqlite_session.py      # SQLite 持久化
│   │   └── openai_conversations_session.py
│   ├── sandbox/                   # 沙箱执行环境
│   │   ├── sandbox_agent.py       # SandboxAgent 
│   │   ├── runtime.py             # Sandbox 运行时
│   │   ├── sandboxes/             # Docker / Unix Local
│   │   ├── capabilities/          # Shell / Filesystem / Patch
│   │   ├── session/               # 沙箱会话管理
│   │   ├── entries/               # Git/File/S3 mount
│   │   └── memory/                # 沙箱记忆
│   ├── realtime/                  # 实时语音 Agent
│   │   ├── openai_realtime.py     # OpenAI Realtime API (1724 行)
│   │   └── session.py             # 实时会话 (1112 行)
│   ├── tracing/                   # 追踪系统
│   │   ├── spans.py / traces.py   # Span / Trace 数据结构
│   │   └── processors.py          # 追踪数据处理
│   ├── voice/                     # 语音管道
│   │   ├── pipeline.py            # STT → Agent → TTS
│   │   └── models/                # OpenAI STT/TTS
│   └── extensions/                # 扩展
│       ├── models/                # LiteLLM / any-llm
│       ├── memory/                # Redis / SQLAlchemy / Dapr
│       └── sandbox/               # E2B / Modal / Blaxel / Daytona / Cloudflare
├── tests/                         # 测试套件 (~158K LOC)
├── examples/                      # 使用示例
├── docs/                          # MkDocs 文档 (含 ja/ko/zh 翻译)
└── pyproject.toml                 # 构建配置 (hatchling)

Core Concepts & Data Flow #

1. Agent — 智能体定义 #


@dataclass
class Agent(AgentBase, Generic[TContext]):
    name: str                           # Agent 名称
    instructions: str | Callable        # 系统提示词 (静态 / 动态)
    model: str | Model | None           # LLM 模型 (默认 gpt-4.1)
    tools: list[Tool]                   # 可用工具
    handoffs: list[Agent | Handoff]     # 可委托的子 Agent
    input_guardrails: list[InputGuardrail]   # 输入护栏
    output_guardrails: list[OutputGuardrail]  # 输出护栏
    output_type: type | None            # 结构化输出类型
    hooks: AgentHooks | None            # 生命周期回调
    tool_use_behavior: ...              # 工具调用后行为

关键设计: Agent 是一个 @dataclass,不是 Pydantic Model。这意味着:

2. Runner — 执行引擎 #


Runner.run(agent, input)
  │
  ├─ 1. 创建 Trace Context
  ├─ 2. 运行 Input Guardrails (并行 / 前置)
  │     └─ 如果 tripwire 触发 → 抛异常停止
  ├─ 3. Turn Loop (最多 max_turns 轮):
  │     ├─ 准备输入 (system prompt + history + tools)
  │     ├─ 调用 Model.get_response() / stream_response()
  │     ├─ 解析模型输出 → NextStep:
  │     │   ├─ NextStepFinalOutput  → 输出护栏 → 返回结果
  │     │   ├─ NextStepHandoff     → 切换到新 Agent, 继续循环
  │     │   ├─ NextStepRunAgain    → 执行工具, 结果送回模型
  │     │   └─ NextStepInterruption → 暂停等待人工审批
  │     └─ 持久化会话到 Session
  └─ 4. 返回 RunResult

关键设计:

3. Tool System — 工具系统 #


Tool (Union type)
  ├─ FunctionTool        → 用户自定义 Python 函数 (@function_tool)
  ├─ WebSearchTool       → OpenAI 托管 Web 搜索
  ├─ FileSearchTool      → OpenAI 托管文件搜索
  ├─ CodeInterpreterTool → OpenAI 托管代码执行
  ├─ ComputerTool        → 计算机操作 (CUA)
  ├─ HostedMCPTool       → OpenAI 托管 MCP
  ├─ CustomTool          → 自定义 Tool JSON
  ├─ ShellTool           → Shell 命令执行
  ├─ ApplyPatchTool      → 文件补丁应用
  ├─ LocalShellTool      → 本地 Shell
  ├─ ImageGenerationTool → 图片生成
  └─ ToolSearchTool      → 工具搜索

FunctionTool 核心流程:

  1. @function_tool 装饰器自动提取参数类型 → JSON Schema
  2. 运行时: 模型输出工具调用 → 反序列化参数 → 执行函数
  3. 支持 needs_approval 审批、is_enabled 动态启用
  4. 支持超时 (timeout_ms)、错误处理 (failure_error_function)
  5. 4. Handoff — Agent 委托 #

    
    triage_agent = Agent(
        name="Triage",
        handoffs=[billing_agent, tech_agent],  # 可直接传 Agent
    )
    

    Handoff vs Agent-as-Tool 的区别:

    HandoffAgent as Tool
    对话历史新 Agent 继承完整对话历史新 Agent 收到生成的输入
    控制权新 Agent 接管对话原 Agent 保持控制权
    使用场景路由分发 (triage)子任务调用

    5. Guardrails — 安全护栏 #

    两种护栏并行于模型执行:

    • InputGuardrail: 运行在第一轮、第一个 Agent,并行检查输入
    • OutputGuardrail: 运行在最终输出上

    如果 tripwire_triggered = True → 抛出异常,立即停止执行。

    6. Session / Memory — 会话持久化 #

    
    Session (ABC)
      ├─ SQLiteSession          → 本地 SQLite
      ├─ OpenAIConversationsSession → OpenAI 服务端存储
      ├─ OpenAIResponsesCompactionSession → 自动压缩上下文
      └─ Extensions:
          ├─ RedisSession
          ├─ SQLAlchemySession
          ├─ DaprSession
          └─ EncryptedSession
    

    7. Sandbox — 沙箱执行 (v0.14 新增) #

    
    SandboxAgent(
        name="Workspace Assistant",
        default_manifest=Manifest(entries={
            "repo": GitRepo(repo="openai/openai-agents-python", ref="main"),
        }),
    )
    

    支持的沙箱后端:

    • UnixLocalSandboxClient — 本地文件系统
    • DockerSandboxClient — Docker 容器
    • Extensions: E2B, Modal, Blaxel, Daytona, Cloudflare, Runloop, Vercel

    核心能力 (Capabilities):

    • Shell 命令执行
    • 文件系统操作
    • apply_patch 文件补丁
    • 上下文压缩 (compaction)
    • 技能注入 (skills)

    8. Tracing — 全链路追踪 #

    
    Trace
      └─ Agent Span
           ├─ Generation Span (LLM 调用)
           ├─ Function Span (工具执行)
           ├─ Guardrail Span
           ├─ Handoff Span
           └─ Custom Span
    

    追踪数据通过 TracingProcessor 上报到 OpenAI Dashboard,支持自定义处理器。

    9. Realtime / Voice — 实时语音 #

    
    Voice Pipeline:
      Microphone → STT → Agent → TTS → Speaker
    
    Realtime Agent:
      WebSocket ↔ gpt-realtime-1.5 (全双工)
    

    Key Design Decisions #

    1. Provider-Agnostic 但 OpenAI-First #

    • 内部数据格式统一为 OpenAI Responses API 格式
    • 其他 LLM 通过 chatcmpl_converter.py 转换 Chat Completions 格式
    • LiteLLM / any-llm 作为 extensions 支持 100+ 模型

    2. Dataclass over Pydantic for Agent #

    • Agent 是 @dataclass 而非 Pydantic BaseModel
    • 好处: 更轻量、支持 dataclasses.replace() 浅拷贝
    • 代价: 类型验证需要手动实现 (__post_init__)

    3. 状态外置 + 可中断恢复 #

    • RunState 可序列化为 JSON → 实现 human-in-the-loop
    • 审批系统: 工具调用可暂停等待人工审批
    • 恢复时将 RunState 传入 Runner.run() 继续执行

    4. 并发模型: asyncio 为主 #

    • 所有核心路径使用 async/await
    • 工具执行并发: asyncio.gather() 并行执行多个工具调用
    • run_sync() 通过 asyncio.run() 桥接同步调用

    5. MCP 作为一等公民 #

    • 原生支持 MCP Server 声明式挂载
    • 每次 Agent 运行时自动发现并注册 MCP 工具
    • 支持 hosted MCP (OpenAI 服务端) 和 local MCP

    Key Data Structures #

    StructureFileLOCPurposeLifetime
    Agentagent.py941Agent 定义 (指令/工具/护栏)用户定义, 长生命周期
    RunStaterun_state.py3304可序列化的运行状态单次 run, 可持久化
    RunResultresult.py896执行结果 (输出/items/usage)单次 run
    FunctionTooltool.py~300函数工具定义用户定义
    Handoffhandoffs/~350Agent 委托关系用户定义
    ModelResponseitems.py-模型响应包装单轮
    ProcessedResponserun_steps.py-解析后的模型输出单轮
    Sessionmemory/-会话状态持久化跨 run
    Span / Tracetracing/-追踪数据单次 run

    Critical Path Analysis #

    
    Runner.run()                           (run.py:194)
      → create_trace_for_run()             (tracing/context.py)
      → prepare_input_with_session()       (session_persistence.py)
      → run_input_guardrails()             (run_loop.py)    ~异步并行
      → TURN LOOP:
        → get_all_tools()                  (run_loop.py)    fetch MCP tools
        → run_single_turn()               (run_loop.py)
          → Model.get_response()          (models/openai_responses.py)  ~1-30s
          → resolve_processed_response()  (agent_runner_helpers.py)
          → NextStep 分支:
            → tool_execution()            (tool_execution.py)  并行执行
            → handoff resolution          (turn_resolution.py)
        → save_turn_items_if_needed()     (session_persistence.py)
      → run_output_guardrails()           (run_loop.py)
      → RunResult
    

    瓶颈: LLM 调用 (Model.get_response()) 是主要延迟源。工具执行通过 asyncio.gather() 并行化。


    Testing & Quality #

    • 测试框架: pytest + pytest-asyncio + pytest-mock
    • 测试规模: ~158K LOC 测试代码 (约 1.8x 源码)
    • 覆盖范围:
    • unit tests for 所有核心模块
    • snapshot tests (inline-snapshot)
    • 真实模型调用可选标记 (@allow_call_model_methods)
    • 类型检查: mypy (strict) + pyright
    • 代码风格: ruff (格式化 + lint)
    • CI: GitHub Actions (tests, lint, typecheck, docs)

    Comparison with Alternatives #

    FeatureOpenAI Agents SDKLangChainCrewAIAutoGen
    轻量级✅ 单包, 87K LOC❌ 生态庞大
    多 Agent 编排✅ Handoffs✅ Chains✅ Crews✅ Conversations
    OpenAI 深度集成✅ 原生❌ 适配层
    MCP 支持✅ 一等公民⚠️ 社区
    沙箱执行✅ (v0.14)✅ Docker
    实时语音
    Human-in-loop✅ RunState 中断恢复⚠️ 手动
    Tracing✅ 内置 → OpenAI Dashboard⚠️ LangSmith
    多 LLM 支持✅ LiteLLM/any-llm
    类型安全✅ Generic[TContext]⚠️

    Tech Debt & Observations #

    1. run_state.py 3304 行 — 状态序列化逻辑过于集中,包含复杂的 schema version 迁移
    2. tool.py 1938 行 — Tool 的 Union 类型定义和 FunctionTool 实现混在一起
    3. OpenAI Responses 格式作为内部数据格式 — 与 OpenAI 绑定过深,其他 provider 需要来回转换
    4. Extensions 目录膨胀 — 7 种 sandbox provider, 2 种 LLM adapter, 5 种 session backend

    5. Verdict & Recommendations #

      应该使用的场景:

      • 如果你的主要 LLM 是 OpenAI → 这是最佳选择
      • 需要 multi-agent handoff + tool calling + guardrails 的组合
      • 需要 human-in-the-loop 中断恢复
      • 需要 MCP 集成

      不适合的场景:

      • 纯 open-source LLM 场景 (与 OpenAI API 格式耦合较深)
      • 需要极简框架 (这个 SDK 功能已经相当丰富)
      • 需要自定义底层调度逻辑 (Runner 的 turn loop 不太容易扩展)

      值得学习的设计:

      1. RunState 可序列化 + 中断恢复 — 生产级 human-in-the-loop 方案
      2. Guardrails 与 Agent 执行并行 — 不增加额外延迟
      3. Agent-as-Tool 模式 — 优雅地实现 Agent 嵌套调用
      4. Tracing 内建而非插件 — 保证了可观测性
      5. Sandbox Capabilities 插件化 — 可扩展的沙箱能力系统