# Graph Engineering with Claude: the 11-Step Roadmap From Loops to Graph Architect
**作者**: 0xRafy
**日期**: 2026-07-21T12:22:25.000Z
**来源**: [https://x.com/0xRafy/status/2079542513317118268](https://x.com/0xRafy/status/2079542513317118268)
---
Most people who try to build an AI agent end up with a single while-loop calling Claude with a growing prompt until it hits the context ceiling and starts hallucinating. 9 out of 10 never split the work across specialized agents.
They don't route. They don't branch. They don't parallelize. They run one loop, pray it works, and debug it for weeks when it doesn't.
This is the 11-step roadmap that turns that single loop into a graph that fans out, verifies itself, and converges on a real output.
> Follow my Substack to get fresh AI alpha: movez.substack.com
Loop engineering was a real step forward. Instead of prompting Claude once and copying the output, you built a system that retried, verified, and improved with each pass. That was era three.
But a loop is still one agent doing one job. When the task gets complex, one loop drowns in context, mixes concerns, and tries to be everything at once. Graph engineering is the next layer: you design the flow.
Which agent runs. What happens when it fails. Where results go next. When a human steps in. This article is the complete playbook.
A loop is one agent getting smarter. A graph is many agents getting coordinated.
# 01. Why one loop is not enough
A loop gives one agent autonomy. It retries, it learns from failures, it improves with each pass. For single-purpose tasks - fix this bug, summarize this doc, clean this dataset - a loop is all you need. You define the goal, attach a verifier, set a stop condition, and let it run.
The problem starts when the task has multiple concerns. Research + analysis + code + review. One agent trying to do all four fills its context window with mixed information, loses track of earlier steps, and produces work that looks complete but falls apart under inspection.
The research bleeds into the analysis. The code ignores the review. The agent is doing four jobs and has the context budget of one.
This is not a model limitation. Sonnet and Opus are more than capable of each individual task. The limitation is architectural. You are asking one worker to be the researcher, the analyst, the builder, and the reviewer simultaneously, all in one conversation.
The fix is not a bigger context window. The fix is splitting the work across agents that each do one thing well, and wiring them together so the output of one becomes the input of the next.
> The shift from loop to graph is the same shift a solo developer makes when they join a team. One person doing everything is fast until the project gets complex. Then you need roles, handoffs, and reviews.
# 02. The four building blocks
Every graph, no matter how complex, is assembled from four primitives. You do not need a framework. You need functions, a dictionary, and if-statements.
- Node. A function that does one thing. A Claude call, a tool execution, a database query, a file write. It takes state in and returns state out. The node does not know what came before it or what comes after. It only knows its own job.
- Edge. A connection between two nodes. "After Research, run Analysis." Edges can be unconditional (always) or conditional (only if state["status"] == "needs_more_data"). Conditional edges are how graphs make decisions.
- Router. A special node that inspects the current state and picks which path to take. It does not do work itself. It classifies the input and sends it to the right specialist. Think of it as a dispatcher, not a worker.
- State. A dictionary that flows through the entire graph. Every node reads from it and writes to it. State is the graph's memory. Without it, each node starts from scratch and the graph has no idea what already happened.
```
# The entire graph abstraction in 20 lines
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)
```
# 03. The four eras
Each era of AI engineering absorbed the one before it. Prompt engineering taught us to write good instructions. Context engineering taught us to feed the right data. Loop engineering taught us to build one autonomous agent.
Graph engineering teaches us to wire many agents into a coordinated system.
Most people are still in era one or two. They write a prompt, maybe attach some context, and expect a single Claude call to do the whole job. The builders who are shipping real products right now are in era three or four. They are not writing better prompts. They are designing better flows.
# 04. Pattern: Sequential chain
The simplest graph. Node A runs, passes its output to Node B, which passes to Node C. No branching, no routing. Most "AI pipelines" are this.
Sequential works when every step always succeeds and the order never changes. Data extraction, format conversion, translate-then-summarize. The moment any step can fail or the path can vary, you need one of the other four patterns.
The biggest risk with sequential chains is error propagation. If Node A produces bad output, Node B builds on that bad output, and Node C builds on that. By the end, the error has compounded through three layers.
This is why even simple sequential graphs benefit from a verification node at the end.
```
def run_sequential(task):
state = {"task": task}
state = research_node(state) # reads task, writes research
state = analyze_node(state) # reads research, writes analysis
state = write_node(state) # reads analysis, writes report
state = verify_node(state) # reads report, writes passed/failed
return state
```
# 05. Pattern: Router
A router inspects the input and picks one of several paths. Not every node runs. The router selects the right specialist for the job. This is how you build a system that handles multiple types of tasks without loading every tool into one agent.
The key insight: routing is a classification task, not a reasoning task. You do not need Sonnet to decide whether a ticket is about billing or a bug. Haiku handles it in milliseconds for a fraction of the cost. Save the expensive model for the actual work inside the specialist.
Each specialist should have its own system prompt and its own tool set. A code agent needs run_tests and write_file. A research agent needs web_search and read_doc. Giving every agent every tool means wrong tool calls, wasted tokens, and confused output.
```
def router(state):
# Cheap model for classification
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, # own prompt, own tools
"research": research_agent, # own prompt, own tools
"writing": writing_agent, # own prompt, own tools
}
agent = routes.get(category.strip(), research_agent)
return agent(state)
```
> Pro nuance: if your router makes wrong calls, the fix is usually a better system prompt, not a bigger model.
> Write 5-6 real examples of each category into the router's prompt. Few-shot classification on Haiku outperforms zero-shot on Sonnet for routing tasks.
# 06. Pattern: parallel fan-out
Multiple nodes run at the same time on the same input. A collector waits for all of them and merges the results. Three agents running in parallel means your total time is the time of the slowest agent, not the sum of all three.
The constraint is independence. If Agent B needs Agent A's output, you cannot parallelize them. If both work on the same input and produce separate outputs, you can. Competitive analysis is the clean example: one agent per competitor, all running simultaneously, results merge at the end.
The merge node is where most people underinvest. Merging three JSON blobs is easy. Merging three research reports into a coherent synthesis requires its own prompt and its own quality bar. Treat the merge node like a real agent, not a string concatenation.
```
import asyncio
async def fan_out(task):
results = await asyncio.gather(
research_competitor(task, "Acme"),
research_competitor(task, "Globex"),
research_competitor(task, "Initech"),
)
# Merge is its own Claude call with a synthesis prompt
merged = call_claude(
system="Synthesize these 3 competitor reports into a comparison matrix.",
prompt=json.dumps(results)
)
return merged
```
# 07. Pattern: loop with gate
The builder agent does the work. A separate reviewer agent checks it. If the review fails, the builder retries with the failure reason appended to state. This is the pattern behind every self-correcting agent system.
The critical rule: the builder and the reviewer must be different agents. Different system prompts, often different model tiers. The builder's job is to produce. The reviewer's job is to reject. If the same agent reviews its own work, it approves its own mistakes, because the same reasoning that produced the flaw is the same reasoning evaluating it.
The reviewer's prompt should be adversarial. Not "is this good?" but "find every flaw, inconsistency, and missing edge case. If you find nothing wrong, respond with {passed: true}. If you are unsure, fail it." A strict reviewer catches real bugs. A polite reviewer waves everything through.
```
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}
```
# 08. Pattern: human-in-the-loop
The graph pauses at a specific node and waits for human approval before continuing. The agent does the research and drafts the action. The human reviews the critical decision. The agent executes after approval.
This pattern is mandatory for anything expensive, irreversible, or high-stakes. Sending emails to customers, deploying to production, executing financial transactions, deleting data. The graph handles the work. The human handles the judgment call.
The approval node should show the human exactly what will happen and what it will cost. Not "do you approve?" but "this action will send 2,400 emails to the billing segment with this subject line and this body. Estimated cost: $12. Approve?"
Specific enough that the human can make a real decision in 5 seconds.
```
def human_gate(state):
# Show exactly what will happen
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
```
# 09. State flow and model tiering
State is a flat dictionary that every node reads from and writes to. Keep it explicit: every node should declare what keys it reads and what keys it writes. When a graph breaks, the first thing you check is state. If you cannot trace which node wrote which key, you cannot debug anything.
Model tiering is the second production concern. Not every node needs the same model. A router classifies: Haiku. A builder reasons: Sonnet. A final quality gate on high-stakes output: Opus. Using Sonnet for routing is like hiring a senior engineer to sort mail. It works, but you are burning budget on the wrong task.
```
# Match the model to the job
TIERS = {
"router": "claude-haiku-4-5", # fast, cheap, classify
"builder": "claude-sonnet-4-6", # reasoning, tools
"reviewer": "claude-sonnet-4-6", # strict, adversarial
"final_qa": "claude-opus-4-6", # highest bar, rare
}
# State contract for every node
# research_node: reads ["task"] writes ["research"]
# analysis_node: reads ["research"] writes ["analysis"]
# writer_node: reads ["analysis"] writes ["draft"]
# reviewer_node: reads ["draft"] writes ["review"]
```
# 10. Fallback paths and error handling
The happy path works. Any unexpected error crashes the whole graph. Production graphs need a fallback edge for every node that can fail.
A fallback is not "ignore the error." It is a separate path in the graph that handles the failure gracefully. If the research agent times out, return a partial result and flag it. If the reviewer crashes, skip automatic review and route to human review. If the whole graph fails after max retries, save state to disk and send an alert. The user gets something useful instead of a stack trace.
```
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
# In the graph
state = safe_node(research_node, state, "research_error")
if state.get("needs_human"):
notify_human(state)
else:
state = analyze_node(state)
```
# 11. Full working example
Here is a complete graph that handles customer support tickets. It classifies the ticket with Haiku, routes to a specialist with Sonnet, the specialist drafts a response, a reviewer checks it, and if approved the response ships.
If rejected, the specialist retries with the reviewer's feedback. All five patterns in one 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.",
messages=[{"role": "user", "content": state["ticket"]}]
)
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"], prompts["general"]),
messages=[{"role": "user", "content": state["ticket"]}]
)
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 in this support response. JSON: {\"passed\": bool, \"issues\": [str]}",
messages=[{"role": "user", "content": state["draft"]}]
)
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
# Fallback: max retries hit, route to human
state["needs_human"] = True
return state
```
Classify (Haiku) -> Route -> Draft (Sonnet) -> Review (Sonnet) -> Ship or retry. Fallback to human after max retries. Under 60 lines. No framework.
# 6 graphs to build with Claude this week
- Ticket triage: Claude classifies incoming tickets by category and urgency. Routes each to a specialist agent with domain-specific instructions. Reviewer checks before sending.
- Competitive analysis: Claude spawns one sub-agent per competitor. All agents research in parallel. A merge node synthesizes findings into a single report.
- PR review pipeline: Claude reads the diff, runs tests, writes review comments. A second agent reviews the review for false positives before posting.
- Blog post pipeline: Research agent gathers sources. Writer agent drafts. Editor agent checks facts and tone. Loop retries until editor passes.
- ETL with validation: Claude extracts data from PDFs, transforms schema, loads to database. Validator agent samples rows and flags anomalies before commit.
- Incident response: Alert triggers the graph. Claude reads logs, identifies root cause, drafts a fix and a postmortem. Human approves the fix before deploy.
# Five ways people break their first graph
- × Giant monolith node. One node doing everything. If it fails, everything fails. Break it into smaller nodes with single responsibilities.
- × No state between nodes. Each node starts from scratch. No shared dictionary, no context passed forward. The graph forgets what the previous node learned.
- × No fallback path. The happy path works. Any error crashes the whole graph. Always build a fallback edge.
- × Sonnet for routing. Routing is classification. Haiku does it faster and cheaper. Use Sonnet where it matters: building and reviewing.
- × Skipping the gate. No review node. The first wrong answer sent to a customer is the last time anyone trusts the system.
# Conclusion:
A loop is an agent. A graph is a team.
Loop engineering was the breakthrough that turned prompting into building. You learned to make one agent that retries, verifies, and improves. That was a real skill. It still is.
But every team outperforms every solo operator once the work gets complex. A graph is how you build a team of agents: a researcher who gathers, an analyst who reasons, a writer who drafts, a reviewer who rejects. Each agent is simple. The graph makes them powerful.
The code in this article has no framework dependency. Functions, dictionaries, if-statements. That is a graph. You do not need LangGraph. You do not need CrewAI. You need to understand the five patterns and wire them together for your use case.
> 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.
## 相关链接
- [0xRafy](https://x.com/0xRafy)
- [@0xRafy](https://x.com/0xRafy)
- [118K](https://x.com/0xRafy/status/2079542513317118268/analytics)
- [movez.substack.com](https://movez.substack.com/)
- [Upgrade to Premium](https://x.com/i/premium_sign_up)
- [8:22 PM · Jul 21, 2026](https://x.com/0xRafy/status/2079542513317118268)
- [118.5K Views](https://x.com/0xRafy/status/2079542513317118268/analytics)
- [View quotes](https://x.com/0xRafy/status/2079542513317118268/quotes)
---
*导出时间: 2026/7/22 08:41:26*
---
## 中文翻译
# 使用 Claude 进行图工程：从循环到图架构师的 11 步路线图
**作者**: 0xRafy
**日期**: 2026-07-21T12:22:25.000Z
**来源**: [https://x.com/0xRafy/status/2079542513317118268](https://x.com/0xRafy/status/2079542513317118268)
---
大多数尝试构建 AI 智能体的人，最终得到的只是一个单一的 while 循环，不断调用 Claude 并伴随着越来越长的提示词，直到触及上下文上限并开始产生幻觉。十个人里有九个从未将工作拆分给专门的智能体。
他们不进行路由。不分叉。不并行化。他们运行一个循环，祈祷它能成功，一旦失败就调试好几个星期。
这就是一份 11 步的路线图，能将那个单一的循环变成一个能够发散、自我验证并收敛为真实输出的图。
> 关注我的 Substack 获取最新的 AI alpha：movez.substack.com
循环工程确实是一个真正的进步。与其提示 Claude 一次并复制输出，你构建了一个在每次传递中重试、验证和改进的系统。那是第三个时代。
但循环仍然是一个智能体在做一份工作。当任务变得复杂时，一个循环会淹没在上下文中，混杂各种关注点，并试图同时包揽一切。图工程是下一层：你设计流程。
哪个智能体运行。失败时发生什么。结果去向何处。人类何时介入。这篇文章就是完整的实战指南。
循环是一个智能体变得更聪明。图是许多智能体变得更协调。
# 01. 为什么一个循环不够
循环赋予一个智能体自主权。它重试，从失败中学习，每次通过都有所改进。对于单一用途的任务——修复这个 bug、总结这份文档、清理这个数据集——一个循环就足够了。你定义目标，附加一个验证器，设置停止条件，然后让它运行。
当任务涉及多个关注点时，问题就出现了。研究 + 分析 + 代码 + 审查。一个试图做这四件事的智能体会用混杂的信息填满其上下文窗口，丢失对先前步骤的跟踪，并产生看起来完整但经不起检查的工作成果。
研究渗透进分析。代码无视审查。这个智能体在做四份工作，却只有一份工作的上下文预算。
这不是模型的局限性。Sonnet 和 Opus 完全有能力完成每项单独的任务。局限性在于架构。你要求一个工人在同一次对话中同时充当研究人员、分析师、构建者和审查者。
解决方法不是更大的上下文窗口。解决方法是将工作拆分给各自擅长一件事的智能体，并将它们连接起来，使一个的输出成为下一个的输入。
> 从循环到图的转变，就像一个独立开发者加入团队时发生的转变一样。一个人做所有事情在项目变复杂之前很快。然后你需要角色、交接和审查。
# 02. 四个构建块
无论多复杂，每个图都是由四个基元组装而成的。你不需要框架。你需要函数、字典和 if 语句。
- 节点。一个做一件事的函数。一次 Claude 调用、一次工具执行、一次数据库查询、一次文件写入。它接收状态并返回状态。节点不知道在它之前发生了什么或之后会发生什么。它只知道自己的工作。
- 边。两个节点之间的连接。“研究之后，运行分析。”边可以是无条件的（总是）或有条件的（仅当 state["status"] == "needs_more_data"）。条件边是图进行决策的方式。
- 路由器。一个检查当前状态并选择走哪条路径的特殊节点。它自己不工作。它对输入进行分类并将其发送给正确的专家。把它想象成一个调度员，而不是工人。
- 状态。流经整个图的字典。每个节点都从中读取并向其写入。状态是图的记忆。没有它，每个节点都要从头开始，图也不知道已经发生了什么。
```
# 20 行代码实现的整个图抽象
def node(func):
"""节点只是一个接收状态并返回状态的函数。"""
def wrapper(state):
return func(state)
return wrapper
def edge(state, next_node):
"""边将状态传递给下一个节点。"""
return next_node(state)
def router(state, condition, if_true, if_false):
"""路由器根据状态选择一条路径。"""
if condition(state):
return if_true(state)
return if_false(state)
```
# 03. 四个时代
AI 工程的每个时代都吸收了前一个时代。提示工程教会我们编写好的指令。上下文工程教会我们输入正确的数据。循环工程教会我们构建一个自主智能体。
图工程教会我们将许多智能体连接成一个协调的系统。
大多数人还停留在第一个或第二个时代。他们编写一个提示词，也许附加上下文，并期望一次 Claude 调用就能完成整个工作。那些正在交付真实产品的构建者处于第三个或第四个时代。他们不是在编写更好的提示词。他们是在设计更好的流程。
# 04. 模式：顺序链
最简单的图。节点 A 运行，将其输出传递给节点 B，节点 B 再传递给节点 C。没有分叉，没有路由。大多数“AI 管道”都是这种形式。
当每一步总是成功且顺序从不改变时，顺序链是有效的。数据提取、格式转换、先翻译后总结。一旦任何步骤可能失败或路径可能变化，你就需要其他四种模式中的一种。
顺序链最大的风险是错误传播。如果节点 A 产生错误的输出，节点 B 建立在该错误的输出上，节点 C 又建立在节点 B 之上。到最后，错误已经通过三层累积了。
这就是为什么即使是简单的顺序图也能从末尾的验证节点中受益。
```
def run_sequential(task):
state = {"task": task}
state = research_node(state) # 读取 task，写入 research
state = analyze_node(state) # 读取 research，写入 analysis
state = write_node(state) # 读取 analysis，写入 report
state = verify_node(state) # 读取 report，写入 passed/failed
return state
```
# 05. 模式：路由器
路由器检查输入并从几条路径中选择一条。不是每个节点都运行。路由器为工作选择合适的专家。这就是你如何构建一个能处理多种任务类型而不必将每个工具都加载到一个智能体中的系统。
关键见解：路由是一个分类任务，而不是推理任务。你不需要 Sonnet 来决定一个工单是关于计费还是 bug。Haiku 可以在几毫秒内以极低的成本处理它。把昂贵的模型留给专家内部的实际工作。
每个专家应该有自己的系统提示词和自己的工具集。代码智能体需要 run_tests 和 write_file。研究智能体需要 web_search 和 read_doc。给每个智能体每个工具意味着错误的工具调用、浪费的令牌和困惑的输出。
```
def router(state):
# 用于分类的廉价模型
category = call_claude(
model="claude-haiku-4-5",
system="将此任务分类。一个词：code、research 或 writing。",
prompt=state["task"]
)
routes = {
"code": code_agent, # 自己的提示词，自己的工具
"research": research_agent, # 自己的提示词，自己的工具
"writing": writing_agent, # 自己的提示词，自己的工具
}
agent = routes.get(category.strip(), research_agent)
return agent(state)
```
> 专业提示：如果你的路由器做出错误的调用，解决方法通常是一个更好的系统提示词，而不是更大的模型。
> 将每个类别的 5-6 个真实示例写入路由器的提示词中。对于路由任务，Haiku 上的少样本分类优于 Sonnet 上的零样本分类。
# 06. 模式：并行发散
多个节点同时基于同一输入运行。一个收集器等待它们全部完成并合并结果。三个智能体并行运行意味着你的总时间是最慢智能体的时间，而不是三个的总和。
约束条件是独立性。如果智能体 B 需要智能体 A 的输出，你就不能并行化它们。如果两者都基于同一输入工作并产生单独的输出，你就可以。竞争分析是一个清晰的例子：每个竞争对手一个智能体，全部同时运行，最后合并结果。
合并节点是大多数人投入不足的地方。合并三个 JSON 块很容易。将三个研究报告合并成一个连贯的综合分析需要它自己的提示词和它自己的质量标准。把合并节点当作一个真正的智能体，而不是字符串连接。
```
import asyncio
async def fan_out(task):
results = await asyncio.gather(
research_competitor(task, "Acme"),
research_competitor(task, "Globex"),
research_competitor(task, "Initech"),
)
# 合并是它自己的 Claude 调用，带有综合提示词
merged = call_claude(
system="将这 3 份竞争对手报告综合成一个对比矩阵。",
prompt=json.dumps(results)
)
return merged
```
# 07. 模式：带守门的循环
构建者智能体执行工作。一个单独的审查者智能体检查它。如果审查失败，构建者会将失败原因附加到状态后重试。这是每个自我纠正智能体系统背后的模式。
关键规则：构建者和审查者必须是不同的智能体。不同的系统提示词，通常是不同的模型层级。构建者的工作是生产。审查者的工作是拒绝。如果同一个智能体审查自己的工作，它会认可自己的错误，因为产生缺陷的推理与评估它的推理是相同的。
审查者的提示词应该是对抗性的。不是“这个好吗？”，而是“找出每一个缺陷、不一致和缺失的边缘情况。如果你没发现任何问题，请响应 {passed: true}。如果你不确定，就让它失败。”严格的审查者能捕捉到真正的 bug。礼貌的审查者会让一切通过。
```
def loop_with_gate(task, max_tries=5):
state = {"task": task, "history": []}
for i in range(max_tries):
result = builder_agent(state) # Sonnet，创造性
check = reviewer_agent(result) # Sonnet，对抗性
if check["passed"]:
return result
state["history"].append({
"attempt": i + 1,
"issues": check["issues"]
})
return {"status": "max_retries", "last_result": result}
```
# 08. 模式：人机协同
图在一个特定节点暂停，在继续之前等待人工批准。智能体进行研究并起草行动。人类审查关键决策。智能体在批准后执行。
对于任何昂贵、不可逆或高风险的事情，这种模式都是强制性的。向客户发送电子邮件、部署到生产环境、执行金融交易、删除数据。图处理工作。人类处理判断。
批准节点应该确切地向人类展示将会发生什么以及它将花费什么。不是“你批准吗？”，而是“此操作将使用此主题行和正文向计费细分群体发送 2,400 封电子邮件。预计成本：12 美元。批准？”
足够具体，以便人类可以在 5 秒内做出真正的决定。
```
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
```
# 09. 状态流和模型分层
状态是一个扁平的字典，每个节点都从中读取并向其写入。保持显式：每个节点都应该声明它读取哪些键以及写入哪些键。当图崩溃时，你首先要检查的就是状态。如果你无法追踪哪个节点写入了哪个键，你就无法调试任何东西。
模型分层是第二个生产关注点。不是每个节点都需要相同的模型。路由器分类：Haiku。构建者推理：Sonnet。高风险输出的最终质量门：Opus。用 Sonnet 做路由就像雇佣一名高级工程师来分拣邮件。这行得通，但你在错误的任务上烧钱。
```
# 将模型与工作匹配
TIERS = {
"router": "claude-haiku-4-5", # 快速，便宜，分类
"builder": "claude-sonnet-4-6", # 推理，工具
"reviewer": "claude-sonnet-4-6", # 严格，对抗性
"final_qa": "claude-opus-4-6", # 最高标准，罕见
}
# 每个节点的状态契约
# research_node: 读取 ["task"] 写入 ["research"]
# analysis_node: 读取 ["research"] 写入 ["analysis"]
# writer_node: 读取 ["analysis"] 写入 ["draft"]
# reviewer_node: 读取 ["draft"] 写入 ["review"]
```
# 10. 回退路径和错误处理
快乐路径是有效的。任何意想不到的错误都会使整个图崩溃。生产图需要为每个可能失败的节点设置一条回退边。
回退不是“忽略错误”。它是图中优雅处理失败的单独路径。如果研究智能体超时，返回部分结果并标记它。如果审查者崩溃，跳过自动审查并路由到人工审查。如果整个图在最大重试次数后失败，将状态保存到磁盘并发送警报。用户得到一些有用的东西，而不是堆栈跟踪。
```
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)
```
# 11. 完整的工作示例
这是一个完整的图，用于处理客户支持工单。它使用 Haiku 对工单进行分类，使用 Sonnet 路由到专家，专家起草回复，审查者检查它，如果批准则发送回复。
如果被拒绝，专家会根据审查者的反馈重试。所有五种模式都在一个图中。
```
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.",
messages=[{"role": "user", "content": state["ticket"]}]
)
state["category"] = r.content[0].text.strip().lower()
return state
def draft(state):
prompts = {
"billing": "Handle billing. Precise on refund pol