‍♂️ 个人主页小李同学_LSH的主页✍ 作者简介LLM学习者 希望大家多多支持我们一起进步如果文章对你有帮助的话欢迎评论 点赞 收藏 加关注目录 我们要做什么 环境准备 整体架构 完整代码实现Step 1基础设置Step 2给每个 Agent 定义工具CriticAgent 的工具问题分析WriterAgent 的工具格式化输出参考文献QAAgent 的工具质量评分Step 4创建 Coordinator核心Step 5运行系统Step 6追问和多轮对话重要 面试必考知识点详解Q1Supervisor 模式 vs Swarm 模式什么区别Q2为什么用 create_react_agent 而不是自己写节点Q3InMemorySaver vs 生产级持久化Q4如何观测每个 Agent 的执行过程Q5如何处理 Agent 失败的情况 进阶加入人工介入节点Human-in-the-Loop 效果演示️ 读者投票 总结速查 最后写在前面多 Agent 协作是 2026 年 AI 工程师面试最高频的话题之一。这篇文章用一个技术报告生成系统作为场景手把手带你搭一套 Coordinator 5 个专家 Agent 的完整系统。代码基于LangGraph v1.02025年10月稳定版全部可直接运行没有废弃 API。 我们要做什么场景给一个技术主题比如RAG 系统优化自动生成一份完整的技术研究报告。这个任务需要多个专家协作完成用户提问帮我研究 RAG 系统优化方案 Coordinator协调者接收任务拆解后依次派发 ↓ ├── ResearchAgent调研专家搜索最新资料 ├── CodeAgent代码专家写示例代码 ├── CriticAgent审查专家找漏洞和问题 ├── WriterAgent写作专家整合成报告 └── QAAgent质检专家最终质量把关 最终输出一份结构完整、有代码、经过审查的技术报告为什么用多 Agent 而不是一个大 Prompt单 Agent 一把梭多 Agent 协作适合场景简单单步任务复杂多阶段任务质量容易遗漏步骤每个专家聚焦自己的领域可调试性黑盒出错难定位每步可观测易排查扩展性修改一个影响全局增删 Agent 不影响其他面试价值基础高频考点 环境准备pip install langgraph1.0 langgraph-supervisor langchain-openai langchain-anthropic langchain-core langsmith.env文件项目根目录# 二选一或同时配置 OPENAI_API_KEYsk-... ANTHROPIC_API_KEYsk-ant-... # 可选LangSmith 追踪强烈推荐面试时可以展示 trace LANGCHAIN_TRACING_V2true LANGCHAIN_API_KEYls__... LANGCHAIN_PROJECTmulti-agent-demo 整体架构在写代码之前先把架构图理清楚核心机制Coordinator 是唯一的路由节点它不执行任务只做决策每个 Worker Agent 完成任务后通过transfer_back_to_supervisor工具把控制权还给 CoordinatorCoordinator 看到汇报后决定下一步派谁还是直接结束 完整代码实现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(modelgpt-4o, temperature0) # 或者用 Claude Sonnet # llm ChatAnthropic(modelclaude-sonnet-4-6, temperature0) # 或者用国产模型DeepSeek # from langchain_openai import ChatOpenAI # llm ChatOpenAI( # modeldeepseek-chat, # base_urlhttps://api.deepseek.com/v1, # api_keyos.getenv(DEEPSEEK_API_KEY), # temperature0 # ) 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 选 1000overlap 200 是通用起点 splitter RecursiveCharacterTextSplitter( chunk_size1000, chunk_overlap200, separators[\n\n, \n, 。, ., ] ) chunks splitter.split_documents(docs) # 3. 向量化 存储 embeddings OpenAIEmbeddings(modeltext-embedding-3-small) vectordb Chroma.from_documents(chunks, embeddings, persist_directory./chroma_db) # 4. 构建检索链 qa RetrievalQA.from_chain_type( llmChatOpenAI(modelgpt-4o-mini), retrievervectordb.as_retriever( search_typemmr, # MMR 减少冗余 search_kwargs{k: 5, fetch_k: 20} ), return_source_documentsTrue ) 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参考文献Lewis et al. (2020).Retrieval-Augmented Generation for Knowledge-Intensive NLP Tasks. NeurIPS.Gao et al. (2024).Retrieval-Augmented Generation for Large Language Models: A Survey. arXiv.Edge et al. (2024).From Local to Global: A Graph RAG Approach. Microsoft Research.LangChain Documentation: https://docs.langchain.comLlamaIndex 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( modelllm, toolsresearch_tools, nameresearch_agent, prompt( 你是一名技术调研专家专注于收集和整理最新的技术资料。\n 你的职责\n 1. 使用 search_arxiv 搜索相关论文\n 2. 使用 fetch_github_trending 获取热门项目\n 3. 整理成结构化的调研报告\n 完成后将调研结果以清晰的格式汇报给协调者。 ) ) code_agent create_react_agent( modelllm, toolscode_tools, namecode_agent, prompt( 你是一名代码专家专注于编写高质量的示例代码。\n 你的职责\n 1. 根据主题生成可运行的代码示例\n 2. 使用 run_code_check 检查代码质量\n 3. 确保代码有完整的注释和异常处理\n 完成后将代码示例汇报给协调者。 ) ) critic_agent create_react_agent( modelllm, toolscritic_tools, namecritic_agent, prompt( 你是一名技术评审专家专注于发现方案的潜在问题。\n 你的职责\n 1. 使用 analyze_weaknesses 分析技术方案的缺陷\n 2. 使用 fact_check 验证关键技术声明\n 3. 提供具体的改进建议\n 完成后将审查结果汇报给协调者。\n 注意你的工作是建设性批评而不是否定一切。 ) ) writer_agent create_react_agent( modelllm, toolswriter_tools, namewriter_agent, prompt( 你是一名技术写作专家专注于将信息整合成高质量的报告。\n 你的职责\n 1. 整合调研结果、代码示例和审查意见\n 2. 使用 format_as_report 生成结构化报告\n 3. 使用 add_references 添加参考文献\n 完成后将完整报告汇报给协调者。 ) ) qa_agent create_react_agent( modelllm, toolsqa_tools, nameqa_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, ], modelllm, 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_modelast_message, # 只返回最后一条消息完整报告 ).compile(checkpointercheckpointer) 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})]}, configconfig, stream_modeupdates ): 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( topicRAG 系统优化从基础检索到 GraphRAG, thread_idrag-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)]}, configconfig ) return result[messages][-1].content # 先生成报告 run_report_generation(RAG 系统优化, thread_idsession-001) # 然后追问系统记得上下文 followup followup_question( 刚才提到的 GraphRAG能让 code_agent 再写一个更完整的实现吗, thread_idsession-001 ) print(followup) 面试必考知识点详解Q1Supervisor 模式 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( modelllm, toolsmy_tools, namemy_agent ) # 内部自动实现了调用LLM → 有工具调用? → 执行工具 → 再调用LLM → ... → 返回create_react_agent封装了 ReActReasoning Acting循环适合大多数场景。当你需要自定义循环逻辑时再手写节点。Q3InMemorySavervs 生产级持久化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(conninfopostgresql://user:passlocalhost/db) checkpointer PostgresSaver(pool) app supervisor.compile(checkpointercheckpointer) # 此后重启服务thread_id 的状态依然存在面试延伸问题如果用户关闭浏览器再打开如何恢复上次的对话答使用PostgresSaver 一个稳定的thread_id比如用户 ID每次对话传入同一个thread_id即可。Q4如何观测每个 Agent 的执行过程# 方法一LangSmith最推荐有可视化界面 # 只需要设置环境变量 # LANGCHAIN_TRACING_V2true # LANGCHAIN_API_KEYls__... # 然后正常运行自动记录所有 trace # 方法二流式输出本文使用的方法 for step in app.stream(inputs, stream_modeupdates): 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( checkpointercheckpointer, # 防止无限循环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(resumeapprove), # 或者 Command(resume请补充部署相关内容) configconfig ) 效果演示运行上面的代码你会看到类似这样的输出 开始生成报告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 [完整报告内容...] 遇到问题欢迎评论我都会回复 总结速查概念关键点create_react_agent创建单个 Worker封装了 ReAct 工具调用循环create_supervisor创建 Coordinator管理多个 WorkerInMemorySaver开发用生产用PostgresSaverstream_modeupdates流式输出看到每步执行过程recursion_limit防止无限循环的安全阀interruptHuman-in-the-Loop 的核心机制thread_id多轮对话的会话 ID保证状态连续 最后如果这篇帮你搞懂了多 Agent 协作的核心机制点赞让更多在学 Agent 的同学看到⭐收藏代码随时复制使用评论遇到报错或者有问题我都会回复关注持续更新 AI Agent 实战一个正在学 AI 的大学生 ‍