LangGraph 实战：一个 Coordinator 带着 5 个专家 Agent 干活（代码全部可运行）

🤵‍♂️ 个人主页：小李同学_LSH的主页

✍🏻 作者简介：LLM学习者
🐋 希望大家多多支持，我们一起进步！😄
如果文章对你有帮助的话，
欢迎评论 💬点赞👍🏻 收藏 📂加关注+

目录

🎯 我们要做什么

🔧 环境准备

📐 整体架构

💻 完整代码实现

Step 1：基础设置

Step 2：给每个 Agent 定义工具

============================================================

CriticAgent 的工具：问题分析

============================================================

============================================================

WriterAgent 的工具：格式化输出

============================================================

参考文献

============================================================

QAAgent 的工具：质量评分

============================================================

Step 4：创建 Coordinator（核心）

Step 5：运行系统

Step 6：追问和多轮对话（重要！）

🔍 面试必考知识点详解

Q1：Supervisor 模式 vs Swarm 模式，什么区别？

Q2：为什么用 create_react_agent 而不是自己写节点？

Q3：InMemorySaver vs 生产级持久化

Q4：如何观测每个 Agent 的执行过程？

Q5：如何处理 Agent 失败的情况？

🚀 进阶：加入人工介入节点（Human-in-the-Loop）

📊 效果演示

🗳️ 读者投票

🎁 总结速查

📣 最后

写在前面：多 Agent 协作是 2026 年 AI 工程师面试最高频的话题之一。这篇文章用一个技术报告生成系统作为场景，手把手带你搭一套 Coordinator + 5 个专家 Agent 的完整系统。代码基于LangGraph v1.0（2025年10月稳定版），全部可直接运行，没有废弃 API。

🎯 我们要做什么

场景：给一个技术主题（比如"RAG 系统优化"），自动生成一份完整的技术研究报告。

这个任务需要多个专家协作完成：

用户提问："帮我研究 RAG 系统优化方案" Coordinator（协调者）接收任务，拆解后依次派发： ↓ ├── ResearchAgent（调研专家）：搜索最新资料 ├── CodeAgent（代码专家）：写示例代码 ├── CriticAgent（审查专家）：找漏洞和问题 ├── WriterAgent（写作专家）：整合成报告 └── QAAgent（质检专家）：最终质量把关 最终输出：一份结构完整、有代码、经过审查的技术报告

为什么用多 Agent 而不是一个大 Prompt？

🔧 环境准备

pip install langgraph>=1.0 langgraph-supervisor langchain-openai langchain-anthropic langchain-core langsmith

.env文件（项目根目录）：

# 二选一，或同时配置 OPENAI_API_KEY=sk-... ANTHROPIC_API_KEY=sk-ant-... # 可选：LangSmith 追踪（强烈推荐，面试时可以展示 trace） LANGCHAIN_TRACING_V2=true LANGCHAIN_API_KEY=ls__... LANGCHAIN_PROJECT=multi-agent-demo

📐 整体架构

在写代码之前，先把架构图理清楚：

核心机制：

• Coordinator 是唯一的路由节点，它不执行任务，只做决策
• 每个 Worker Agent 完成任务后，通过transfer_back_to_supervisor工具把控制权还给 Coordinator
• Coordinator 看到汇报后，决定下一步派谁还是直接结束

💻 完整代码实现

Step 1：基础设置

# multi_agent_system.py import os from dotenv import load_dotenv load_dotenv() from langchain_openai import ChatOpenAI from langchain_anthropic import ChatAnthropic from langchain_core.tools import tool from langgraph.prebuilt import create_react_agent from langgraph.checkpoint.memory import InMemorySaver from langgraph_supervisor import create_supervisor # ---- 选你的模型 ---- # 用 GPT-4o（推荐，工具调用稳定） llm = ChatOpenAI(model="gpt-4o", temperature=0) # 或者用 Claude Sonnet # llm = ChatAnthropic(model="claude-sonnet-4-6", temperature=0) # 或者用国产模型（DeepSeek） # from langchain_openai import ChatOpenAI # llm = ChatOpenAI( # model="deepseek-chat", # base_url="https://api.deepseek.com/v1", # api_key=os.getenv("DEEPSEEK_API_KEY"), # temperature=0 # ) print("模型初始化完成")

Step 2：给每个 Agent 定义工具

每个 Agent 只拥有自己专属的工具，职责单一是多 Agent 设计的核心原则。

# ============================================================ # ResearchAgent 的工具：搜索 + 摘要 # ============================================================ @tool def search_arxiv(query: str) -> str: """搜索 arxiv 上与查询相关的最新论文，返回标题、摘要和链接""" # 实际生产中接入真实 API，这里用 Mock 演示结构 mock_results = f""" 搜索查询：{query} 论文1：《Adaptive RAG: Learning to Adapt Retrieval》(2025) 摘要：提出了自适应检索策略，根据查询复杂度动态决定是否需要检索... 链接：https://arxiv.org/abs/2403.14403 论文2：《RAPTOR: Recursive Abstractive Processing for Tree-Organized Retrieval》(2024) 摘要：构建递归摘要树来改善长文档的检索效果... 链接：https://arxiv.org/abs/2401.18059 论文3：《GraphRAG: Knowledge Graph-Augmented Generation》(2025) 摘要：利用知识图谱增强 RAG 的推理能力，解决多跳问题... 链接：https://arxiv.org/abs/2404.16130 """ return mock_results @tool def fetch_github_trending(topic: str) -> str: """获取 GitHub 上与主题相关的热门开源项目""" mock_results = f""" 主题：{topic} 相关热门项目 1. LlamaIndex (35k stars) - 企业级 RAG 框架，支持 GraphRAG 2. Chroma (15k stars) - 轻量级向量数据库，适合原型开发 3. Milvus (30k stars) - 生产级向量数据库，支持亿级向量 4. RAGAS (8k stars) - RAG 系统评估框架 5. FastRAG (3k stars) - 英特尔优化的高性能 RAG 库 """ return mock_results research_tools = [search_arxiv, fetch_github_trending] # ============================================================ # CodeAgent 的工具：代码生成 + 执行 # ============================================================ @tool def generate_code_example(task_description: str, language: str = "python") -> str: """根据任务描述生成可运行的代码示例""" # 实际生产中让 LLM 生成，这里 Mock 演示 if "rag" in task_description.lower(): return ''' ```python # RAG 系统核心实现示例 from langchain.document_loaders import PyPDFLoader from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain_openai import OpenAIEmbeddings, ChatOpenAI from langchain_chroma import Chroma from langchain.chains import RetrievalQA def build_rag_system(pdf_path: str): # 1. 加载文档 loader = PyPDFLoader(pdf_path) docs = loader.load() # 2. 切割（chunk_size 选 1000，overlap 200 是通用起点） splitter = RecursiveCharacterTextSplitter( chunk_size=1000, chunk_overlap=200, separators=["\n\n", "\n", "。", ".", " "] ) chunks = splitter.split_documents(docs) # 3. 向量化 + 存储 embeddings = OpenAIEmbeddings(model="text-embedding-3-small") vectordb = Chroma.from_documents(chunks, embeddings, persist_directory="./chroma_db") # 4. 构建检索链 qa = RetrievalQA.from_chain_type( llm=ChatOpenAI(model="gpt-4o-mini"), retriever=vectordb.as_retriever( search_type="mmr", # MMR 减少冗余 search_kwargs={"k": 5, "fetch_k": 20} ), return_source_documents=True ) return qa # 使用 qa_system = build_rag_system("your_doc.pdf") result = qa_system.invoke({"query": "RAG 的核心流程是什么？"}) print(result["result"])

''' return f"# {language} 代码示例\n# 待实现: {task_description}" @tool def run_code_check(code: str) -> str: """对代码进行语法检查和基本质量分析""" issues = [] if "import" not in code: issues.append("警告：未发现 import 语句") if "try" not in code and "except" not in code: issues.append("建议：添加异常处理（try/except）") if len(code.split('\n')) > 50 and "def " not in code: issues.append("建议：代码较长，考虑拆分为函数")

if not issues: return "代码检查通过，未发现明显问题" return "代码检查结果：\n" + "\n".join(f"- {i}" for i in issues)

code_tools = [generate_code_example, run_code_check]

============================================================

CriticAgent 的工具：问题分析

============================================================

@tool def analyze_weaknesses(content: str) -> str: """分析技术方案的潜在缺陷和改进点""" return f""" 对以下内容的批判性分析：

已识别的潜在问题： 1. 【技术局限】向量检索本质上是语义相似度匹配，对精确匹配（如代码、公式）效果有限 2. 【性能瓶颈】大规模向量库（亿级）的实时检索延迟可能超过 100ms，影响用户体验 3. 【一致性问题】文档更新后，向量库需要同步更新，存在数据一致性挑战 4. 【评估困难】RAG 系统的效果难以自动化评估，需要人工标注数据 5. 【成本问题】大量 Embedding 调用会产生 API 费用，需要合理缓存策略 改进建议： - 混合检索（BM25 + 向量）提升精确匹配能力 - 添加 RAGAS 自动评估框架 - 实现增量更新机制避免全量重建 """

@tool def fact_check(claims: str) -> str: """验证技术声明的准确性""" return f""" 事实核查结果： - 声明"RAG 比 Fine-tuning 更适合知识更新频繁的场景"：✅ 准确 - 声明"向量相似度搜索时间复杂度 O(1)"：⚠️ 不准确， 暴力搜索是 O(n)，近似搜索（HNSW）是 O(log n) - 声明"chunk_size 越小越好"：❌ 错误， 太小会丢失上下文，太大会引入噪音，需要根据文档类型调整 """ critic_tools = [analyze_weaknesses, fact_check]

============================================================

WriterAgent 的工具：格式化输出

============================================================

@tool def format_as_report( title: str, sections: str, include_toc: bool = True ) -> str: """将内容格式化为结构化的技术报告""" toc = "## 目录\n1. 背景与问题定义\n2. 技术方案\n3. 代码示例\n4. 优缺点分析\n5. 最佳实践建议\n\n" if include_toc else "" return f"""# {title} {toc}--- {sections} @tool def add_references(content: str, topic: str) -> str: """为报告添加参考文献部分""" return content + f"""

参考文献

1. Lewis et al. (2020).Retrieval-Augmented Generation for Knowledge-Intensive NLP Tasks. NeurIPS.
2. Gao et al. (2024).Retrieval-Augmented Generation for Large Language Models: A Survey. arXiv.
3. Edge et al. (2024).From Local to Global: A Graph RAG Approach. Microsoft Research.
4. LangChain Documentation: https://docs.langchain.com
5. LlamaIndex Documentation: https://docs.llamaindex.ai """

writer_tools = [format_as_report, add_references]

QAAgent 的工具：质量评分

@tool def score_report_quality(report: str) -> str: """对报告进行多维度质量评分（0-10分）""" scores = { "完整性": 8, # 是否覆盖了所有关键方面 "准确性": 9, # 技术事实是否正确 "可读性": 8, # 结构清晰、语言流畅 "实用性": 9, # 代码示例是否可用 "深度": 7, # 是否有足够的技术深度 } avg = sum(scores.values()) / len(scores) details = "\n".join(f" - {k}: {v}/10" for k, v in scores.items())

feedback = [] if scores["深度"] < 8: feedback.append("建议：增加更多技术细节，如算法复杂度分析") if scores["完整性"] < 9: feedback.append("建议：补充部署和监控相关内容") return f""" 质量评分报告：

{details} 综合评分：{avg:.1f}/10

改进建议： {chr(10).join(feedback) if feedback else " 无，报告质量良好"} 结论：{"通过" if avg >= 7.5 else "需要修改"} """

@tool def check_completeness(report: str, required_sections: list[str]) -> str: """检查报告是否包含所有必要章节""" missing = [s for s in required_sections if s.lower() not in report.lower()] if missing: return f"缺少以下章节：{', '.join(missing)}" return "完整性检查通过，所有必要章节均已包含"

qa_tools = [score_report_quality, check_completeness]

--- ### Step 3：创建 5 个专家 Agent ```python # ============================================================ # 创建专家 Worker Agents # 每个 Agent = LLM + 专属工具 + 专属 System Prompt # ============================================================ research_agent = create_react_agent( model=llm, tools=research_tools, name="research_agent", prompt=( "你是一名技术调研专家，专注于收集和整理最新的技术资料。\n" "你的职责：\n" "1. 使用 search_arxiv 搜索相关论文\n" "2. 使用 fetch_github_trending 获取热门项目\n" "3. 整理成结构化的调研报告\n" "完成后，将调研结果以清晰的格式汇报给协调者。" ) ) code_agent = create_react_agent( model=llm, tools=code_tools, name="code_agent", prompt=( "你是一名代码专家，专注于编写高质量的示例代码。\n" "你的职责：\n" "1. 根据主题生成可运行的代码示例\n" "2. 使用 run_code_check 检查代码质量\n" "3. 确保代码有完整的注释和异常处理\n" "完成后，将代码示例汇报给协调者。" ) ) critic_agent = create_react_agent( model=llm, tools=critic_tools, name="critic_agent", prompt=( "你是一名技术评审专家，专注于发现方案的潜在问题。\n" "你的职责：\n" "1. 使用 analyze_weaknesses 分析技术方案的缺陷\n" "2. 使用 fact_check 验证关键技术声明\n" "3. 提供具体的改进建议\n" "完成后，将审查结果汇报给协调者。\n" "注意：你的工作是建设性批评，而不是否定一切。" ) ) writer_agent = create_react_agent( model=llm, tools=writer_tools, name="writer_agent", prompt=( "你是一名技术写作专家，专注于将信息整合成高质量的报告。\n" "你的职责：\n" "1. 整合调研结果、代码示例和审查意见\n" "2. 使用 format_as_report 生成结构化报告\n" "3. 使用 add_references 添加参考文献\n" "完成后，将完整报告汇报给协调者。" ) ) qa_agent = create_react_agent( model=llm, tools=qa_tools, name="qa_agent", prompt=( "你是一名质量保证专家，负责对报告进行最终审核。\n" "你的职责：\n" "1. 使用 score_report_quality 对报告打分\n" "2. 使用 check_completeness 检查报告完整性\n" "3. 如果评分低于 7.5，明确指出需要改进的地方\n" "完成后，给出最终的质检结论。" ) ) print("5 个专家 Agent 创建完成")

Step 4：创建 Coordinator（核心）

# ============================================================ # 创建 Coordinator（协调者） # ============================================================ checkpointer = InMemorySaver() # 对话状态持久化 supervisor_app = create_supervisor( agents=[ research_agent, code_agent, critic_agent, writer_agent, qa_agent, ], model=llm, prompt=( "你是一个技术报告生成系统的协调者（Coordinator）。\n\n" "你管理以下 5 名专家：\n" "- research_agent（调研专家）：负责搜索论文和开源项目\n" "- code_agent（代码专家）：负责编写示例代码\n" "- critic_agent（审查专家）：负责找出方案的问题和漏洞\n" "- writer_agent（写作专家）：负责整合所有内容生成报告\n" "- qa_agent（质检专家）：负责最终的质量把关\n\n" "标准工作流程（按顺序执行）：\n" "1. 先派 research_agent 收集资料\n" "2. 再派 code_agent 编写示例代码\n" "3. 再派 critic_agent 审查发现问题\n" "4. 再派 writer_agent 整合所有内容生成报告\n" "5. 最后派 qa_agent 进行质量验收\n" "6. qa_agent 通过后输出 FINISH\n\n" "重要原则：\n" "- 每次只派一个 Agent 工作\n" "- 等 Agent 汇报完成后再决定下一步\n" "- 如果 qa_agent 发现严重问题，可以重新派相关专家修改\n" "- 最终输出要包含完整的技术报告内容" ), output_mode="last_message", # 只返回最后一条消息（完整报告） ).compile(checkpointer=checkpointer) print("Coordinator 创建完成，系统就绪！")

Step 5：运行系统

# ============================================================ # 运行多 Agent 系统 # ============================================================ def run_report_generation(topic: str, thread_id: str = "demo-001"): """ 生成技术研究报告 Args: topic: 研究主题 thread_id: 会话 ID（用于状态持久化，同一 ID 可继续上次对话） """ print(f"\n{'='*60}") print(f"开始生成报告：{topic}") print(f"{'='*60}\n") config = {"configurable": {"thread_id": thread_id}} # 流式输出，可以看到每个 Agent 的工作过程 for step in supervisor_app.stream( {"messages": [("user", f"请为以下主题生成完整的技术研究报告：{topic}")]}, config=config, stream_mode="updates" ): for node_name, node_output in step.items(): if node_name == "__end__": continue # 打印当前工作的 Agent print(f"\n🔄 [{node_name}] 正在工作...") # 打印最新消息 if "messages" in node_output: last_msg = node_output["messages"][-1] if hasattr(last_msg, "content") and last_msg.content: # 只打印前 300 字，避免输出过长 content = str(last_msg.content) print(f" {content[:300]}{'...' if len(content) > 300 else ''}") # 获取最终结果 final_state = supervisor_app.get_state(config) final_messages = final_state.values.get("messages", []) if final_messages: final_report = final_messages[-1].content print(f"\n{'='*60}") print("最终报告：") print(f"{'='*60}") print(final_report) return final_report return "报告生成失败" # 运行！ if __name__ == "__main__": report = run_report_generation( topic="RAG 系统优化：从基础检索到 GraphRAG", thread_id="rag-report-001" )

Step 6：追问和多轮对话（重要！）

状态持久化让系统支持多轮对话，这在实际使用中非常有价值：

def followup_question(question: str, thread_id: str): """ 对已生成的报告追问 使用相同的 thread_id，系统会记住之前的上下文 """ config = {"configurable": {"thread_id": thread_id}} result = supervisor_app.invoke( {"messages": [("user", question)]}, config=config ) return result["messages"][-1].content # 先生成报告 run_report_generation("RAG 系统优化", thread_id="session-001") # 然后追问（系统记得上下文！） followup = followup_question( "刚才提到的 GraphRAG，能让 code_agent 再写一个更完整的实现吗？", thread_id="session-001" ) print(followup)

🔍 面试必考知识点详解

Q1：Supervisor 模式 vs Swarm 模式，什么区别？

这是面试最高频的问题之一，必须能流畅回答：

Supervisor 模式（本文使用）： ┌─────────────────────────────────┐ │ Coordinator │ │ （中央大脑，所有决策在这里） │ └──────┬──────┬──────┬──────┬────┘ ↓ ↓ ↓ ↓ Agent1 Agent2 Agent3 Agent4 优点：决策集中，逻辑清晰，容易调试 缺点：Coordinator 是单点，复杂任务可能成瓶颈 适合：任务有明确先后顺序，需要全局状态管理 Swarm 模式（去中心化）： Agent1 ←→ Agent2 ↕ ↕ Agent3 ←→ Agent4 Agent 之间直接 handoff，通过 Command(goto=...) 跳转 优点：更灵活，延迟更低 缺点：调试困难，需要每个 Agent 都知道何时转交 适合：任务流程不固定，需要动态协作

Q2：为什么用create_react_agent而不是自己写节点？

# 手动写节点（繁琐，但灵活）： def my_agent_node(state: AgentState): response = llm.invoke(state["messages"]) if response.tool_calls: # 手动处理工具调用 tool_results = execute_tools(response.tool_calls) return {"messages": state["messages"] + [response] + tool_results} return {"messages": state["messages"] + [response]} # create_react_agent（简洁，自动处理工具循环）： agent = create_react_agent( model=llm, tools=my_tools, name="my_agent" ) # 内部自动实现了：调用LLM → 有工具调用? → 执行工具 → 再调用LLM → ... → 返回

create_react_agent封装了 ReAct（Reasoning + Acting）循环，适合大多数场景。当你需要自定义循环逻辑时再手写节点。

Q3：InMemorySavervs 生产级持久化

from langgraph.checkpoint.memory import InMemorySaver # 开发测试用 from langgraph.checkpoint.postgres import PostgresSaver # 生产用 from langgraph.checkpoint.redis import RedisSaver # 高并发用 # 生产级配置示例： from psycopg_pool import ConnectionPool pool = ConnectionPool(conninfo="postgresql://user:pass@localhost/db") checkpointer = PostgresSaver(pool) app = supervisor.compile(checkpointer=checkpointer) # 此后重启服务，thread_id 的状态依然存在

面试延伸问题："如果用户关闭浏览器再打开，如何恢复上次的对话？"

答：使用PostgresSaver+ 一个稳定的thread_id（比如用户 ID），每次对话传入同一个thread_id即可。

Q4：如何观测每个 Agent 的执行过程？

# 方法一：LangSmith（最推荐，有可视化界面） # 只需要设置环境变量： # LANGCHAIN_TRACING_V2=true # LANGCHAIN_API_KEY=ls__... # 然后正常运行，自动记录所有 trace # 方法二：流式输出（本文使用的方法） for step in app.stream(inputs, stream_mode="updates"): for node, output in step.items(): print(f"节点 [{node}] 完成") # 方法三：获取完整执行历史 history = list(app.get_state_history(config)) for state in history: print(state.config, state.values)

Q5：如何处理 Agent 失败的情况？

# 在 Coordinator 的 prompt 中加入错误处理指令： coordinator_prompt = """ ...（原有 prompt）... 错误处理原则： - 如果某个 Agent 报告工具调用失败，先重试一次 - 如果连续失败两次，跳过该步骤并在报告中注明 - 如果 writer_agent 无法生成完整报告，请求 research_agent 补充资料后重试 - 任何情况下都不要无限循环，最多总步骤不超过 20 步 """ # 在代码层面设置递归限制： app = supervisor.compile( checkpointer=checkpointer, # 防止无限循环（LangGraph v1.0 支持） ) config = { "configurable": {"thread_id": "xxx"}, "recursion_limit": 25 # 最大节点执行次数 }

🚀 进阶：加入人工介入节点（Human-in-the-Loop）

面试加分项——让系统在关键节点等待人工确认：

from langgraph.types import interrupt # 在 writer_agent 完成后，加入一个人工审核步骤 @tool def request_human_approval(draft_report: str) -> str: """请求人工审核报告草稿""" # interrupt 会暂停图的执行，等待人工输入 human_feedback = interrupt({ "type": "approval_request", "report_preview": draft_report[:500], "question": "请审核报告草稿，输入'approve'通过，或输入修改意见" }) return f"人工反馈：{human_feedback}" # 恢复执行（人工审核后）： app.invoke( Command(resume="approve"), # 或者 Command(resume="请补充部署相关内容") config=config )

📊 效果演示

运行上面的代码，你会看到类似这样的输出：

============================================================ 开始生成报告：RAG 系统优化：从基础检索到 GraphRAG ============================================================ 🔄 [supervisor] 正在工作... 决定派 research_agent 开始收集资料... 🔄 [research_agent] 正在工作... 搜索 arxiv 论文：RAG optimization GraphRAG... 获取 GitHub 热门项目：RAG... 已完成调研，整理了 3 篇核心论文和 5 个热门项目... 🔄 [supervisor] 正在工作... 调研完成。现在派 code_agent 编写示例代码... 🔄 [code_agent] 正在工作... 生成 RAG 核心实现代码... 代码检查：通过... 🔄 [supervisor] 正在工作... 代码完成。派 critic_agent 进行审查... 🔄 [critic_agent] 正在工作... 分析技术方案缺陷... 进行事实核查... 发现 3 个需要注意的问题... 🔄 [supervisor] 正在工作... 审查完成。派 writer_agent 整合内容... 🔄 [writer_agent] 正在工作... 整合调研结果、代码示例和审查意见... 生成结构化报告... 添加参考文献... 🔄 [supervisor] 正在工作... 报告生成完成。派 qa_agent 进行质检... 🔄 [qa_agent] 正在工作... 质量评分：综合 8.2/10... 完整性检查：通过... 质检结论：通过！ 🔄 [supervisor] 正在工作... 所有步骤完成，输出最终报告... # RAG 系统优化：从基础检索到 GraphRAG [完整报告内容...]

🔔 遇到问题欢迎评论，我都会回复！

🎁 总结速查

📣 最后

如果这篇帮你搞懂了多 Agent 协作的核心机制：

• 👍点赞让更多在学 Agent 的同学看到
• ⭐收藏代码随时复制使用
• 💬评论遇到报错或者有问题，我都会回复
• 🔔关注持续更新 AI Agent 实战，一个正在学 AI 的大学生 👨‍🎓