Agent 核心循环与事件驱动架构

AI-01: Agent 核心循环与事件驱动架构

学习目标

• 理解 Runner.run() 驱动的 turn-based 循环架构
• 掌握 NextStep 四种类型的流转语义
• 区分同步/异步/流式三种运行模式
• 理解 turn 计数规则与 max_turns 机制

前置知识

本章涉及 Agent 核心循环的通用原理与流式传输模型，建议先阅读：

• Agent 核心循环设计
• 流式传输架构

下文假设你已理解上述概念，直接聚焦 OpenAI Agents SDK 的具体实现。

项目实践

Runner 三入口模式

SDK 提供三个入口点，覆盖不同使用场景：

# 异步模式（推荐）
result = await Runner.run(agent, "你好")

# 同步模式（阻塞包装）
result = Runner.run_sync(agent, "你好")

# 流式模式（立即返回，后台执行）
result = Runner.run_streamed(agent, "你好")
async for event in result.stream_events():
    handle(event)

三者的底层关系：run_sync() 是 Runner.run() 的同步包装器（asyncio.run()），而 run_streamed() 返回 RunResultStreaming 对象并立即启动后台 asyncio.Task。

核心循环流程

AgentRunner.run() 的主循环可以概括为以下步骤：

关键设计：current_turn 在循环顶部递增，这意味着——每次模型调用（连同其所有工具调用）计为 一个 turn。恢复执行时（从 RunState），turn 计数器从暂停点恢复，不会重复计数。

NextStep 四种类型

每次 run_single_turn() 结束时，通过 discriminated union NextStep 决定下一步：

类型	含义	触发条件
NextStepFinalOutput	模型产生最终文本输出	无 handoff、无待执行工具、tool_use_behavior 判定为终态
NextStepHandoff	模型调用 handoff 工具，需要转移控制权	模型响应中包含 handoff tool call
NextStepRunAgain	工具执行完毕，需要再次调用模型	工具结果需要回传给模型继续推理
NextStepInterruption	有工具需要人类审批	某工具声明了 needs_approval

这四种类型覆盖了 Agent 循环的所有可能分支。这种设计将复杂的控制流收敛为一个可测试的 discriminated union。

Turn 计数与 max_turns

• 默认值：DEFAULT_MAX_TURNS = 10
• 计数规则：每次模型调用计 1 个 turn（包含其所有工具调用）
• 恢复规则：从 RunState 恢复时，current_turn 从已持久化的值继续
• 禁用：传入 max_turns=None 可完全取消上限

# 自定义 turn 上限
result = await Runner.run(agent, "输入", run_config=RunConfig(max_turns=20))

# 无上限
result = await Runner.run(agent, "输入", run_config=RunConfig(max_turns=None))

流式 vs 非流式

非流式路径等待完整响应后返回 RunResult。流式路径的差异在于：

1. RunResultStreaming 对象立即返回，实际循环在后台 asyncio.Task 中运行
2. 每个事件推入 asyncio.Queue，通过 stream_events() 异步消费
3. 支持 cancel_mode："immediate"（立即取消）或 "after_turn"（等待当前 turn 完成）
4. 事件类型包括：RawResponsesStreamEvent（原始模型流）、RunItemStreamEvent（结构化 run item）、AgentUpdatedStreamEvent（Agent 切换通知）

问题与规避

问题	规避方案
MaxTurnsExceeded 未捕获直接中断运行	使用 error_handlers={"max_turns": handler} 配置回退策略
流式模式后台任务泄漏	始终消费 stream_events() 直到结束，或显式调用 cancel()
并行 guardrails tripwire 触发时 token 已消耗	对成本敏感场景使用 run_in_parallel=False 顺序执行
恢复执行时 guardrails 不应重新触发	输入 guardrails 仅首轮运行，恢复时跳过

设计取舍

为什么 run_internal/ 独立分包

run.py 仅 93KB，但核心逻辑外包到 run_internal/ 下的多个模块（run_loop.py 80KB、turn_resolution.py 82KB、tool_execution.py 87KB）。这样做的好处：

• 可读性：run.py 仅保留循环编排和公共流程控制
• 可测试性：内部模块可独立测试，不依赖 Runner 全量上下文
• 可维护性：新增工具类型只需修改 run_internal/，不影响 run.py

代价：添加新工具/输出类型时，需要同步更新 run_internal/ 下的 7 个文件（items.py、run_steps.py、turn_resolution.py、tool_execution.py、tool_planning.py、stream_events.py、run_state.py）。

为什么 input guardrails 仅首轮运行

如果每次 turn 都运行输入 guardrails，会导致：(1) 模型自身生成的中间输入被反复检查，增加延迟；(2) handoff 后的新 Agent 对已被上游检查过的输入重复验证。将输入 guardrails 限定为 starting_agent 的 turn 0，既保证入口安全，又避免冗余。

风险：handoff input filter 可能修改输入，使其绕过新 Agent 的 guardrails。缓解方式是在 handoff filter 中内嵌验证逻辑。

参考来源

• OpenAI Agents SDK 文档: Running Agents
• OpenAI Agents SDK 文档: Streaming