最近在尝试构建一个复杂的AI应用时你是否遇到过这样的困境单个大模型LLM能力有限处理复杂任务时逻辑混乱、容易出错而手动编写多个Agent智能体之间的协作流程又异常繁琐代码耦合度高难以维护和扩展如果你正为此头疼那么LangGraph就是你一直在寻找的解决方案。它不是一个全新的框架而是建立在LangChain之上的一个专门用于构建有状态、多智能体Multi-Agent工作流的库。它用“图”Graph的思维来编排智能体让复杂的多智能体协作变得像搭积木一样清晰、可控。本文将为你带来一份2025年最新、最全面的LangGraph实战教程。我们将从零开始深入剖析其核心架构与组件并通过一个完整的代码实战项目带你彻底掌握如何用它构建一个高效、稳定的多智能体系统。无论你是想入门Agent开发还是希望优化现有的智能体架构这篇文章都能提供一条清晰的路径。1. 背景与核心概念为什么需要LangGraph在深入代码之前我们首先要理解LangGraph要解决的根本问题以及它与LangChain的关系。1.1 多智能体系统Multi-Agent System, MAS的挑战传统的单智能体模型在处理以下场景时显得力不从心复杂决策链比如一个需求从分析、规划、编码到测试的完整软件开发流程。专业化分工需要不同特长的智能体协作如一个负责检索资料Research Agent一个负责撰写报告Writer Agent一个负责审核Review Agent。长期记忆与状态管理任务执行过程中产生的中间状态、历史对话、工具调用结果需要在多个步骤间传递和共享。循环与条件分支根据上一步的结果决定下一步是继续、重试还是转向另一个分支。如果只用基础的LangChain来拼接这些流程代码会迅速变得冗长且难以调试状态管理更是噩梦。1.2 LangGraph是什么LangGraph是 LangChain 生态系统中的一个库它扩展了 LangChain 的Runnable协议核心思想是将智能体和工作流建模为有向图Directed Graph。你可以把整个任务流程想象成一张流程图节点Nodes代表一个执行单元。可以是一个简单的函数、一个LangChain Chain、一个Tool的调用或者一个完整的Agent。边Edges定义了节点之间的流转逻辑。它决定了上一个节点执行完后下一步该去哪个节点。边可以是有条件的Conditional Edge根据当前状态决定下一步。关键特性有状态Stateful整个图共享一个持久化的状态对象所有节点都可以读取和修改它。这是实现多步骤协作的基石。循环Cycles图支持循环这是实现类似“ReAct”模式思考-行动-观察循环或迭代优化流程的关键。人类介入Human-in-the-Loop可以在特定节点暂停等待人工输入或审核再继续执行。持久化Persistence支持将图的执行状态保存到数据库实现长时运行的工作流。1.3 LangGraph vs. LangChain这是一个常见的困惑点简单区分如下LangChain是一个用于构建LLM应用程序的框架。它提供了连接LLM、提示词、记忆、索引和工具的基础模块。它的AgentExecutor也可以运行多步骤任务但编排逻辑相对内聚和固定。LangGraph是LangChain的一个库专注于编排Orchestration。它不提供新的LLM连接器或工具而是利用现有的LangChain组件如LLM、Tools、Chains并以“图”这种更灵活、更强大的方式来编排它们之间的交互。类比如果把构建AI应用比作造车LangChain提供了发动机LLM、轮胎Tools、方向盘Chains等所有零件。而LangGraph则是那套精密的传动和控制系统它决定何时启动发动机、如何分配动力到各个轮胎以完成复杂的驾驶动作。2. 环境准备与版本说明在开始实战前我们需要搭建好开发环境。本文将使用Python进行演示。2.1 基础环境要求操作系统Windows 10/11, macOS, 或 Linux (如 Ubuntu 20.04)。Python版本建议使用 Python 3.10 或 3.11。3.12可能需要关注一些库的兼容性。包管理工具pip或poetry。2.2 安装核心依赖我们主要需要安装langchain,langgraph以及一个LLM的接入库。这里我们使用OpenAI的GPT模型作为示例你也可以替换为其他兼容的模型如通过Ollama本地部署的模型。打开终端创建一个新的虚拟环境并安装依赖# 创建并激活虚拟环境 (可选但推荐) python -m venv langgraph-env source langgraph-env/bin/activate # Linux/macOS # langgraph-env\Scripts\activate # Windows # 安装核心库 pip install langchain langgraph langchain-openai # 如果你需要用到更丰富的工具可以安装社区包 # pip install langchain-community重要版本说明langgraph的API在快速迭代本文代码基于2025年初的稳定版本。如果遇到API不兼容请参考官方文档调整。langchain-openai是LangChain官方维护的OpenAI集成包用于替代旧的openai集成方式。2.3 设置API密钥为了调用OpenAI API你需要设置环境变量。切勿将密钥硬编码在代码中在Linux/macOS的终端中export OPENAI_API_KEY你的-sk-开头的密钥在Windows的CMD中set OPENAI_API_KEY你的-sk-开头的密钥在Python脚本中不推荐用于生产环境import os os.environ[OPENAI_API_KEY] 你的-sk-开头的密钥最佳实践使用.env文件配合python-dotenv库来管理密钥。pip install python-dotenv在项目根目录创建.env文件OPENAI_API_KEY你的-sk-开头的密钥在代码开头加载from dotenv import load_dotenv load_dotenv() # 这会从 .env 文件加载环境变量3. LangGraph核心组件深度剖析理解LangGraph的架构关键在于掌握其四大核心组件状态State、节点Nodes、边Edges和图Graph。3.1 状态State工作流的共享记忆状态是一个类似字典TypedDict的对象它定义了在整个图执行过程中流动和存储的所有数据。它是节点之间通信的唯一媒介。定义状态 通常使用TypedDict来明确状态的结构这有助于类型检查和代码清晰度。from typing import TypedDict, Annotated, List from langgraph.graph.message import add_messages import operator class State(TypedDict): # 消息历史一个由系统、用户、AI消息组成的列表 # Annotated 和 add_messages 是LangGraph提供的语法糖用于自动合并消息避免重复。 messages: Annotated[List, add_messages] # 其他自定义状态字段 query: str research_findings: str final_answer: str # 用于控制流程的字段 needs_human_review: boolmessages这是LangGraph为基于聊天的Agent设计的一个特殊字段。Annotated[List, add_messages]表示这个列表会被add_messages函数自动处理新的消息会被追加到列表末尾而不是覆盖。你可以定义任何需要的字段如query用户问题、research_findings研究结果、final_answer最终答案等。状态字段也可以是控制流标志如needs_human_review。3.2 节点Nodes任务的执行单元节点是一个普通的Python函数或可调用对象它接收当前的状态State作为输入返回一个对该状态的更新字典。关键规则节点函数返回的字典中的键必须对应状态中定义的字段。返回的值会用于更新状态中的对应字段对于messages是追加对于其他字段通常是替换。def research_node(state: State) - dict: 研究节点根据查询获取信息 query state[“query”] # 模拟一个研究过程实际中这里可能调用搜索API、数据库查询等 findings f”根据对‘{query}’的研究发现相关知识点A、B、C。” # 返回要更新到状态中的字段 return {“research_findings”: findings} def write_node(state: State) - dict: 写作节点根据研究结果撰写报告 findings state[“research_findings”] report f”报告摘要\n基于以下发现{findings}\n总结为...” # 同时更新最终答案和消息历史模拟AI回复 return { “final_answer”: report, “messages”: [{“role”: “assistant”, “content”: report}] }3.3 边Edges流程的导航规则边决定了执行完一个节点后接下来应该执行哪个节点。有两种主要类型的边普通边Normal Edge无条件地指向下一个节点。条件边Conditional Edge根据当前状态的某些条件动态决定下一个节点。这是实现分支和循环的核心。条件边通常与一个路由器函数Router Function配合使用。路由器函数检查状态返回下一个要执行的节点的名称字符串。def should_continue(state: State) - str: 路由器函数决定是否循环还是结束 # 检查消息历史中最后一条是否是用户要求继续 last_message state[“messages”][-1] if last_message[“role”] “user” and “继续” in last_message[“content”]: return “research_node” # 返回节点名称继续研究 else: return “__end__” # 特殊关键字表示结束图执行 def human_review_router(state: State) - str: 路由器函数决定是否需要人工审核 if state.get(“needs_human_review”, False): return “human_review_node” else: return “publish_node”3.4 图Graph组装的蓝图图是将节点和边组装起来的容器。创建图的基本步骤如下from langgraph.graph import StateGraph, END # 1. 创建一个图构建器并指定状态模式 workflow StateGraph(State) # 2. 添加节点 workflow.add_node(“research”, research_node) workflow.add_node(“write”, write_node) workflow.add_node(“human_review”, human_review_node) # 3. 设置入口点 workflow.set_entry_point(“research”) # 4. 添加边 workflow.add_edge(“research”, “write”) # 研究完直接写作 workflow.add_conditional_edges( “write”, # 从哪个节点出发 human_review_router, # 路由器函数 { # 路由器返回值到节点名称的映射 “human_review_node”: “human_review”, “publish_node”: END # END是特殊节点表示结束 } ) workflow.add_edge(“human_review”, “publish_node”) # 人工审核后发布 # 5. 编译图得到一个可执行的对象 app workflow.compile()编译后的app就是一个可以运行的工作流。你可以通过app.invoke(initial_state)来启动它。4. 完整实战案例构建一个多智能体协作的“技术博客助手”现在我们将综合运用以上知识构建一个相对复杂的多智能体系统。这个系统包含三个智能体技术研究员Researcher负责搜索和收集给定技术主题的最新资料。内容写手Writer根据研究员提供的资料撰写一篇结构清晰的博客草稿。质量审核员Reviewer对博客草稿进行审核检查技术准确性和可读性并提出修改建议。如果需要重大修改则返回给写手重写。4.1 项目结构与设计我们创建一个简单的项目结构tech_blog_assistant/ ├── agents/ │ ├── __init__.py │ ├── researcher.py │ ├── writer.py │ └── reviewer.py ├── graph/ │ ├── __init__.py │ └── workflow.py ├── state.py ├── tools.py ├── main.py └── .env4.2 定义状态、工具和智能体1. 定义状态 (state.py)# state.py from typing import TypedDict, Annotated, List, Optional from langgraph.graph.message import add_messages class BlogWorkflowState(TypedDict): 博客助手工作流的状态定义 # 核心数据流 topic: str # 用户输入的主题 research_materials: List[str] # 研究员收集的资料列表 blog_draft: str # 写手生成的博客草稿 review_comments: str # 审核员提出的意见 final_output: str # 最终成品 # 控制流与元数据 needs_revision: bool # 审核员标记是否需要重写 revision_count: int # 重写次数防止无限循环 max_revisions: int # 最大重写次数 # 消息历史用于与用户交互或记录内部对话 messages: Annotated[List, add_messages]2. 创建模拟工具 (tools.py)在实际项目中这里会集成真实的搜索引擎API、知识库等。我们先用模拟工具代替。# tools.py from langchain.tools import tool tool def search_web(query: str) - str: 模拟网页搜索工具。输入搜索关键词返回模拟的搜索结果。 # 模拟不同查询返回不同结果 if “langgraph” in query.lower(): return “””搜索结果摘要 1. LangGraph 是LangChain的扩展库用于构建有状态、多智能体工作流。 2. 它使用图Graph来定义节点智能体/函数和边流转逻辑。 3. 核心优势在于处理复杂、多步骤的协作任务。 “”” elif “llm” in query.lower(): return “””搜索结果摘要 1. LLM (Large Language Model) 指大语言模型如GPT-4。 2. 它们通过预测下一个词来生成文本具有强大的理解和生成能力。 3. 应用场景包括聊天、翻译、代码生成等。 “”” else: return f”关于‘{query}’的模拟搜索结果这是一个非常重要的技术主题涉及多个核心概念和最佳实践。”3. 实现研究员智能体 (agents/researcher.py)研究员使用搜索工具来收集信息。# agents/researcher.py from langchain_openai import ChatOpenAI from langchain.agents import create_react_agent, AgentExecutor from langchain import hub from ..tools import search_web # 导入工具 def create_researcher_agent(): 创建并返回一个研究员智能体的执行器 llm ChatOpenAI(model“gpt-4o-mini”, temperature0) # 使用一个较小、较便宜的模型 tools [search_web] # 从LangChain Hub拉取一个适合“计划-执行”的提示词 prompt hub.pull(“hwchase17/react”) # 创建ReAct模式的Agent agent create_react_agent(llm, tools, prompt) # 包装成执行器 agent_executor AgentExecutor( agentagent, toolstools, handle_parsing_errorsTrue, # 优雅处理解析错误 verboseFalse # 设为True可以看到详细思考过程 ) return agent_executor # 研究员节点函数 def research_node(state: BlogWorkflowState) - dict: 研究员节点根据主题收集资料 topic state[“topic”] researcher create_researcher_agent() # 构建查询让Agent去搜索 query f”请搜索并总结关于‘{topic}’的最新技术资料、核心概念和关键应用场景。” try: response researcher.invoke({“input”: query}) findings response[“output”] except Exception as e: findings f”研究员在搜索‘{topic}’时遇到错误{str(e)}。将使用备用知识。” # 更新状态 return { “research_materials”: [findings], # 包装成列表 “messages”: [{“role”: “assistant”, “content”: f”研究员已完成对‘{topic}’的资料收集。”}] }4. 实现写手智能体 (agents/writer.py)写手利用研究员提供的资料来撰写博客。# agents/writer.py from langchain_openai import ChatOpenAI from langchain.prompts import ChatPromptTemplate from langchain.schema import SystemMessage, HumanMessage def create_writer_chain(): 创建一个专门用于写作的LangChain Chain llm ChatOpenAI(model“gpt-4”, temperature0.7) # 写作需要一些创造性 prompt ChatPromptTemplate.from_messages([ SystemMessage(content“你是一位资深技术博客作者擅长将复杂的技术概念转化为通俗易懂、结构清晰的教程文章。你的写作风格严谨且友好。”), HumanMessage.from_template(“”” 请根据以下研究资料撰写一篇关于【{topic}】的技术博客文章。 **研究资料** {materials} **写作要求** 1. 文章需包含引言、核心概念详解、实战代码示例、总结与展望。 2. 代码示例需完整、可运行并附有解释。 3. 语言流畅层次分明。 4. 字数在800-1200字左右。 请直接输出博客正文无需额外说明。 “””) ]) chain prompt | llm return chain def write_node(state: BlogWorkflowState) - dict: 写手节点根据研究资料撰写博客草稿 topic state[“topic”] materials “\n---\n”.join(state[“research_materials”]) writer_chain create_writer_chain() try: draft writer_chain.invoke({“topic”: topic, “materials”: materials}) draft_content draft.content except Exception as e: draft_content f”写手在创作时遇到错误{str(e)}。未能生成草稿。” return { “blog_draft”: draft_content, “messages”: [{“role”: “assistant”, “content”: “写手已完成博客初稿。”}] }5. 实现审核员智能体 (agents/reviewer.py)审核员评估博客草稿的质量。# agents/reviewer.py from langchain_openai import ChatOpenAI from langchain.prompts import ChatPromptTemplate from langchain.schema import SystemMessage, HumanMessage def create_reviewer_chain(): 创建一个专门用于审核的LangChain Chain llm ChatOpenAI(model“gpt-4”, temperature0.2) # 审核需要严谨温度调低 prompt ChatPromptTemplate.from_messages([ SystemMessage(content“你是一位严格的技术内容审核专家。你的任务是评估技术博客草稿的质量并给出具体、可操作的修改意见。”), HumanMessage.from_template(“”” 请审核以下关于【{topic}】的技术博客草稿 **博客草稿** {draft} **请从以下维度进行审核** 1. **技术准确性**内容是否有事实性错误或过时信息 2. **逻辑与结构**文章结构是否清晰逻辑是否连贯 3. **代码质量**示例代码是否正确、完整、有解释 4. **可读性**语言是否通顺是否过于晦涩或啰嗦 **输出格式要求** - 首先给出总体评价通过/需要修改。 - 然后分点列出具体的修改意见。 - 如果问题严重需要大改请在最后一行明确指出‘NEEDS_MAJOR_REVISION’。 - 如果只需小修小补请在最后一行明确指出‘NEEDS_MINOR_REVISION’。 - 如果完全合格请在最后一行明确指出‘APPROVED’。 “””) ]) chain prompt | llm return chain def review_node(state: BlogWorkflowState) - dict: 审核员节点审核博客草稿 topic state[“topic”] draft state[“blog_draft”] reviewer_chain create_reviewer_chain() try: review reviewer_chain.invoke({“topic”: topic, “draft”: draft}) review_comments review.content except Exception as e: review_comments f”审核员在评估时遇到错误{str(e}。审核未完成。” # 解析审核结果决定是否需要重写 needs_revision False if “NEEDS_MAJOR_REVISION” in review_comments or “NEEDS_MINOR_REVISION” in review_comments: needs_revision True return { “review_comments”: review_comments, “needs_revision”: needs_revision, “revision_count”: state.get(“revision_count”, 0) (1 if needs_revision else 0), “messages”: [{“role”: “assistant”, “content”: f”审核员已完成审核。需要修改{needs_revision}”}] }4.3 组装工作流图 (graph/workflow.py)这是LangGraph的核心我们将所有节点和边连接起来。# graph/workflow.py from langgraph.graph import StateGraph, END from ..state import BlogWorkflowState from ..agents.researcher import research_node from ..agents.writer import write_node from ..agents.reviewer import review_node def create_blog_workflow(): 创建并编译博客助手工作流图 # 1. 初始化图构建器 workflow StateGraph(BlogWorkflowState) # 2. 添加三个核心节点 workflow.add_node(“researcher”, research_node) workflow.add_node(“writer”, write_node) workflow.add_node(“reviewer”, review_node) # 3. 设置入口点 workflow.set_entry_point(“researcher”) # 4. 添加主要边研究 - 写作 - 审核 workflow.add_edge(“researcher”, “writer”) workflow.add_edge(“writer”, “reviewer”) # 5. 定义条件路由函数审核后决定下一步 def review_router(state: BlogWorkflowState) - str: 根据审核结果和重试次数决定路由 needs_revision state[“needs_revision”] revision_count state.get(“revision_count”, 0) max_revisions state.get(“max_revisions”, 2) # 默认最多重试2次 if not needs_revision: # 审核通过结束流程 return “__end__” elif revision_count max_revisions: # 超过最大重试次数强制结束可能输出当前版本或报错 print(f”警告已达到最大重写次数{max_revisions}次流程终止。”) return “__end__” else: # 需要修改返回给写手重写 return “writer” # 6. 添加条件边从审核员节点出发根据路由函数决定去向 workflow.add_conditional_edges( “reviewer”, review_router, { “writer”: “writer”, # 返回写手节点 “__end__”: END, # 结束图执行 } ) # 7. 编译图 app workflow.compile() return app4.4 运行与验证 (main.py)最后我们创建一个主程序来运行整个工作流。# main.py import asyncio from dotenv import load_dotenv from graph.workflow import create_blog_workflow # 加载环境变量OPENAI_API_KEY load_dotenv() async def main(): # 1. 创建图应用 app create_blog_workflow() # 2. 定义初始状态 initial_state { “topic”: “LangGraph多智能体系统架构详解”, “research_materials”: [], “blog_draft”: “”, “review_comments”: “”, “final_output”: “”, “needs_revision”: False, “revision_count”: 0, “max_revisions”: 2, # 最多允许重写2次 “messages”: [] # 初始消息为空 } print(“ 开始执行‘技术博客助手’工作流...”) print(f” 主题{initial_state[‘topic’]}”) print(“-” * 50) # 3. 运行图异步方式 try: # LangGraph的invoke是同步的但内部可能调用异步LLM。 # 这里我们直接使用同步调用。 final_state app.invoke(initial_state) # 4. 输出最终结果 print(“\n✅ 工作流执行完成”) print(“-” * 50) print(“ 最终生成的博客内容”) print(“-” * 50) print(final_state.get(“blog_draft”, “未生成内容”)) print(“-” * 50) print(“ 审核意见”) print(final_state.get(“review_comments”, “无审核意见”)) print(“-” * 50) print(f” 重写次数{final_state.get(‘revision_count’, 0)}”) except Exception as e: print(f”❌ 工作流执行失败{e}”) if __name__ “__main__”: # 如果是异步环境可以用 asyncio.run(main()) asyncio.run(main())4.5 运行结果说明在终端运行python main.py你将看到类似以下的输出具体内容因模型和随机性而异 开始执行‘技术博客助手’工作流... 主题LangGraph多智能体系统架构详解 -------------------------------------------------- ✅ 工作流执行完成 -------------------------------------------------- 最终生成的博客内容 -------------------------------------------------- 这里会输出一篇关于LangGraph的结构化技术博客包含引言、架构解析、代码示例等 ... -------------------------------------------------- 审核意见 总体评价需要修改。 修改意见 1. 技术准确性第三部分代码示例中StateGraph的导入路径需要明确写出from langgraph.graph import StateGraph。 2. 逻辑与结构‘核心组件’小节可以再增加一个‘图编译与执行’的子标题使流程更清晰。 3. 代码质量第二个代码块的缩进有误请修正。 4. 可读性第二段开头稍显啰嗦建议精简。 NEEDS_MINOR_REVISION -------------------------------------------------- 重写次数1这个例子展示了工作流如何自动运行研究员搜索 - 写手创作 - 审核员提出修改意见 - 由于是NEEDS_MINOR_REVISION流程根据条件边自动返回到写手节点进行修改 - 修改后再次审核 - 审核通过(APPROVED) - 流程结束。5. 常见问题与排查思路在使用LangGraph构建复杂工作流时你可能会遇到一些典型问题。问题现象常见原因解决思路KeyError当访问状态字段1. 状态TypedDict中未定义该字段。2. 节点返回的更新字典键名拼写错误。1. 检查State类的定义确保字段存在。2. 检查节点函数return的字典键名是否与State中定义的完全一致。图陷入无限循环1. 条件边路由逻辑有误总是返回非结束节点。2. 缺少循环终止条件如最大重试次数。1. 仔细检查路由器函数的逻辑确保存在到达END或”__end__”的路径。2. 在状态中增加计数器如revision_count并在路由器中判断。智能体Agent执行失败或卡住1. LLM API调用失败网络、密钥、额度。2. Agent提示词不佳导致无法解析工具调用。3. 工具函数抛出异常。1. 检查API密钥和环境变量确认网络通畅。2. 在AgentExecutor中设置handle_parsing_errorsTrue和verboseTrue观察详细过程。3. 为工具函数添加完善的错误处理和日志。状态更新不符合预期1. 对messages字段使用了错误的更新方式直接赋值而非追加。2. 多个节点并发修改同一字段产生冲突在复杂图中。1. 对于messages字段务必使用Annotated[List, add_messages]定义并在更新时返回{“messages”: [new_message]}。2. 考虑使用更精细的状态设计或将冲突字段拆分为多个。langgraph或langchain导入错误1. 包未正确安装。2. 版本不兼容。1. 使用pip list检查包版本。确保安装的是langgraph而非langgraph-sdk等。2. 查看官方文档确认使用的API与当前版本匹配。降级或升级到指定版本。工作流执行速度慢1. 串行调用多个耗时节点如多次LLM调用。2. 网络延迟。1. 评估节点间依赖。对于可并行的节点LangGraph支持并行节点add_node后配置但需要仔细设计状态拆分与合并。2. 考虑使用异步版本的LLM客户端和节点函数。6. 最佳实践与工程建议将LangGraph用于实际项目时遵循以下最佳实践可以大幅提升系统的可维护性、健壮性和性能。6.1 状态设计原则最小化与清晰化只将需要在节点间传递的数据放入状态。避免将整个上下文或大型对象塞进去。使用TypedDict强烈推荐使用TypedDict来定义状态这为IDE提供类型提示能提前发现许多字段名错误。区分数据流与控制流像topic、result属于数据流字段像needs_review、step_count属于控制流字段。在设计时明确其用途。6.2 节点设计原则单一职责每个节点只做一件事。例如一个节点负责调用搜索API另一个节点负责解析结果。这提高了可测试性和复用性。纯函数化节点函数应尽可能保持“纯”即输出完全由输入状态决定避免依赖和修改外部全局变量。这使工作流的行为更可预测。完善的错误处理节点内部必须用try...except包裹核心逻辑并返回错误信息到状态中让后续节点或路由函数能处理失败情况而不是让整个图崩溃。6.3 图结构设计先画图再编码在编写代码前用纸笔或绘图工具画出工作流的节点和边。这有助于理清逻辑避免循环错误。合理使用条件边条件边是LangGraph的灵魂但过度使用会使图变得复杂难懂。确保每个条件边的路由函数逻辑简单明了。设置安全阀对于任何可能循环的路径如审核-重写一定要设置最大迭代次数并在状态中体现防止无限循环消耗资源。6.4 生产环境考量持久化Persistence对于长时间运行或需要中断恢复的工作流必须使用LangGraph的持久化功能。它可以将图的状态保存到数据库如SQLite, Postgres之后通过thread_id恢复执行。from langgraph.checkpoint.sqlite import SqliteSaver memory SqliteSaver.from_conn_string(“:memory:”) # 或文件路径 app workflow.compile(checkpointermemory) # 调用时传入config包含thread_id config {“configurable”: {“thread_id”: “user-123-session-1”}} app.invoke(initial_state, configconfig)可视化与监控利用app.get_graph().draw_mermaid()输出Mermaid图表便于文档化和调试。对于生产系统需要在关键节点添加日志记录和指标收集如执行时间、成功率。配置化管理将LLM模型类型、温度、最大重试次数等参数提取为配置项如使用Pydantic Settings避免硬编码。测试策略对每个节点函数进行单元测试。对于整个图可以编写集成测试使用模拟Mock的LLM和工具来验证流程是否正确。6.5 性能优化异步支持如果节点涉及大量I/O操作如网络请求、数据库查询将其定义为async函数并使用ainvoke异步执行图可以显著提升吞吐量。缓存对于昂贵的操作如调用某些收费API考虑在节点中引入缓存机制如functools.lru_cache或外部Redis避免重复计算。LLM调用优化合理选择模型非核心任务使用小型/快速模型如gpt-4o-mini核心任务再用大模型。合并相似的提示词减少不必要的LLM调用次数。通过本教程你不仅学会了LangGraph的核心概念和基本用法还亲手构建了一个具备循环、条件分支和专业化分工的多智能体系统。LangGraph将复杂的协作逻辑可视化、模块化是开发下一代AI应用的有力工具。下一步你可以尝试将其与你的实际业务结合例如构建智能客服流水线、自动化报告生成系统或复杂的决策支持工具。记住从简单的图开始逐步迭代和复杂化是掌握LangGraph的最佳路径。 LangGraph实战：构建多智能体工作流，解决复杂AI任务编排难题
最近在尝试构建一个复杂的AI应用时你是否遇到过这样的困境单个大模型LLM能力有限处理复杂任务时逻辑混乱、容易出错而手动编写多个Agent智能体之间的协作流程又异常繁琐代码耦合度高难以维护和扩展如果你正为此头疼那么LangGraph就是你一直在寻找的解决方案。它不是一个全新的框架而是建立在LangChain之上的一个专门用于构建有状态、多智能体Multi-Agent工作流的库。它用“图”Graph的思维来编排智能体让复杂的多智能体协作变得像搭积木一样清晰、可控。本文将为你带来一份2025年最新、最全面的LangGraph实战教程。我们将从零开始深入剖析其核心架构与组件并通过一个完整的代码实战项目带你彻底掌握如何用它构建一个高效、稳定的多智能体系统。无论你是想入门Agent开发还是希望优化现有的智能体架构这篇文章都能提供一条清晰的路径。1. 背景与核心概念为什么需要LangGraph在深入代码之前我们首先要理解LangGraph要解决的根本问题以及它与LangChain的关系。1.1 多智能体系统Multi-Agent System, MAS的挑战传统的单智能体模型在处理以下场景时显得力不从心复杂决策链比如一个需求从分析、规划、编码到测试的完整软件开发流程。专业化分工需要不同特长的智能体协作如一个负责检索资料Research Agent一个负责撰写报告Writer Agent一个负责审核Review Agent。长期记忆与状态管理任务执行过程中产生的中间状态、历史对话、工具调用结果需要在多个步骤间传递和共享。循环与条件分支根据上一步的结果决定下一步是继续、重试还是转向另一个分支。如果只用基础的LangChain来拼接这些流程代码会迅速变得冗长且难以调试状态管理更是噩梦。1.2 LangGraph是什么LangGraph是 LangChain 生态系统中的一个库它扩展了 LangChain 的Runnable协议核心思想是将智能体和工作流建模为有向图Directed Graph。你可以把整个任务流程想象成一张流程图节点Nodes代表一个执行单元。可以是一个简单的函数、一个LangChain Chain、一个Tool的调用或者一个完整的Agent。边Edges定义了节点之间的流转逻辑。它决定了上一个节点执行完后下一步该去哪个节点。边可以是有条件的Conditional Edge根据当前状态决定下一步。关键特性有状态Stateful整个图共享一个持久化的状态对象所有节点都可以读取和修改它。这是实现多步骤协作的基石。循环Cycles图支持循环这是实现类似“ReAct”模式思考-行动-观察循环或迭代优化流程的关键。人类介入Human-in-the-Loop可以在特定节点暂停等待人工输入或审核再继续执行。持久化Persistence支持将图的执行状态保存到数据库实现长时运行的工作流。1.3 LangGraph vs. LangChain这是一个常见的困惑点简单区分如下LangChain是一个用于构建LLM应用程序的框架。它提供了连接LLM、提示词、记忆、索引和工具的基础模块。它的AgentExecutor也可以运行多步骤任务但编排逻辑相对内聚和固定。LangGraph是LangChain的一个库专注于编排Orchestration。它不提供新的LLM连接器或工具而是利用现有的LangChain组件如LLM、Tools、Chains并以“图”这种更灵活、更强大的方式来编排它们之间的交互。类比如果把构建AI应用比作造车LangChain提供了发动机LLM、轮胎Tools、方向盘Chains等所有零件。而LangGraph则是那套精密的传动和控制系统它决定何时启动发动机、如何分配动力到各个轮胎以完成复杂的驾驶动作。2. 环境准备与版本说明在开始实战前我们需要搭建好开发环境。本文将使用Python进行演示。2.1 基础环境要求操作系统Windows 10/11, macOS, 或 Linux (如 Ubuntu 20.04)。Python版本建议使用 Python 3.10 或 3.11。3.12可能需要关注一些库的兼容性。包管理工具pip或poetry。2.2 安装核心依赖我们主要需要安装langchain,langgraph以及一个LLM的接入库。这里我们使用OpenAI的GPT模型作为示例你也可以替换为其他兼容的模型如通过Ollama本地部署的模型。打开终端创建一个新的虚拟环境并安装依赖# 创建并激活虚拟环境 (可选但推荐) python -m venv langgraph-env source langgraph-env/bin/activate # Linux/macOS # langgraph-env\Scripts\activate # Windows # 安装核心库 pip install langchain langgraph langchain-openai # 如果你需要用到更丰富的工具可以安装社区包 # pip install langchain-community重要版本说明langgraph的API在快速迭代本文代码基于2025年初的稳定版本。如果遇到API不兼容请参考官方文档调整。langchain-openai是LangChain官方维护的OpenAI集成包用于替代旧的openai集成方式。2.3 设置API密钥为了调用OpenAI API你需要设置环境变量。切勿将密钥硬编码在代码中在Linux/macOS的终端中export OPENAI_API_KEY你的-sk-开头的密钥在Windows的CMD中set OPENAI_API_KEY你的-sk-开头的密钥在Python脚本中不推荐用于生产环境import os os.environ[OPENAI_API_KEY] 你的-sk-开头的密钥最佳实践使用.env文件配合python-dotenv库来管理密钥。pip install python-dotenv在项目根目录创建.env文件OPENAI_API_KEY你的-sk-开头的密钥在代码开头加载from dotenv import load_dotenv load_dotenv() # 这会从 .env 文件加载环境变量3. LangGraph核心组件深度剖析理解LangGraph的架构关键在于掌握其四大核心组件状态State、节点Nodes、边Edges和图Graph。3.1 状态State工作流的共享记忆状态是一个类似字典TypedDict的对象它定义了在整个图执行过程中流动和存储的所有数据。它是节点之间通信的唯一媒介。定义状态 通常使用TypedDict来明确状态的结构这有助于类型检查和代码清晰度。from typing import TypedDict, Annotated, List from langgraph.graph.message import add_messages import operator class State(TypedDict): # 消息历史一个由系统、用户、AI消息组成的列表 # Annotated 和 add_messages 是LangGraph提供的语法糖用于自动合并消息避免重复。 messages: Annotated[List, add_messages] # 其他自定义状态字段 query: str research_findings: str final_answer: str # 用于控制流程的字段 needs_human_review: boolmessages这是LangGraph为基于聊天的Agent设计的一个特殊字段。Annotated[List, add_messages]表示这个列表会被add_messages函数自动处理新的消息会被追加到列表末尾而不是覆盖。你可以定义任何需要的字段如query用户问题、research_findings研究结果、final_answer最终答案等。状态字段也可以是控制流标志如needs_human_review。3.2 节点Nodes任务的执行单元节点是一个普通的Python函数或可调用对象它接收当前的状态State作为输入返回一个对该状态的更新字典。关键规则节点函数返回的字典中的键必须对应状态中定义的字段。返回的值会用于更新状态中的对应字段对于messages是追加对于其他字段通常是替换。def research_node(state: State) - dict: 研究节点根据查询获取信息 query state[“query”] # 模拟一个研究过程实际中这里可能调用搜索API、数据库查询等 findings f”根据对‘{query}’的研究发现相关知识点A、B、C。” # 返回要更新到状态中的字段 return {“research_findings”: findings} def write_node(state: State) - dict: 写作节点根据研究结果撰写报告 findings state[“research_findings”] report f”报告摘要\n基于以下发现{findings}\n总结为...” # 同时更新最终答案和消息历史模拟AI回复 return { “final_answer”: report, “messages”: [{“role”: “assistant”, “content”: report}] }3.3 边Edges流程的导航规则边决定了执行完一个节点后接下来应该执行哪个节点。有两种主要类型的边普通边Normal Edge无条件地指向下一个节点。条件边Conditional Edge根据当前状态的某些条件动态决定下一个节点。这是实现分支和循环的核心。条件边通常与一个路由器函数Router Function配合使用。路由器函数检查状态返回下一个要执行的节点的名称字符串。def should_continue(state: State) - str: 路由器函数决定是否循环还是结束 # 检查消息历史中最后一条是否是用户要求继续 last_message state[“messages”][-1] if last_message[“role”] “user” and “继续” in last_message[“content”]: return “research_node” # 返回节点名称继续研究 else: return “__end__” # 特殊关键字表示结束图执行 def human_review_router(state: State) - str: 路由器函数决定是否需要人工审核 if state.get(“needs_human_review”, False): return “human_review_node” else: return “publish_node”3.4 图Graph组装的蓝图图是将节点和边组装起来的容器。创建图的基本步骤如下from langgraph.graph import StateGraph, END # 1. 创建一个图构建器并指定状态模式 workflow StateGraph(State) # 2. 添加节点 workflow.add_node(“research”, research_node) workflow.add_node(“write”, write_node) workflow.add_node(“human_review”, human_review_node) # 3. 设置入口点 workflow.set_entry_point(“research”) # 4. 添加边 workflow.add_edge(“research”, “write”) # 研究完直接写作 workflow.add_conditional_edges( “write”, # 从哪个节点出发 human_review_router, # 路由器函数 { # 路由器返回值到节点名称的映射 “human_review_node”: “human_review”, “publish_node”: END # END是特殊节点表示结束 } ) workflow.add_edge(“human_review”, “publish_node”) # 人工审核后发布 # 5. 编译图得到一个可执行的对象 app workflow.compile()编译后的app就是一个可以运行的工作流。你可以通过app.invoke(initial_state)来启动它。4. 完整实战案例构建一个多智能体协作的“技术博客助手”现在我们将综合运用以上知识构建一个相对复杂的多智能体系统。这个系统包含三个智能体技术研究员Researcher负责搜索和收集给定技术主题的最新资料。内容写手Writer根据研究员提供的资料撰写一篇结构清晰的博客草稿。质量审核员Reviewer对博客草稿进行审核检查技术准确性和可读性并提出修改建议。如果需要重大修改则返回给写手重写。4.1 项目结构与设计我们创建一个简单的项目结构tech_blog_assistant/ ├── agents/ │ ├── __init__.py │ ├── researcher.py │ ├── writer.py │ └── reviewer.py ├── graph/ │ ├── __init__.py │ └── workflow.py ├── state.py ├── tools.py ├── main.py └── .env4.2 定义状态、工具和智能体1. 定义状态 (state.py)# state.py from typing import TypedDict, Annotated, List, Optional from langgraph.graph.message import add_messages class BlogWorkflowState(TypedDict): 博客助手工作流的状态定义 # 核心数据流 topic: str # 用户输入的主题 research_materials: List[str] # 研究员收集的资料列表 blog_draft: str # 写手生成的博客草稿 review_comments: str # 审核员提出的意见 final_output: str # 最终成品 # 控制流与元数据 needs_revision: bool # 审核员标记是否需要重写 revision_count: int # 重写次数防止无限循环 max_revisions: int # 最大重写次数 # 消息历史用于与用户交互或记录内部对话 messages: Annotated[List, add_messages]2. 创建模拟工具 (tools.py)在实际项目中这里会集成真实的搜索引擎API、知识库等。我们先用模拟工具代替。# tools.py from langchain.tools import tool tool def search_web(query: str) - str: 模拟网页搜索工具。输入搜索关键词返回模拟的搜索结果。 # 模拟不同查询返回不同结果 if “langgraph” in query.lower(): return “””搜索结果摘要 1. LangGraph 是LangChain的扩展库用于构建有状态、多智能体工作流。 2. 它使用图Graph来定义节点智能体/函数和边流转逻辑。 3. 核心优势在于处理复杂、多步骤的协作任务。 “”” elif “llm” in query.lower(): return “””搜索结果摘要 1. LLM (Large Language Model) 指大语言模型如GPT-4。 2. 它们通过预测下一个词来生成文本具有强大的理解和生成能力。 3. 应用场景包括聊天、翻译、代码生成等。 “”” else: return f”关于‘{query}’的模拟搜索结果这是一个非常重要的技术主题涉及多个核心概念和最佳实践。”3. 实现研究员智能体 (agents/researcher.py)研究员使用搜索工具来收集信息。# agents/researcher.py from langchain_openai import ChatOpenAI from langchain.agents import create_react_agent, AgentExecutor from langchain import hub from ..tools import search_web # 导入工具 def create_researcher_agent(): 创建并返回一个研究员智能体的执行器 llm ChatOpenAI(model“gpt-4o-mini”, temperature0) # 使用一个较小、较便宜的模型 tools [search_web] # 从LangChain Hub拉取一个适合“计划-执行”的提示词 prompt hub.pull(“hwchase17/react”) # 创建ReAct模式的Agent agent create_react_agent(llm, tools, prompt) # 包装成执行器 agent_executor AgentExecutor( agentagent, toolstools, handle_parsing_errorsTrue, # 优雅处理解析错误 verboseFalse # 设为True可以看到详细思考过程 ) return agent_executor # 研究员节点函数 def research_node(state: BlogWorkflowState) - dict: 研究员节点根据主题收集资料 topic state[“topic”] researcher create_researcher_agent() # 构建查询让Agent去搜索 query f”请搜索并总结关于‘{topic}’的最新技术资料、核心概念和关键应用场景。” try: response researcher.invoke({“input”: query}) findings response[“output”] except Exception as e: findings f”研究员在搜索‘{topic}’时遇到错误{str(e)}。将使用备用知识。” # 更新状态 return { “research_materials”: [findings], # 包装成列表 “messages”: [{“role”: “assistant”, “content”: f”研究员已完成对‘{topic}’的资料收集。”}] }4. 实现写手智能体 (agents/writer.py)写手利用研究员提供的资料来撰写博客。# agents/writer.py from langchain_openai import ChatOpenAI from langchain.prompts import ChatPromptTemplate from langchain.schema import SystemMessage, HumanMessage def create_writer_chain(): 创建一个专门用于写作的LangChain Chain llm ChatOpenAI(model“gpt-4”, temperature0.7) # 写作需要一些创造性 prompt ChatPromptTemplate.from_messages([ SystemMessage(content“你是一位资深技术博客作者擅长将复杂的技术概念转化为通俗易懂、结构清晰的教程文章。你的写作风格严谨且友好。”), HumanMessage.from_template(“”” 请根据以下研究资料撰写一篇关于【{topic}】的技术博客文章。 **研究资料** {materials} **写作要求** 1. 文章需包含引言、核心概念详解、实战代码示例、总结与展望。 2. 代码示例需完整、可运行并附有解释。 3. 语言流畅层次分明。 4. 字数在800-1200字左右。 请直接输出博客正文无需额外说明。 “””) ]) chain prompt | llm return chain def write_node(state: BlogWorkflowState) - dict: 写手节点根据研究资料撰写博客草稿 topic state[“topic”] materials “\n---\n”.join(state[“research_materials”]) writer_chain create_writer_chain() try: draft writer_chain.invoke({“topic”: topic, “materials”: materials}) draft_content draft.content except Exception as e: draft_content f”写手在创作时遇到错误{str(e)}。未能生成草稿。” return { “blog_draft”: draft_content, “messages”: [{“role”: “assistant”, “content”: “写手已完成博客初稿。”}] }5. 实现审核员智能体 (agents/reviewer.py)审核员评估博客草稿的质量。# agents/reviewer.py from langchain_openai import ChatOpenAI from langchain.prompts import ChatPromptTemplate from langchain.schema import SystemMessage, HumanMessage def create_reviewer_chain(): 创建一个专门用于审核的LangChain Chain llm ChatOpenAI(model“gpt-4”, temperature0.2) # 审核需要严谨温度调低 prompt ChatPromptTemplate.from_messages([ SystemMessage(content“你是一位严格的技术内容审核专家。你的任务是评估技术博客草稿的质量并给出具体、可操作的修改意见。”), HumanMessage.from_template(“”” 请审核以下关于【{topic}】的技术博客草稿 **博客草稿** {draft} **请从以下维度进行审核** 1. **技术准确性**内容是否有事实性错误或过时信息 2. **逻辑与结构**文章结构是否清晰逻辑是否连贯 3. **代码质量**示例代码是否正确、完整、有解释 4. **可读性**语言是否通顺是否过于晦涩或啰嗦 **输出格式要求** - 首先给出总体评价通过/需要修改。 - 然后分点列出具体的修改意见。 - 如果问题严重需要大改请在最后一行明确指出‘NEEDS_MAJOR_REVISION’。 - 如果只需小修小补请在最后一行明确指出‘NEEDS_MINOR_REVISION’。 - 如果完全合格请在最后一行明确指出‘APPROVED’。 “””) ]) chain prompt | llm return chain def review_node(state: BlogWorkflowState) - dict: 审核员节点审核博客草稿 topic state[“topic”] draft state[“blog_draft”] reviewer_chain create_reviewer_chain() try: review reviewer_chain.invoke({“topic”: topic, “draft”: draft}) review_comments review.content except Exception as e: review_comments f”审核员在评估时遇到错误{str(e}。审核未完成。” # 解析审核结果决定是否需要重写 needs_revision False if “NEEDS_MAJOR_REVISION” in review_comments or “NEEDS_MINOR_REVISION” in review_comments: needs_revision True return { “review_comments”: review_comments, “needs_revision”: needs_revision, “revision_count”: state.get(“revision_count”, 0) (1 if needs_revision else 0), “messages”: [{“role”: “assistant”, “content”: f”审核员已完成审核。需要修改{needs_revision}”}] }4.3 组装工作流图 (graph/workflow.py)这是LangGraph的核心我们将所有节点和边连接起来。# graph/workflow.py from langgraph.graph import StateGraph, END from ..state import BlogWorkflowState from ..agents.researcher import research_node from ..agents.writer import write_node from ..agents.reviewer import review_node def create_blog_workflow(): 创建并编译博客助手工作流图 # 1. 初始化图构建器 workflow StateGraph(BlogWorkflowState) # 2. 添加三个核心节点 workflow.add_node(“researcher”, research_node) workflow.add_node(“writer”, write_node) workflow.add_node(“reviewer”, review_node) # 3. 设置入口点 workflow.set_entry_point(“researcher”) # 4. 添加主要边研究 - 写作 - 审核 workflow.add_edge(“researcher”, “writer”) workflow.add_edge(“writer”, “reviewer”) # 5. 定义条件路由函数审核后决定下一步 def review_router(state: BlogWorkflowState) - str: 根据审核结果和重试次数决定路由 needs_revision state[“needs_revision”] revision_count state.get(“revision_count”, 0) max_revisions state.get(“max_revisions”, 2) # 默认最多重试2次 if not needs_revision: # 审核通过结束流程 return “__end__” elif revision_count max_revisions: # 超过最大重试次数强制结束可能输出当前版本或报错 print(f”警告已达到最大重写次数{max_revisions}次流程终止。”) return “__end__” else: # 需要修改返回给写手重写 return “writer” # 6. 添加条件边从审核员节点出发根据路由函数决定去向 workflow.add_conditional_edges( “reviewer”, review_router, { “writer”: “writer”, # 返回写手节点 “__end__”: END, # 结束图执行 } ) # 7. 编译图 app workflow.compile() return app4.4 运行与验证 (main.py)最后我们创建一个主程序来运行整个工作流。# main.py import asyncio from dotenv import load_dotenv from graph.workflow import create_blog_workflow # 加载环境变量OPENAI_API_KEY load_dotenv() async def main(): # 1. 创建图应用 app create_blog_workflow() # 2. 定义初始状态 initial_state { “topic”: “LangGraph多智能体系统架构详解”, “research_materials”: [], “blog_draft”: “”, “review_comments”: “”, “final_output”: “”, “needs_revision”: False, “revision_count”: 0, “max_revisions”: 2, # 最多允许重写2次 “messages”: [] # 初始消息为空 } print(“ 开始执行‘技术博客助手’工作流...”) print(f” 主题{initial_state[‘topic’]}”) print(“-” * 50) # 3. 运行图异步方式 try: # LangGraph的invoke是同步的但内部可能调用异步LLM。 # 这里我们直接使用同步调用。 final_state app.invoke(initial_state) # 4. 输出最终结果 print(“\n✅ 工作流执行完成”) print(“-” * 50) print(“ 最终生成的博客内容”) print(“-” * 50) print(final_state.get(“blog_draft”, “未生成内容”)) print(“-” * 50) print(“ 审核意见”) print(final_state.get(“review_comments”, “无审核意见”)) print(“-” * 50) print(f” 重写次数{final_state.get(‘revision_count’, 0)}”) except Exception as e: print(f”❌ 工作流执行失败{e}”) if __name__ “__main__”: # 如果是异步环境可以用 asyncio.run(main()) asyncio.run(main())4.5 运行结果说明在终端运行python main.py你将看到类似以下的输出具体内容因模型和随机性而异 开始执行‘技术博客助手’工作流... 主题LangGraph多智能体系统架构详解 -------------------------------------------------- ✅ 工作流执行完成 -------------------------------------------------- 最终生成的博客内容 -------------------------------------------------- 这里会输出一篇关于LangGraph的结构化技术博客包含引言、架构解析、代码示例等 ... -------------------------------------------------- 审核意见 总体评价需要修改。 修改意见 1. 技术准确性第三部分代码示例中StateGraph的导入路径需要明确写出from langgraph.graph import StateGraph。 2. 逻辑与结构‘核心组件’小节可以再增加一个‘图编译与执行’的子标题使流程更清晰。 3. 代码质量第二个代码块的缩进有误请修正。 4. 可读性第二段开头稍显啰嗦建议精简。 NEEDS_MINOR_REVISION -------------------------------------------------- 重写次数1这个例子展示了工作流如何自动运行研究员搜索 - 写手创作 - 审核员提出修改意见 - 由于是NEEDS_MINOR_REVISION流程根据条件边自动返回到写手节点进行修改 - 修改后再次审核 - 审核通过(APPROVED) - 流程结束。5. 常见问题与排查思路在使用LangGraph构建复杂工作流时你可能会遇到一些典型问题。问题现象常见原因解决思路KeyError当访问状态字段1. 状态TypedDict中未定义该字段。2. 节点返回的更新字典键名拼写错误。1. 检查State类的定义确保字段存在。2. 检查节点函数return的字典键名是否与State中定义的完全一致。图陷入无限循环1. 条件边路由逻辑有误总是返回非结束节点。2. 缺少循环终止条件如最大重试次数。1. 仔细检查路由器函数的逻辑确保存在到达END或”__end__”的路径。2. 在状态中增加计数器如revision_count并在路由器中判断。智能体Agent执行失败或卡住1. LLM API调用失败网络、密钥、额度。2. Agent提示词不佳导致无法解析工具调用。3. 工具函数抛出异常。1. 检查API密钥和环境变量确认网络通畅。2. 在AgentExecutor中设置handle_parsing_errorsTrue和verboseTrue观察详细过程。3. 为工具函数添加完善的错误处理和日志。状态更新不符合预期1. 对messages字段使用了错误的更新方式直接赋值而非追加。2. 多个节点并发修改同一字段产生冲突在复杂图中。1. 对于messages字段务必使用Annotated[List, add_messages]定义并在更新时返回{“messages”: [new_message]}。2. 考虑使用更精细的状态设计或将冲突字段拆分为多个。langgraph或langchain导入错误1. 包未正确安装。2. 版本不兼容。1. 使用pip list检查包版本。确保安装的是langgraph而非langgraph-sdk等。2. 查看官方文档确认使用的API与当前版本匹配。降级或升级到指定版本。工作流执行速度慢1. 串行调用多个耗时节点如多次LLM调用。2. 网络延迟。1. 评估节点间依赖。对于可并行的节点LangGraph支持并行节点add_node后配置但需要仔细设计状态拆分与合并。2. 考虑使用异步版本的LLM客户端和节点函数。6. 最佳实践与工程建议将LangGraph用于实际项目时遵循以下最佳实践可以大幅提升系统的可维护性、健壮性和性能。6.1 状态设计原则最小化与清晰化只将需要在节点间传递的数据放入状态。避免将整个上下文或大型对象塞进去。使用TypedDict强烈推荐使用TypedDict来定义状态这为IDE提供类型提示能提前发现许多字段名错误。区分数据流与控制流像topic、result属于数据流字段像needs_review、step_count属于控制流字段。在设计时明确其用途。6.2 节点设计原则单一职责每个节点只做一件事。例如一个节点负责调用搜索API另一个节点负责解析结果。这提高了可测试性和复用性。纯函数化节点函数应尽可能保持“纯”即输出完全由输入状态决定避免依赖和修改外部全局变量。这使工作流的行为更可预测。完善的错误处理节点内部必须用try...except包裹核心逻辑并返回错误信息到状态中让后续节点或路由函数能处理失败情况而不是让整个图崩溃。6.3 图结构设计先画图再编码在编写代码前用纸笔或绘图工具画出工作流的节点和边。这有助于理清逻辑避免循环错误。合理使用条件边条件边是LangGraph的灵魂但过度使用会使图变得复杂难懂。确保每个条件边的路由函数逻辑简单明了。设置安全阀对于任何可能循环的路径如审核-重写一定要设置最大迭代次数并在状态中体现防止无限循环消耗资源。6.4 生产环境考量持久化Persistence对于长时间运行或需要中断恢复的工作流必须使用LangGraph的持久化功能。它可以将图的状态保存到数据库如SQLite, Postgres之后通过thread_id恢复执行。from langgraph.checkpoint.sqlite import SqliteSaver memory SqliteSaver.from_conn_string(“:memory:”) # 或文件路径 app workflow.compile(checkpointermemory) # 调用时传入config包含thread_id config {“configurable”: {“thread_id”: “user-123-session-1”}} app.invoke(initial_state, configconfig)可视化与监控利用app.get_graph().draw_mermaid()输出Mermaid图表便于文档化和调试。对于生产系统需要在关键节点添加日志记录和指标收集如执行时间、成功率。配置化管理将LLM模型类型、温度、最大重试次数等参数提取为配置项如使用Pydantic Settings避免硬编码。测试策略对每个节点函数进行单元测试。对于整个图可以编写集成测试使用模拟Mock的LLM和工具来验证流程是否正确。6.5 性能优化异步支持如果节点涉及大量I/O操作如网络请求、数据库查询将其定义为async函数并使用ainvoke异步执行图可以显著提升吞吐量。缓存对于昂贵的操作如调用某些收费API考虑在节点中引入缓存机制如functools.lru_cache或外部Redis避免重复计算。LLM调用优化合理选择模型非核心任务使用小型/快速模型如gpt-4o-mini核心任务再用大模型。合并相似的提示词减少不必要的LLM调用次数。通过本教程你不仅学会了LangGraph的核心概念和基本用法还亲手构建了一个具备循环、条件分支和专业化分工的多智能体系统。LangGraph将复杂的协作逻辑可视化、模块化是开发下一代AI应用的有力工具。下一步你可以尝试将其与你的实际业务结合例如构建智能客服流水线、自动化报告生成系统或复杂的决策支持工具。记住从简单的图开始逐步迭代和复杂化是掌握LangGraph的最佳路径。