用 Claude 做图工程:从循环到图架构师的 11 步路线图

Graph Engineering with Claude: the 11-Step Roadmap From Loops to Graph Architect

中文译文 · 18k 字

一句话摘要

从循环进阶图架构师的路线图

用 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,本页为站内中文整理)