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

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

前言：你的 Multi-Agent 项目，80% 的时间花在了架构纠结上

如果你正在用 LangGraph 构建多智能体系统，大概率遇到过这些问题：

• 两个 Agent 配合写文章，该用循环还是单次？循环几次合适？
• 5 个专家同时分析一个问题，是串行、并行、还是投票？
• 加了一个安全审核 Agent，怎么设计 approve / block / 重试的路由？
• 看了十几篇教程，每篇都说自己的方案好，但没人告诉我什么场景用什么模式。

工具不是问题——LangGraph 的 API 足够强大。真正的难点在于架构决策。

所以我做了 AgentFlow：一本多智能体设计模式的参考手册。不是框架，不是教程合集，而是 10 种经过实战验证的协作模式，每种都包含完整代码、架构图、单元测试和中英双语文档。

一、10 种模式一览

先看全景。AgentFlow 包含以下 10 种模式，覆盖了从最简单的自我修正到去中心化群体智能的完整谱系：

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

这大概是整个项目最实用的部分。我画了一棵决策树，从你的场景需求出发，4 层二元判断，精确导航到对应模式：

简单总结几条规则：

• 需要人审批？ → Human-in-the-Loop，没有其他选择
• 简单任务 + 需要打磨质量？ → Reflection，最简单也最有效
• 多源并行处理？ → MapReduce，用 Send API 动态扇出
• 有正反方观点需要权衡？ → Debate，让 Agent 互相挑战
• 高风险输出需要检查？ → GuardRail，approve/block/redirect 三路由

三、深入两个核心模式

模式 1：Reflection（反思循环）

这是最简单的模式，但效果惊人地好。核心思想是人类编辑流程的 Agent 版本：

写手出初稿 → 审稿人打分+反馈 → 写手修改 → 审稿人再打分 → ...直到分数达标或达到最大轮次

核心代码非常简洁。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 个 debater 同时调用 LLM，耗时约等于 1 次调用而不是 3 次。在 Agent 数量多的场景下差异非常明显。

4.2 RAG-Agent 的可注入检索器

很多 RAG 示例项目都是硬编码 mock 数据，看完 demo 就没法用了。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

默认用 mock 跑 demo，但接入真实向量库只需一行：

pattern = RAGAgentPattern(retriever=my_chroma_retriever)

4.3 真实的 LLM 调用计数

Benchmark 不应该用"估算"。AgentFlow 通过 LangChain 的 CallbackHandler 实现精确计数：

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

每个 pattern 的 run() 方法返回值里都包含 llm_call_count 字段，Benchmark 框架直接读取真实数据。

五、Pattern 组合：1+1 > 2

单个 Pattern 已经有用，但真正的威力在于组合。项目的 examples/ 目录包含两个组合案例：

AI Newsroom（AI 新闻编辑部） = MapReduce + Debate + Reflection：

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

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

Research Team（研究团队） = Hierarchical + Chain-of-Experts：

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

六、项目规格

说几个硬指标：

• 10 种模式，每种 200-320 行 Python，总计约 2400 行核心代码
• 140 个单元测试，全部使用 mock LLM，不需要 API key 就能跑
• Benchmark 框架，支持跨模式对比 LLM 调用次数、耗时、输出质量
• CI 全绿：pytest + MkDocs build，每次 push 自动跑
• 中英双语文档：每个 pattern 都有独立的中文和英文 README
• 3 分钟上手：clone → set API key → 运行任意 example

git clone 
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 都是独立的，需要哪个就复制哪个到你自己的项目里
• 没有黑箱，每个节点的行为、每条边的路由逻辑都清清楚楚

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

八、后续计划

• Streaming 支持（目前所有模式都是 batch 模式）
• 更多组合案例（GuardRail + RAG-Agent = 安全知识问答）
• 跑一轮完整 Benchmark，公开各模式的实测对比数据
• LangGraph Cloud 部署指南

如果你觉得这个项目有用，欢迎在 GitHub 上点个 Star：

也欢迎提 Issue 或 PR。如果你有实际项目中遇到的多 Agent 架构问题，评论区聊聊，我可以帮你分析适合哪种模式。