多智能体架构设计模式排行榜：主管与流水线模式深度评测

为什么需要多 Agent 协同？

前八篇内容围绕单 Agent 系统展开：一个 LLM、一组工具、一条对话历史。这套方案能覆盖大多数场景，这是事实。

但有些任务天生需要“多专家”分工：

• 撰写技术文章：研究员收集素材、写手起草、编辑润色——三个角色，三种思考模式
• 处理用户工单：意图识别、知识库检索、回复生成——三个阶段，各自独立可验证
• 代码审查：静态分析、安全扫描、可读性评估——三个维度，互不影响

单 Agent 当然也能处理这些任务，但你会发现 System Prompt 越堆越长，输出质量却越来越不稳定——原因很简单：你强迫一个角色同时扮演所有专家。

多 Agent 的核心价值在于职责分离：每个 Agent 只专注一件事，并把这件事做到极致。

两种主流架构模式

多 Agent 系统存在多种拓扑结构，其中两种最常见：

Supervisor 模式（动态路由）和 Pipeline 模式（固定顺序）。通俗地说，Supervisor 模式像有一个“指挥中心”决定下一步调用哪个专家；而 Pipeline 模式像流水线作业，执行路径从一开始就固化在代码里，每个 Agent 只知道自己当前的上下文和下一节点。

两种模式并非互斥，它们适用于完全不同的问题域。

Demo 1：Supervisor 模式

设计思路

Supervisor 模式的核心挑战是路由可靠性。如果让 LLM 每一步都去决策“下一个调谁”，常见问题会反复出现：

• 重复调用同一个 Worker
• 遗漏记录已调用过的 Worker
• 不确定何时终止流程

更优的设计是两阶段混合策略：

Phase 1：让 LLM 完成一次任务分类（判断属于 simple_fact 还是 full_article）
Phase 2：Python 根据分类结果和已调用列表执行确定性路由

换句话说，LLM 负责“识别任务类型”，Python 负责“按规则调度”。这个分工非常合理。

LangGraph 实现

class SupervisorState(TypedDict):
messages: Annotated[list, add_messages]
task: str
task_type: str # "simple_fact" or "full_article"
called: list[str]
next: str

def classify_node(state: SupervisorState) -> SupervisorState:
"""LLM 做一次分类，结果写入 state，后续路由全程可用"""
decision = _ask(
"Classify this task:\n"
"simple_fact— a factual question with a direct short answer\n"
"full_article — needs research, writing, and editorial review\n"
"Output one word only: simple_fact / full_article",
f"Task: {state['task']}",
).strip().lower()
task_type = "full_article" if "full_article" in decision else "simple_fact"
return {**state, "task_type": task_type}

def supervisor_node(state: SupervisorState) -> SupervisorState:
"""纯 Python 路由，不调 LLM，不会循环"""
called = state["called"]
task_type = state["task_type"]
if "researcher" not in called:
next_worker = "researcher"
elif task_type == "simple_fact":
next_worker = "FINISH" # 简单问题：研究完就结束
elif "writer" not in called:
next_worker = "writer"
elif "reviewer" not in called:
next_worker = "reviewer"
else:
next_worker = "FINISH"
return {**state, "next": next_worker}

图的拓扑结构

g = StateGraph(SupervisorState)
g.set_entry_point("classify")
g.add_edge("classify", "supervisor")
g.add_conditional_edges(
"supervisor",
route_supervisor,
{
"researcher": "researcher",
"writer": "writer",
"reviewer": "reviewer",
"FINISH": END
},
)
g.add_edge("researcher", "supervisor")
g.add_edge("writer", "supervisor")
g.add_edge("reviewer", "supervisor")

整个执行流程：classify → supervisor → [workers] → supervisor → ... → FINISH，形成一个可控循环。

实测执行结果

任务："Write a short article about Python list comprehensions"

[classify] task_type = full_article
[supervisor] → researcher
[researcher] working...
[supervisor] → writer
[writer] working...
[supervisor] → reviewer
[reviewer] working...
[supervisor] → FINISH
Workers called: ['researcher', 'writer', 'reviewer']
Task type : full_article

classify 只执行一次，后续路由全程由 Python 控制，执行链非常清晰：researcher → writer → reviewer → FINISH。

Demo 2：Pipeline 模式

Pipeline 的代码量比 Supervisor 少很多——因为完全不需要路由逻辑。

class PipelineState(TypedDict):
topic: str
outline: str
draft: str
polished: str
stage_log: list[str]

def outline_agent(state: PipelineState) -> PipelineState:
outline = _ask("创建一个 5 点大纲...", state["topic"])
return {**state, "outline": outline, "stage_log": [...]}

def draft_agent(state: PipelineState) -> PipelineState:
draft = _ask("基于大纲写 200 字草稿...", state["outline"])
return {**state, "draft": draft, "stage_log": [...]}

def polish_agent(state: PipelineState) -> PipelineState:
polished = _ask("润色...", state["draft"])
return {**state, "polished": polished, "stage_log": [...]}

# 拓扑：outline → draft → polish → END
g.add_edge("outline_agent", "draft_agent")
g.add_edge("draft_agent", "polish_agent")
g.add_edge("polish_agent", END)

实测执行结果

[outline_agent] 957 chars
[draft_agent] 1846 chars
[polish_agent] 2168 chars
最终输出（前 300 字）："### Unveiling the Power of List Comprehensions in Python
Python's lists are dynamic and powerful data containers...

每个阶段的输出作为下一阶段的输入，内容逐步丰富：大纲 957 字符 → 草稿 1846 → 润色 2168。