单一 ReAct 循环是 Agent 的"单核"——但很多真实任务天然就是图（DAG），而不是直线。本章把 Agent 从"循环"扩展到"有向图"，让你能把分支、并行、汇合、回退写到系统里，而不是让模型现编。

11.1 为什么单 Loop 不够

让 AI 做一个"自检 + 修复一束文件"的任务，在单 loop 里跑大概是这样：

thinking: 看 5 个文件 (生成 list)
↓
tool: read 文件 1
↓
thinking: 文件 1 没问题
↓
tool: read 文件 2
↓
thinking: 文件 2 有 lint 错
↓
tool: edit 文件 2
↓
...重复 5 轮，每次都让"thinking"来做路由决策

问题：你反复让模型决定"下一步该看哪一支"，但其实流程是确定的：每个文件都看一遍、有问题就修。这把"决策权"浪费在了不必要的地方，还让模型容易漏一两个文件。

这是 Graph 工作流要解决的事：把"必然要做的形状"画成 DAG，模型只做每个节点的"实质推理"，流程通过图去接管。

11.2 Graph Agent 的三件套

┌─────────────────────────────────────────────────────┐
│  Node    : 一个 f(input) -> output 的小 Agent    │
│  Edge    : 节点之间的连接，可以是固定的         │
│             也可以是 condition 提供的分支选择 ──┘    │
│  State   : 在所有节点之间流动的共享字典             │
└─────────────────────────────────────────────────────┘

11.3 一段 30 行的极简 Graph Loop

不依赖任何框架你能写一个：

from typing import TypedDict, Callable

class State(TypedDict):
    files: list[str]
    failed: list[str]
    fixed: list[str]

def node_check(state: State) -> State:
    # 让 AI 检查这批文件，输出 failed 列表
    state["failed"] = ai_check_files(state["files"])
    return state

def node_fix(state: State) -> State:
    state["fixed"] = []
    for f in state["failed"]:
        state["fixed"].append(ai_fix_file(f))
    return state

# 边：根据 state 决定下一步去哪
def route(state: State) -> str:
    return "fix" if state["failed"] else "done"

graph = {
    "check": (node_check, route),
    "fix":   (node_fix,   "check_after_fix"),         # 固定边
    "check_after_fix": (node_check, route),
    "done":  (lambda s: s, None),
}

def run_graph(initial: State) -> State:
    state = initial
    cur, runner, router = "check", *graph["check"]
    while runner:
        state = runner(state)
        nxt   = router(state) if callable(router) else router
        runner_pair = graph.get(nxt)
        if not runner_pair:
            break
        runner, router = runner_pair
    return state

经过几百行框架代劳后的同济图结构，你看 LangGraph / OpenAI Swarm / Anthropic Computer Use 之类项目里都能识别出来这个内核。

11.4 何时把任务画成 Graph

简单决策树：

是单一 ReAct Loop 能 30 步解决的吗？
   ├─ 是  → 单 loop（用 Claude Code / OpenCode 原生就行）
   └─ 否  → 任务形状有"多个固定路径 / 并行分支 / 需要 fan-out fan-in"吗？
            ├─ 是 → Graph
            └─ 否 → 是否有所谓的 "需要在 N 个尝试中选一个 best"？有 → Graph+ 路由

11.4.1 Graph 适用场景

  • ** mandated Pipeline**：你的 CI/CD 像流水：lint → test → build → deploy → notification
  • ** fan-out / fan-in**：对 N 个文件做同样分析，再 merge
  • 分支决策：单测失败 vs 成功走不同修复路径
  • 多轮验证：fix → retest → fix → retest 这些是有 schema 的

11.4.2 不适用 Graph 的场景

  • 开放式探索：让 Agent 自己决定下一动作——回到 ReAct 即可，画成图反而死板
  • 测试级 prompt 探索：会浪费一半 model turn

11.5 三种典型 Graph 形态

11.5.1 Linear（手拉手排队）

A → B → C → D

适合简单 pipeline：

graph = {
    "extract": (node_extract, "transform"),
    "transform": (node_transform, "load"),
    "load":     (node_load,     None),
}

11.5.2 Fan-out / Fan-in（并行）

       ┌── A_1 ──┐
data ──┤── A_2 ──┤── merge
       └── A_3 ──┘

并行跑 3 路节点，merge 节点收集结果。节省墙钟时间的经典套路：

import asyncio

async def node_fanout(state: State) -> State:
    results = await asyncio.gather(*[ai_analyze(f) for f in state["files"]])
    state["analyses"] = results
    return state

11.5.3 Loop with Branch（有条件回环）

check ──┬─ failed → fix ──→ check ──┐
        └─ passed → done            ↓
          (订阅一直循环到通过)

工程上的两个底层：

def route(state: State) -> str:
    if state["failed"]:
        return "fix"
    if state["loop_count"] > 3:                       # 熔断
        return "escalate"
    return "done"

不要让 LLM 在"路由判断上"消耗 token —— 这个判断的逻辑正好用普通编程做掉，到下一节点才让 AI 认真思考。

11.6 写 Graph：组手令 vs 编程嵌入

6.1 用框架（LangGraph 风格）

from langgraph.graph import StateGraph, END

g = StateGraph(State)
g.add_node("check", node_check)
g.add_node("fix",   node_fix)
g.add_conditional_edges("check", route, {"fix": "fix", "done": END})
g.add_edge("fix", "check")
app = g.compile()

框架帮你做了"路由 + state 传播 + 错误重试"。代价是抽象的概念多（“node”、“checkpointer”、“interrupt”），学习曲线不低。

6.2 不用框架，纯 Python 协程

async def run_etl_pipeline(initial: State) -> State:
    state = initial
    state = await node_extract(state)
    state = await node_transform(state)
    while state["errors"]:
        state = await node_repair(state)
        state = await node_transform(state)
        if state["attempt"] > 5:
            raise RuntimeError("etl stalled")
    return await node_load(state)

简洁、易测。如果节点 ≤5、分支少、不需要 checkpoint 维护，推举纯协程。Graph 是视觉化辅助，不要为画图而画图。

11.7 跟 Agent harness 的关系

Claude Code / OpenCode / Codex 这些 harness 是 multi-loop + 部分 graph 路由：内部 loop 是 9/10 章的 ReAct，外部在 subagent dispatch 时是 graph（11.x） + Loop。你不用学所有框架，理解这两层就够。

下面 12 章 Harness 框架 会专门讲这些"外部架构"。

11.8 小结

  • Graph 适合：DAG pipeline / fan-out / 分支循环
  • 不适合：开放探索、单任务 30 步内能完成
  • 30 行极简 Graph 即可上手；不需要 LangGraph 也能做
  • 决策点优先交给 if/else，复杂推理才交 LLM

下一篇：《12 Harness 框架：OpenCode 与 superpowers》。

Summary: Graph 把"固定形态"从模型里拿出来当 DAG 画，模型只做干货节点的推理。