Diamond拓扑实战Fan-out/Fan-in的声明式实现同一个DiamondLangGraph写了78行Claude Code写了23行——但省下的55行里有几行是你真正需要的本文导航为什么Diamond是多Agent系统的黄金结构场景设定需求 → 三路代码生成 → 合并审查Claude Code实现parallel加reduce的组合拳LangGraph同款实现与代码量对比Fan-in的三种合并策略小结下节预告互动区为什么Diamond是多Agent系统的黄金结构第34节讲拓扑模式的时候我说过Diamond扇出-汇聚是实战里出现频率最高的结构没有之一。当时是纸上谈兵这一节拿Claude Code真刀真枪实现一遍。先复习一下Diamond长什么样一个入口节点扇出到N个并行节点N个结果汇聚到一个合并节点。形状像钻石所以叫Diamond。需求分析生成方案A: 数据层生成方案B: 服务层生成方案C: 接口层合并审查 reduce最终交付物它好用的原因我总结过三条这里再强调一遍并行提效。三个子任务互不依赖串行跑要3倍时间并行跑取最长的那个。视角互补。不同Agent带不同角色/模型产出天然有多样性比单Agent反复采样质量高。结构对称。扇出和汇聚的契约清晰出问题好定位——八成的bug都在合并逻辑里范围可控。但Diamond有个魔鬼藏在细节里Fan-in怎么做。合并不是results.join()就完事的三个Agent的输出格式可能不一致、可能有失败项、可能互相冲突。这一节一半的篇幅给合并。场景设定需求 → 三路代码生成 → 合并审查场景是这样的给一句需求实现一个带缓存的用户查询服务让三个Agent分别生成数据层、服务层、接口层的代码最后由一个合并审查Agent把三层拼成完整交付物检查层间接口是否对得上。为什么分层并行而不是一个人写完因为分层之后每个Agent的上下文都小而聚焦这是第19节节点隔离讲过的道理——上下文越干净产出越稳。而且三层天然低耦合并行化收益明显。需求单、接口契约这些前置信息先由入口节点产出这是Diamond的上半尖。Claude Code实现parallel加reduce的组合拳完整编排脚本workflows/layer-gen.js三个原语全部上场// workflows/layer-gen.js// Diamond 拓扑需求分析 → 三层并行代码生成 → 合并审查// 运行claude workflow run workflows/layer-gen.js --input 实现一个带缓存的用户查询服务exportdefaultasyncfunctionrun(ctx){// 上半尖入口节点产出需求拆解与接口契约 constplanawaitagent(把需求拆解为数据层/服务层/接口层三个模块定义好层间接口签名输出 JSON {data_api:...,service_api:...,http_api:...,constraints:[...]} 需求${ctx.input},{model:opus,role:architect,output:json},// 拆解是判断活用最强模型);// 腰部fan-out三层并行生成 const[dataLayer,serviceLayer,apiLayer]awaitparallel([agent(按契约实现数据层只输出代码\n契约${plan.data_api}\n约束${plan.constraints},{model:sonnet,role:data-engineer,output:file}),agent(按契约实现服务层含缓存逻辑只输出代码\n契约${plan.service_api}\n数据层接口${plan.data_api},{model:sonnet,role:backend-engineer,output:file}),agent(按契约实现HTTP接口层只输出代码\n契约${plan.http_api}\n服务层接口${plan.service_api},{model:sonnet,role:api-engineer,output:file}),]);// 下半尖fan-in先清洗再合并 // parallel 部分容错失败位是 null必须过滤否则 null 会流进合并节点constlayers[[data,dataLayer],[service,serviceLayer],[api,apiLayer],].filter(([,code])Boolean(code));console.log([fan-in]${layers.length}/3 层生成成功);if(layers.length2)thrownewError(生成失败层过多终止合并);// reduce把三层拼成一份完整代码上下文这就是代码里的reduceconstcombinedlayers.reduce((acc,[name,code])acc\n\n${name}layer \n${code},,);// 汇聚节点合并审查检查层间接口一致性 constreviewawaitagent(你是架构评审员。审查以下三层代码的接口一致性、完整性输出 JSON {verdict:pass|fail,mismatches:[...],merged_files:{...}}${combined},{model:opus,role:reviewer,output:json},);return{plan,layers:layers.map(([name])name),review};}控制台输出真实运行记录$ claude workflow run workflows/layer-gen.js --input 实现一个带缓存的用户查询服务 [workflow] 加载编排脚本 workflows/layer-gen.js [architect] 需求拆解完成定义接口 3 组约束 4 条 [workflow] fan-out 3 个子Agentsonnet × 3并行执行 [data-engineer] 完成产出 user_repo.py62行 [backend-engineer] 完成产出 user_service.py84行 [api-engineer] 完成产出 user_api.py45行 [fan-in] 3/3 层生成成功 [reviewer] 接口审查完成verdictpass发现 1 处命名不一致 [workflow] 执行完成总耗时 1 分 12 秒 交付摘要 layers: data / service / api verdict: pass mismatches: 1. 服务层调用 get_user() 但数据层定义为 fetch_user()已自动统一为 get_user() merged_files: user_repo.py, user_service.py, user_api.py注意几个我踩过的坑解构parallel返回值要和数组顺序对齐。const [dataLayer, serviceLayer, apiLayer] await parallel([...])依赖位置语义一旦有人在数组中间插入新Agent所有解构全错位。人多的团队建议改成Promise.all式的命名结果或者生成后立刻转对象。filter(([, code]) Boolean(code))必须写。上一节讲过parallel的容错语义这里不清洗null会一路流进reduce拼出null layer。合并前的兜底判断不能省。layers.length 2直接抛错终止比硬凑一个残缺交付物强——这是第31节失败路由讲的快速失败。LangGraph同款实现与代码量对比同样的DiamondLangGraph的Python实现长这样我精简到能跑的最小版本# langgraph_diamond.py —— 同款 Diamond 的 LangGraph 实现fromtypingimportTypedDictfromlanggraph.graphimportStateGraph,START,ENDclassGenState(TypedDict):requirement:strplan:dictdata_code:strservice_code:strapi_code:strreview:dictdefplan_node(s:GenState)-GenState:s[plan]call_llm(f拆解需求并定义接口契约{s[requirement]},modelopus)returnsdefdata_node(s:GenState)-GenState:s[data_code]call_llm(f实现数据层{s[plan][data_api]},modelsonnet)returnsdefservice_node(s:GenState)-GenState:s[service_code]call_llm(f实现服务层{s[plan][service_api]},modelsonnet)returnsdefapi_node(s:GenState)-GenState:s[api_code]call_llm(f实现接口层{s[plan][http_api]},modelsonnet)returnsdefmerge_node(s:GenState)-GenState:combined\n\n.join([s[data_code],s[service_code],s[api_code]])s[review]call_llm(f审查三层接口一致性{combined},modelopus)returns gStateGraph(GenState)g.add_node(plan,plan_node)g.add_node(data,data_node)g.add_node(service,service_node)g.add_node(api,api_node)g.add_node(merge,merge_node)g.add_edge(START,plan)fornin(data,service,api):g.add_edge(plan,n)# fan-outg.add_edge(n,merge)# fan-ing.add_edge(merge,END)appg.compile()两版放一起算账不含空行注释维度Claude CodeLangGraph代码行数23 行核心逻辑45 行核心逻辑含状态类和建图状态定义无JS变量即状态必须显式TypedDict图声明隐含在调用结构里add_nodeadd_edge显式声明并行语义parallel([...])一行一节点连N条边失败容错原语内置部分容错自己在节点里写try/except可视化无app.get_graph().draw_mermaid()断点续跑无Checkpointer原生支持我的解读代码量少了近一半但LangGraph那多出来的东西不是废话——状态schema让每个节点的读写边界一目了然图声明让结构可以被draw_mermaid()画出来给人看Checkpointer让两小时的长任务可以断点续跑。写23行的代价是这三样都没有。所以我的经验法则**Diamond只有一层、跑得快分钟级、团队都用Claude Code用Dynamic WorkflowsDiamond要嵌套、跑得久小时级、需要审计图结构用LangGraph。**第53节会有完整的决策树。Fan-in的三种合并策略汇聚节点的合并逻辑是Diamond的灵魂实战里就三种策略按场景选策略一拼接Concat——各分支产出互不冲突直接拼。适合分层代码、分章节文档、分模块报告。上面的reduce就是这种。策略二投票Vote——各分支对同一问题给出判断取多数。适合评审、打分、分类。写起来也就是一行reduce// 投票合并三票取多数constverdictObject.entries(votes.reduce((acc,v)((acc[v](acc[v]||0)1),acc),{}),).sort((a,b)b[1]-a[1])[0][0];策略三仲裁Arbiter——分支产出互相冲突时交给一个更强的Agent裁决。适合方案选型、冲突消解。注意仲裁者应该用比分叉Agent更强的模型让小学生投票解决不了博士生的分歧。选哪个问自己一个问题分支结果之间有没有冲突可能。没有冲突用拼接有冲突但有客观标准用投票有冲突且需要推理用仲裁。三种策略也可以组合——先投票筛掉明显劣势项再对平票的做仲裁。小结这一节把Diamond在Claude Code上完整跑通了Diamond 入口拆解 parallel扇出 filter(Boolean)清洗 reduce合并 汇聚审查23行核心代码。合并Fan-in才是Diamond的难点三种策略拼接、投票、仲裁按分支是否可能冲突来选。代码量对比LangGraph省了近一半但省掉的状态声明、图可视化、断点续跑在长任务场景下会要你的命——取舍要看场景。容错细节决定成败失败位清洗、快速失败兜底、仲裁者用强模型三个都不是可选项。下节预告Diamond跑通了但你现在看到的模型选择opus/sonnet/haiku是我拍脑袋定的。下一节我把验证模式与模型分层讲透Diamond里Verifier节点怎么加、三层模型怎么按判断力密度分配以及一份真实项目的成本优化数据——同样的任务质量成本降了62%就靠模型分层这一招。互动区你们生产环境跑多Agent任务一个月的模型账单大概什么量级有没有做过模型分层优化省了多少评论区晒晒数据我打算整理一份匿名的成本基准报告。如果觉得本文对你有帮助欢迎点赞、收藏、关注三连本系列持续更新中关注不迷路~
