最近在整理团队的技术栈发现一个很有意思的现象很多同事在讨论“AI应用开发”时第一反应就是去查LangChain的文档然后试图用它的各种组件去“拼”出一个解决方案。结果往往是一个简单的需求被拆解成十几个步骤引入了大量抽象层代码变得臃肿调试起来像在走迷宫。更关键的是当需求稍微复杂一点比如需要多个AI模型协作、或者任务执行有前后依赖时传统的LangChain链式结构就显得力不从心代码里充满了各种if-else和状态管理维护成本直线上升。这其实引出了一个更深层的问题我们到底需要一个什么样的框架来构建复杂的AI应用是继续在“链”的思维里打补丁还是换一种更符合AI Agent工作方式的范式这正是LangGraph出现并迅速获得关注的原因。它不是一个替代品而是一次关键的范式升级——从线性的“链”转向了图结构的“工作流”。理解这个转变远比学会调用几个新API重要得多。这篇文章我将结合最新的技术动态和实战经验为你系统梳理从LangChain到LangGraph的演进逻辑并重点拆解如何用LangGraph构建真正可用的多智能体Multi-Agent系统。我们不会停留在概念层面而是深入到设计模式、状态管理、错误处理这些决定项目成败的工程细节中。1. 重新理解LangGraph它解决的远不止是“画个图”很多人第一次接触LangGraph会把它简单理解为“用图的方式画LangChain的流程”。这个理解太浅了也低估了它的价值。LangGraph的核心是引入了一套基于状态State和节点Node的、显式声明式的编程模型来应对AI应用开发中最棘手的问题复杂、有状态、可能循环的工作流。1.1 从“链”到“图”思维模式的根本转变在LangChain的“链”式思维里我们关注的是“下一步调用什么工具或LLM”。数据像流水一样从一个环节流到下一个环节。这种模式对于简单的问答、文本转换、单次检索增强生成RAG非常有效。但是当你的应用需要条件分支根据AI的回复内容决定下一步是查询数据库、调用API还是直接结束。循环迭代一个任务需要反复执行直到满足某个条件比如持续优化一段代码或者多轮对话澄清用户意图。多参与者协作让不同的AI智能体一个负责规划一个负责执行一个负责审核共同完成一个任务它们之间有信息传递和依赖关系。持久化状态在整个工作流执行过程中需要维护和更新一个共享的上下文比如已收集的信息、已执行的操作列表。这时用“链”来硬编码就会非常痛苦。你不得不在代码里手动维护状态变量用复杂的逻辑判断来控制流程走向。LangGraph把这种“控制流”的复杂性通过“图”的结构显式地定义了出来。一个简单的类比LangChain像是给你一套标准的乐高积木LLM调用、工具、记忆、检索器你可以按顺序搭建一条直线轨道。而LangGraph不仅给了你积木还给了你转接头、分岔路口和循环轨道的图纸让你能搭建出更复杂、更智能的铁路网络系统。这个“图纸”就是图结构。1.2 核心概念拆解State, Node, Edge要玩转LangGraph必须吃透三个核心概念它们共同构成了工作流的骨架。1. State状态这是LangGraph工作的基石。State是一个字典或Pydantic模型定义了在整个工作流执行过程中所有节点共享和修改的数据。from typing import TypedDict, Annotated from langgraph.graph import add_messages import operator class State(TypedDict): # 消息历史通常由LangGraph内置的add_messages函数管理 messages: Annotated[list, add_messages] # 自定义状态用户查询 user_query: str # 自定义状态从知识库检索到的上下文 retrieved_context: list[str] # 自定义状态最终答案 final_answer: str # 自定义状态标记工作流是否应该继续 should_continue: boolState的设计是关键。它决定了不同节点之间能交换什么信息。好的State设计应该是“最小化但完整”的只包含工作流必需的数据。2. Node节点节点是实际执行工作的单元。每个节点都是一个函数它接收当前的State执行一些操作调用LLM、使用工具、处理数据然后返回一个更新后的State。def retrieve_node(state: State): 检索节点根据用户查询从向量库获取上下文 query state[“user_query”] # 假设我们有一个检索函数 contexts my_retriever.invoke(query) return {“retrieved_context”: contexts} def generate_node(state: State): 生成节点结合查询和上下文生成最终答案 query state[“user_query”] context state[“retrieved_context”] prompt f“基于以下信息{context}\n\n请回答{query}” response chat_model.invoke(prompt) return {“final_answer”: response.content, “should_continue”: False}节点函数应该保持单一职责。一个节点只做一件事这样易于测试和复用。3. Edge边边决定了工作流的走向。它连接节点并根据条件决定下一个执行哪个节点。这是实现分支和循环的关键。条件边Conditional Edge根据State中的某个值如should_continue或一个函数的返回值决定下一步。普通边无条件地指向下一个节点。LangGraph通过Graph对象来组装这一切from langgraph.graph import StateGraph, END # 1. 创建图并指定State的类型 workflow StateGraph(State) # 2. 添加节点 workflow.add_node(“retrieve”, retrieve_node) workflow.add_node(“generate”, generate_node) # 3. 设置入口点 workflow.set_entry_point(“retrieve”) # 4. 添加边定义流程 workflow.add_edge(“retrieve”, “generate”) # retrieve完成后无条件执行generate workflow.add_edge(“generate”, END) # generate完成后工作流结束 # 5. 编译图 app workflow.compile()编译后的app就是一个可执行的工作流。你可以通过app.invoke({“user_query”: “你的问题”})来运行它。整个流程的走向完全由你定义的图结构控制而不是隐藏在复杂的业务逻辑代码里。2. 构建实战级多智能体系统超越“Hello World”理解了基础我们来看一个更贴近实战的场景构建一个多智能体协作系统。假设我们要开发一个“技术调研助手”它由三个智能体组成规划者Planner分析用户模糊的需求拆解成具体的、可执行的研究子任务。执行者Executor负责执行子任务比如搜索网络使用工具、阅读文档、总结信息。审核者Reviewer评估执行者收集信息的质量和完整性决定是否需要继续深入或返回给规划者调整任务。用传统的链式编码这三个角色的状态同步和流程跳转会是一场噩梦。用LangGraph我们可以清晰地建模。2.1 设计智能体状态与节点首先设计一个足够承载多轮协作的Statefrom typing import TypedDict, Annotated, List, Optional from langgraph.graph import add_messages class ResearchState(TypedDict): # 核心对话历史 messages: Annotated[List, add_messages] # 原始用户需求 original_query: str # 规划者产生的调研计划任务列表 research_plan: List[str] # 当前正在执行的任务索引 current_task_index: int # 执行者收集到的所有信息 gathered_information: List[dict] # 每个元素可能是 {“task”: “xxx”, “content”: “yyy”} # 审核者的结论 review_verdict: Optional[str] # 例如“complete”, “need_more”, “change_plan” # 最终报告 final_report: Optional[str]然后实现三个核心节点函数规划者节点接收原始需求输出一个任务列表。def planner_node(state: ResearchState): from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI llm ChatOpenAI(model“gpt-4”) prompt ChatPromptTemplate.from_messages([ (“system”, “你是一个资深技术调研规划者。请将用户模糊的需求拆解成3-5个具体、可独立搜索和回答的子任务。”), (“human”, “用户需求{query}”) ]) chain prompt | llm # 假设LLM返回一个用换行符分隔的任务列表 plan_text chain.invoke({“query”: state[“original_query”]}).content plan_list [task.strip() for task in plan_text.split(‘\n’) if task.strip()] return { “research_plan”: plan_list, “current_task_index”: 0, “gathered_information”: [] }执行者节点执行当前任务可以使用网络搜索工具。def executor_node(state: ResearchState): from langchain_community.tools import DuckDuckGoSearchRun search DuckDuckGoSearchRun() current_task state[“research_plan”][state[“current_task_index”]] search_result search.invoke(current_task) new_info { “task”: current_task, “content”: search_result } # 更新收集到的信息列表 updated_info state[“gathered_information”] [new_info] return { “gathered_information”: updated_info }审核者节点评估当前信息是否足够生成报告。def reviewer_node(state: ResearchState): llm ChatOpenAI(model“gpt-4”, temperature0) prompt ChatPromptTemplate.from_messages([ (“system”, “你是质量审核员。根据原始需求、已完成的计划、已收集的信息判断\n” “1. ‘complete’信息已足够可以生成最终报告。\n” “2. ‘need_more’信息不足需要继续执行下一个任务。\n” “3. ‘change_plan’当前计划有问题需要重新规划。\n” “只返回上述三个词中的一个。”), (“human”, f“需求{state[‘original_query’]}\n计划{state[‘research_plan’]}\n已收集信息{state[‘gathered_information’]}”) ]) chain prompt | llm verdict chain.invoke({}).content.strip().lower() return {“review_verdict”: verdict}2.2 组装工作流图实现条件逻辑与循环这是LangGraph最精彩的部分。我们需要定义智能体之间的协作规则。from langgraph.graph import StateGraph, END # 创建图 research_graph StateGraph(ResearchState) # 添加节点 research_graph.add_node(“planner”, planner_node) research_graph.add_node(“executor”, executor_node) research_graph.add_node(“reviewer”, reviewer_node) # 设置入口点 research_graph.set_entry_point(“planner”) # 定义边和条件逻辑 # 1. 规划后总是先执行第一个任务 research_graph.add_edge(“planner”, “executor”) # 2. 执行完一个任务后必须经过审核 research_graph.add_edge(“executor”, “reviewer”) # 3. 审核后根据结果决定下一步继续执行、重新规划还是结束 def decide_next_step(state: ResearchState): verdict state.get(“review_verdict”) if verdict “need_more”: # 还有任务吗 next_index state[“current_task_index”] 1 if next_index len(state[“research_plan”]): # 更新索引返回执行者节点 return “executor” else: # 计划已执行完但审核者还觉得不够可能也需要重新规划 return “planner” elif verdict “change_plan”: return “planner” # 回到规划者重新开始 elif verdict “complete”: return “generate_report” # 进入一个我们还没定义的“生成报告”节点 else: return END # 安全兜底 # 添加条件边 research_graph.add_conditional_edges( “reviewer”, decide_next_step, # 这个函数决定下一个节点 { “executor”: “executor”, “planner”: “planner”, “generate_report”: “generate_report”, END: END } ) # 还需要一个节点来更新当前任务索引并在执行前检查 def prepare_executor(state: ResearchState): # 这个节点可以在执行前被调用用于更新状态 # 或者我们可以把索引更新逻辑直接放在decide_next_step返回executor时处理。 # 更清晰的做法是创建一个advance_task节点。 pass # 为了简化我们可以在decide_next_step里直接修改state吗不行函数只应返回下一个节点名。 # 因此我们需要一个额外的节点来更新current_task_index。 # 让我们重构一下增加一个advance_task节点。 research_graph.add_node(“advance_task”, advance_task_node) # 修改边planner - advance_task - executor research_graph.add_edge(“planner”, “advance_task”) research_graph.add_edge(“advance_task”, “executor”) # 同时在decide_next_step返回executor时实际应该先到advance_task # 我们需要调整条件边的映射。这个例子展示了多智能体工作流的核心状态驱动、条件路由、循环反馈。规划、执行、审核形成了一个闭环直到审核者认为信息足够为止。整个流程清晰可见修改协作逻辑只需要调整图的结构和条件函数而不是深入每个智能体的内部代码。3. 与LangChain生态集成RAG、MCP与长期记忆LangGraph不是孤岛它完美继承了LangChain庞大的工具生态。将RAG、MCP等热门技术集成到LangGraph工作流中能极大提升智能体的能力。3.1 集成RAG从静态检索到动态、迭代式检索传统的RAG是“一次检索一次生成”。在智能体工作流中检索可以是动态的、多轮的。规划阶段智能体可能先检索一些背景知识来帮助拆解任务。执行阶段根据具体的子任务去检索不同的专业知识。审核阶段可能为了验证信息的准确性再次进行检索。在LangGraph中你可以为不同的节点配备不同的检索器。例如为“技术调研助手”的executor_node集成一个网络搜索工具如上例同时再增加一个vectorstore_retriever_node专门从内部知识库检索公司技术文档。关键点RAG的retriever和llm都是LangGraph节点的“工具”。节点函数调用这些工具并更新State。这使得RAG从一个独立的链变成了智能体工作流中的一个可插拔、可重复使用的组件。3.2 理解MCPModel Context Protocol扩展智能体的“感官”MCP是一个新兴但非常重要的协议。你可以把它理解为智能体的“外挂感官和手”。它定义了一套标准让任何工具、数据源或服务都能以统一的方式向AI模型或智能体暴露其功能。在LangGraph中集成MCP Server将MCP Server作为工具一个MCP Server可能提供了“查询数据库”、“操作GitHub Issue”、“读取Figma设计稿”等一系列功能。你可以用LangChain的MCPToolkit将这些功能包装成智能体可以调用的工具。在节点中调用在你的executor_node或专门的tool_node中根据State里的任务描述决定调用哪个MCP工具。动态工具调用高级用法是让智能体在运行时根据需求动态发现并调用可用的MCP工具。这需要智能体具备一定的工具描述理解能力。# 示例集成一个假设的“代码仓库分析”MCP工具 from langchain_mcp_adapters.tools import MCPToolkit # 连接到MCP Server toolkit MCPToolkit(server_url“http://localhost:8080”) tools toolkit.get_tools() # 获取该Server提供的所有工具 # 在节点函数中你可以遍历tools选择最合适的进行调用 def code_analysis_node(state: State): task state[“current_task”] # 简单演示假设我们只有一个工具 if “分析代码依赖” in task: result tools[0].invoke({“repo_url”: state[“repo_url”]}) return {“analysis_result”: result}MCP的价值在于标准化和生态。未来可能会有大量现成的MCP Server用于Jira、Slack、内部CRM等你的LangGraph智能体可以轻松集成它们无需为每个服务单独编写适配代码。3.3 实现长期记忆让智能体拥有“上下文”智能体如果每次对话都失忆那就不是真正的智能体。LangGraph通过Checkpointer机制提供了长期记忆支持。对话记忆使用add_messages注解可以自动管理消息历史。自定义状态持久化通过配置Checkpointer你可以将整个State或其中一部分如gathered_information保存到数据库如SQLite、Postgres。流程恢复对于长时间运行的工作流如果中断可以从检查点恢复执行。这意味著你可以构建能处理跨会话复杂任务的智能体比如一个需要几天时间、分多个阶段完成的“项目代码审查助手”。4. 从开发到生产避坑指南与工程化实践能用LangGraph跑通一个Demo和能把它用于生产环境中间隔着巨大的鸿沟。以下是几个关键的工程化考量点。4.1 错误处理与鲁棒性智能体工作流可能在任何节点失败LLM调用超时、工具API异常、网络问题。必须有完善的错误处理。节点级Try-Catch在每个节点函数内部进行异常捕获并更新State来反映错误例如设置一个error字段。图级错误处理LangGraph允许你定义interrupts和triggers但更实用的方式是在条件边函数中检查State中的错误标志并路由到一个专门的error_handler_node。重试机制对于瞬时的失败如网络超时可以在节点内实现简单的重试逻辑。对于更复杂的重试如更换工具、简化查询可能需要设计一个子工作流。4.2 状态管理与数据流设计State的设计是系统的核心契约。糟糕的State设计会导致节点耦合过紧、难以调试。保持State扁平化避免过深的嵌套结构这会让节点函数变得复杂。清晰的字段命名如user_input,agent_scratchpad,final_output。区分“工作区”和“输出区”有些状态是中间过程如draft_response有些是最终结果approved_answer。在设计节点时要明确读写的是哪部分。使用Pydantic模型进行验证用pydantic.BaseModel代替TypedDict来定义State可以利用其强大的数据验证和序列化能力提前发现数据格式错误。4.3 调试、监控与可观测性当工作流变得复杂调试不能只靠print。可视化图结构LangGraph自带可视化功能app.get_graph().draw_mermaid()生成Mermaid图直观看到你的工作流。记录完整执行轨迹利用LangSmithLangChain官方平台或自定义日志记录每个节点的输入State、输出State、耗时和LLM调用详情。这对于理解智能体的决策过程、定位性能瓶颈和错误至关重要。为关键节点添加metadata在节点返回State时可以附带一些元数据如处理数据的版本、使用的模型名称等。4.4 性能与成本优化并发执行如果多个节点间没有依赖关系LangGraph支持并发执行add_edge时指定多个目标。合理设计图结构可以缩短总执行时间。LLM调用优化避免在每个节点都调用LLM。对于一些简单的路由或判断可以尝试用更小的模型或启发式规则。缓存对于昂贵的检索操作或工具调用考虑引入缓存层如langchain.cache。流式输出对于生成最终答案的节点使用流式响应Streaming可以提升用户体验。从LangChain到LangGraph本质是从构建“智能函数”到构建“智能系统”的跃迁。它迫使开发者以更高层次的抽象来思考问题如何设计状态如何划分智能体职责如何定义协作规则这带来的不仅是代码的清晰更是系统设计思维的提升。当你下次再面对一个复杂的AI应用需求时不妨先问自己这是一个线性流程还是一个需要条件判断、循环和多方协作的图如果是后者那么从LangGraph开始设计很可能会让你事半功倍。真正的挑战不在于学会一个新的API而在于掌握这种用“图”来编排智能的新范式。
最近在整理团队的技术栈发现一个很有意思的现象很多同事在讨论“AI应用开发”时第一反应就是去查LangChain的文档然后试图用它的各种组件去“拼”出一个解决方案。结果往往是一个简单的需求被拆解成十几个步骤引入了大量抽象层代码变得臃肿调试起来像在走迷宫。更关键的是当需求稍微复杂一点比如需要多个AI模型协作、或者任务执行有前后依赖时传统的LangChain链式结构就显得力不从心代码里充满了各种if-else和状态管理维护成本直线上升。这其实引出了一个更深层的问题我们到底需要一个什么样的框架来构建复杂的AI应用是继续在“链”的思维里打补丁还是换一种更符合AI Agent工作方式的范式这正是LangGraph出现并迅速获得关注的原因。它不是一个替代品而是一次关键的范式升级——从线性的“链”转向了图结构的“工作流”。理解这个转变远比学会调用几个新API重要得多。这篇文章我将结合最新的技术动态和实战经验为你系统梳理从LangChain到LangGraph的演进逻辑并重点拆解如何用LangGraph构建真正可用的多智能体Multi-Agent系统。我们不会停留在概念层面而是深入到设计模式、状态管理、错误处理这些决定项目成败的工程细节中。1. 重新理解LangGraph它解决的远不止是“画个图”很多人第一次接触LangGraph会把它简单理解为“用图的方式画LangChain的流程”。这个理解太浅了也低估了它的价值。LangGraph的核心是引入了一套基于状态State和节点Node的、显式声明式的编程模型来应对AI应用开发中最棘手的问题复杂、有状态、可能循环的工作流。1.1 从“链”到“图”思维模式的根本转变在LangChain的“链”式思维里我们关注的是“下一步调用什么工具或LLM”。数据像流水一样从一个环节流到下一个环节。这种模式对于简单的问答、文本转换、单次检索增强生成RAG非常有效。但是当你的应用需要条件分支根据AI的回复内容决定下一步是查询数据库、调用API还是直接结束。循环迭代一个任务需要反复执行直到满足某个条件比如持续优化一段代码或者多轮对话澄清用户意图。多参与者协作让不同的AI智能体一个负责规划一个负责执行一个负责审核共同完成一个任务它们之间有信息传递和依赖关系。持久化状态在整个工作流执行过程中需要维护和更新一个共享的上下文比如已收集的信息、已执行的操作列表。这时用“链”来硬编码就会非常痛苦。你不得不在代码里手动维护状态变量用复杂的逻辑判断来控制流程走向。LangGraph把这种“控制流”的复杂性通过“图”的结构显式地定义了出来。一个简单的类比LangChain像是给你一套标准的乐高积木LLM调用、工具、记忆、检索器你可以按顺序搭建一条直线轨道。而LangGraph不仅给了你积木还给了你转接头、分岔路口和循环轨道的图纸让你能搭建出更复杂、更智能的铁路网络系统。这个“图纸”就是图结构。1.2 核心概念拆解State, Node, Edge要玩转LangGraph必须吃透三个核心概念它们共同构成了工作流的骨架。1. State状态这是LangGraph工作的基石。State是一个字典或Pydantic模型定义了在整个工作流执行过程中所有节点共享和修改的数据。from typing import TypedDict, Annotated from langgraph.graph import add_messages import operator class State(TypedDict): # 消息历史通常由LangGraph内置的add_messages函数管理 messages: Annotated[list, add_messages] # 自定义状态用户查询 user_query: str # 自定义状态从知识库检索到的上下文 retrieved_context: list[str] # 自定义状态最终答案 final_answer: str # 自定义状态标记工作流是否应该继续 should_continue: boolState的设计是关键。它决定了不同节点之间能交换什么信息。好的State设计应该是“最小化但完整”的只包含工作流必需的数据。2. Node节点节点是实际执行工作的单元。每个节点都是一个函数它接收当前的State执行一些操作调用LLM、使用工具、处理数据然后返回一个更新后的State。def retrieve_node(state: State): 检索节点根据用户查询从向量库获取上下文 query state[“user_query”] # 假设我们有一个检索函数 contexts my_retriever.invoke(query) return {“retrieved_context”: contexts} def generate_node(state: State): 生成节点结合查询和上下文生成最终答案 query state[“user_query”] context state[“retrieved_context”] prompt f“基于以下信息{context}\n\n请回答{query}” response chat_model.invoke(prompt) return {“final_answer”: response.content, “should_continue”: False}节点函数应该保持单一职责。一个节点只做一件事这样易于测试和复用。3. Edge边边决定了工作流的走向。它连接节点并根据条件决定下一个执行哪个节点。这是实现分支和循环的关键。条件边Conditional Edge根据State中的某个值如should_continue或一个函数的返回值决定下一步。普通边无条件地指向下一个节点。LangGraph通过Graph对象来组装这一切from langgraph.graph import StateGraph, END # 1. 创建图并指定State的类型 workflow StateGraph(State) # 2. 添加节点 workflow.add_node(“retrieve”, retrieve_node) workflow.add_node(“generate”, generate_node) # 3. 设置入口点 workflow.set_entry_point(“retrieve”) # 4. 添加边定义流程 workflow.add_edge(“retrieve”, “generate”) # retrieve完成后无条件执行generate workflow.add_edge(“generate”, END) # generate完成后工作流结束 # 5. 编译图 app workflow.compile()编译后的app就是一个可执行的工作流。你可以通过app.invoke({“user_query”: “你的问题”})来运行它。整个流程的走向完全由你定义的图结构控制而不是隐藏在复杂的业务逻辑代码里。2. 构建实战级多智能体系统超越“Hello World”理解了基础我们来看一个更贴近实战的场景构建一个多智能体协作系统。假设我们要开发一个“技术调研助手”它由三个智能体组成规划者Planner分析用户模糊的需求拆解成具体的、可执行的研究子任务。执行者Executor负责执行子任务比如搜索网络使用工具、阅读文档、总结信息。审核者Reviewer评估执行者收集信息的质量和完整性决定是否需要继续深入或返回给规划者调整任务。用传统的链式编码这三个角色的状态同步和流程跳转会是一场噩梦。用LangGraph我们可以清晰地建模。2.1 设计智能体状态与节点首先设计一个足够承载多轮协作的Statefrom typing import TypedDict, Annotated, List, Optional from langgraph.graph import add_messages class ResearchState(TypedDict): # 核心对话历史 messages: Annotated[List, add_messages] # 原始用户需求 original_query: str # 规划者产生的调研计划任务列表 research_plan: List[str] # 当前正在执行的任务索引 current_task_index: int # 执行者收集到的所有信息 gathered_information: List[dict] # 每个元素可能是 {“task”: “xxx”, “content”: “yyy”} # 审核者的结论 review_verdict: Optional[str] # 例如“complete”, “need_more”, “change_plan” # 最终报告 final_report: Optional[str]然后实现三个核心节点函数规划者节点接收原始需求输出一个任务列表。def planner_node(state: ResearchState): from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI llm ChatOpenAI(model“gpt-4”) prompt ChatPromptTemplate.from_messages([ (“system”, “你是一个资深技术调研规划者。请将用户模糊的需求拆解成3-5个具体、可独立搜索和回答的子任务。”), (“human”, “用户需求{query}”) ]) chain prompt | llm # 假设LLM返回一个用换行符分隔的任务列表 plan_text chain.invoke({“query”: state[“original_query”]}).content plan_list [task.strip() for task in plan_text.split(‘\n’) if task.strip()] return { “research_plan”: plan_list, “current_task_index”: 0, “gathered_information”: [] }执行者节点执行当前任务可以使用网络搜索工具。def executor_node(state: ResearchState): from langchain_community.tools import DuckDuckGoSearchRun search DuckDuckGoSearchRun() current_task state[“research_plan”][state[“current_task_index”]] search_result search.invoke(current_task) new_info { “task”: current_task, “content”: search_result } # 更新收集到的信息列表 updated_info state[“gathered_information”] [new_info] return { “gathered_information”: updated_info }审核者节点评估当前信息是否足够生成报告。def reviewer_node(state: ResearchState): llm ChatOpenAI(model“gpt-4”, temperature0) prompt ChatPromptTemplate.from_messages([ (“system”, “你是质量审核员。根据原始需求、已完成的计划、已收集的信息判断\n” “1. ‘complete’信息已足够可以生成最终报告。\n” “2. ‘need_more’信息不足需要继续执行下一个任务。\n” “3. ‘change_plan’当前计划有问题需要重新规划。\n” “只返回上述三个词中的一个。”), (“human”, f“需求{state[‘original_query’]}\n计划{state[‘research_plan’]}\n已收集信息{state[‘gathered_information’]}”) ]) chain prompt | llm verdict chain.invoke({}).content.strip().lower() return {“review_verdict”: verdict}2.2 组装工作流图实现条件逻辑与循环这是LangGraph最精彩的部分。我们需要定义智能体之间的协作规则。from langgraph.graph import StateGraph, END # 创建图 research_graph StateGraph(ResearchState) # 添加节点 research_graph.add_node(“planner”, planner_node) research_graph.add_node(“executor”, executor_node) research_graph.add_node(“reviewer”, reviewer_node) # 设置入口点 research_graph.set_entry_point(“planner”) # 定义边和条件逻辑 # 1. 规划后总是先执行第一个任务 research_graph.add_edge(“planner”, “executor”) # 2. 执行完一个任务后必须经过审核 research_graph.add_edge(“executor”, “reviewer”) # 3. 审核后根据结果决定下一步继续执行、重新规划还是结束 def decide_next_step(state: ResearchState): verdict state.get(“review_verdict”) if verdict “need_more”: # 还有任务吗 next_index state[“current_task_index”] 1 if next_index len(state[“research_plan”]): # 更新索引返回执行者节点 return “executor” else: # 计划已执行完但审核者还觉得不够可能也需要重新规划 return “planner” elif verdict “change_plan”: return “planner” # 回到规划者重新开始 elif verdict “complete”: return “generate_report” # 进入一个我们还没定义的“生成报告”节点 else: return END # 安全兜底 # 添加条件边 research_graph.add_conditional_edges( “reviewer”, decide_next_step, # 这个函数决定下一个节点 { “executor”: “executor”, “planner”: “planner”, “generate_report”: “generate_report”, END: END } ) # 还需要一个节点来更新当前任务索引并在执行前检查 def prepare_executor(state: ResearchState): # 这个节点可以在执行前被调用用于更新状态 # 或者我们可以把索引更新逻辑直接放在decide_next_step返回executor时处理。 # 更清晰的做法是创建一个advance_task节点。 pass # 为了简化我们可以在decide_next_step里直接修改state吗不行函数只应返回下一个节点名。 # 因此我们需要一个额外的节点来更新current_task_index。 # 让我们重构一下增加一个advance_task节点。 research_graph.add_node(“advance_task”, advance_task_node) # 修改边planner - advance_task - executor research_graph.add_edge(“planner”, “advance_task”) research_graph.add_edge(“advance_task”, “executor”) # 同时在decide_next_step返回executor时实际应该先到advance_task # 我们需要调整条件边的映射。这个例子展示了多智能体工作流的核心状态驱动、条件路由、循环反馈。规划、执行、审核形成了一个闭环直到审核者认为信息足够为止。整个流程清晰可见修改协作逻辑只需要调整图的结构和条件函数而不是深入每个智能体的内部代码。3. 与LangChain生态集成RAG、MCP与长期记忆LangGraph不是孤岛它完美继承了LangChain庞大的工具生态。将RAG、MCP等热门技术集成到LangGraph工作流中能极大提升智能体的能力。3.1 集成RAG从静态检索到动态、迭代式检索传统的RAG是“一次检索一次生成”。在智能体工作流中检索可以是动态的、多轮的。规划阶段智能体可能先检索一些背景知识来帮助拆解任务。执行阶段根据具体的子任务去检索不同的专业知识。审核阶段可能为了验证信息的准确性再次进行检索。在LangGraph中你可以为不同的节点配备不同的检索器。例如为“技术调研助手”的executor_node集成一个网络搜索工具如上例同时再增加一个vectorstore_retriever_node专门从内部知识库检索公司技术文档。关键点RAG的retriever和llm都是LangGraph节点的“工具”。节点函数调用这些工具并更新State。这使得RAG从一个独立的链变成了智能体工作流中的一个可插拔、可重复使用的组件。3.2 理解MCPModel Context Protocol扩展智能体的“感官”MCP是一个新兴但非常重要的协议。你可以把它理解为智能体的“外挂感官和手”。它定义了一套标准让任何工具、数据源或服务都能以统一的方式向AI模型或智能体暴露其功能。在LangGraph中集成MCP Server将MCP Server作为工具一个MCP Server可能提供了“查询数据库”、“操作GitHub Issue”、“读取Figma设计稿”等一系列功能。你可以用LangChain的MCPToolkit将这些功能包装成智能体可以调用的工具。在节点中调用在你的executor_node或专门的tool_node中根据State里的任务描述决定调用哪个MCP工具。动态工具调用高级用法是让智能体在运行时根据需求动态发现并调用可用的MCP工具。这需要智能体具备一定的工具描述理解能力。# 示例集成一个假设的“代码仓库分析”MCP工具 from langchain_mcp_adapters.tools import MCPToolkit # 连接到MCP Server toolkit MCPToolkit(server_url“http://localhost:8080”) tools toolkit.get_tools() # 获取该Server提供的所有工具 # 在节点函数中你可以遍历tools选择最合适的进行调用 def code_analysis_node(state: State): task state[“current_task”] # 简单演示假设我们只有一个工具 if “分析代码依赖” in task: result tools[0].invoke({“repo_url”: state[“repo_url”]}) return {“analysis_result”: result}MCP的价值在于标准化和生态。未来可能会有大量现成的MCP Server用于Jira、Slack、内部CRM等你的LangGraph智能体可以轻松集成它们无需为每个服务单独编写适配代码。3.3 实现长期记忆让智能体拥有“上下文”智能体如果每次对话都失忆那就不是真正的智能体。LangGraph通过Checkpointer机制提供了长期记忆支持。对话记忆使用add_messages注解可以自动管理消息历史。自定义状态持久化通过配置Checkpointer你可以将整个State或其中一部分如gathered_information保存到数据库如SQLite、Postgres。流程恢复对于长时间运行的工作流如果中断可以从检查点恢复执行。这意味著你可以构建能处理跨会话复杂任务的智能体比如一个需要几天时间、分多个阶段完成的“项目代码审查助手”。4. 从开发到生产避坑指南与工程化实践能用LangGraph跑通一个Demo和能把它用于生产环境中间隔着巨大的鸿沟。以下是几个关键的工程化考量点。4.1 错误处理与鲁棒性智能体工作流可能在任何节点失败LLM调用超时、工具API异常、网络问题。必须有完善的错误处理。节点级Try-Catch在每个节点函数内部进行异常捕获并更新State来反映错误例如设置一个error字段。图级错误处理LangGraph允许你定义interrupts和triggers但更实用的方式是在条件边函数中检查State中的错误标志并路由到一个专门的error_handler_node。重试机制对于瞬时的失败如网络超时可以在节点内实现简单的重试逻辑。对于更复杂的重试如更换工具、简化查询可能需要设计一个子工作流。4.2 状态管理与数据流设计State的设计是系统的核心契约。糟糕的State设计会导致节点耦合过紧、难以调试。保持State扁平化避免过深的嵌套结构这会让节点函数变得复杂。清晰的字段命名如user_input,agent_scratchpad,final_output。区分“工作区”和“输出区”有些状态是中间过程如draft_response有些是最终结果approved_answer。在设计节点时要明确读写的是哪部分。使用Pydantic模型进行验证用pydantic.BaseModel代替TypedDict来定义State可以利用其强大的数据验证和序列化能力提前发现数据格式错误。4.3 调试、监控与可观测性当工作流变得复杂调试不能只靠print。可视化图结构LangGraph自带可视化功能app.get_graph().draw_mermaid()生成Mermaid图直观看到你的工作流。记录完整执行轨迹利用LangSmithLangChain官方平台或自定义日志记录每个节点的输入State、输出State、耗时和LLM调用详情。这对于理解智能体的决策过程、定位性能瓶颈和错误至关重要。为关键节点添加metadata在节点返回State时可以附带一些元数据如处理数据的版本、使用的模型名称等。4.4 性能与成本优化并发执行如果多个节点间没有依赖关系LangGraph支持并发执行add_edge时指定多个目标。合理设计图结构可以缩短总执行时间。LLM调用优化避免在每个节点都调用LLM。对于一些简单的路由或判断可以尝试用更小的模型或启发式规则。缓存对于昂贵的检索操作或工具调用考虑引入缓存层如langchain.cache。流式输出对于生成最终答案的节点使用流式响应Streaming可以提升用户体验。从LangChain到LangGraph本质是从构建“智能函数”到构建“智能系统”的跃迁。它迫使开发者以更高层次的抽象来思考问题如何设计状态如何划分智能体职责如何定义协作规则这带来的不仅是代码的清晰更是系统设计思维的提升。当你下次再面对一个复杂的AI应用需求时不妨先问自己这是一个线性流程还是一个需要条件判断、循环和多方协作的图如果是后者那么从LangGraph开始设计很可能会让你事半功倍。真正的挑战不在于学会一个新的API而在于掌握这种用“图”来编排智能的新范式。
的多种实现方案)
