2026 年，每个构建多 Agent 系统的人仍然在写直线。第一步，然后第二步，然后第三步——每一步都等着上一步完成。这就是为什么它会慢，以及如何解决它。
没人检查的问题
你构建了一个多步骤 Agent。它能工作。但也慢。
你以为模型是瓶颈。其实不是。
瓶颈是你画出的形状。一条链——步骤 1 等步骤 2，步骤 2 等步骤 3——即使其中一半步骤彼此毫无关系，也强制顺序执行。
"总结这份文档，然后查一下天气"是两个独立的任务，却披着一件工作流的外衣。天气任务不需要总结。从来不需要。但如果你把它写成一条链，它还是会等着。
这种浪费的等待，乘以几十个步骤，就是你大部分运行时间消失的地方。
第一章 - 循环 vs 图
一个循环是一个自我改进的单位：
1尝试某事 → 检查结果 → 调整 → 再试一次
这就是原子。一个 Agent，一个指标，循环直到收敛。
循环有一个已知的失效模式：它们只优化你衡量的东西，其他一概不管。一个旨在快速关闭工单的客服机器人会快速关闭工单——同时客户满意度悄然暴跌。循环看不到自身指标之外的东西。这就是古德哈特定律在你的 Agent 架构中的体现。
图通过设计解决了这个问题。你不是用一个循环去追逐一个数字，而是构建一个由多个循环组成的网络，它们相互监视和纠正。节点 A 的输出馈送给节点 B。节点 C 独立运行并检查两者。整个系统不由单一指标驱动——而是由结构驱动。
对于 Agent 系统，这意味着一个具体的转变：停止编写一个从上到下包揽一切的 Agent。首先设计工作的形状——什么必须在什么之前发生，什么可以同时运行，什么实际上需要等待。
第二章 - 节点、边，以及区分它们的测试
一个图正好有两个组成部分：
节点 - 一个工作单元。一个 Agent，一个任务，一个输入，一个输出。
边 - 一个真正的依赖关系。节点 B 的输入需要节点 A 的输出。
几乎每个人都会犯的错误：默认把"然后"当成一条边。
1"阅读这份代码库，然后编写更新日志"2"获取定价页面，然后总结竞品功能"
为你工作流中的每一个"然后"问一个问题：
下一步骤是否真的读取了上一步骤的输出？
如果是 → 真正的边。保持顺序执行。
如果否 → 没有边。等待是浪费的。让它们并行运行。
如果两个任务之间没有数据跨越边界，它们就是独立的——而你顺序运行的每一对独立任务，都是在白白浪费运行时间。
以下是在代码中应用的测试：
1from dataclasses import dataclass23@dataclass4class TaskNode:5 id: str6 prompt: str7 depends_on: list[str] # 该节点实际需要的节点 ID89def has_real_edge(node_a: TaskNode, node_b: TaskNode) -> bool:10 """11 核心图工程测试：12 node_b 的 prompt 是否真的需要 node_a 的输出？13 """14 return node_a.id in node_b.depends_on1516# 示例：大多数"链"会坍缩成 2-3 个真正的依赖组17nodes = [18 TaskNode("audit_routes", "列出所有 API 路由文件", []),19 TaskNode("check_auth", "检查认证中间件覆盖率", []),20 TaskNode("fetch_weather", "获取今日天气", []),21 TaskNode("summarize", "总结路由和认证发现",22 depends_on=["audit_routes", "check_auth"]),23]2425# audit_routes, check_auth, fetch_weather 之间没有边26# 它们并行运行。只有 "summarize" 有真正的边——它会等待。
你当前"做 A，然后 B，然后 C"的 Agent 在技术上已经是一个图了。只是它是最差的那种——一个单一的链，以至于如果 C 停住了，下游就永远不会运行。
第三章 - 构建你的第一个图
要求：
- Claude Code（支持动态工作流的最新版本）。
- Max、Team 或 Enterprise 计划——工作流默认开启。Pro 计划需手动启用。
打开一个真实的仓库。不要用玩具示例——只有真正规模才能体现回报。
启动你第一个图的提示词：
1创建一个工作流，审计此代码库中的每个路由文件。23对于每个路由文件，独立检查：4- 是否存在认证中间件5- 所有参数是否有输入验证6- 是否配置了限流7- 错误处理是否泄露堆栈跟踪89并行运行这些检查，覆盖所有路由文件——10它们彼此不依赖。1112所有文件检查完成后，生成一份合并报告，13按严重程度分组：严重、警告、信息。1415合并步骤应等待所有检查完成。16在此之前的所有步骤都不应等待。
注意提示词本身嵌入的结构：明确指出的并行工作，以及唯一真正的依赖关系（合并等待所有检查完成）。你不是在指望 Agent 推断出图——你是在描述它。
底层发生的事情——编排的简化版本：
1import asyncio2from anthropic import Anthropic34client = Anthropic()56async def audit_route_file(filepath: str) -> dict:7 """一个节点。独立于其他所有路由文件运行。"""8 response = await client.messages.create(9 model="claude-sonnet-5",10 max_tokens=1000,11 messages=[{12 "role": "user",13 "content": f"""审计此路由文件，检查：14 - 认证中间件，输入验证，15 限流，错误处理1617 文件：{filepath}1819 返回 JSON：{{"file": "", "issues": [], "severity": ""}}"""20 }]21 )22 return {"file": filepath, "result": response.content[0].text}2324async def consolidate(results: list[dict]) -> str:25 """唯一的真正边——等待每个审计节点完成。"""26 response = await client.messages.create(27 model="claude-opus-4-8",28 max_tokens=2000,29 messages=[{30 "role": "user",31 "content": f"""将这 {len(results)} 个路由审计结果32 合并成一个按严重程度分组的报告：3334 {results}"""35 }]36 )37 return response.content[0].text3839async def run_graph(route_files: list[str]):40 # 扇出——所有独立节点并发运行41 audit_tasks = [audit_route_file(f) for f in route_files]42 results = await asyncio.gather(*audit_tasks)4344 # 扇入——唯一具有真正依赖关系的节点45 report = await consolidate(results)46 return report4748# 40 个路由文件，一个提示词，一次并行传递49results = asyncio.run(run_graph([50 f"routes/{f}.py" for f in ["auth", "users", "billing", "orders"]51 # ...另外 36 个
40 次顺序 API 调用，每次约 8 秒，总共超过 5 分钟。同样的 40 次调用并行扇出：不到 15 秒，延迟由最慢的单个文件决定，而不是所有文件的总和。
第四章 - 图实际在什么地方失效
图工程在三个可预测的地方失败。在遇到它们之前了解它们。
上下文坍缩。 扇出 1000 个节点，然后尝试将所有 1000 个输出喂入一个合并步骤，在合成开始之前你就会超出任何上下文窗口。
修复： 分层扇入。将节点分组为 20-50 个一批，总结每批，然后合并总结结果——而不是原始输出。
1async def layered_consolidate(results: list[dict], batch_size: int = 30):2 """分层扇入——绝不在规模上合成原始输出。"""3 batches = [results[i:i+batch_size]4 for i in range(0, len(results), batch_size)]56 batch_summaries = await asyncio.gather(*[7 summarize_batch(batch) for batch in batches8 ])910 # 最终合并基于总结，而不是 1000 个原始结果11 return await consolidate(batch_summaries)
虚假独立性。 你会假设两个节点是独立的，因为它们的提示词没有相互引用——但它们都写入同一个文件，或者命中同一个受限流的 API。这就是一个隐藏的边。
修复： 审计共享资源，而不仅仅是共享数据。两个有写入冲突的节点，即使没有数据依赖，也需要一条边。
静默节点失败。 在链中，一个失败会停止一切——烦人但显而易见。在图里，200 个节点中的一个失败可能消失在看起来完整的报告中。
修复： 每个扇入步骤在合成前检查节点数量是否与预期数量相符，并明确标记缺口，而不是悄悄使用部分数据。
1async def safe_consolidate(results: list[dict], expected_count: int):2 if len(results) < expected_count:3 missing = expected_count - len(results)4 print(f"警告：{missing} 个节点静默失败。"5 f"报告将不完整。")6 return await consolidate(results)
第五章 - 扩展到真正的集群
一旦模式在 40 个节点上成功运行，扩展到数百个节点就只是一个配置更改，而不是重新设计——前提是你从第二章开始就正确地构建了图。
完整的生产形态：
1 编排器2 |3 +--------+-------+-------+--------+4 v v v v v5 节点 1 节点 2 节点 3 ... 节点 N6 (并行，它们之间没有边)7 | | | |8 +--------+-------+-------+-------+9 v10 批次总结 <- 分层扇入11 (每 30 个一组)12 v13 最终报告 <- 唯一真正的边
编排器的唯一工作：将任务分解为节点，识别真正的边，并调度。它自己不执行任何工作——它绘制图。
1async def orchestrate(task: str, resources: list[str]):2 """3 编排器节点——分解，不执行。4 """5 plan = await client.messages.create(6 model="claude-opus-4-8",7 max_tokens=2000,8 messages=[{9 "role": "user",10 "content": f"""任务：{task}11 可用资源：{resources}1213 分解为一个图：14 - 列出每个独立节点（无共享边）15 - 列出节点间的任何真正依赖关系16 - 如果数量 > 50，将节点分组为扇入批次1718 返回 JSON，包含：nodes, edges, batch_groups"""19 }]20 )2122 graph = parse_plan(plan.content[0].text)2324 # 并行执行独立节点25 node_results = await asyncio.gather(*[26 execute_node(n) for n in graph["nodes"] if not n["depends_on"]27 ])2829 # 然后执行依赖节点，仅尊重真正的边30 final = await execute_dependent_chain(graph["edges"], node_results)3132 return final
这就是图工程代表的真正转变：你不再是一个编写每一步的人，而是成为一个设计依赖结构的人。Agent 填充节点。你拥有边。
当你用图而不是线来思考时，会发生什么变化
一个具有 40 个步骤的线性 Agent 有 40 个顺序故障点，并且延迟是其最慢的单个步骤的 40 倍。
一个具有相同 40 个工作单元但采用图结构，其并行故障点的数量等于你拥有的真正依赖关系的数量——在大多数工作流中通常是 3 到 5 个——并且延迟受限于你最慢的层，而不是你的总步骤数。
这不是边际加速。这是执行完全相同的基础工作，一个工作流需要 5 分钟，而另一个只需要 15 秒的区别。
模型从来不是瓶颈。你画的那条线才是。
这是截至 2026 年 7 月多 Agent 编排模式的技术分解。代码示例仅供说明——在规模部署之前，请根据你的生产环境调整错误处理、限流和重试逻辑。
感谢阅读。
![[备忘录] 老板正在放弃表现不佳的下属](/cdn-cgi/image/width=1920,quality=90,format=auto,metadata=none/https%3A%2F%2Fcms-assets.youmind.com%2Fmedia%2F1784827522698_408j7z_HN3Kb76awAAvvjF.jpg)