SDK 核心概念一览

📑 目录

概念全景图

Runner 调度算法的深度解析

Runner 作为 Agents SDK 的中央调度器,其内部实现了一个基于协程的异步事件循环。与传统的线程池调度不同,Runner 充分利用了 Python 的 asyncio,使得大量 I/O 密集型操作(如 LLM 调用、工具执行)可以并发进行而不阻塞主线程。

在调度策略上,Runner 采用了贪婪执行模式:当模型在一次响应中请求调用多个工具时,Runner 会立即并发启动所有工具的执行,而不是串行等待。这种设计显著降低了多工具场景下的总延迟。例如,如果 Agent 需要同时查询天气和股价,串行执行可能需要 2+2=4 秒,而并发执行只需 max(2,2)=2 秒。

然而,贪婪执行也带来了潜在的资源竞争问题。如果两个工具同时写入同一个文件或修改同一条数据库记录,就可能产生竞态条件。SDK 目前不内置工具间的锁机制,这需要开发者在自己的工具实现中通过数据库事务或文件锁来保证一致性。

从 token 消耗的角度来看,每次 Runner 循环都会将完整的对话历史(包括之前的工具调用和结果)发送给 LLM。这意味着随着对话轮数的增加,输入 token 数量会快速增长。对于长对话场景,建议启用 Session 的自动压缩功能,或者手动截断历史消息。

┌─────────────────────────────────────────────────────────────┐
│                      OpenAI Agents SDK                       │
├──────────┬──────────┬──────────┬──────────┬─────────────────┤
│  Agent   │  Tools   │ Handoffs │Guardrails│   Sessions      │
├──────────┴──────────┴──────────┴──────────┴─────────────────┤
│                         Runner                               │
├─────────────────────────────────────────────────────────────┤
│  Tracing  │  Human-in-the-loop  │  Sandbox  │  Realtime     │
└─────────────────────────────────────────────────────────────┘

1. Agent

Agent 是 SDK 的核心构建块。它是一个配置了**指令(instructions)、工具(tools)、模型(model)**以及可选的运行时行为(handoffs、guardrails、结构化输出)的 LLM。

agent = Agent(
    name="Assistant",
    instructions="You are a helpful assistant.",
    model="gpt-5-nano",
    tools=[...],
    handoffs=[...],
    input_guardrails=[...],
    output_type=MyOutput,
)

2. Tools

Tools 让 Agent 能够执行动作:获取数据、运行代码、调用外部 API。SDK 支持五类工具:

  • Hosted OpenAI 工具:WebSearchTool、FileSearchTool 等
  • 本地运行时工具:ComputerTool、ApplyPatchTool、ShellTool
  • 函数工具:用 @function_tool 包装 Python 函数
  • Agent 作为工具Agent.as_tool()
  • 实验性工具:Codex 工具

3. Handoffs

Handoffs 允许一个 Agent 将任务委托给另一个 Agent。特别适用于不同 Agent 专精于不同领域的场景。

billing_agent = Agent(name="Billing agent")
refund_agent = Agent(name="Refund agent")
triage_agent = Agent(
    name="Triage agent",
    handoffs=[billing_agent, refund_agent]
)

4. Guardrails

Guardrails 用于对用户输入和 Agent 输出进行校验。分为三种:

  • Input Guardrails:在首个 Agent 执行前校验用户输入
  • Output Guardrails:在最终输出返回前校验
  • Tool Guardrails:在每次函数工具调用前后校验

5. Sessions

Sessions 提供内置的会话记忆,自动维护跨多次运行的对话历史,无需手动处理 to_input_list()

from agents import SQLiteSession

session = SQLiteSession("conversation_123")
result = await Runner.run(agent, "Hello", session=session)

6. Human in the loop

内置的人工介入机制,在敏感操作前暂停执行,等待人工审批。

@function_tool(needs_approval=True)
def delete_file(path: str) -> str:
    ...

7. Tracing

内置追踪系统,自动记录 Agent 运行过程中的事件:LLM 生成、工具调用、交接、防护栏等。

# 默认开启,可通过以下方式禁用
# export OPENAI_AGENTS_DISABLE_TRACING=1

8. Sandbox Agents

预配置在容器中执行长期任务的 Agent,可操作真实文件系统、运行命令、应用补丁。

from agents.sandbox import SandboxAgent, Manifest

agent = SandboxAgent(
    name="Code Assistant",
    instructions="Help with coding tasks.",
    default_manifest=Manifest(entries={...}),
)

9. Realtime Agents

基于 gpt-realtime-1.5 和 WebSocket 的低延迟语音 Agent。

from agents.realtime import RealtimeAgent, RealtimeRunner

agent = RealtimeAgent(name="Voice Assistant", instructions="...")
runner = RealtimeRunner(starting_agent=agent, config={...})

概念关系图

User Input → [Input Guardrails] → Agent → LLM Call
                                              ↓
                    ┌──────────────────── Tool Call
                    ↓                         ↓
              [Handoff] → New Agent    [Tool Guardrails]
                    ↓                         ↓
              Agent Output ← [Output Guardrails]
                    ↓
              [Session Save] → [Trace Record]

SDK 核心组件架构图

以下架构图展示了 Agents SDK 中各核心组件的关系和数据流向:

flowchart LR
    subgraph Input["输入层"]
        I1[用户输入]
        I2[输入 Guardrails]
    end
    subgraph Core["核心引擎"]
        R[Runner]
        A[Agent]
        C[Context]
        H[Hooks]
    end
    subgraph Tools["工具层"]
        T1[Function Tool]
        T2[OpenAI 托管工具]
        T3[Agent 作为工具]
        T4[MCP 工具]
    end
    subgraph Output["输出层"]
        O1[输出 Guardrails]
        O2[结构化输出]
    end
    I1 --> I2
    I2 --> R
    R --> A
    A --> C
    A --> H
    A --> T1
    A --> T2
    A --> T3
    A --> T4
    T1 --> C
    T2 --> C
    T3 --> C
    T4 --> C
    C --> R
    R --> O1
    O1 --> O2
    O2 --> I1

上图中的箭头方向揭示了 SDK 的设计哲学:Runner 是中心调度器,所有数据流都经过它;Agent 是配置容器,定义了行为策略;Context 是状态载体,存储了对话历史和工具结果。

概念辨析:Agent、Runner 与 Model

许多初学者容易混淆这三个核心概念的区别。简单来说:

Agent 是策略配置。它本身不执行任何计算,只是定义了应该怎么做——使用什么模型、遵循什么指令、拥有什么工具。你可以把 Agent 看作一份岗位说明书,描述了这个角色的职责和能力边界。同一个 Agent 可以被多个 Runner 实例使用,也可以被序列化后存储到数据库中供后续加载。

Runner 是执行引擎。它负责将 Agent 的配置转化为实际的 API 调用和工具执行。Runner 管理着整个执行生命周期:从接收输入、构建 prompt、调用 LLM、解析响应、执行工具、到最终返回结果。一个 Runner 实例通常对应一次用户会话,会话结束后 Runner 的生命周期也随之结束。

Model 是计算能力来源。它是实际的 LLM(如 GPT-5、Claude 4),通过 HTTP API 提供服务。Agent 通过 model 参数指定使用哪个模型,Runner 负责将 Agent 的指令和上下文转换为该模型 API 所需的格式。模型的选择直接影响输出质量、延迟和成本,是系统调优的首要杠杆。

理解这三者的关系,是设计复杂 Agent 系统的关键。例如,在多租户场景中,你可以为每个租户创建独立的 Agent 配置(不同的 instructions 和 tools),但共用同一个 Runner 执行引擎和模型连接池,实现配置隔离与资源复用的平衡。

在这个流程中,Context 是贯穿始终的纽带。它不仅存储对话历史,还承载工具执行结果、Handoff 状态和追踪信息。设计良好的上下文管理机制,是构建复杂多 Agent 系统的基石。值得注意的是,Handoff 并非简单的函数调用,而是控制权转移。当 Agent A Handoff 到 Agent B 后,Agent A 不再参与后续对话,直到用户明确切回。这与 Agent 作为工具模式有本质区别,后者是同步调用并立即返回。

下一步

选择你感兴趣的章节深入阅读,建议从 Agent 配置详解 开始。

完整实战示例:全链路监控的 Agent 系统

以下示例展示了如何在 Runner 层统一注入日志、限流和 metrics,适用于生产环境的可观测性需求:

import time
import asyncio
from collections import defaultdict
from agents import Agent, Runner, RunContextWrapper


class MetricsCollector:
    """简单的内存指标收集器,生产环境可替换为 Prometheus。"""
    def __init__(self):
        self.calls = defaultdict(int)
        self.latency = defaultdict(list)

    def record(self, agent_name: str, duration: float, turns: int):
        self.calls[agent_name] += 1
        self.latency[agent_name].append(duration)
        print(f"[METRICS] {agent_name}: {duration:.2f}s, {turns} turns")


metrics = MetricsCollector()


async def run_with_monitoring(agent: Agent, user_input: str):
    """包装 Runner.run,注入监控逻辑。"""
    start = time.perf_counter()
    try:
        result = await Runner.run(agent, user_input)
        duration = time.perf_counter() - start
        metrics.record(agent.name, duration, len(result.raw_responses))
        return result
    except Exception as e:
        duration = time.perf_counter() - start
        print(f"[ERROR] Agent {agent.name} failed after {duration:.2f}s: {e}")
        raise


async def main():
    billing_agent = Agent(name="Billing", instructions="Handle billing questions.")
    refund_agent = Agent(name="Refund", instructions="Handle refund requests.")
    triage = Agent(
        name="Triage",
        instructions="Route to Billing or Refund based on user intent.",
        handoffs=[billing_agent, refund_agent])

常见问题与调试

问题一:概念混淆导致架构错误

最常见的架构错误是将 Agent 直接当作可执行对象使用。记住:Agent 是配置,Runner 是引擎。Agent 不包含任何状态,也不能直接调用 .run();所有的执行都必须通过 Runner.run(agent, ...) 发起。在代码审查中,如果发现直接修改 Agent 内部属性或试图缓存 Agent 实例来保存状态,都是典型的反模式。

问题二:Guardrails 与主流程的时序误解

输入 Guardrails 默认与 Agent 主流程并行执行,这意味着即使 Guardrails 检测到问题,Agent 的 LLM 调用可能已经开始。如果需要阻塞式校验,必须在 Guardrails 配置中显式设置。调试时可以通过 trace 查看 Guardrails 和 LLM 调用的时间线,确认它们是否真的并行。

问题三:多 Agent 系统中的 Session 污染

当多个 Agent 共享同一个 Session 时,一个 Agent 产生的工具调用记录会影响另一个 Agent 的上下文理解。在多 Agent 编排中,建议为每个用户会话维护一个 Session,但在 Agent 切换时通过 input_filter 清理不相关的历史记录。

性能优化建议

对于高并发场景,Runner 的协程模型可以支撑数百个并发对话,但底层 LLM API 的速率限制会成为瓶颈。建议在 Runner 层之上增加一个请求队列和令牌桶限流器,避免因触发 OpenAI 的 rate limit 而导致大面积请求失败。

与其他方案对比

维度Agents SDKLangChainCrewAI
概念数量9 大核心概念,关系清晰数十个抽象概念,学习曲线陡角色、任务、流程三大概念
架构风格中心调度(Runner)链式组合(LCEL)去中心化协作
上手难度中(需理解 Runner 循环)高(需掌握大量抽象)低(高层封装)
可定制性高(底层控制精细)极高(组件可任意组合)中(受限于预设模式)

Agents SDK 的概念体系围绕 Runner 构建,形成了一颗以调度器为根的概念树。这种设计的好处是职责清晰,缺点是灵活性略逊于 LangChain 的链式组合。CrewAI 在概念上做了大量简化,适合快速搭建原型,但在需要精细控制对话流和工具执行时可能会感到束缚。

核心概念依赖关系与扩展模式

mermaid
flowchart TD
subgraph InputLayer [输入层]
IG[Input Guardrails]
end
subgraph CoreLayer [核心调度层]
R[Runner]
A[Agent]
T[Tools]
H[Handoffs]
OG[Output Guardrails]
end
subgraph DataLayer [数据持久层]
S[Sessions]
Tr[Tracing]
end
subgraph ExtLayer [扩展能力层]
HL[Human-in-the-loop]
SB[Sandbox Agents]
RT[Realtime Agents]
end
IG --> R
R --> A
A --> T
A --> H
T --> OG
H --> A
R --> S
R --> Tr
HL -.-> T
SB -.-> A
RT -.-> R


上图以分层架构的方式呈现了 9 大核心概念的职责边界。Runner 处于核心调度层的中枢位置,所有数据流都必须经过它;扩展能力层则通过虚线表示可选的增强能力,不会破坏核心层的稳定性。理解这种分层有助于在架构设计时做出合理的取舍,避免过度设计导致系统臃肿。

### 基于装饰器模式的 Runner 扩展

当需要在不修改 SDK 源码的前提下增强 Runner 的行为时,装饰器模式是最佳选择。例如,为所有 Agent 调用统一注入限流、日志和重试逻辑:

```python
import time
import asyncio
from functools import wraps
from agents import Agent, Runner, RunResult


def runner_instrument(log_prefix: str = "[AGENT]"):
    """Runner 装饰器工厂:为 run / run_sync 添加日志和耗时统计。"""
    def decorator(func):
        @wraps(func)
        async def async_wrapper(agent: Agent, user_input: str, **kwargs):
            start = time.perf_counter()
            print(f"{log_prefix} Starting {agent.name} with input: {user_input[:50]}")
            try:
                result: RunResult = await func(agent, user_input, **kwargs)
                duration = time.perf_counter() - start
                print(f"{log_prefix} Finished {agent.name} in {duration:.2f}s")
                return result
            except Exception as e:
                duration = time.perf_counter() - start
                print(f"{log_prefix} Failed {agent.name} after {duration:.2f}s: {e}")
                raise

        return async_wrapper
    return decorator


# 使用示例:包装 Runner.run
monitored_run = runner_instrument("[MONITOR]")(Runner.run)

# 现在所有通过 monitored_run 发起的调用都会自动记录日志
# result = await monitored_run(my_agent, "user query")

装饰器模式的优势在于职责分离:限流、审计、缓存等横切关注点从业务逻辑中剥离,使 Agent 配置和 Runner 调用保持简洁。生产环境中,可以将这类装饰器进一步抽象为 AOP 中间件,配合依赖注入框架实现更灵活的插件化架构。当团队需要同时支持多个品牌的 Agent 时,这种插件化能力尤为重要,它允许你在不改动核心业务代码的前提下切换不同的监控后端。

多概念组合的设计模式

在实际项目中,单一概念往往不足以支撑复杂业务。一个典型的多概念组合场景是:用户输入先经过 Input Guardrails 进行敏感词过滤,然后由 Triage Agent 分发到不同领域的 Specialist Agent,每个 Specialist 配备独立的 Tools 和 Output Guardrails,同时 Session 负责持久化对话历史,Tracing 记录全链路日志。这种架构本质上是责任链模式策略模式的结合。设计时需要特别注意概念之间的数据流方向:Guardrails 只能拦截或修改数据,不能改变控制流;Handoffs 则是控制流的显式转移点。混淆这两者的职责是新手最常见的架构错误。建议在绘制系统架构图时,用不同颜色区分"数据过滤路径"和"控制转移路径",强迫自己在设计阶段就理清边界。另一个关键决策是 Session 的粒度:按用户会话隔离是最常见的选择,但在多租户 SaaS 场景中,还需要叠加租户级别的隔离策略,防止数据泄露。Session 的存储后端选择也会影响系统扩展性,SQLite 适合原型验证,PostgreSQL 适合生产环境,Redis 则适合高并发短会话场景。

生产环境部署与性能优化

系统架构设计的实践要点

将本章节的技术应用到生产环境时,首要考虑的是稳定性与可观测性。建议采用渐进式 rollout 策略:先在开发环境验证核心逻辑,再迁移到预发布环境进行压力测试,最后才全量上线。部署过程中应配置完善的日志收集和指标监控,确保任何问题都能被快速发现和定位。

具体来说,需要在基础设施层面做好以下准备:容器资源限制(CPU/内存)、网络策略配置(防火墙规则、服务网格)、持久化存储选型(SSD vs 标准盘)以及备份恢复方案。对于高可用要求严格的场景,建议部署多实例并配置负载均衡,避免单点故障导致服务中断。

全链路可观测性的关键指标

监控是生产系统的生命线。针对本章节涉及的功能,建议重点跟踪以下指标:请求延迟(P50/P95/P99)、错误率(4xx/5xx/超时)、吞吐量(QPS/TPS)以及资源利用率(CPU/内存/磁盘/网络)。这些指标应接入统一的监控大盘,并设置合理的告警阈值。

除了基础指标,还应关注业务层面的指标。例如功能成功率、用户满意度、成本消耗趋势等。通过将技术指标与业务指标关联分析,可以更准确地评估系统改进的实际价值,避免陷入"为了优化而优化"的陷阱。

概念映射到实现的架构考量

随着业务规模增长,单实例部署很快会成为瓶颈。扩展性设计应在项目初期就纳入考量,而非事后补救。水平扩展通常比垂直扩展更具成本效益,但也引入了分布式系统的复杂性(数据一致性、服务发现、负载均衡等)。

在扩展过程中,建议遵循"无状态优先"原则:将状态外置到独立的存储层(如 Redis、PostgreSQL),使计算层可以随时水平扩容。对于无法避免的状态(如会话、缓存),采用分布式一致性协议或最终一致性模型来管理。定期进行容量规划和压力测试,确保系统在流量峰值时仍能稳定运行。

运维团队的协作建议

技术方案的落地离不开高效的团队协作。建议建立清晰的运维手册(Runbook),涵盖常见故障的诊断步骤、应急处理流程和升级路径。同时,通过定期的复盘会议,将线上事故转化为团队的学习素材,持续完善系统的健壮性。

在工具链方面,推荐将本章节的配置和脚本纳入版本控制(Git),并使用 Infrastructure as Code(IaC)工具(如 Terraform、Ansible)管理基础设施变更。这不仅能提高部署效率,还能确保环境一致性,减少"在我机器上能跑"的问题。