链式、扇出、菱形、条件路由、循环——掌握这五种，90%的场景够用
一、引言
图的形状不是装饰，它是决定运行时间的最大杠杆。掌握五种基础拓扑，就能组合出任何复杂的图。
二、链式（Chain）
`
A → B → C → D
`
最简单，每个节点只有一条边进、一条边出。
适用场景： 线性流程，步骤之间有强依赖。
`python
graph.add_edge(“scan”, “extract”)
graph.add_edge(“extract”, “analyze”)
graph.add_edge(“analyze”, “store”)
`
注意： 如果链里有两三根箭头没有携带数据，它们只是顺序——剪掉，链会塌缩成更宽的并行结构。
三、扇出（Fan-out）
`
┌→ B₁ ─┐
A →├→ B₂ ─┤→ C
└→ B₃ ─┘
`
把活儿一次性派出去并行执行。这是图比Loop快的核心原因。
两个关键细节：
- 扇出是一道屏障，会等所有函数都跑完才返回
- 一个抛错的函数会被解析成null，不拖垮整个批次
`python
# 并行提取多个文件
from concurrent.futures import ThreadPoolExecutor, as_completed
def fan_out_extract(state: GraphState):
files = state[“pending_files”]
with ThreadPoolExecutor(max_workers=4) as executor:
futures = {
executor.submit(extract_single, f): f
for f in files
}
results = []
for future in as_completed(futures):
try:
results.append(future.result())
except Exception as e:
results.append(None) # 失败的不拖垮其他人
return {“extracted”: [r for r in results if r is not None]}
`
适用场景： N个独立信源、N个文件、N条路由。
四、菱形（Diamond）
`
┌→ B ─┐
A →┤ ├→ D
└→ C ─┘
`
派发 → 归约 → 合成。 这是最常用的高级拓扑。
市场扫描、依赖审计、代码评审、研究报告，背后都是这个形状。
`python
# 菱形：并行验证 → 合并结果
graph.add_node(“dispatch”, dispatch_tasks)
graph.add_node(“verify_a”, verify_source_a)
graph.add_node(“verify_b”, verify_source_b)
graph.add_node(“verify_c”, verify_source_c)
graph.add_node(“merge”, merge_results)
# 扇出
graph.add_edge(“dispatch”, “verify_a”)
graph.add_edge(“dispatch”, “verify_b”)
graph.add_edge(“dispatch”, “verify_c”)
# 汇入
graph.add_edge(“verify_a”, “merge”)
graph.add_edge(“verify_b”, “merge”)
graph.add_edge(“verify_c”, “merge”)
`
标准写法： 派发 → 归约 → 合成。先派出去收集广度，用普通代码归约压缩，再用最后一个agent合成写出答案。
五、条件路由（Conditional Routing）
`
A → [判断] → B（路径1）
→ C（路径2）
`
不是每张图都是固定的。有时候走哪条边，取决于某个节点发现了什么。
`python
def route_by_type(state: GraphState) -> str:
file_type = state[“detected_type”]
if file_type == “document”:
return “extract_doc”
elif file_type == “image”:
return “extract_image”
elif file_type == “video”:
return “extract_video”
return “skip”
graph.add_conditional_edges(
“identify”,
route_by_type,
{
“document”: “extract_doc”,
“image”: “extract_image”,
“video”: “extract_video”,
“skip”: END,
}
)
`
关键点： 路由的判断可以由Agent完成，但路由本身是代码——同样的分类结果每次都走同一条路。
六、循环（Loop with Convergence）
`
A → B → [检查] → A（继续）
→ C（完成）
`
能收敛的写法叫”跑到干为止：
持续派出发现者，直到连续K轮都没发现新东西，才停下来。
`python
seen = set()
confirmed = []
dry = 0
while dry < 2: # 连续两轮空手而归就停下
found = await parallel(FINDERS)
fresh = [f for f in found if key(f) not in seen]
if not fresh:
dry += 1
continue
dry = 0
for f in fresh:
seen.add(key(f)) # 对”见过的一切”去重
# 验证后加入confirmed
verified = await verify(fresh)
confirmed.extend(verified)
def should_continue(state: GraphState) -> str:
if state[“consecutive_empty”] >= 2:
return “finalize”
if state[“total_iterations”] >= 10: # 硬上限
return “finalize”
return “continue_loop”
`
关键细节： 要对着”见过的一切”去重，而不是只对着”已确认的结果”去重。不然被否掉的发现每一轮都会重新冒出来，循环永远跑不干。
七、拓扑选择指南
| 场景 | 推荐拓扑 | 原因 |
|---|
|——|———-|——|
| 线性流程 | 链式 | 步骤强依赖 |
|---|---|---|
| N个独立任务 | 扇出 | 并行加速 |
| 收集+分析+合成 | 菱形 | 广度→压缩→输出 |
| 按条件分流 | 条件路由 | 不同类型不同处理 |
| 规模未知的探索 | 循环 | 跑到干为止 |
默认用Pipeline（链式+扇出），只在必须同步时用屏障。
屏障会等最慢的节点。”代码更干净”和”这些阶段感觉是分开的”都不是理由，屏障带来的延迟是真实的、可测量的、被浪费掉的时间。
八、总结
`
链式: A → B → C → D
扇出: A → [B,C,D] → E
菱形: A → [B,C,D] → E（派发→归约→合成）
路由: A → [判断] → B 或 C
循环: A → B → [检查] → A 或 C
`
掌握这五种，90%的场景够用。剩下的，组合就好。
下一章预告：框架选型——LangGraph vs CrewAI vs AutoGen vs 原生Python。
发表回复