Graph Engineering with Claude — 从单循环到多 Agent 图

大多数人构建 Agent 的方式是一个 while 循环不断调用 Claude，prompt 越涨越大直到 context 窗口打满开始幻觉。90% 的人从不把工作切分给专门的子 agent——不做路由、不分支、不并行。他们跑一个循环，祈祷它能工作，然后花几周 debug 它不行的情况。

从 Loop 到 Graph 的转变，就像独立开发者加入团队——一个人单干在项目简单时很快，但一旦变复杂，就需要角色、交接和审查。

"A loop is one agent getting smarter. A graph is many agents getting coordinated."

四大构建块

无论多复杂的 Graph 都由 4 个原语组成。你不需要框架——只需要函数、字典、if 语句。

• Node（节点） — 一个只做一件事的函数。Claude 调用、工具执行、DB 查询、文件写入。取 state 进，返 state 出。不知道前后是什么。
• Edge（边） — 节点之间的连接。可以是"始终执行"或"条件执行"(if state==X)。条件边是 Graph 做决策的方式。
• Router（路由器） — 检查当前状态，选择走哪条路。不做实际工作，只做分类和分发。相当于团队的调度员。
• State（状态） — 贯穿整个 Graph 的字典。每个节点读/写它。是 Graph 的记忆——没有它，每个节点都从零开始。

# 20 行的完整 Graph 抽象
def node(func):
    """A node is just a function that takes state and returns state."""
    def wrapper(state):
        return func(state)
    return wrapper

def edge(state, next_node):
    """An edge passes state to the next node."""
    return next_node(state)

def router(state, condition, if_true, if_false):
    """A router picks a path based on state."""
    if condition(state):
        return if_true(state)
    return if_false(state)

四大时代

时代	核心技能	说明
Era 1 — Prompt Engineering	写好指令	大多数人停留在这一层
Era 2 — Context Engineering	喂对数据	开始关注上下文质量
Era 3 — Loop Engineering	构建自主 Agent	重试、验证、改进的单循环
Era 4 — Graph Engineering	协调多个 Agent	本文目标

真正在发货的人已经进入 Era 3 或 4。他们不是在写更好的 prompt，而是在设计更好的 flow。

五大 Graph 模式

1. Sequential Chain — 顺序链

Node A 运行，传给 Node B，再传给 Node C。最简单也最危险的模式——错误会三级放大传播。

def run_sequential(task):
    state = {"task": task}
    state = research_node(state)
    state = analyze_node(state)
    state = write_node(state)
    state = verify_node(state)       # 末尾验证节点不可少！
    return state

即使最简单的顺序链，末尾也必须有 verification node。如果 A 产出糟糕，B 在此基础上构建，C 再叠加——错误已经累积了三层。

2. Router — 路由器

Router 检查输入选择路径。路由是分类任务，不是推理任务。用 Haiku 做路由，用 Sonnet 做实际工作。

def router(state):
    category = call_claude(
        model="claude-haiku-4-5",  # 便宜又快
        system="Classify this task. One word: code, research, or writing.",
        prompt=state["task"]
    )
    routes = {
        "code": code_agent,       # 自己的 prompt + tools
        "research": research_agent,
        "writing": writing_agent,
    }
    return routes.get(category.strip())(state)

每个 specialist 应该有自己的 system prompt 和 tool set。给所有 agent 所有工具 = 错误工具调用 + 浪费 token。

如果路由做错，修复方法是更好的 prompt 而不是更大的模型。在 Router prompt 里写 5-6 个真实 Few-shot 示例，远超零样本 Sonnet。

3. Parallel Fan-Out — 并行扇出

多个节点同时处理同一输入，Collector 等待全部合并。总时间 = 最慢节点，不是总和。

import asyncio

async def fan_out(task):
    results = await asyncio.gather(
        research_competitor(task, "Acme"),
        research_competitor(task, "Globex"),
        research_competitor(task, "Initech"),
    )
    # Merge 节点有它自己的 prompt 和质量标准
    merged = call_claude(
        system="Synthesize these 3 competitor reports.",
        prompt=json.dumps(results)
    )
    return merged

Merge 节点最容易被轻视——合并 3 个 JSON blob 容易，合并 3 份研究报告需要自己的 prompt 和质量标准。把 merge 节点当作真正的 agent，而不是字符串拼接。

4. Loop with Gate — 循环 + 审查门

Builder 做工作 → 独立 Reviewer 检查 → 不通过则重试。自纠正 agent 系统的基石。

def loop_with_gate(task, max_tries=5):
    state = {"task": task, "history": []}
    for i in range(max_tries):
        result = builder_agent(state)      # Sonnet, creative
        check = reviewer_agent(result)     # Sonnet, adversarial
        if check["passed"]:
            return result
        state["history"].append({"attempt": i+1, "issues": check["issues"]})
    return {"status": "max_retries", "last_result": result}

关键规则：Builder 和 Reviewer 必须是不同 Agent。同一个 agent 审查自己会批准自己的错误——产生缺陷的推理也会评估它。

Reviewer 的 prompt 应该是对抗式的——"找到每一个缺陷、不一致、遗漏的边界情况"。严格的 Reviewer 抓到真正的 bug，礼貌的 Reviewer 放行一切。

5. Human-in-the-Loop — 人在回路

Graph 在特定节点暂停，等待人工批准。对昂贵、不可逆、高风险操作必须使用。

def human_gate(state):
    print(f"Action: {state['action']}")
    print(f"Scope:  {state['scope']}")
    print(f"Cost:   {state['est_cost']}")
    approval = input("Approve? [y/n]: ")
    if approval == "y":
        state["approved"] = True
    else:
        state["approved"] = False
        state["human_feedback"] = input("Reason: ")
    return state

审批节点必须展示具体信息——"发 2400 封邮件，预估 $12，审批？"要具体到人能在 5 秒 内做出真实决策。

状态流与模型分层

# 按工作匹配模型
TIERS = {
    "router":   "claude-haiku-4-5",  # 快速、便宜、分类
    "builder":  "claude-sonnet-4-6", # 推理、工具
    "reviewer": "claude-sonnet-4-6", # 严格、对抗式
    "final_qa": "claude-opus-4-6",  # 最高标准、罕见
}

# 每个节点的 State 契约
# research_node:  reads ["task"]         writes ["research"]
# analysis_node:  reads ["research"]     writes ["analysis"]
# writer_node:    reads ["analysis"]     writes ["draft"]
# reviewer_node:  reads ["draft"]        writes ["review"]

State 是扁平字典，每个节点声明它读/写哪些 key。Graph 出问题时，先检查 state——无法追踪谁写了哪个 key，就无法 debug。

用 Sonnet 做路由就像雇高级工程师去分拣邮件——可行但浪费预算。

错误处理与 Fallback 路径

def safe_node(func, state, fallback_key="error"):
    try:
        return func(state)
    except Exception as e:
        state[fallback_key] = str(e)
        state["needs_human"] = True
        return state

state = safe_node(research_node, state, "research_error")
if state.get("needs_human"):
    notify_human(state)
else:
    state = analyze_node(state)

Fallback 不是"忽略错误"，而是 Graph 中的另一条路径。用户得到的是有用的信息，而不是堆栈跟踪。

完整示例：客服工单系统

import anthropic, json
client = anthropic.Anthropic()

def classify(state):
    r = client.messages.create(
        model="claude-haiku-4-5", max_tokens=50,
        system="Classify: billing, technical, general. One word.")
    state["category"] = r.content[0].text.strip().lower()
    return state

def draft(state):
    prompts = {
        "billing": "Handle billing. Precise on refund policy.",
        "technical": "Handle tech. Ask for logs first.",
        "general": "Handle general inquiries. Brief.",
    }
    r = client.messages.create(
        model="claude-sonnet-4-6", max_tokens=1024,
        system=prompts.get(state["category"]))
    state["draft"] = r.content[0].text
    return state

def review(state):
    r = client.messages.create(
        model="claude-sonnet-4-6", max_tokens=200,
        system='Find flaws. JSON: {"passed": bool, "issues": [str]}')
    state["review"] = json.loads(r.content[0].text)
    return state

def run(ticket, max_retries=3):
    state = {"ticket": ticket, "attempts": 0}
    state = classify(state)          # Haiku router
    for _ in range(max_retries):
        state = draft(state)         # Sonnet specialist
        state = review(state)        # Sonnet reviewer
        state["attempts"] += 1
        if state["review"]["passed"]:
            send_response(state["draft"])
            return state
    state["needs_human"] = True      # Fallback
    return state

模式组合：Classify (Haiku) → Route → Draft (Sonnet) → Review (Sonnet) → Ship or retry。Fallback 到人工。不到 60 行。无框架。

6 个可立即构建的 Graph

用例	模式组合	说明
Ticket Triage	Router + Specialist + Loop	按分类/紧急度分派，各自走 specialist，通过审查才发送
Competitive Analysis	Parallel Fan-Out + Merge	每竞对一个子 agent，并行研究，merge 合成报告
PR Review Pipeline	Sequential + Reviewer	读 diff → 跑测试 → 写审查；第二 agent 再审查审查意见
Blog Post Pipeline	Sequential + Loop-with-gate	研究 → 写稿 → 编辑审查，循环直到通过
ETL with Validation	Sequential + Validator	PDF 提取 → 转换 → 加载；验证 agent 抽检行
Incident Response	Research → Gate → Execute	告警 → 读日志 → 定位根因 → 草拟修复 → 人工审批部署

5 个常见错误

❌ 巨量单块节点：一个节点做所有事。失败了全盘皆炸。拆成单一职责的小节点。

❌ 节点间无 State：每个节点从零开始，无共享字典。Graph 忘了前一个节点学过什么。

❌ 无 Fallback 路径：Happy path 工作，任何错误让整个 graph 崩溃。每个可能失败的节点建一条 fallback。

❌ Sonnet 做路由：路由是分类，Haiku 更快更便宜。把 Sonnet 用在真正需要的地方。

❌ 跳过审查门：第一个错误答案发送给客户后，没人再信任这个系统。

Grok Build 实践对照

文章概念	Grok Build 实践	位置
Node	Subagent 系统 — 3 个内置 type	16-subagents.md
Edge	resume_from + background	subagent 参数
Router	Agent Types + 模型切换	[subagents.models]
State	Opendray Memory — 16 个工具	MCP server
Sequential	resume_from 链式调用	16-subagents.md
Router Pattern	deepseek-flash → deepseek-pro → qwen3.5-reviewer	config.toml
Parallel Fan-Out	背景 subagent 并发	20-background-tasks.md
Loop with Gate	Goal Orchestrator — 6 模块 ~15K 行	benchmarks 分析
Human-in-Loop	Permission modes — Yolo/Auto/Manual	15-agent-mode.md
Model Tiering	7 个模型分层配置	config.toml
Fallback Paths	capability_mode + worktree	16-subagents.md
Reviewer/Verifier	N skepticism 并行验证	benchmarks

✅ Sequential Chain · ✅ Router Pattern · ✅ Parallel Fan-Out · ✅ Loop with Gate · ✅ Human-in-Loop

建议采纳的最佳实践

• 🔴 Router Few-shot 优化（低优先级）— Router prompt 中加入 5-6 个真实分类示例，替代纯零样本分类。
• 🟡 Merge 节点专项化（中优先级）— Parallel Fan-Out 后的 Merge 应享有一级 agent 地位——自己的 prompt + model tier。
• 🔴 Reviewer 对抗式 Prompt（低优先级）— GoalClassifier 中的 skeptics 采用"对抗式审查"system prompt 模板。
• 🟡 State Contract 显式声明（中优先级）— 每个 subagent 声明读/写的 key（类似接口签名），方便调试 state 问题。
• 🔴 Fallback 路径可视化（低优先级）— 对单个 graph 的 fallback 路径做可视化文档。

---

"Two kinds of builders in 2026. Ones still running one loop for everything. Ones designing the flow. The model is the same. The architecture is not."