一、理论基础从单智能体到多智能体1.1 单智能体的能力边界与瓶颈经过前面的学习我们已经掌握了工业级单智能体的开发但在面对复杂任务时单智能体存在以下无法突破的瓶颈瓶颈类型具体表现工业级场景示例能力单一单个智能体无法同时精通多个领域一份财务报告需要数据分析师、财务专家、文案专家共同完成注意力有限任务复杂度提升时容易遗漏关键信息分析 100 页的行业报告单智能体容易出现信息遗漏和幻觉效率低下只能串行执行任务无法并行处理同时检索多个数据源、生成多个章节的报告容错性差单个步骤出错会导致整个任务失败工具调用失败后单智能体容易陷入死循环或给出错误答案可扩展性差新增能力需要修改原有代码违反开闭原则为智能体添加代码生成能力需要修改整个思考逻辑1.2 多智能体系统的核心价值与本质多智能体系统Multi-Agent System, MAS的本质是 分而治之 的软件工程思想在 AI 领域的延伸 。它将一个复杂的大任务分解为多个简单的小任务每个小任务由一个专门的智能体负责通过智能体之间的协作共同完成目标。核心价值专业专精每个智能体只专注于一个领域通过提示词工程和微调可以达到专家级水平并行处理多个智能体可以同时执行独立的子任务大幅提升系统效率容错性强单个智能体出错不会影响整个系统其他智能体可以发现并纠正错误可扩展性好新增能力只需要添加新的智能体不需要修改原有代码复杂度可控将复杂任务分解为多个简单子任务降低系统设计和调试难度团队协作模拟人类团队的工作方式更容易被企业用户理解和接受1.3 多智能体系统的核心架构与协作模式1.3.1 三种核心架构深度对比架构类型核心思想适用场景优点缺点工业级成熟度主从式架构一个主智能体负责全局协调和任务分配多个从智能体负责执行具体任务任务分配、流程控制、项目管理结构简单、易于控制、逻辑清晰主智能体成为性能瓶颈和单点故障★★★★★对等式架构所有智能体地位平等自主决策通过消息传递进行通信和协作分布式决策、群体智能、去中心化系统容错性强、可扩展性好、负载均衡协调难度大、容易出现冲突、通信开销大★★★☆☆流水线式架构智能体按顺序执行任务前一个智能体的输出是后一个智能体的输入内容生成、数据处理、流水线作业逻辑清晰、易于调试、性能稳定灵活性差、无法并行处理、瓶颈在最慢的环节★★★★★1.3.2 四种常见协作模式深度解析分工协作模式核心思想将任务分解为多个独立的子任务每个智能体负责一个子任务示例研究员收集数据 → 分析师分析数据 → 写作家生成报告 → 设计师制作图表适用场景结构化、流程化的任务关键技术任务分解、结果汇总、阶段切换评审迭代模式核心思想一个智能体生成结果另一个智能体评审并提出修改意见迭代优化直到满足要求示例程序员写代码 → 代码审查员审查代码 → 程序员修改代码 → 再次审查适用场景对质量要求高的任务关键技术评审标准、迭代控制、终止条件并行处理模式核心思想多个智能体同时执行独立的子任务最后汇总结果示例同时检索市场数据、竞品数据、用户数据然后汇总分析适用场景数据收集、信息检索、多源数据聚合关键技术并行执行、结果合并、冲突解决协商决策模式核心思想多个智能体通过协商达成一致决策示例多个专家智能体共同诊断疾病投票得出最终结论适用场景复杂决策、风险评估、专家系统关键技术投票机制、协商协议、冲突解决1.4 LangGraph 多智能体实现原理深度解析LangGraph 并没有提供专门的 多智能体 类而是通过 状态共享 节点分工 路由控制 的方式实现多智能体系统这也是 LangGraph 最强大和最优雅的设计之一。核心原理状态共享所有智能体共享同一个全局状态State这是智能体之间通信的唯一方式节点分工每个智能体对应图中的一个节点负责执行一个特定的任务路由控制路由函数根据当前状态决定下一个执行哪个智能体节点工具共享所有智能体可以共享同一个工具集也可以拥有自己的私有工具特性复用原生支持 Checkpoint、人类介入、流式输出等所有单智能体的高级特性LangGraph 多智能体 vs 其他框架无需学习新的 API完全复用单智能体的知识和经验状态统一管理避免了复杂的通信协议和消息传递性能更高没有额外的通信开销和序列化开销调试更简单可以通过 LangSmith 追踪整个系统的执行过程1.5 多智能体系统设计的五大原则单一职责原则每个智能体只能有一个明确的角色和职责不要让一个智能体做太多事情明确边界原则每个智能体的任务边界要清晰输入输出要明确状态最小化原则状态应该只存储必要的信息避免状态过大导致性能问题可观测性原则每个智能体的输入输出都要记录日志便于调试和审计容错性原则设计时要考虑到智能体可能出错的情况要有错误处理和降级机制二、核心实战研究员 写作家 双智能体系统我们将实现一个工业级的双智能体报告生成系统这是企业中最常见的多智能体应用场景。整个系统采用流水线式架构分为两个阶段研究阶段和写作阶段。2.1 第一步系统设计与状态定义2.1.1 角色与职责设计核心理论为什么选择 研究员 写作家 的分工这是人类社会最经典的分工模式之一经过了长期的实践验证研究和写作需要完全不同的能力研究需要严谨、准确、逻辑性强写作需要创造力、表达能力、结构化思维这种分工可以让每个智能体专注于自己擅长的领域大幅提升输出质量便于后续扩展可以添加评审员、设计师等角色而不需要修改原有代码详细的角色与职责定义智能体角色定位核心能力输入输出拥有的工具温度设置研究员信息收集与分析专家信息检索、数据分析、要点整理、信息评估用户的报告需求结构化的研究要点企业知识库检索、计算器、当前时间0.1低温度保证准确性写作家内容生成与润色专家大纲生成、内容撰写、文字润色、结构化输出研究员的研究要点完整的商业报告无只负责生成内容0.7高温度提升创造性2.1.2 状态设计核心理论多智能体状态设计的核心原则复用优先尽可能复用已有的状态字段避免重复定义分层设计将状态分为基础层、多智能体层、业务层增量更新消息历史等字段使用增量更新避免覆盖历史数据可追踪性添加任务阶段、迭代次数等字段清晰记录系统的执行过程状态定义与代码实现# core/multi_agent.py from typing import List, Annotated from pydantic import BaseModel, Field from langchain_core.messages import BaseMessage import operator from enum import Enum # 复用第8天的审核状态 from core.react_agent import ApprovalStatus class TaskPhase(str, Enum): 任务阶段枚举清晰标记系统当前处于哪个阶段 RESEARCH research # 研究阶段 WRITING writing # 写作阶段 REVIEW review # 评审阶段预留扩展 COMPLETED completed # 完成阶段 class MultiAgentState(BaseModel): 双智能体系统状态分层设计 # 基础层复用单智能体的通用字段 messages: Annotated[List[BaseMessage], operator.add] Field(default_factorylist) iteration_count: int Field(default0) max_iterations: int Field(default10) # 多智能体迭代次数更多 user_id: str Field(defaultdefault_user) # 多智能体层多智能体系统通用字段 task_phase: TaskPhase Field(defaultTaskPhase.RESEARCH) current_agent: str Field(defaultresearcher) # 业务层报告生成业务特有字段 research_notes: str Field(default, description研究员整理的结构化研究要点) report_outline: str Field(default, description报告大纲预留扩展) report_draft: str Field(default, description报告草稿预留扩展) final_report: str Field(default, description最终生成的完整报告) # 审核层复用第8天的审核字段 approval_status: ApprovalStatus Field(defaultApprovalStatus.PENDING) approver: str Field(default) approval_comment: str Field(default)2.2 第二步研究员智能体实现2.2.1 研究员智能体设计核心理论研究员的核心能力模型需求理解能力准确理解用户的报告需求明确需要收集哪些信息信息检索能力熟练使用各种工具检索相关信息信息筛选能力从大量信息中筛选出有用的信息信息整理能力将筛选后的信息整理成结构化的要点信息评估能力评估信息是否足够生成一份完整的报告研究员提示词设计原则明确任务边界明确告诉研究员只负责收集和整理信息不生成完整报告强调准确性所有信息必须来自工具检索绝对不能编造结构化输出要求研究员输出结构化的要点便于写作家使用明确终止条件告诉研究员什么时候应该结束研究进入写作阶段工具使用指导详细说明每个工具的使用场景和方法2.2.2 核心代码实现from core.llm_factory import LLMFactory from core.tools import PRODUCTION_TOOLS from langgraph.prebuilt import ToolNode from utils.logger import logger # 初始化LLM并绑定工具研究员需要使用工具 researcher_llm LLMFactory.get_llm(temperature0.1) # 低温度保证准确性 researcher_llm_with_tools researcher_llm.bind_tools(PRODUCTION_TOOLS) # 研究员系统提示词 RESEARCHER_PROMPT 你是一位专业的企业研究员擅长收集和分析信息。 你的任务是 1. 深入理解用户的报告需求 2. 使用提供的工具检索相关信息 3. 整理成清晰、结构化的研究要点 4. 评估信息是否足够生成一份完整的报告 5. 如果信息不足继续检索如果足够将研究要点传递给写作家 你必须严格遵守以下规则 1. 所有信息必须来自工具检索绝对不能编造 2. 研究要点要全面、准确、有条理 3. 每个要点都要有对应的信息来源 4. 当你认为信息足够时明确说研究完成可以开始写作 5. 不要生成完整的报告只生成研究要点 你可以使用以下工具 {tools_description} tools_description \n.join([f- {tool.name}: {tool.description} for tool in PRODUCTION_TOOLS]) researcher_system_message SystemMessage(contentRESEARCHER_PROMPT.format(tools_descriptiontools_description)) def researcher_node(state: MultiAgentState) - dict: 研究员智能体节点 logger.info(f 研究员开始工作迭代次数{state.iteration_count}/{state.max_iterations}) # 构建消息列表系统提示词 对话历史 messages [researcher_system_message] state.messages # 调用LLM response researcher_llm_with_tools.invoke(messages) # 检查是否超过最大迭代次数 if state.iteration_count state.max_iterations: logger.warning(⚠️ 超过最大迭代次数强制进入写作阶段) return { messages: [response], iteration_count: state.iteration_count 1, task_phase: TaskPhase.WRITING, research_notes: response.content } # 检查是否研究完成 if 研究完成 in response.content and 可以开始写作 in response.content: logger.info(✅ 研究完成进入写作阶段) return { messages: [response], iteration_count: state.iteration_count 1, task_phase: TaskPhase.WRITING, research_notes: response.content } # 否则继续研究 return { messages: [response], iteration_count: state.iteration_count 1 }关键设计要点低温度设置研究员需要准确性所以设置 temperature0.1明确的任务边界研究员只负责收集和整理信息不生成完整报告自动阶段切换当研究员说 研究完成 时自动切换到写作阶段安全限制设置最大迭代次数防止无限研究2.3 第三步写作家智能体实现写作家智能体负责将研究员的研究要点转化为一份专业、结构化的报告。# 初始化写作家LLM不需要工具高温度提升创造性 writer_llm LLMFactory.get_llm(temperature0.7) # 写作家系统提示词 WRITER_PROMPT 你是一位专业的企业文案写作专家擅长撰写结构化、专业的商业报告。 你的任务是 1. 根据研究员提供的研究要点生成一份完整的商业报告 2. 报告结构要清晰包含标题、摘要、正文、结论等部分 3. 语言要专业、准确、简洁符合企业内部沟通规范 4. 所有内容必须基于研究员提供的研究要点不能添加任何外部信息 5. 生成完成后明确说报告生成完成 报告结构要求 # 报告标题 ## 一、摘要 简要概括报告的核心内容和结论 ## 二、研究背景 说明报告的目的和背景 ## 三、核心发现 分点列出最重要的研究发现 ## 四、详细分析 对每个核心发现进行详细分析 ## 五、结论与建议 总结报告内容提出具体的建议 研究员的研究要点 {research_notes} def writer_node(state: MultiAgentState) - dict: 写作家智能体节点 logger.info(✍️ 写作家开始工作) # 构建提示词包含研究员的研究要点 prompt WRITER_PROMPT.format(research_notesstate.research_notes) # 调用LLM生成报告 response writer_llm.invoke(prompt) logger.info(✅ 报告生成完成) return { messages: [response], iteration_count: state.iteration_count 1, task_phase: TaskPhase.COMPLETED, final_report: response.content }关键设计要点高温度设置写作家需要一定的创造性所以设置 temperature0.7结构化输出明确要求报告的结构保证输出的一致性严格的信息边界所有内容必须基于研究员的研究要点防止幻觉自动完成标记生成完成后明确标记便于系统识别2.4 第四步协作流程与路由函数设计路由函数是多智能体系统的大脑它决定了下一个执行哪个智能体。from langgraph.graph import END def multi_agent_router(state: MultiAgentState) - str: 多智能体路由函数决定下一个执行的节点 last_message state.messages[-1] # 检查是否有工具调用研究员需要调用工具 if last_message.tool_calls: logger.info(f 研究员需要调用工具{[tc[name] for tc in last_message.tool_calls]}) return tools # 根据任务阶段路由 if state.task_phase TaskPhase.RESEARCH: return researcher elif state.task_phase TaskPhase.WRITING: return writer elif state.task_phase TaskPhase.COMPLETED: return END else: logger.error(f❌ 未知的任务阶段{state.task_phase}) return END路由逻辑如果有工具调用先执行工具节点如果处于研究阶段执行研究员节点如果处于写作阶段执行写作家节点如果任务完成结束执行2.5 第五步构建与编译双智能体图from langgraph.graph import StateGraph from core.checkpoint import postgres_checkpointer # 复用第8天的Checkpoint def build_research_writer_agent() - StateGraph: 构建研究员写作家双智能体系统 builder StateGraph(MultiAgentState) # 添加节点 builder.add_node(researcher, researcher_node) builder.add_node(tools, ToolNode(PRODUCTION_TOOLS)) builder.add_node(writer, writer_node) # 添加边 builder.add_edge(tools, researcher) # 工具执行完回到研究员 builder.add_edge(writer, END) # 写作家执行完结束 # 添加条件边 builder.add_conditional_edges(researcher, multi_agent_router) # 设置入口点 builder.set_entry_point(researcher) # 编译图配置Checkpoint和中断点 graph builder.compile( checkpointerpostgres_checkpointer, interrupt_before[writer] # 在写作前添加中断点支持人工审核研究结果 ) logger.info(✅ 研究员写作家双智能体系统构建完成) return graph # 全局单例 research_writer_agent build_research_writer_agent()三、项目整合将双智能体集成到现有系统3.1 升级 RAGService支持双智能体# core/rag_service.py # 在原有导入基础上添加 from core.multi_agent import research_writer_agent, TaskPhase from langchain_core.messages import HumanMessage class RAGService: # 原有方法保持不变... # # 新增双智能体报告生成 # def generate_report( self, topic: str, user_id: str, auto_approve: bool False ) - dict: 生成报告双智能体系统 :param topic: 报告主题 :param user_id: 用户ID :param auto_approve: 是否自动批准研究结果跳过人工审核 config {configurable: {thread_id: freport_{user_id}_{int(time.time())}}} logger.info(f用户[{user_id}]请求生成报告{topic}) # 启动双智能体系统 result research_writer_agent.invoke( { messages: [HumanMessage(contenttopic)], user_id: user_id }, configconfig ) state research_writer_agent.get_state(config) # 如果需要人工审核 if state.next and writer in state.next: if auto_approve: logger.info(自动批准研究结果继续生成报告) result research_writer_agent.invoke(None, configconfig) return { task_id: config[configurable][thread_id], status: completed, report: result[final_report] } else: return { task_id: config[configurable][thread_id], status: pending_review, research_notes: state.values[research_notes], message: 研究完成等待人工审核 } # 如果已经完成 return { task_id: config[configurable][thread_id], status: completed, report: result[final_report] } def approve_research(self, task_id: str) - dict: 批准研究结果继续生成报告 config {configurable: {thread_id: task_id}} state research_writer_agent.get_state(config) if not state.next or writer not in state.next: raise ValueError(当前任务不需要审核或已被处理) logger.info(f批准任务{task_id}的研究结果开始生成报告) result research_writer_agent.invoke(None, configconfig) return { task_id: task_id, status: completed, report: result[final_report] } def reject_research(self, task_id: str, reason: str) - dict: 拒绝研究结果要求重新研究 config {configurable: {thread_id: task_id}} logger.info(f拒绝任务{task_id}的研究结果原因{reason}) # 更新状态添加拒绝消息让研究员重新研究 research_writer_agent.update_state( config, { messages: [HumanMessage(contentf研究结果被拒绝原因{reason}。请重新研究。)], task_phase: TaskPhase.RESEARCH } ) # 重新执行研究员节点 result research_writer_agent.invoke(None, configconfig) state research_writer_agent.get_state(config) return { task_id: task_id, status: pending_review, research_notes: state.values[research_notes], message: 已重新研究等待再次审核 }3.2 新增双智能体 API 接口# main8.py # 在原有请求模型基础上添加 class ReportRequest(BaseModel): topic: str user_id: str auto_approve: bool False class TaskIdRequest(BaseModel): task_id: str reason: str # 在原有接口基础上添加 app.post(/api/agent/report/generate) async def generate_report(request: ReportRequest): 生成报告双智能体系统 try: return rag_service.generate_report( topicrequest.topic, user_idrequest.user_id, auto_approverequest.auto_approve ) except Exception as e: raise HTTPException(status_code500, detailstr(e)) app.post(/api/agent/report/approve) async def approve_report(request: TaskIdRequest): 批准研究结果生成报告 try: return rag_service.approve_research(request.task_id) except Exception as e: raise HTTPException(status_code500, detailstr(e)) app.post(/api/agent/report/reject) async def reject_report(request: TaskIdRequest): 拒绝研究结果要求重新研究 try: return rag_service.reject_research(request.task_id, request.reason) except Exception as e: raise HTTPException(status_code500, detailstr(e))访问时遇到以下问题下一节处理。
