一、什么是多智能体系统多智能体系统Multi-Agent System, MAS是由多个独立的、具有特定角色和能力的智能体组成通过相互通信、协作共同完成一个复杂任务的系统。在 LangGraph 中多智能体的本质就是每个智能体是一个独立的子图主图作为调度器管理智能体之间的交互和任务流转。多智能体 vs 单智能体 vs 子图表格类型核心特点适用场景单智能体一个大脑做所有事全能但不专业简单工具调用、单轮问答子图按功能拆分模块代码复用功能模块化、代码解耦多智能体按角色拆分每个智能体有独立的人格、目标和能力相互协作复杂写作、项目开发、团队协作、多角色决策多智能体的核心优势角色专业化每个智能体只做自己擅长的事输出质量远高于单智能体可扩展性新增角色只需新增一个子图不影响现有系统容错性一个智能体出错其他智能体可以纠正或兜底复杂任务分解天然适配需要多角色分工的复杂任务写报告、做方案、代码开发二、双智能体协作系统架构我们要实现的研究员 作家双智能体系统是最经典、最实用的多智能体架构1. 角色分工表格智能体核心职责输出研究员围绕主题收集资料、整理要点、提供事实依据结构化的研究报告、资料要点作家基于研究员的资料进行内容创作、润色、排版完整的文章、报告、文案2. 标准协作流程用户输入写作任务 → 研究员收集整理资料 → 作家基于资料写初稿 ↓需要补充资料 研究员补充资料 → 作家修改完善 → 最终输出完成3. 核心 State 设计多智能体的 State 必须包含所有智能体共享的全局信息是智能体之间通信的唯一载体from typing import TypedDict, Annotated, Sequence from langchain_core.messages import BaseMessage from langgraph.graph import add_messages class MultiAgentState(TypedDict): # 全局任务信息 task: str # 用户的原始写作任务 topic: str # 写作主题 # 智能体输出 research_notes: str # 研究员的研究笔记 draft: str # 作家的初稿 final_article: str # 最终完成的文章 # 交互控制 current_agent: Literal[researcher, writer, user] # 当前轮到谁工作 turn_count: int # 交互轮次防死循环 max_turns: int # 最大交互轮次 # 对话历史智能体之间的通信记录 messages: Annotated[Sequence[BaseMessage], add_messages] # 错误处理 error: str | None三、完整可运行代码实现基于你之前的代码风格实现了角色明确、交互自然、带错误处理、带持久化的双智能体协作系统完全兼容智谱 API 和国内网络环境。第一步环境准备bash运行pip install -U langgraph langchain langchain-openai pydantic_settings python-dotenv第二步完整代码import os from typing import TypedDict, Annotated, Sequence, Literal from dotenv import load_dotenv from pydantic_settings import BaseSettings, SettingsConfigDict from langchain_core.messages import BaseMessage, HumanMessage, AIMessage from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI from langgraph.graph import add_messages, StateGraph, START, END from langgraph.checkpoint.memory import MemorySaver # 1. 配置管理 load_dotenv() class Settings(BaseSettings): ZHIPU_API_KEY: str ZHIPU_BASE_URL: str LLM_MODEL: str glm-4.6 LLM_BACKUP_MODEL: str glm-4-flash LLM_TIMEOUT: int 30 MAX_TURNS: int 5 # 最大交互轮次防死循环 model_config SettingsConfigDict( env_file.env, env_file_encodingutf-8, extraignore ) settings Settings() # 2. 全局状态定义 class MultiAgentState(TypedDict): task: str topic: str research_notes: str draft: str final_article: str current_agent: Literal[researcher, writer] turn_count: int max_turns: int messages: Annotated[Sequence[BaseMessage], add_messages] error: str | None # 3. LLM初始化 # 通用LLM两个智能体共享 llm ChatOpenAI( api_keysettings.ZHIPU_API_KEY, base_urlsettings.ZHIPU_BASE_URL, model_namesettings.LLM_MODEL, temperature0.7, # 写作需要一点创造性 timeoutsettings.LLM_TIMEOUT ) # 4. 智能体1研究员 def create_researcher_agent(): 创建研究员智能体负责收集资料、整理要点 researcher_prompt ChatPromptTemplate.from_messages([ (system, 你是专业的资深研究员擅长围绕特定主题收集、整理、提炼准确、全面的资料。 你的职责 1. 严格围绕用户的写作主题收集相关的核心概念、发展历程、关键技术、应用场景、优缺点等信息 2. 输出结构化的研究笔记分点清晰逻辑严谨 3. 只提供事实性的资料不要进行创作或润色 4. 如果资料不足明确指出需要补充的内容 5. 研究笔记要足够详细能够支撑作家完成一篇高质量的文章 当前写作任务{task} 写作主题{topic} ), (human, {input}) ]) def researcher_node(state: MultiAgentState): print(\n-- 研究员开始工作 --) try: # 构建提示词 prompt researcher_prompt.format( taskstate[task], topicstate[topic], input请基于当前主题提供详细的研究笔记 ) # 调用LLM生成研究笔记 response llm.invoke(prompt) research_notes response.content print(✅ 研究员完成研究笔记) print(f研究笔记长度{len(research_notes)} 字符) return { research_notes: research_notes, current_agent: writer, turn_count: state[turn_count] 1, messages: [AIMessage(contentf研究员已完成研究笔记\n{research_notes})] } except Exception as e: print(f❌ 研究员工作失败{str(e)}) return { error: str(e), turn_count: state[turn_count] 1 } # 研究员是一个简单的单节点子图 builder StateGraph(MultiAgentState) builder.add_node(researcher, researcher_node) builder.set_entry_point(researcher) return builder.compile() # 5. 智能体2作家 def create_writer_agent(): 创建作家智能体负责基于研究笔记进行内容创作 writer_prompt ChatPromptTemplate.from_messages([ (system, 你是专业的资深作家擅长基于研究员提供的资料创作高质量、结构清晰、语言流畅的文章。 你的职责 1. 严格基于研究员提供的研究笔记进行创作绝对禁止编造内容 2. 文章结构要完整包含引言、正文、结论三个部分 3. 语言要通顺、专业、有逻辑性适合目标读者阅读 4. 如果研究笔记中有不足或缺失的内容明确指出需要研究员补充的信息 5. 输出完整的文章初稿格式规范段落清晰 当前写作任务{task} 写作主题{topic} 研究员提供的研究笔记 {research_notes} ), (human, {input}) ]) def writer_node(state: MultiAgentState): print(\n-- ✍️ 作家开始工作 --) try: # 构建提示词 prompt writer_prompt.format( taskstate[task], topicstate[topic], research_notesstate[research_notes], input请基于研究笔记写一篇完整的文章初稿 ) # 调用LLM生成初稿 response llm.invoke(prompt) draft response.content print(✅ 作家完成文章初稿) print(f初稿长度{len(draft)} 字符) return { draft: draft, current_agent: researcher, # 写完初稿后交给研究员审核 turn_count: state[turn_count] 1, messages: [AIMessage(contentf作家已完成文章初稿\n{draft})] } except Exception as e: print(f❌ 作家工作失败{str(e)}) return { error: str(e), turn_count: state[turn_count] 1 } # 作家也是一个简单的单节点子图 builder StateGraph(MultiAgentState) builder.add_node(writer, writer_node) builder.set_entry_point(writer) return builder.compile() # 6. 主图智能体调度器 def build_multi_agent_system(): 构建双智能体协作系统 # 创建两个智能体子图 researcher_agent create_researcher_agent() writer_agent create_writer_agent() # 主图构建 builder StateGraph(MultiAgentState) # 把子图作为节点加入主图 builder.add_node(researcher, researcher_agent) builder.add_node(writer, writer_agent) # 最终汇总节点生成最终文章 def finalize_node(state: MultiAgentState): print(\n-- 任务完成生成最终文章 --) return { final_article: state[draft], messages: [AIMessage(contentf最终文章\n{state[draft]})] } builder.add_node(finalize, finalize_node) # 条件路由决定下一步由哪个智能体工作还是结束 def router(state: MultiAgentState) - Literal[researcher, writer, finalize, end]: # 优先处理错误 if state.get(error): print(f❌ 系统出错{state[error]}任务结束) return end # 防死循环超过最大轮次强制结束 if state[turn_count] state[max_turns]: print(⚠️ 超过最大交互轮次强制结束任务) return finalize # 根据当前智能体决定下一步 current_agent state[current_agent] if current_agent researcher: # 如果还没有研究笔记研究员先工作 if not state.get(research_notes): return researcher # 如果已经有初稿研究员审核后需要补充资料 elif state.get(draft): # 简单判断如果初稿长度足够直接结束否则让研究员补充资料 if len(state[draft]) 500: return finalize else: print( 初稿内容不足研究员补充资料) return researcher else: return researcher elif current_agent writer: # 有研究笔记作家开始写作 if state.get(research_notes): return writer else: return researcher else: return finalize # 入口开始 → 路由 builder.add_edge(START, researcher) # 条件路由 builder.add_conditional_edges( researcher, router, { writer: writer, finalize: finalize, end: END } ) builder.add_conditional_edges( writer, router, { researcher: researcher, finalize: finalize, end: END } ) # 汇总完成 → 结束 builder.add_edge(finalize, END) # 编译主图带持久化 checkpointer MemorySaver() return builder.compile(checkpointercheckpointer) # 7. 测试运行 if __name__ __main__: print( LangGraph 双智能体协作系统 启动 ) # 初始化多智能体系统 multi_agent build_multi_agent_system() # 会话配置 config { configurable: { thread_id: multi_agent_session_001, user_id: user_001 } } # 测试写作任务 test_task 写一篇关于LangGraph多智能体系统的技术文章 test_topic LangGraph多智能体系统的原理、架构与实现 # 初始化状态 initial_state { task: test_task, topic: test_topic, research_notes: , draft: , final_article: , current_agent: researcher, turn_count: 0, max_turns: settings.MAX_TURNS, messages: [HumanMessage(contenttest_task)], error: None } # 运行多智能体系统 result multi_agent.invoke(initial_state, configconfig) # 输出最终结果 print(\n *80) print( 最终完成的文章) print(*80) print(result[final_article]) print(*80) print(f\n✅ 任务完成共交互 {result[turn_count]} 轮)四、运行效果演示 LangGraph 双智能体协作系统 启动 -- 研究员开始工作 -- ✅ 研究员完成研究笔记 研究笔记长度1200 字符 -- ✍️ 作家开始工作 -- ✅ 作家完成文章初稿 初稿长度850 字符 -- 任务完成生成最终文章 -- 最终完成的文章 # LangGraph多智能体系统的原理、架构与实现 ## 一、什么是LangGraph多智能体系统 LangGraph多智能体系统是基于LangGraph状态机构建的由多个独立的、具有特定角色和能力的智能体组成通过相互通信、协作共同完成复杂任务的系统。 ...文章正文... ## 五、总结 LangGraph多智能体系统通过角色专业化和协作机制大幅提升了复杂任务的处理能力。研究员负责收集资料作家负责内容创作两者分工明确优势互补能够产出比单智能体更高质量的结果。 ✅ 任务完成共交互 3 轮五、核心要点总结多智能体 多个子图 主图调度每个智能体是独立的子图主图负责管理交互流程角色提示词是关键明确每个智能体的职责、输出格式和行为准则是多智能体协作成功的核心状态是通信的唯一载体所有智能体通过共享的 State 传递信息不要在智能体之间直接调用防死循环机制必不可少设置最大交互轮次避免智能体无限对话持久化支持主图绑定 Checkpointer 后支持会话恢复和断点续跑六、进阶扩展方向1. 增加评审员智能体新增第三个智能体「评审员」负责审核文章质量提出修改意见形成「研究员→作家→评审员→作家修改」的完整闭环。2. 给智能体增加工具调用能力给研究员增加网络搜索工具可以实时获取最新资料给作家增加格式转换工具可以输出 Markdown、PDF 等格式的文章3. 更复杂的交互模式循环反馈作家写完初稿后研究员可以提出修改意见作家修改后再提交并行执行多个研究员同时收集不同方面的资料提升效率人工介入在关键步骤暂停等待用户审核或提供反馈后继续执行4. 多智能体消息总线用messages字段作为智能体之间的消息总线每个智能体都可以读取和写入消息实现更自然的对话式协作。这个双智能体系统是多智能体开发的基础你可以基于这个架构轻松扩展出更多角色和更复杂的协作流程。
LangGraph 多智能体基础全解
一、什么是多智能体系统多智能体系统Multi-Agent System, MAS是由多个独立的、具有特定角色和能力的智能体组成通过相互通信、协作共同完成一个复杂任务的系统。在 LangGraph 中多智能体的本质就是每个智能体是一个独立的子图主图作为调度器管理智能体之间的交互和任务流转。多智能体 vs 单智能体 vs 子图表格类型核心特点适用场景单智能体一个大脑做所有事全能但不专业简单工具调用、单轮问答子图按功能拆分模块代码复用功能模块化、代码解耦多智能体按角色拆分每个智能体有独立的人格、目标和能力相互协作复杂写作、项目开发、团队协作、多角色决策多智能体的核心优势角色专业化每个智能体只做自己擅长的事输出质量远高于单智能体可扩展性新增角色只需新增一个子图不影响现有系统容错性一个智能体出错其他智能体可以纠正或兜底复杂任务分解天然适配需要多角色分工的复杂任务写报告、做方案、代码开发二、双智能体协作系统架构我们要实现的研究员 作家双智能体系统是最经典、最实用的多智能体架构1. 角色分工表格智能体核心职责输出研究员围绕主题收集资料、整理要点、提供事实依据结构化的研究报告、资料要点作家基于研究员的资料进行内容创作、润色、排版完整的文章、报告、文案2. 标准协作流程用户输入写作任务 → 研究员收集整理资料 → 作家基于资料写初稿 ↓需要补充资料 研究员补充资料 → 作家修改完善 → 最终输出完成3. 核心 State 设计多智能体的 State 必须包含所有智能体共享的全局信息是智能体之间通信的唯一载体from typing import TypedDict, Annotated, Sequence from langchain_core.messages import BaseMessage from langgraph.graph import add_messages class MultiAgentState(TypedDict): # 全局任务信息 task: str # 用户的原始写作任务 topic: str # 写作主题 # 智能体输出 research_notes: str # 研究员的研究笔记 draft: str # 作家的初稿 final_article: str # 最终完成的文章 # 交互控制 current_agent: Literal[researcher, writer, user] # 当前轮到谁工作 turn_count: int # 交互轮次防死循环 max_turns: int # 最大交互轮次 # 对话历史智能体之间的通信记录 messages: Annotated[Sequence[BaseMessage], add_messages] # 错误处理 error: str | None三、完整可运行代码实现基于你之前的代码风格实现了角色明确、交互自然、带错误处理、带持久化的双智能体协作系统完全兼容智谱 API 和国内网络环境。第一步环境准备bash运行pip install -U langgraph langchain langchain-openai pydantic_settings python-dotenv第二步完整代码import os from typing import TypedDict, Annotated, Sequence, Literal from dotenv import load_dotenv from pydantic_settings import BaseSettings, SettingsConfigDict from langchain_core.messages import BaseMessage, HumanMessage, AIMessage from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI from langgraph.graph import add_messages, StateGraph, START, END from langgraph.checkpoint.memory import MemorySaver # 1. 配置管理 load_dotenv() class Settings(BaseSettings): ZHIPU_API_KEY: str ZHIPU_BASE_URL: str LLM_MODEL: str glm-4.6 LLM_BACKUP_MODEL: str glm-4-flash LLM_TIMEOUT: int 30 MAX_TURNS: int 5 # 最大交互轮次防死循环 model_config SettingsConfigDict( env_file.env, env_file_encodingutf-8, extraignore ) settings Settings() # 2. 全局状态定义 class MultiAgentState(TypedDict): task: str topic: str research_notes: str draft: str final_article: str current_agent: Literal[researcher, writer] turn_count: int max_turns: int messages: Annotated[Sequence[BaseMessage], add_messages] error: str | None # 3. LLM初始化 # 通用LLM两个智能体共享 llm ChatOpenAI( api_keysettings.ZHIPU_API_KEY, base_urlsettings.ZHIPU_BASE_URL, model_namesettings.LLM_MODEL, temperature0.7, # 写作需要一点创造性 timeoutsettings.LLM_TIMEOUT ) # 4. 智能体1研究员 def create_researcher_agent(): 创建研究员智能体负责收集资料、整理要点 researcher_prompt ChatPromptTemplate.from_messages([ (system, 你是专业的资深研究员擅长围绕特定主题收集、整理、提炼准确、全面的资料。 你的职责 1. 严格围绕用户的写作主题收集相关的核心概念、发展历程、关键技术、应用场景、优缺点等信息 2. 输出结构化的研究笔记分点清晰逻辑严谨 3. 只提供事实性的资料不要进行创作或润色 4. 如果资料不足明确指出需要补充的内容 5. 研究笔记要足够详细能够支撑作家完成一篇高质量的文章 当前写作任务{task} 写作主题{topic} ), (human, {input}) ]) def researcher_node(state: MultiAgentState): print(\n-- 研究员开始工作 --) try: # 构建提示词 prompt researcher_prompt.format( taskstate[task], topicstate[topic], input请基于当前主题提供详细的研究笔记 ) # 调用LLM生成研究笔记 response llm.invoke(prompt) research_notes response.content print(✅ 研究员完成研究笔记) print(f研究笔记长度{len(research_notes)} 字符) return { research_notes: research_notes, current_agent: writer, turn_count: state[turn_count] 1, messages: [AIMessage(contentf研究员已完成研究笔记\n{research_notes})] } except Exception as e: print(f❌ 研究员工作失败{str(e)}) return { error: str(e), turn_count: state[turn_count] 1 } # 研究员是一个简单的单节点子图 builder StateGraph(MultiAgentState) builder.add_node(researcher, researcher_node) builder.set_entry_point(researcher) return builder.compile() # 5. 智能体2作家 def create_writer_agent(): 创建作家智能体负责基于研究笔记进行内容创作 writer_prompt ChatPromptTemplate.from_messages([ (system, 你是专业的资深作家擅长基于研究员提供的资料创作高质量、结构清晰、语言流畅的文章。 你的职责 1. 严格基于研究员提供的研究笔记进行创作绝对禁止编造内容 2. 文章结构要完整包含引言、正文、结论三个部分 3. 语言要通顺、专业、有逻辑性适合目标读者阅读 4. 如果研究笔记中有不足或缺失的内容明确指出需要研究员补充的信息 5. 输出完整的文章初稿格式规范段落清晰 当前写作任务{task} 写作主题{topic} 研究员提供的研究笔记 {research_notes} ), (human, {input}) ]) def writer_node(state: MultiAgentState): print(\n-- ✍️ 作家开始工作 --) try: # 构建提示词 prompt writer_prompt.format( taskstate[task], topicstate[topic], research_notesstate[research_notes], input请基于研究笔记写一篇完整的文章初稿 ) # 调用LLM生成初稿 response llm.invoke(prompt) draft response.content print(✅ 作家完成文章初稿) print(f初稿长度{len(draft)} 字符) return { draft: draft, current_agent: researcher, # 写完初稿后交给研究员审核 turn_count: state[turn_count] 1, messages: [AIMessage(contentf作家已完成文章初稿\n{draft})] } except Exception as e: print(f❌ 作家工作失败{str(e)}) return { error: str(e), turn_count: state[turn_count] 1 } # 作家也是一个简单的单节点子图 builder StateGraph(MultiAgentState) builder.add_node(writer, writer_node) builder.set_entry_point(writer) return builder.compile() # 6. 主图智能体调度器 def build_multi_agent_system(): 构建双智能体协作系统 # 创建两个智能体子图 researcher_agent create_researcher_agent() writer_agent create_writer_agent() # 主图构建 builder StateGraph(MultiAgentState) # 把子图作为节点加入主图 builder.add_node(researcher, researcher_agent) builder.add_node(writer, writer_agent) # 最终汇总节点生成最终文章 def finalize_node(state: MultiAgentState): print(\n-- 任务完成生成最终文章 --) return { final_article: state[draft], messages: [AIMessage(contentf最终文章\n{state[draft]})] } builder.add_node(finalize, finalize_node) # 条件路由决定下一步由哪个智能体工作还是结束 def router(state: MultiAgentState) - Literal[researcher, writer, finalize, end]: # 优先处理错误 if state.get(error): print(f❌ 系统出错{state[error]}任务结束) return end # 防死循环超过最大轮次强制结束 if state[turn_count] state[max_turns]: print(⚠️ 超过最大交互轮次强制结束任务) return finalize # 根据当前智能体决定下一步 current_agent state[current_agent] if current_agent researcher: # 如果还没有研究笔记研究员先工作 if not state.get(research_notes): return researcher # 如果已经有初稿研究员审核后需要补充资料 elif state.get(draft): # 简单判断如果初稿长度足够直接结束否则让研究员补充资料 if len(state[draft]) 500: return finalize else: print( 初稿内容不足研究员补充资料) return researcher else: return researcher elif current_agent writer: # 有研究笔记作家开始写作 if state.get(research_notes): return writer else: return researcher else: return finalize # 入口开始 → 路由 builder.add_edge(START, researcher) # 条件路由 builder.add_conditional_edges( researcher, router, { writer: writer, finalize: finalize, end: END } ) builder.add_conditional_edges( writer, router, { researcher: researcher, finalize: finalize, end: END } ) # 汇总完成 → 结束 builder.add_edge(finalize, END) # 编译主图带持久化 checkpointer MemorySaver() return builder.compile(checkpointercheckpointer) # 7. 测试运行 if __name__ __main__: print( LangGraph 双智能体协作系统 启动 ) # 初始化多智能体系统 multi_agent build_multi_agent_system() # 会话配置 config { configurable: { thread_id: multi_agent_session_001, user_id: user_001 } } # 测试写作任务 test_task 写一篇关于LangGraph多智能体系统的技术文章 test_topic LangGraph多智能体系统的原理、架构与实现 # 初始化状态 initial_state { task: test_task, topic: test_topic, research_notes: , draft: , final_article: , current_agent: researcher, turn_count: 0, max_turns: settings.MAX_TURNS, messages: [HumanMessage(contenttest_task)], error: None } # 运行多智能体系统 result multi_agent.invoke(initial_state, configconfig) # 输出最终结果 print(\n *80) print( 最终完成的文章) print(*80) print(result[final_article]) print(*80) print(f\n✅ 任务完成共交互 {result[turn_count]} 轮)四、运行效果演示 LangGraph 双智能体协作系统 启动 -- 研究员开始工作 -- ✅ 研究员完成研究笔记 研究笔记长度1200 字符 -- ✍️ 作家开始工作 -- ✅ 作家完成文章初稿 初稿长度850 字符 -- 任务完成生成最终文章 -- 最终完成的文章 # LangGraph多智能体系统的原理、架构与实现 ## 一、什么是LangGraph多智能体系统 LangGraph多智能体系统是基于LangGraph状态机构建的由多个独立的、具有特定角色和能力的智能体组成通过相互通信、协作共同完成复杂任务的系统。 ...文章正文... ## 五、总结 LangGraph多智能体系统通过角色专业化和协作机制大幅提升了复杂任务的处理能力。研究员负责收集资料作家负责内容创作两者分工明确优势互补能够产出比单智能体更高质量的结果。 ✅ 任务完成共交互 3 轮五、核心要点总结多智能体 多个子图 主图调度每个智能体是独立的子图主图负责管理交互流程角色提示词是关键明确每个智能体的职责、输出格式和行为准则是多智能体协作成功的核心状态是通信的唯一载体所有智能体通过共享的 State 传递信息不要在智能体之间直接调用防死循环机制必不可少设置最大交互轮次避免智能体无限对话持久化支持主图绑定 Checkpointer 后支持会话恢复和断点续跑六、进阶扩展方向1. 增加评审员智能体新增第三个智能体「评审员」负责审核文章质量提出修改意见形成「研究员→作家→评审员→作家修改」的完整闭环。2. 给智能体增加工具调用能力给研究员增加网络搜索工具可以实时获取最新资料给作家增加格式转换工具可以输出 Markdown、PDF 等格式的文章3. 更复杂的交互模式循环反馈作家写完初稿后研究员可以提出修改意见作家修改后再提交并行执行多个研究员同时收集不同方面的资料提升效率人工介入在关键步骤暂停等待用户审核或提供反馈后继续执行4. 多智能体消息总线用messages字段作为智能体之间的消息总线每个智能体都可以读取和写入消息实现更自然的对话式协作。这个双智能体系统是多智能体开发的基础你可以基于这个架构轻松扩展出更多角色和更复杂的协作流程。
