我用 LangGraph 实现了 10 种 Multi-Agent 设计模式，附完整代码和架构图

我用 LangGraph 实现了 10 种 Multi-Agent 设计模式，附完整代码和架构图

构建多智能体系统时，最耗时的往往不是编码，而是架构决策。如果你正在使用 LangGraph，大概率也纠结过这些问题：两个 Agent 协作写文章，该用循环还是单次流程？循环几次才算合适？当五个专家需要同时分析一个问题时，是串行、并行，还是投票机制更优？引入一个安全审核 Agent 后，如何优雅地设计批准、拦截或重试的路由逻辑？

市面上教程不少，但大多只展示自己的方案，很少告诉你什么场景该用什么模式。工具本身不是瓶颈——LangGraph 的 API 足够强大。真正的挑战在于，如何根据具体需求，选择并组合出最合适的协作架构。

为此，我们创建了 AgentFlow。它不是一个新框架，也不是教程合集，而是一本多智能体设计模式的参考手册。里面包含了 10 种经过实战验证的协作模式，每种都配有完整的代码、清晰的架构图、详尽的单元测试以及中英双语文档。

一、10 种模式一览

先来看看全景。AgentFlow 涵盖的 10 种模式，构成了一个从简单自我修正到复杂去中心化群体智能的完整谱系：

模式	核心思想	LangGraph 关键技术
? Reflection	写 → 审 → 改，循环直到达标	条件边 + 状态循环
⚔️ Debate	N 方辩论 + 主持人裁决	异步并发 + 结构化输出解析
?️ MapReduce	并行分发 → 汇总合成	Send API 动态扇出
? Hierarchical	经理拆任务 → 员工执行 → 经理汇总	嵌套子图 + Send
?️ Voting	多 Agent 独立投票 → 聚合结果	广播扇出 + 加权统计
?️ GuardRail	主 Agent + 安全守门员	approve/block/redirect 三路由
? RAG-Agent	Agent 自主决定是否检索	条件检索循环 + 可注入 retriever
? Chain-of-Experts	任务流经多个专家节点	顺序路由 + 上下文累积
? Human-in-the-Loop	关键节点暂停等人类确认	interrupt + resume
? Swarm	去中心化群体协作	异步消息传递 + 聚合器

二、不知道选哪个？看决策树

这可能是整个项目中最实用的部分。我们绘制了一棵决策树，从你的具体场景需求出发，通过四层二元判断，可以精确导航到对应的模式。

简单总结几条核心规则：

• 需要人工审批？
  → 直接选择 Human-in-the-Loop，这是唯一选择。

• 处理简单任务但需要打磨质量？
  → Reflection 模式最简单有效。

• 需要从多个源头并行处理信息？
  → MapReduce 模式，利用 Send API 实现动态扇出。

• 有正反方观点需要权衡？
  → Debate 模式，让 Agent 互相挑战。

• 输出存在高风险，需要额外检查？
  → GuardRail 模式，提供批准、拦截、重定向三条路由。

三、深入两个核心模式

模式 1：Reflection（反思循环）

这是最简单的模式，但效果往往出奇地好。它的核心思想，其实是人类编辑流程的自动化版本：写手产出初稿，审稿人打分并提供反馈，写手据此修改，审稿人再次评估……如此循环，直到分数达标或达到最大轮次限制。

其核心代码非常简洁。LangGraph 的条件边天然适合这种“满足条件就退出，否则继续循环”的逻辑：

class ReflectionPattern:
    def build_graph(self) -> StateGraph:
        graph = StateGraph(ReflectionState)
        graph.add_node("write", self._write_node)
        graph.add_node("review", self._review_node)
        graph.add_edge(START, "write")
        graph.add_edge("write", "review")
        graph.add_conditional_edges(
            "review",
            self._should_continue,
            {"continue": "write", "end": END},
        )
        return graph.compile()

    def _should_continue(self, state: ReflectionState) -> str:
        if state["score"] >= self.score_threshold:
            return "end"
        if state["iteration"] >= self.max_iterations:
            return "end"
        return "continue"

仅仅几行条件判断，就构建了一个完整的写作-审阅循环。实际运行中，初稿可能只有 5/10 分，但经过 2-3 轮修改，通常能达到 8/10 分以上。

文章写作、代码生成后的自我审查、报告迭代优化。

模式 2：MapReduce（并行扇出）

当你需要从 N 个数据源并行收集信息，再进行汇总时，这个模式是最优解。关键在于 LangGraph 的 Send API——它允许在运行时动态决定并行分支的数量：

class MapReducePattern:
    def _dispatch(self, state: MapReduceState) -> list[Send]:
        """运行时动态创建 N 个并行 mapper"""
        return [
            Send("mapper", {"source": source, "topic": state["topic"]})
            for source in state["sources"]
        ]

    def build_graph(self) -> StateGraph:
        graph = StateGraph(MapReduceState)
        graph.add_node("mapper", self._mapper)
        graph.add_node("reducer", self._reducer)
        # 关键：用 conditional_edges + Send 实现动态并行
        graph.add_conditional_edges(START, self._dispatch, ["mapper"])
        graph.add_edge("mapper", "reducer")
        graph.add_edge("reducer", END)
        return graph.compile()

3 个数据源就并行启动 3 个 mapper，10 个就启动 10 个——分支数量完全由输入数据动态决定，无需在代码中硬编码。

多源新闻聚合、并行文档摘要、分布式数据提取。

四、几个技术细节的处理

4.1 Debate 和 Swarm 的真正异步并发

许多“多 Agent”项目看似并行，实则是 for agent in agents: llm.invoke() 的串行调用。在 AgentFlow 中，Debate 和 Swarm 模式使用 asyncio.gather 配合 ainvoke 实现了真正的并发：

async def _debate_round(self, state: DebateState) -> dict:
    tasks = [
        self._call_debater(debater, state["topic"], ...)
        for debater in state["debaters"]
    ]
    results = await asyncio.gather(*tasks)  # 真正的并发
    return {"debate_history": list(results)}

这样一来，3 个辩论者同时调用 LLM，总耗时大约等于 1 次调用，而非 3 次。在 Agent 数量较多的场景下，性能差异会非常明显。

4.2 RAG-Agent 的可注入检索器

很多 RAG 示例项目使用硬编码的模拟数据，演示完便难以投入实际使用。AgentFlow 的 RAG-Agent 模式将检索函数设计为可注入的参数：

RetrieverFunc = Callable[[list[str]], list[dict]]

class RAGAgentPattern:
    def __init__(self, retriever: RetrieverFunc | None = None):
        self.retriever = retriever or _default_mock_retriever

默认配置使用模拟检索器运行演示，但要接入真实的向量数据库，只需一行代码：

pattern = RAGAgentPattern(retriever=my_chroma_retriever)

4.3 真实的 LLM 调用计数

性能基准测试不应依赖“估算”。AgentFlow 通过集成 LangChain 的 CallbackHandler 实现了精确的调用计数：

class LLMCallCounterHandler(BaseCallbackHandler):
    def on_chat_model_start(self, serialized, messages, **kwargs):
        self._count[0] += 1

每个模式的 run() 方法返回值中都包含 llm_call_count 字段，基准测试框架可以直接读取这些真实数据。

五、Pattern 组合：1+1 > 2

单个模式已经很有用，但真正的威力在于组合。项目的 examples/ 目录下提供了两个组合案例：

= MapReduce + Debate + Reflection：

多源并行采集新闻 → 正反方辩论新闻价值 → 编辑循环打磨成稿

整个流程模拟了一个真实新闻编辑部的工作方式：记者分头采集信息，编辑部内部讨论新闻角度，主编反复打磨稿件。三种模式各司其职，通过 LangGraph 的状态流无缝串联。

= Hierarchical + Chain-of-Experts：

项目经理拆分子任务 → 各领域专家顺序分析 → 项目经理汇总报告

六、项目规格

说几个硬指标：

• 10 种模式
  ，每种包含 200-320 行 Python 核心代码，总计约 2400 行。

• 140 个单元测试
  ，全部使用模拟 LLM，无需 API 密钥即可运行。

• 基准测试框架
  ，支持跨模式对比 LLM 调用次数、耗时和输出质量。

• CI 全绿
  ：集成 pytest 和 MkDocs 构建，每次推送代码自动运行。

• 中英双语文档
  ：每个模式都有独立的中文和英文 README。

• 3 分钟上手
  ：克隆仓库 → 设置 API 密钥 → 运行任意示例。

git clone https://github.com/iuyup/AgentFlow.git
cd AgentFlow && uv sync
cp .env.example .env # 填入你的 API key
python -m patterns.reflection.example

七、为什么不用 CrewAI / AutoGen？

这个问题肯定有人会问，这里直接给出回答：

AgentFlow 的定位不是框架，而是参考手册。CrewAI 和 AutoGen 这类工具旨在帮助你“快速搭建”多 Agent 系统，它们在 LLM 之上增加了一层抽象。而 AgentFlow 的哲学截然不同——它直接使用 LangGraph 的原生 API，不添加任何中间层。

这意味着：

• 你学习到的是 LangGraph 本身的能力，而非某个封装层的特定用法。
• 每个模式都是独立的，你需要哪个，就直接复制哪个到自己的项目中。
• 没有黑箱，每个节点的行为、每条边的路由逻辑都清晰可见。

用设计模式来类比：你不会因为学习了“观察者模式”就必须引入一个观察者框架。Pattern 是思想，不是强制的依赖。

八、后续计划

• Streaming 支持
  （目前所有模式均为批处理模式）。

• 更多组合案例
  （例如 GuardRail + RAG-Agent = 安全知识问答）。

• 运行完整的基准测试
  ，并公开各模式的实测对比数据。

• LangGraph Cloud 部署指南
  。