一句话摘要
从循环进阶图架构师的路线图
用 Claude 做图工程:从循环到图架构师的 11 步路线图 July 21, 2026 · 14分钟阅读 · 查看原文 ↗ Claude AI 循环 金融
大多数尝试构建 AI agent 的人,最终都会得到一个单一的 while 循环——用一个不断膨胀的 prompt 调用 Claude,直到它撞上上下文上限、开始幻觉。十个里有九个,从没把工作拆分到专业化的 agent 身上。
他们不做路由。他们不分叉。他们不并行。他们跑一个循环,祈祷它能成,成不了就调试几个星期。
这就是那 11 步路线图,它把那个单一循环,变成一个能扇出、能自我验证、并收敛到一个真实输出的图。
> 关注我的 Substack,获取新鲜的 AI alpha:movez.substack.com
循环工程是真正的进步。你不是给 Claude 一次 prompt 然后复制输出,而是构建一个每次迭代都重试、验证、改进的系统。那是第三个时代。
但一个循环,仍然是一个 agent 在做一份工作。当任务变复杂时,一个循环会在上下文里溺水,把关注点混在一起,试图同时成为一切。图工程是下一层:你设计流程。
哪个 agent 运行。它失败时会发生什么。结果接下来去哪。人什么时候介入。这篇文章就是完整的打法。
一个循环,是一个 agent 变得更聪明。一个图,是许多 agent 被协调起来。
01. 为什么一个循环不够
一个循环给一个 agent 自主权。它重试,它从失败中学习,它每次迭代都在改进。对于单一目的的任务——修这个 bug、总结这份文档、清洗这个数据集——一个循环就够了。你定义目标,挂一个验证器,设一个停止条件,让它跑。
问题始于任务有多个关注点的时候。研究 + 分析 + 代码 + 审阅。一个 agent 试图做全部四件事,会用混杂的信息填满它的上下文窗口,忘掉前面的步骤,产出看起来完整、但一经检查就散架的工作。
研究渗进了分析。代码无视了审阅。这个 agent 在做四份工作,却只有一个工作的上下文预算。
这不是模型的局限。Sonnet 和 Opus 完全有能力做好每一项单独的任务。局限是架构性的。你在要求一个工人,在同一个对话里,同时当研究者、分析师、构建者和审阅者。
修复的办法不是一个更大的上下文窗口。修复的办法是把工作拆分到各自把一件事做好的 agent 身上,并把它们串联起来,让一个的输出成为下一个的输入。
> 从循环到图的转变,和一个独立开发者加入团队时的转变是一样的。一个人做所有事,在项目变复杂之前是快的。然后你就需要角色、交接和审阅。
02. 四个构建块
每一个图,无论多复杂,都由四个原语组装而成。你不需要一个框架。你需要函数、一个字典、和 if 语句。
节点(Node)。一个只做一件事的函数。一次 Claude 调用、一次工具执行、一次数据库查询、一次文件写入。它接收状态、返回状态。这个节点不知道它之前是什么、之后是什么。它只知道自己的工作。
边(Edge)。两个节点之间的连接。"研究之后,运行分析。"边可以是无条件的(总是),也可以是有条件的(只有当 state["status"] == "needs_more_data" 时)。有条件的边,是图做决策的方式。
路由器(Router)。一个特殊节点,检查当前状态,选择走哪条路径。它自己不做工作。它对输入做分类,把它送到正确的专家那里。把它想成一个调度员,而不是工人。
状态(State)。一个流经整个图的字典。每个节点都从它读、向它写。状态是图的记忆。没有它,每个节点都从零开始,图就完全不知道已经发生了什么。
# 整个图抽象,20 行代码
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. 四个时代
AI 工程的每个时代,都吸收了它之前的那个。prompt 工程教我们写好指令。上下文工程教我们喂对数据。循环工程教我们构建一个自主的 agent。
图工程教我们把许多 agent 连成一个协调的系统。
大多数人还在第一或第二个时代。他们写一个 prompt,也许附上一点上下文,然后指望一次 Claude 调用就把整件事做完。现在真正在交付产品的构建者,处于第三或第四个时代。他们不是在写更好的 prompt。他们在设计更好的流程。
04. 模式:顺序链
最简单的图。节点 A 运行,把输出传给节点 B,节点 B 再传给节点 C。不分叉,不路由。大多数"AI 流水线"都是这个。
顺序在每一步都必然成功、顺序永不改变的时候才成立。数据抽取、格式转换、先翻译再总结。一旦某一步可能失败、或者路径可能变化,你就需要另外四种模式里的某一种。
顺序链最大的风险是错误传播。如果节点 A 产出坏输出,节点 B 就在坏输出上构建,节点 C 又在上面构建。到最后,错误已经复合了三层。
这就是为什么即便简单的顺序图,也受益于在末尾放一个验证节点。
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. 模式:路由器
路由器检查输入,从几条路径里挑一条。不是每个节点都运行。路由器为这份工作选出正确的专家。这就是你构建一个"能处理多种任务、又不用把所有工具塞进一个 agent"的系统的方式。
关键洞察:路由是一个分类任务,不是一个推理任务。你不需要 Sonnet 来决定一个工单是关于账单还是关于 bug。Haiku 在几毫秒内、用几分之一的成本就搞定了。把昂贵的模型留给专家内部的真正工作。
每个专家都应该有自己的 system prompt 和自己的工具集。一个代码 agent 需要 run_tests 和 write_file。一个研究 agent 需要 web_search 和 read_doc。给每个 agent 所有工具,意味着错误的工具调用、浪费的 token、和混乱的输出。
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)
> 进阶细节:如果你的路由器判断错了,修复办法通常是一个更好的 system prompt,而不是一个更大的模型。
> 往路由器的 prompt 里写进每一类的 5-6 个真实例子。对路由任务来说,Haiku 上的少样本分类,胜过 Sonnet 上的零样本。
06. 模式:并行扇出
多个节点在同一个输入上同时运行。一个收集器等待所有结果并把它们合并。三个 agent 并行跑,意味着你的总时间是其中最慢那个 agent 的时间,而不是三者之和。
约束是独立性。如果 agent B 需要 agent A 的输出,你就不能并行它们。如果两者在同一个输入上工作、产出各自独立的输出,你就可以。竞品分析是最干净的例子:每个竞品一个 agent,全部同时跑,结果在最后合并。
合并节点是大多数人投入不足的地方。合并三个 JSON blob 很容易。把三份研究报告合并成一份连贯的综合,需要它自己的 prompt 和它自己的质量标准。把合并节点当成一个真正的 agent 来对待,而不是一次字符串拼接。
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. 模式:带闸门的循环
构建者 agent 做工作。一个独立的审阅者 agent 检查它。如果审阅失败,构建者带着追加到状态里的失败原因重试。这就是每个自我纠错 agent 系统背后的模式。
关键规则:构建者和审阅者必须是不同的 agent。不同的 system prompt,常常还是不同的模型层级。构建者的工作是产出。审阅者的工作是拒绝。如果同一个 agent 审阅自己的工作,它会批准自己的错误,因为产出缺陷的同一套推理,正是评估它的那套推理。
审阅者的 prompt 应该是对抗性的。不是"这个好吗?"而是"找出每一个缺陷、不一致和遗漏的边缘情况。如果你找不到任何问题,就回复 {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, 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. 模式:人在环中
图在某个特定节点暂停,等待人的批准才继续。agent 做研究、起草动作。人审阅那个关键决定。agent 在批准后执行。
这个模式对任何昂贵、不可逆、或高风险的事情都是强制性的。给客户发邮件、部署到生产、执行金融交易、删除数据。图处理工作。人处理判断。
批准节点应该向人确切展示将要发生什么、以及它的代价。不是"你批准吗?"而是"这个动作将向账单分段发送 2,400 封邮件,主题行是这个、正文是这个。预估成本:$12。批准?"
具体到让人能在 5 秒内做一个真正的决定。
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. 状态流与模型分层
状态是一个扁平字典,每个节点都从它读、向它写。保持它显式:每个节点都应该声明它读哪些键、写哪些键。当一个图坏了,你第一件要检查的就是状态。如果你追踪不到哪个节点写了哪个键,你就什么都调试不了。
模型分层是第二个生产层面的关注点。不是每个节点都需要同一个模型。路由器做分类:Haiku。构建者做推理:Sonnet。高风险输出上的最终质量闸门:Opus。用 Sonnet 做路由,就像雇一个高级工程师去分拣邮件。它行得通,但你在把预算烧在错误的任务上。
# 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. 回退路径与错误处理
顺利路径能跑。任何意外错误都会让整个图崩掉。生产级的图,需要给每一个可能失败的节点配一条回退边。
回退不是"无视错误"。它是图里一条独立的路径,优雅地处理那个失败。如果研究 agent 超时,返回一个部分结果并标出来。如果审阅者崩了,跳过自动审阅、路由到人工审阅。如果整个图在最大重试后失败,把状态存到磁盘并发送警报。用户得到的是某个有用的东西,而不是一个堆栈跟踪。
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. 完整可运行示例
下面是一个处理客服工单的完整图。它用 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 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
分类(Haiku)-> 路由 -> 起草(Sonnet)-> 审阅(Sonnet)-> 发出或重试。超过最大重试后回退到人。60 行以内。没有框架。
这周用 Claude 构建 6 个图
工单分诊:Claude 按类别和紧急度对进来的工单分类。把每个路由到一个带领域专属指令的专家 agent。发送前由审阅者检查。
竞品分析:Claude 为每个竞品生成一个子 agent。所有 agent 并行研究。一个合并节点把发现综合成一份报告。
PR 审阅流水线:Claude 读 diff、跑测试、写审阅评论。第二个 agent 在发布之前,审阅这份审阅,找出误报。
博客文章流水线:研究 agent 收集来源。写作 agent 起草。编辑 agent 检查事实和语气。循环重试,直到编辑通过。
带校验的 ETL:Claude 从 PDF 抽取数据,转换 schema,加载到数据库。校验 agent 抽样行,在提交前标出异常。
事故响应:警报触发图。Claude 读日志、定位根因、起草修复和一份事后复盘。人在部署前批准修复。
人们搞砸第一个图的五种方式
× 巨型单体节点。一个节点做所有事。它一失败,所有都失败。把它拆成更小、职责单一的节点。
× 节点之间没有状态。每个节点从零开始。没有共享字典、没有向前传递的上下文。图忘掉了上一个节点学到的东西。
× 没有回退路径。顺利路径能跑。任何错误都让整个图崩掉。永远建一条回退边。
× 用 Sonnet 做路由。路由是分类。Haiku 更快更便宜。把 Sonnet 用在它真正重要的地方:构建和审阅。
× 跳过闸门。没有审阅节点。第一个被发给客户的错误答案,就是人们最后一次信任这个系统。
结论:
一个循环,是一个 agent。一个图,是一支团队。
循环工程是那个把"写 prompt"变成"构建系统"的突破。你学会了做一个会重试、验证、改进的 agent。那是一项真技能。现在依然是。
但一旦工作变复杂,每一支团队都胜过每一个单干的人。一个图,就是你构建一支 agent 团队的方式:一个收集资料的研究者、一个推理的分析师、一个起草的写作者、一个拒绝的审阅者。每个 agent 都很简单。是图让他们强大。
这篇文章里的代码没有任何框架依赖。函数、字典、if 语句。那就是一个图。你不需要 LangGraph。你不需要 CrewAI。你需要的是理解这五种模式,并为你的用例把它们连起来。
> 2026 年有两种构建者。一种还在为所有事情跑一个循环。另一种在设计流程。模型是一样的。架构不是。
标签:# X # Claude # AI # 循环 # 金融 # 指南 # Sonnet
相关文章:循环工程:/loop 傻瓜指南
TL;DR:任何人都能借助循环把自己的 Claude 生产力提升 10 倍——哪怕你完全不懂技术。
Claude AI 循环 金融
原文参考:https://maxed.wiki/posts/graph-engineering-with-claude-the-11-step-roadmap-from-loops-to-graph-architect/ (Maxed.wiki,本页为站内中文整理)