在实际 AI 应用开发中构建一个能处理复杂、多步骤任务的智能体系统远比实现单一功能调用要困难。开发者常常面临状态管理混乱、流程控制复杂、多智能体协作困难等挑战。LangGraph 作为 LangChain 生态中用于构建有状态、多参与者应用的工作流库提供了一种基于图Graph的清晰范式来编排智能体Agent和工具Tool尤其擅长处理带有循环、分支和状态共享的复杂任务。本文将深入解析 LangGraph 的核心组件并通过一个从零开始的多智能体协作项目手把手展示如何构建一个可运行、可调试的智能体系统。无论你是希望将现有 LangChain 应用升级为更健壮的工作流还是计划从零设计一个多智能体协作架构本文提供的概念、代码和排错路径都将帮助你建立清晰的技术认知。1. 理解 LangGraph 的核心图、状态与节点在开始编码之前必须理解 LangGraph 解决问题的基本模型。它并非要取代 LangChain而是对其在复杂流程编排能力上的重要补充。1.1 为什么需要图Graph来管理智能体传统的链式调用Chain适用于线性任务例如解析用户问题 - 检索知识 - 生成回答。然而现实中的任务往往是非线性的。以一个“研究助手”任务为例用户要求“分析一下 LangGraph 的最新特性并写一份报告”。这个任务可能包含1联网搜索最新信息2总结搜索到的多篇文档3根据总结草拟报告大纲4检查大纲完整性若不完整则返回步骤1补充搜索5撰写完整报告。这个过程存在明显的循环检查-补充和条件分支。如果只用简单的链开发者需要手动维护大量的中间变量和if-else逻辑代码会迅速变得难以维护和调试。LangGraph 将整个工作流抽象为一个有向图图中的节点代表一个执行单元如调用一个 LLM、运行一个工具边代表执行路径。通过明确定义状态State的结构和在节点间如何流转它使得复杂、有状态的工作流变得清晰、可预测。1.2 核心三要素State、Node、EdgeState状态这是 LangGraph 工作流的“记忆体”和“共享白板”。它是一个字典或 Pydantic 模型定义了工作流执行过程中需要传递和修改的所有数据。例如一个研究助手的状态可能包含input用户原始问题、search_results网络搜索结果列表、summary内容摘要、report_outline报告大纲、report最终报告。每个节点都可以读取和修改状态中的特定字段。Node节点节点是一个可调用对象函数它接收当前整个 State 作为输入执行特定操作如调用 LLM、运行工具并返回一个包含对 State 更新内容的字典。例如一个search_node函数会读取state[‘input’]调用搜索引擎工具然后将结果写入state[‘search_results’]。Edge边边定义了控制流即决定执行完一个节点后下一步该去哪个节点。边分为两种条件边Conditional Edge根据 State 中的某个条件例如检查大纲是否完整决定下一个节点。这实现了分支逻辑。普通边Normal Edge无条件地指向下一个节点实现线性执行。一个特殊节点是END表示工作流正常终止。1.3 LangGraph 与 LangChain 的关系这是一个常见的困惑点。你可以这样理解LangChain提供了构建 AI 应用所需的丰富“乐高积木”如各种 LLM 封装、提示模板、文档加载器、检索器、工具和基础的链。LangGraph提供了组装这些“乐高积木”以构建复杂动态流程的“设计图”和“组装说明书”。它尤其擅长处理那些需要循环、多参与者智能体协作的场景。在实践中你通常会同时使用两者用 LangChain 的组件ChatOpenAITool作为节点内的实现用 LangGraph 来编排这些组件的执行顺序和状态流转。2. 环境准备与项目初始化我们将构建一个“多智能体协作写作助手”。这个系统包含两个智能体一个“研究员”负责搜索和总结资料一个“作家”负责根据资料撰写文章。它们通过共享的 State 进行协作。2.1 环境与依赖配置首先确保你的 Python 环境版本在 3.8 以上。建议使用虚拟环境。通过pip安装核心依赖。# 创建并激活虚拟环境可选 python -m venv venv source venv/bin/activate # Linux/Mac # venv\Scripts\activate # Windows # 安装核心库 pip install langgraph langchain-openai langchain-community关键依赖说明langgraph: 核心工作流编排库。langchain-openai: 官方维护的 OpenAI 集成用于调用 GPT 模型。langchain-community: 包含大量社区贡献的第三方集成如网络搜索工具。你还需要准备一个可用的 OpenAI API 密钥并将其设置为环境变量。# Linux/Mac export OPENAI_API_KEYyour-api-key-here # Windows (PowerShell) $env:OPENAI_API_KEYyour-api-key-here2.2 项目结构规划一个清晰的项目结构有助于管理复杂度。建议如下multi_agent_writer/ ├── agents/ │ ├── __init__.py │ ├── researcher.py # 研究员智能体定义 │ └── writer.py # 作家智能体定义 ├── graph/ │ ├── __init__.py │ └── workflow.py # LangGraph 工作流定义 ├── state.py # 共享状态定义 ├── tools.py # 自定义工具定义如搜索 ├── config.py # 配置管理API密钥等 └── main.py # 应用入口我们先从定义共享状态开始。3. 定义工作流状态与智能体节点状态是工作流的基石必须首先明确。3.1 使用 TypedDict 定义 State在state.py中我们使用TypedDict来清晰地定义状态结构。这比普通字典更利于类型检查和代码提示。# state.py from typing import TypedDict, List, Optional class AgentState(TypedDict): 多智能体写作助手的工作流状态。 所有节点都读写此状态的字段。 # 输入 topic: str # 用户指定的写作主题 # 研究员智能体的输出 research_materials: List[str] # 搜索到的原始材料列表 research_summary: str # 研究摘要 # 作家智能体的输出 article_outline: str # 文章大纲 final_article: str # 最终文章 # 控制流标志 needs_more_research: bool # 是否需要进一步研究用于循环控制3.2 构建研究员智能体节点研究员智能体的职责是根据主题进行搜索并生成一份摘要。我们在agents/researcher.py中实现。首先需要定义一个搜索工具。这里我们使用 LangChain 社区的 DuckDuckGo 搜索工具作为示例注意生产环境可能需要更稳定或付费的搜索 API。# tools.py from langchain_community.tools import DuckDuckGoSearchRun # 实例化一个搜索工具 search_tool DuckDuckGoSearchRun()然后实现研究员节点# agents/researcher.py from langchain_openai import ChatOpenAI from langchain.prompts import ChatPromptTemplate from langchain.schema.output_parser import StrOutputParser from .tools import search_tool # 导入搜索工具 from typing import Dict # 初始化 LLM llm ChatOpenAI(modelgpt-4o-mini, temperature0.5) # 使用一个较小、较快的模型进行研究 def research_node(state: Dict) - Dict: 研究员节点执行搜索并总结。 输入state (包含 ‘topic‘) 输出更新后的 state (包含 ‘research_materials‘, ‘research_summary‘) topic state[‘topic‘] # 1. 执行搜索 print(f“[研究员] 正在搜索主题: {topic}“) search_results search_tool.invoke(topic) # 注意实际返回可能是大段文本这里简单处理为列表的一项 materials [search_results] if search_results else [] # 2. 生成研究摘要 prompt_template ChatPromptTemplate.from_messages([ (“system“, “你是一个专业的研究员。请根据提供的网络搜索材料生成一份简洁、关键点突出的摘要。“), (“human“, “主题{topic}\n\n原始材料{materials}\n\n请生成研究摘要“) ]) research_chain prompt_template | llm | StrOutputParser() summary research_chain.invoke({“topic“: topic, “materials“: materials}) print(f“[研究员] 研究摘要生成完成长度{len(summary)} 字符“) # 3. 返回状态更新 return { “research_materials“: materials, “research_summary“: summary, “needs_more_research“: False # 默认一次研究足够后续可由其他节点修改 }3.3 构建作家智能体节点作家智能体的职责是基于研究摘要先撰写大纲再撰写完整文章。我们在agents/writer.py中实现。# agents/writer.py from langchain_openai import ChatOpenAI from langchain.prompts import ChatPromptTemplate from langchain.schema.output_parser import StrOutputParser from typing import Dict # 作家可以使用一个更有创造力的模型 writer_llm ChatOpenAI(model“gpt-4“, temperature0.7) def outline_node(state: Dict) - Dict: 作家节点 - 步骤1撰写大纲。 输入state (包含 ‘topic‘, ‘research_summary‘) 输出更新后的 state (包含 ‘article_outline‘) topic state[‘topic‘] summary state[‘research_summary‘] prompt ChatPromptTemplate.from_messages([ (“system“, “你是一位经验丰富的技术作家。请根据研究摘要为即将撰写的文章创建一份逻辑清晰、结构完整的大纲。“), (“human“, “文章主题{topic}\n研究摘要{summary}\n\n请输出文章大纲“) ]) outline_chain prompt | writer_llm | StrOutputParser() outline outline_chain.invoke({“topic“: topic, “summary“: summary}) print(f“[作家] 文章大纲已创建“) return {“article_outline“: outline} def write_node(state: Dict) - Dict: 作家节点 - 步骤2撰写完整文章。 输入state (包含 ‘topic‘, ‘research_summary‘, ‘article_outline‘) 输出更新后的 state (包含 ‘final_article‘) topic state[‘topic‘] summary state[‘research_summary‘] outline state[‘article_outline‘] prompt ChatPromptTemplate.from_messages([ (“system“, “你是一位优秀的作家。请严格按照提供的大纲并充分参考研究摘要撰写一篇关于给定主题的完整、流畅、信息丰富的文章。文章应面向技术开发者。“), (“human“, “主题{topic}\n研究摘要{summary}\n文章大纲{outline}\n\n请开始撰写文章“) ]) write_chain prompt | writer_llm | StrOutputParser() article write_chain.invoke({“topic“: topic, “summary“: summary, “outline“: outline}) print(f“[作家] 文章撰写完成长度{len(article)} 字符“) return {“final_article“: article}4. 组装 LangGraph 工作流这是最关键的一步我们将把分散的节点和状态组装成一个可执行的工作流图。在graph/workflow.py中操作。4.1 创建图并添加节点# graph/workflow.py from langgraph.graph import StateGraph, END from typing import Dict # 导入状态定义和节点函数 from ..state import AgentState from ..agents.researcher import research_node from ..agents.writer import outline_node, write_node def create_workflow() - StateGraph: 创建并返回多智能体写作工作流图。 # 1. 初始化图并指定状态结构为 AgentState workflow StateGraph(AgentState) # 2. 添加节点 # 节点名是后续连接边时的标识符 workflow.add_node(“researcher“, research_node) workflow.add_node(“outline_writer“, outline_node) workflow.add_node(“article_writer“, write_node) # 3. 设置入口点工作流从 ‘researcher‘ 节点开始 workflow.set_entry_point(“researcher“) # 4. 添加边定义执行顺序 # 研究员完成后进入大纲撰写 workflow.add_edge(“researcher“, “outline_writer“) # 大纲撰写完成后进入文章撰写 workflow.add_edge(“outline_writer“, “article_writer“) # 文章撰写完成后工作流结束 workflow.add_edge(“article_writer“, END) # 5. 编译图得到一个可执行对象 compiled_workflow workflow.compile() return compiled_workflow目前我们构建的是一个简单的线性工作流研究员 - 大纲作家 - 文章作家 - END。这已经是一个可运行的多智能体系统。4.2 引入条件边与循环为了让系统更智能我们可以增加一个“评审”环节让研究员评估作家生成的大纲如果认为资料不足则触发新一轮研究。这需要用到条件边。首先在agents/researcher.py中添加一个评审节点# agents/researcher.py (新增函数) def review_outline_node(state: Dict) - Dict: 评审节点评估大纲是否基于充分的研究。 输入state (包含 ‘research_summary‘, ‘article_outline‘) 输出更新后的 state (主要更新 ‘needs_more_research‘ 标志) summary state[‘research_summary‘] outline state[‘article_outline‘] prompt ChatPromptTemplate.from_messages([ (“system“, “你是一个严格的评审员。请判断当前的文章大纲是否已经充分利用了已有的研究摘要。如果大纲中的关键点在研究摘要中缺乏足够依据则判定为需要更多研究。只回答 ‘是‘ 或 ‘否‘。“), (“human“, “研究摘要{summary}\n\n文章大纲{outline}\n\n是否需要更多研究来支撑大纲是/否“) ]) review_chain prompt | llm | StrOutputParser() decision review_chain.invoke({“summary“: summary, “outline“: outline}) needs_more decision.strip().lower() ‘是‘ print(f“[评审员] 判定 ‘需要更多研究‘: {needs_more}“) return {“needs_more_research“: needs_more}然后修改graph/workflow.py引入条件逻辑# graph/workflow.py (更新版) from langgraph.graph import StateGraph, END from langgraph.graph import START from typing import Dict, Literal # ... 其他导入 ... def create_workflow() - StateGraph: workflow StateGraph(AgentState) # 添加所有节点 workflow.add_node(“researcher“, research_node) workflow.add_node(“outline_writer“, outline_node) workflow.add_node(“reviewer“, review_outline_node) # 新增评审节点 workflow.add_node(“article_writer“, write_node) workflow.set_entry_point(“researcher“) # 修改边连接 # 研究员 - 大纲作家 workflow.add_edge(“researcher“, “outline_writer“) # 大纲作家 - 评审员 workflow.add_edge(“outline_writer“, “reviewer“) # 关键从评审员出发的条件边 def decide_after_review(state: AgentState) - Literal[“more_research“, “write_article“]: 根据评审结果决定下一步 if state.get(“needs_more_research“, False): return “more_research“ # 需要跳回研究员节点 else: return “write_article“ # 可以继续写文章 workflow.add_conditional_edges( “reviewer“, # 源节点 decide_after_review, # 路由函数 { “more_research“: “researcher“, # 如果返回 “more_research“ 则前往 researcher 节点 “write_article“: “article_writer“, # 如果返回 “write_article“ 则前往 article_writer 节点 } ) # 文章作家 - END workflow.add_edge(“article_writer“, END) # 注意当流程跳回 ‘researcher‘ 时它会再次经过 ‘outline_writer‘ - ‘reviewer‘。 # 这形成了一个潜在的循环直到评审通过。 # 为了避免无限循环可以在 research_node 中增加逻辑限制研究次数。 compiled_workflow workflow.compile() return compiled_workflow现在工作流具备了基本的反馈循环能力研究员 - 大纲作家 - 评审员 - (若需要) 研究员 … - 文章作家 - END。5. 运行、验证与结果分析5.1 创建应用入口并运行在main.py中我们初始化工作流并传入初始状态。# main.py from graph.workflow import create_workflow def main(): # 1. 编译工作流 print(“正在编译工作流...“) app create_workflow() # 2. 定义初始状态 initial_state { “topic“: “LangGraph 在多智能体系统中的应用与最佳实践“, “research_materials“: [], “research_summary“: ““, “article_outline“: ““, “final_article“: ““, “needs_more_research“: False, } # 3. 运行工作流 print(f“开始执行工作流主题: {initial_state[‘topic‘]}“) print(“-“ * 50) final_state app.invoke(initial_state) # 4. 输出结果 print(“\n“ ““ * 50) print(“工作流执行完成“) print(““ * 50) print(f“\n最终生成的文章预览前500字符:\n{final_state[‘final_article‘][:500]}...“) print(f“\n文章总长度: {len(final_state[‘final_article‘])} 字符“) # 5. 可选保存结果到文件 with open(‘output_article.md‘, ‘w‘, encoding‘utf-8‘) as f: f.write(f“# {final_state[‘topic‘]}\n\n“) f.write(final_state[‘final_article‘]) print(“\n文章已保存至 ‘output_article.md‘“) if __name__ “__main__“: main()运行python main.py观察控制台输出。你会看到类似下面的日志清晰地展示了智能体间的协作过程正在编译工作流... 开始执行工作流主题: LangGraph 在多智能体系统中的应用与最佳实践 -------------------------------------------------- [研究员] 正在搜索主题: LangGraph 在多智能体系统中的应用与最佳实践 [研究员] 研究摘要生成完成长度1200 字符 [作家] 文章大纲已创建 [评审员] 判定 ‘需要更多研究‘: False [作家] 文章撰写完成长度3200 字符 工作流执行完成 ...5.2 状态流转可视化LangGraph 的一个强大功能是可视化。你可以将编译后的图导出为图片直观理解工作流。# 在 main.py 中添加需要安装 graphviz try: # 显示图结构 from IPython.display import Image, display # 如果你在 Jupyter 环境中 display(Image(app.get_graph().draw_mermaid_png())) except: # 或者保存为文件 app.get_graph().draw_mermaid_png(output_file_path“workflow_graph.png“) print(“工作流图已保存为 ‘workflow_graph.png‘“)生成的图会清晰显示节点、普通边和条件边是理解和调试复杂工作流的利器。6. 常见问题与深度排查在实际开发中你可能会遇到以下典型问题。6.1 状态State相关错误问题现象可能原因检查与解决KeyError提示状态中缺少某个键1. 节点返回的更新字典键名与State定义不匹配。2. 前序节点未生成该字段但后续节点尝试读取。1.检查节点返回值确保每个return的字典键名与TypedDict中的字段名完全一致。2.检查执行顺序确保读取某个字段的节点其前序节点已经正确写入了该字段。使用print(state)在节点开始处打印状态进行调试。状态值被意外覆盖多个节点修改了同一个状态字段且逻辑冲突。1.明确字段职责为每个字段定义清晰的“所有者”节点。例如research_summary只应由research_node写入。2.使用更细粒度状态考虑将状态拆分为多个子状态或使用命名空间。6.2 图编译与执行错误问题现象可能原因检查与解决编译时错误Node ‘xxx‘ already exists重复添加了同名节点。确保add_node时每个节点名称唯一。编译时错误Edge references unknown node边的目标节点名称拼写错误或未添加。仔细检查add_edge和add_conditional_edges中引用的节点名。运行时陷入无限循环条件边逻辑有误导致在几个节点间死循环。1.检查路由函数确保decide_after_review这类函数逻辑正确在某个条件下能跳出循环。2.添加循环计数器在State中增加loop_count字段在节点中递增并在路由函数中判断是否超过最大限制。工作流未按预期路径执行条件边的路由函数返回值与映射字典的键不匹配。确保路由函数返回的字符串如”more_research”与add_conditional_edges中path_map的键完全一致。6.3 工具与模型调用错误问题现象可能原因检查与解决OpenAI API调用失败认证错误或超时1. API 密钥未设置或错误。2. 网络问题。3. 达到速率限制。1.验证环境变量print(os.getenv(‘OPENAI_API_KEY‘))。2.检查网络连接。3.添加重试机制使用tenacity库或 LangChain 内置的Retry回调。工具调用失败如搜索1. 工具初始化错误。2. 工具依赖的第三方服务不可用。1.单独测试工具在节点外写一个简单脚本调用工具确认其本身可用。2.添加降级逻辑在节点内用try-except包裹工具调用失败时返回一个默认值或错误信息到状态中供后续节点判断。6.4 性能与成本优化控制循环次数对于可能循环的路径如研究-评审循环务必在状态中设置max_research_cycles并在路由函数中检查防止因模型判断偏差导致无限循环和 API 费用激增。选择性使用大模型在成本敏感的场景下可以将不同的节点分配给不同能力的模型。例如研究摘要使用gpt-4o-mini而创造性写作使用gpt-4。状态序列化如果状态对象很大例如包含长文本列表频繁在节点间传递会影响性能。考虑只传递必要的引用或 ID。7. 生产环境最佳实践与扩展方向将 LangGraph 工作流用于生产环境需要考虑更多工程化因素。7.1 配置与密钥管理切勿将 API 密钥硬编码在代码中。使用环境变量或专业的配置管理工具如python-dotenv,pydantic-settings。# config.py from pydantic_settings import BaseSettings class Settings(BaseSettings): openai_api_key: str openai_base_url: str | None None # 如需使用代理 model_for_research: str “gpt-4o-mini“ model_for_writing: str “gpt-4“ class Config: env_file “.env“ settings Settings()然后在节点中从settings对象获取配置来初始化 LLM。7.2 持久化与检查点LangGraph 内置了检查点Checkpoint机制可以持久化工作流状态这对于长时间运行或可能中断的任务至关重要。from langgraph.checkpoint import MemorySaver # 在编译图时传入检查点存储器 memory MemorySaver() compiled_workflow workflow.compile(checkpointermemory) # 调用时传入一个 thread_id用于标识会话 config {“configurable“: {“thread_id“: “user_123_session_1“}} final_state compiled_workflow.invoke(initial_state, configconfig) # 后续可以从检查点恢复执行 # final_state compiled_workflow.invoke(new_input, configconfig)生产环境应使用如Redis、PostgreSQL等外部存储的检查点实现。7.3 日志、监控与可观测性结构化日志使用logging模块替代print记录节点开始/结束、状态变化、工具调用结果和耗时。链路追踪集成 OpenTelemetry 等工具追踪一次工作流调用在所有节点和外部服务LLM API、工具 API上的性能数据。状态快照在关键节点后将重要状态字段记录到数据库或监控系统便于事后分析和调试。7.4 扩展方向动态工具调用将 LangChain 的ToolExecutor和AgentExecutor作为节点构建可以自主选择工具的智能体节点。并行执行LangGraph 支持Pregel风格的并发执行。对于相互无依赖的节点可以配置并行分支提升整体效率。人工干预节点在流程中插入一个“人工审核”节点当模型置信度低或遇到敏感内容时暂停工作流等待人工输入后再继续。与 Web 框架集成将编译好的app对象封装成 FastAPI 或 Django 的一个服务端点提供 RESTful API 供前端调用。更复杂的状态管理使用Pydantic模型替代TypedDict获得更强大的数据验证和序列化能力。使用Annotation语法更精细地控制每个节点对状态的读写权限。通过本文的步骤你不仅搭建了一个可运行的多智能体系统更重要的是掌握了 LangGraph 以状态为中心的编排思想。在实际项目中应从最简单的线性流开始逐步引入条件分支和循环并始终关注状态的结构设计和节点的单一职责。当智能体数量和交互逻辑变得复杂时清晰定义的图和状态将成为维护系统可理解性的最重要工具。