本文目录13 个章节
模型不需要决定每个步骤
状态、检查点、审批和安全边界稳定后，控制流成为下一处需要整理的代码。加载配置、校验任务、合并来源、检查引用和等待审批都有确定规则。把这些步骤交给模型会增加调用次数和随机失败。证据是否充足、冲突资料要补查哪一类，才需要开放判断。
选择编排方式时，可以沿着下面的树走。
步骤是否固定？
├── 是 ─▶ 下一步是否只有一个？
│ ├── 是 ─▶ 串行
│ └── 否 ─▶ 分支能否枚举？
│ ├── 是 ─▶ 条件路由
│ └── 否 ─▶ 重新检查需求
└── 否 ─▶ 子任务是否互相独立？
├── 是 ─▶ 受控并行 + reducer
└── 否 ─▶ 有预算的动态规划四种模式放进简报流程
串行：验证任务 → 检索 → 生成 → 校验。后一步依赖前一步输出。
路由：没有主题就请求补充；没有证据就返回缺口；证据充足才生成。
并行：本地资料与允许的远程资料可以同时检索，分支只读共享输入。
动态规划：模型根据证据冲突选择补查版本说明、API 文档或安全公告，步数和工具受预算限制。
并行分支用 reducer 合并
from dataclasses import dataclass
@dataclass(frozen=True)
class Source:
source_id: str
score: float
text: str
def merge_sources(groups: list[list[Source]], limit: int = 6) -> list[Source]:
best: dict[str, Source] = {}
for group in groups:
for source in group:
current = best.get(source.source_id)
if current is None or source.score > current.score:
best[source.source_id] = source
return sorted(best.values(), key=lambda item: (-item.score, item.source_id))[:limit]分支不修改同一个 state 字典。它们返回独立列表，由一个纯函数稳定去重和排序。这样并发顺序变化不会改变最终来源集。
路由条件使用结构化字段
def choose_next(task_valid: bool, evidence_count: int,
citations_valid: bool) -> str:
if not task_valid:
return "request_clarification"
if evidence_count == 0:
return "report_evidence_gap"
if not citations_valid:
return "repair_citations"
return "request_save_approval"自由文本适合呈现，枚举状态适合路由。解析模型一句“资料可能够了”来决定写入，会让测试和恢复变得脆弱。
重试从安全节点开始
纯读取节点可以重跑；带幂等键的写节点先查状态；不可逆动作等待人工处理。重试继承全局轮数、工具调用和费用预算，不能在每个节点重新计数。
故障注入覆盖四个位置：一个并行搜索超时、一个分支返回坏数据、合并后引用不足、保存前进程退出。恢复后只执行未确认的安全节点。
画出项目当前控制流
validate_task
│
▼
local_search ─┐
├─▶ merge_sources ─▶ assess_evidence
remote_search ┘ │
┌─────────────┴─────────────┐
▼ ▼
search_more draft_and_check
│
▼
wait_for_approval这个图已经比最小循环复杂，手写运行时开始承担图状态、Session 和 Trace 等通用工作。工作流整理完成后，下面在行为测试保护下迁移到 OpenAI Agents SDK，并说明 LangGraph 在显式图和检查点场景中的位置。
迁移从对照表开始
没有手写 Loop 就直接使用框架，很难看清 API 背后的控制流。前面的实现已经覆盖模型循环、工具分发、错误、状态和审批，现在可以逐项对照 SDK 接走的通用代码。
手写运行时 Agents SDK
────────── ──────────
for turn in range(...) ───▶ Runner.run
解析 function_call ───▶ function_tool
拼接历史 ───▶ Session / response chain
暂停对象 ───▶ interruption + RunState
事件日志 ───▶ Trace
任务规格、权限、业务预算、结果验收 ───▶ 仍在应用层先冻结 Runtime 协议
领域层定义 run(task) -> RunResult。手写版和 SDK 版都返回结束状态、最终输出、来源 ID、工具事件和费用信息。测试不依赖 SDK 内部类名。
迁移期间不修改 Prompt、工具语义和任务预算。否则结果变化无法归因于运行时。
用 function_tool 包装现有工具
import os
from agents import Agent, Runner, SQLiteSession, function_tool
SOURCES = {"agent-loop#1": "A tool result returns to the next model turn."}
@function_tool
def get_document(source_id: str) -> dict:
"""Read one approved source by its exact source ID."""
text = SOURCES.get(source_id)
if text is None:
return {"ok": False, "error": "source_not_found"}
return {"ok": True, "source_id": source_id, "text": text}
brief_agent = Agent(
name="Research brief agent",
instructions=(
"Use approved sources. Preserve source IDs. "
"Return insufficient_evidence when support is missing."
),
model=os.environ["OPENAI_MODEL"],
tools=[get_document],
)工具继续返回应用定义的领域结果。领域层不接收 SDK 专用对象，未来替换运行时不需要重写检索和验收。
Runner 与 Session 怎样组合
async def run_brief(thread_id: str, question: str) -> str:
session = SQLiteSession(thread_id, "brief_sessions.db")
result = await Runner.run(
brief_agent,
question,
session=session,
max_turns=6,
)
return result.final_outputSession、完整历史、previousResponseId 都能续接上下文，应选择一种主策略。重复传入完整历史并同时使用 Session，会让消息重复。thread_id 绑定租户和用户作用域，不能直接使用前端传来的短字符串作为全局键。
共享行为测试怎样保护迁移
同一组用例分别传给 ManualRuntime 和 AgentsSdkRuntime：搜索后读取、未知来源、证据不足、轮数耗尽、等待审批、拒绝保存。比较领域结果与环境副作用，不比较内部 Trace 的具体事件名称。
┌──────────────────┐
任务夹具 ───────▶ │ Runtime contract │
└───────┬──────────┘
│
┌───────────┴───────────┐
▼ ▼
ManualRuntime AgentsSdkRuntime
│ │
└───────────┬───────────┘
▼
同一结果检查器LangGraph 放在扩展作业
当前项目的动态部分仍然较小，Agents SDK 足以承载。若流程需要显式节点、复杂分支、图级检查点或可视化状态，可以用 LangGraph 重写 retrieve → assess → draft → approve，继续复用 BriefTask、ToolRegistry 和结果检查器。
选框架时记录它替代了哪段代码，以及引入了哪些状态格式和升级成本。框架名称不会自动提升 Agent 质量。
迁移完成的验收线
- 手写版与 SDK 版通过同一行为测试。
- Session 只在同一 thread 内续接。
- 应用预算与 Runner 的
max_turns同时生效。 - 审批、路径权限和结果验收没有移进 Prompt。
- SDK 升级先跑回归，再更新教程代码。
运行时迁移完成后，第 11 篇接入进程外能力。MCP 让 Host 发现 Server 暴露的 Tools、Resources 和 Prompts，应用仍需保留现有白名单、审批和不可信内容边界。
REFERENCES
参考链接
所属系列
AI AGENT 入门教程