
精读 OpenAI Agents SDK 的 `deterministic.py` 示例源码(84 行),讲清「代码驱动」与「LLM 驱动」两种编排哲学的本质区别——前者用 Python 原生控制流(`if/else`、`asyncio`)决定 Agent 执行顺序,后者让 LLM 自主规划。重点拆解三段式流水线(`story_outline_agent → outline_checker_agent → story_agent`)、Pydantic 结构化输出配合 gate 机制、`trace()` 上下文管理器,以及 Runner 的 Agent Loop 工作原理,附三条可落地实践建议。
agent_patterns 里把常见模式分了六类1:"A common tactic is to break down a task into a series of smaller steps. Each task can be performed by an agent, and the output of one agent is used as input to the next."
"Orchestrating via code makes tasks more deterministic and predictable, in terms of speed, cost and performance."
| 维度 | 代码驱动(Orchestrating via code) | LLM 驱动(Orchestrating via LLM) |
|---|---|---|
| 流程控制权 | Python 代码,if/else、while、asyncio | LLM 自主决策,通过 tools / handoffs |
| Agent 间通信 | result.final_output 直接传递 | tool call → tool result → LLM 综合 |
| 适用场景 | 步骤已知、顺序固定、需要确定性 | 开放域任务、多意图、需要推理能力 |
| 典型代表 | deterministic.py、parallelization.py | agents_as_tools.py、handoffs.py |
# frontline_agent 通过 as_tool() 决定调用哪个子 Agent
# "The frontline agent receives a user message and then picks which agents to call, as tools."
frontline_agent = Agent(
name="frontline_agent",
tools=[
spanish_agent.as_tool(...),
french_agent.as_tool(...),
italian_agent.as_tool(...),
]
)if "西班牙语" in prompt 更合理。先生成大纲 → 检查大纲质量 → 质量不过关就停止 → 质量通过就写故事
bool 类型),「停止 / 继续」的逻辑是确定的。让 LLM 在这里规划流程,是在为非问题买单。examples/agent_patterns/deterministic.py 里的完整示例4,84 行(65 loc),源码注释直接写着:"This example demonstrates a deterministic flow, where each step is performed by an agent."
story_outline_agent = Agent(
name="story_outline_agent",
instructions="Generate a very short story outline based on the user's input.",
)
class OutlineCheckerOutput(BaseModel):
good_quality: bool
is_scifi: bool
outline_checker_agent = Agent(
name="outline_checker_agent",
instructions="Read the given story outline, and judge the quality. Also, determine if it is a scifi story.",
output_type=OutlineCheckerOutput,
)
story_agent = Agent(
name="story_agent",
instructions="Write a short story based on the given outline.",
output_type=str,
)story_outline_agent:生成故事大纲(自由文本输出)outline_checker_agent:检验大纲——注意这里用了 output_type=OutlineCheckerOutput,强制 LLM 输出结构化的 Pydantic 对象,包含两个 bool 字段story_agent:根据大纲写完整故事outline_checker_agent 的 output_type 是关键设计。LLM 不是返回「这个大纲不错,是科幻风格」这样的自然语言——而是返回 {"good_quality": true, "is_scifi": true},这样后续的 if 判断才能精确、稳定。async def main():
input_prompt = input_with_fallback(
"What kind of story do you want? ",
"Write a short sci-fi story.",
)
with trace("Deterministic story flow"):
# Step 1: 生成大纲
outline_result = await Runner.run(
story_outline_agent,
input_prompt,
)
# Step 2: 检查大纲
outline_checker_result = await Runner.run(
outline_checker_agent,
outline_result.final_output, # 上一步的输出直接作为输入
)
# Step 3: Gate — 代码控制流程,不是 LLM 控制
assert isinstance(outline_checker_result.final_output, OutlineCheckerOutput)
if not outline_checker_result.final_output.good_quality:
print("Outline is not good quality, so we stop here.")
exit(0)
if not outline_checker_result.final_output.is_scifi:
print("Outline is not a scifi story, so we stop here.")
exit(0)
# Step 4: 写故事
story_result = await Runner.run(
story_agent,
outline_result.final_output,
)
print(f"Story: {story_result.final_output}")outline_result.final_output 传递Runner.run() 返回的 RunResult 对象,.final_output 是 Agent 的最终输出(已经过类型验证)。在代码驱动模式里,Agent 间通信就是把这个值作为下一个 Runner.run() 的 input 参数。没有 handoff,没有 tool result,就是 Python 的变量传递。"Add a gate to stop if the outline is not good quality or not a scifi story"
if not ... exit(0) 是 Python 原生的流程控制。检查条件是否满足——不满足就终止,完全不需要 LLM 参与这个决策。这个 gate 的逻辑可以是任意复杂的业务规则:正则校验、数据库查询、外部 API 调用,全都行。trace("Deterministic story flow") 上下文管理器trace() 里,确保它们在 OpenAI 的可观测性后台里归为同一条 trace,而不是三条独立的、看不出关联的请求。这对生产调试非常有价值4。Runner 的工作方式5。result = await Runner.run(agent, input) # async,推荐
result = Runner.run_sync(agent, input) # 同步封装,非 async 环境用
async for event in Runner.run_streamed(agent): # 流式,逐 token 消费"When you use the run method in Runner, you pass in a starting agent and input. The runner then runs a loop."
final_output → 结束循环,返回结果handoff → 切换 Agent,重新进入循环tool_calls → 执行工具,追加结果,重新循环"If we exceed the max_turns passed, we raise a MaxTurnsExceeded exception. Pass max_turns=None to disable this turn limit."
Runner.run() 调用对应一个独立的 Agent Loop。三步流水线 = 三次 Runner.run() = 三个独立的 Agent Loop 串行执行。流程控制权在你的 async def main() 里,不在任何一个 Agent Loop 内部。| 文件 | 模式 | 核心机制 |
|---|---|---|
parallelization.py | 并行执行 | asyncio.gather() 同时跑多个 Agent |
llm_as_a_judge.py | 评估循环 | while 循环 + 评估 Agent 打分,直到通过 |
routing.py | 条件路由 | 结构化输出 + if/match 决定走哪条路 |
asyncio.gather 变体特别值得提一下:# 并行跑多个不相关任务,速度显著快于串行
results = await asyncio.gather(
Runner.run(agent_a, input_a),
Runner.run(agent_b, input_b),
Runner.run(agent_c, input_c),
){"pass": true} 比让它输出「可以继续」更稳定。字符串解析会出错,Pydantic 类型校验不会。output_type=YourModel 是控制流的前提,不只是格式偏好。await Runner.run() 串行。跑通、正确之后,识别哪些步骤没有数据依赖,再改成 asyncio.gather。过早优化为并行会让调试难度成倍上升。trace() 里with trace("你的流水线名字"): 是一行代码,但它让你的可观测性数据从「三个不相关请求」变成「一条有完整上下文的流水线记录」。在生产环境里,这个差别会在你第一次排查问题时体现出来。当前版本:本文基于 OpenAI Agents SDK v0.17.0(2026-05-08 发布)。默认模型已在 v0.16.0 更新为gpt-5.4-mini6,如需维持旧行为可显式设置model="gpt-4.1"。
下期预告 #21:进入agent_patterns的下一个模式——LLM-as-a-Judge:让一个 Agent 评估另一个 Agent 的输出,并在while循环里迭代直到通过。
github.com
openai.github.io
github.com
github.com
openai.github.io
github.com
This story was produced automatically by a channel. One sentence is all it takes for Neodrop to keep producing for you.