从 loop 到 graph :Claude 图表工程 11 步路线图

0xrafy 发布于 2026-07-22 阅读 19

本文介绍了从单智能体循环(loop)到多智能体图(graph)的工程范式转变。作者指出,单一循环在复杂任务中会因上下文窗口超限而失效,而图工程通过将任务拆分为多个专用智能体并协调它们工作,实现更可靠的输出。文章详细讲解了四种基本构建块(节点、边、路由器、状态)和五种关键模式(顺序链、路由器、并行扇出、带门循环、人机循环),并给出了完整的客服工单处理图实现示例。同时强调了模型分层(Haiku用于路由、Sonnet用于构建和审查、Opus用于最终质量门控)、状态追踪、错误回退路径等生产级最佳实践。最后总结了六个可立即构建的图应用场景,如工单分类、竞品分析、PR审查等。

01. 为什么一个循环不够

大多数尝试构建 AI Agent 的人,最终都会得到一个单循环——用一个不断变长的提示词反复调用 Claude,直到上下文窗口撑爆、开始胡言乱语。十之八九的人从未把任务拆分给专门的子 Agent。

他们没有路由。没有分支。没有并行。只跑一个循环,祈祷它工作,等它出问题时再花几周调试。

这是一份 11 步路线图,帮你把单循环变成一张图表——它能向外展开、自我校验,最终收敛到真实产出。

循环工程是个真正的进步。不再是一次性地提示 Claude 然后复制结果,而是构建一个系统:每次迭代都重试、校验、改进。那是第三个时代。

图像

但一个循环仍只有一个 Agent、只做一件事。任务一复杂,单循环就淹死在上下文里,关注点相互混乱,试图包揽一切。图表工程是下一层:你来设计流程。

哪个 Agent 运行?它失败时怎么办?结果送去哪?什么时候让人类介入?本文将给出完整操作手册。

一个循环是一个 Agent 变聪明;一张图表是多个 Agent 协同配合。

一个循环给单个 Agent 自主权:它会重试,从失败中学习,每轮都改进。对于单一目的的任务(修复这个 Bug、总结这篇文档、清理这个数据集),循环就够了。你定义目标,挂一个校验器,设好停止条件,然后让它跑。

图像

问题出现在任务跨多个领域时:研究 + 分析 + 代码 + 审阅。一个 Agent 想同时做这四件事,上下文窗口会被混杂的信息塞满,丢失前面步骤的上下文,产出的东西看似完整,一检查就散架。

研究会渗进分析,代码会忽略审阅。这个 Agent 干着四份活,却只有一份活的上下文预算。

这不是模型的限制。Sonnet 和 Opus 完全有能力处理每一件独立的事。瓶颈在架构:你让一个工人在同一段对话里同时充当研究员、分析师、开发者和审阅者。

解决方案不是更大的上下文窗口,而是把任务拆分给多个 Agent,每个只做好一件事,再把它们串联起来,让上一个的输出变成下一个的输入。

图像

从循环到图的转变,和独立开发者加入团队时的转变完全一样:一个人单干在项目简单时很快,但一复杂起来,你就需要角色、交接和审阅。

02. 四个构建块

任何图表,无论多复杂,都由这四个原语组装而成。你不需要框架,只需要函数、一个字典和 if 语句。

图像

节点(Node):一个只做一件事的函数——一次 Claude 调用、一次工具执行、一次数据库查询、一次文件写入。它接收状态,返回状态,不关心之前或之后发生了什么,只管自己的任务。

边(Edge):连接两个节点。"研究之后运行分析"。边可以是无条件的(始终执行)或有条件的(仅当 state["status"] == "needs_more_data")。条件边让图能够做决策。

路由器(Router):一个特殊节点,检查当前状态并选择走哪条路。它自己不干活,只负责分类输入并分发给正确的专家。可以把它看作调度员,而不是工人。

状态(State):一个字典,流经整个图。每个节点都从中读取并向它写入。状态是图的内存——没有状态,每个节点都从零开始,图根本不知道之前发生过什么。

图像

## 整个图抽象,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 工程的每一个时代都吸收了前一个时代。提示工程教会我们写好指令;上下文工程教会我们喂对数据;循环工程教会我们构建一个自治的 Agent。

图工程教我们如何把多个 Agent 接成一个协调系统。

图像

大多数人还停留在第一或第二时代:写一段 prompt,也许加点上下文,指望一次 Claude 调用就搞定全部工作。而那些正在交付真正产品的 Builder 已经进入了第三或第四时代——他们不是在写更好的 prompt,而是在设计更好的流程。

图像

04. 模式:顺序链

最简单的图:节点 A 执行,输出传递给节点 B,再传给节点 C。没有分支,没有路由。绝大多数"AI 流水线"就是这种。

图像

顺序链在每一步都一定成功、顺序从不改变时有效。数据提取、格式转换、翻译后总结都适用。一旦某一步可能失败或路径可能变化,你就得改用其他四种模式之一。

顺序链最大的风险是错误传播。如果节点 A 产出糟糕的结果,节点 B 在坏结果上继续构建,节点 C 又往上叠加,最后错误经过三层累积放大了好几倍。

因此,即使最简单的顺序图,也最好在末尾加一个校验节点。

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. 模式:路由器

路由器检查输入,然后从多条路径中选一条执行。并非所有节点都会运行,路由器负责为当前任务挑出最合适的那个专家。这样你就能构建一个处理多种类型任务的系统,而不用把所有工具塞进一个 Agent。

图像

关键洞察:路由是分类任务,不是推理任务。你不需要 Sonnet 来判断工单是计费问题还是 Bug——Haiku 只需几毫秒、成本极低就能搞定。把贵的模型留给专家内部的实际工作吧。

每个专家应该有自己独立的 system prompt 和工具集。代码 Agent 需要 run_testswrite_file;研究 Agent 需要 web_searchread_doc。如果给每个 Agent 都配上所有工具,就会产生错误的工具调用、浪费 token、输出混乱。

图像

def router(state):
    # 用便宜模型做分类
    category = call_claude(
        model="claude-haiku-4-5",
        system="对此任务进行分类。一个词:代码、研究或写作。",
        prompt=state["task"]
    )
    routes = {
        "code": code_agent,       # 自己的 prompt,自己的工具
        "research": research_agent, # 自己的 prompt,自己的工具
        "writing": writing_agent,   # 自己的 prompt,自己的工具
    }
    agent = routes.get(category.strip(), research_agent)
    return agent(state)

进阶细节:如果你的路由器老是选错,通常不是模型不够大,而是 system prompt 不够好。

把每个类别的 5–6 个真实示例写进路由器的 prompt。在路由任务上,Haiku 的少样本分类往往比 Sonnet 的零样本分类效果更好。

06. 模式:并行扇出

多个节点同时处理同一个输入。一个收集器等待所有节点完成后合并结果。三个 Agent 并行运行,总耗时等于最慢的那个 Agent 的耗时,而不是三者的时间之和。

图像

约束条件是独立性。如果 Agent B 需要 Agent A 的输出,那就不能并行化。如果它们都在同一输入上工作、产出独立的输出,则可以并行。竞争分析就是一个干净的例子:每个竞争对手对应一个 Agent,同时运行,最后合并结果。

合并节点是最容易被低估的环节。合并三个 JSON 对象很容易,但把三份研究报告综合成一份连贯的综述,就需要自己的 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 是单独的 Claude 调用,带综合 prompt
    merged = call_claude(
        system="将这三份竞争对手报告综合成一份对比矩阵。",
        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,创造性
        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. 模式:人在回路中

图在某个特定节点暂停,等待人类批准后再继续。Agent 负责研究和草拟行动方案,人类审阅关键决策,Agent 在获得批准后执行。

图像

这种模式对于任何昂贵、不可逆或高风险的操作都是强制的:向客户发邮件、部署到生产环境、执行金融交易、删除数据。图处理工作,人类处理判断。

批准节点应当准确告诉人类接下来会发生什么、会花多少钱。不是"你批准吗?",而是"此操作将向计费细分群体发送 2,400 封邮件,主题和正文如下。预估成本:$12。批准?"

要足够具体,让人类能在 5 秒内做出真实决策。

def human_gate(state):
    # 准确显示将会发生什么
    print(f"操作: {state['action']}")
    print(f"范围: {state['scope']}")
    print(f"成本: {state['est_cost']}")

    approval = input("批准? [y/n]: ")
    if approval == "y":
        state["approved"] = True
    else:
        state["approved"] = False
        state["human_feedback"] = input("原因: ")
    return state

09. 状态流与模型分级

状态是一个扁平的字典,每个节点都从中读、向它写。要让它保持显式:每个节点应该声明它读哪些 key、写哪些 key。图出问题时,第一件事就是检查状态。如果你搞不清哪个节点写了哪个 key,那就没法调试。

图像

模型分级是第二个生产级关注点。不是所有节点都需要同一个模型。路由器做分类:用 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. 回退路径与错误处理

快乐路径能跑通,但任何意外错误都会让整个图崩溃。生产环境中的图需要为每个可能失败的节点准备一条回退边。

图像

回退不是"忽略错误",而是图中另一条路径,用于优雅地处理失败:如果研究 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

## 在图里
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="分类: billing, technical, general. 一个词。",
        messages=[{"role": "user", "content": state["ticket"]}]
    )
    state["category"] = r.content[0].text.strip().lower()
    return state

def draft(state):
    prompts = {
        "billing": "处理计费问题。退款政策要精确。",
        "technical": "处理技术问题。先要求日志。",
        "general": "处理一般询问。简洁。",
    }
    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="找出此支持回复中的缺陷。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 路由器

    for _ in range(max_retries):
        state = draft(state)         # Sonnet 专家
        state = review(state)        # Sonnet 审阅者
        state["attempts"] += 1

        if state["review"]["passed"]:
            send_response(state["draft"])
            return state

    # 回退:达到最大重试次数,转给人工
    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。一张图是一个团队。

循环工程是把提示词变成可运行系统的突破——你学会了制造一个会重试、会校验、会改进的 Agent。那曾经是(现在也是)真正的技能。

但一旦工作变复杂,每个团队的表现都会超过任何单打独斗的个体。图就是你组建 Agent 团队的方式:一个收集信息的研究员,一个推理的分析师,一个起草的写手,一个负责拒绝的审阅者。每个 Agent 很简单,但图让它们强大。

本文的代码没有框架依赖——函数、字典、if 语句。那就是一张图。你不需要 LangGraph,也不需要 CrewAI。你只需要理解那五种模式,然后按自己的用例把它们接起来。

2026 年有两种 Builder:一种还在用一个循环包揽一切;另一种在设计流程。模型相同,架构不同。

  • 原文链接: x.com/0xRafy/status/2079...
  • 登链社区 AI 助手,为大家转译优秀英文文章,如有翻译不通的地方,还请包涵~

相关文章

0 条评论