LangGraph实战：从链式思维到图思维，构建复杂多Agent工作流

1. 从“链”到“图”为什么我们需要新的协作范式如果你在过去一两年里折腾过基于大语言模型的应用开发那么“LangChain”这个名字你一定不陌生。它通过“链”的概念将一个个独立的LLM调用、工具调用、数据检索等环节串联起来构建出功能强大的应用。这就像是在组装一条流水线每个工位节点按顺序处理任务最终交付成品。在很长一段时间里这种线性的、确定性的“链式思维”是构建AI应用的主流思路。然而当我们试图构建更复杂、更智能的Agent智能体系统时“链”的局限性就暴露无遗了。想象一下你要开发一个能处理复杂客户咨询的客服Agent。它可能需要先理解用户意图然后根据意图决定是查询知识库、调用计算工具还是将问题转交给更专业的子Agent处理。这个决策过程不是线性的它充满了分支、循环和状态依赖。用户可能中途改变问题或者一个子任务的结果会直接影响下一个步骤的走向。用“链”来硬编码这种逻辑很快就会变成一堆难以维护的if-else嵌套代码臃肿且脆弱。这正是“图思维”登场的时刻。与其将流程视为一条有去无回的直线不如将其视为一张由节点和边构成的“图”。节点代表一个个具体的操作单元如调用LLM、执行工具、检查条件边则定义了这些操作单元之间的流转路径和逻辑。这种范式天然适合描述带有分支、循环、并行和状态共享的复杂工作流。而LangGraph正是LangChain团队为了将这种“图思维”落地而专门设计的库。它不是要取代LangChain而是对其核心编排能力的一次重大升级和补充专门用于解决多Agent协作、复杂状态机等高级场景。简单说LangChain帮你造零件和定义简单的组装顺序而LangGraph则为你提供了一张可以动态演进的“电路图”或“交通网”让你能设计出真正具备“智能”行为逻辑的智能体系统。2. LangGraph核心三要素State、Node、Edge要理解并上手LangGraph你必须吃透它的三个核心抽象状态State、节点Node和边Edge。这是构建任何图式工作流的基石。2.1 状态工作流的共享记忆体在链式结构中数据通常从一个节点传递到下一个节点形式可能是一个字典或一个特定的Pydantic对象。在图结构中由于存在分支、合并和循环我们需要一个中心化的、结构化的“记忆体”来记录整个工作流的进展和中间结果。这就是State。在LangGraph中State通常被定义为一个TypedDict或Pydantic模型。它包含了工作流运行过程中所有需要共享和更新的变量。例如对于一个问答Agent其State可能包含from typing import TypedDict, List, Annotated from langgraph.graph import add_messages import operator class AgentState(TypedDict): # 对话消息历史 messages: Annotated[List[str], add_messages] # 用户当前查询 query: str # 从知识库检索到的上下文 context: List[str] # Agent的最终回答 answer: str # 记录调用了哪些工具 used_tools: List[str] # 决定下一步该走哪个分支的标志 next_step: strAnnotated类型提示和add_messages是一个简化状态更新的语法糖它告诉LangGraph如何自动合并列表这里是消息列表。State的设计是第一步也是最关键的一步因为它定义了你的工作流所能“看见”和“操作”的全部世界。2.2 节点具体的工作单元节点是一个可调用的函数它接收当前的State执行一些操作并返回一个更新后的State或包含更新部分的字典。节点可以干任何事情调用LLM、执行Python函数、查询数据库、调用外部API或者仅仅是做逻辑判断。一个典型的节点函数看起来像这样def retrieve_node(state: AgentState) - dict: 检索节点根据用户查询从知识库获取相关信息 query state[“query”] # 模拟检索过程 retrieved_docs knowledge_base.search(query, top_k3) # 返回要更新到State中的部分 return {“context”: retrieved_docs, “next_step”: “generate”}关键点在于节点函数只关心它需要读什么和写什么它不需要知道整个图的全局结构。这种解耦使得每个节点的功能非常纯粹易于单独测试和复用。2.3 边控制流转的逻辑边决定了工作流从一个节点执行完毕后接下来该去往哪个节点。这是图思维最精髓的部分它让工作流“活”了起来。LangGraph提供了几种定义边的方式起始边与结束边start和end是特殊边标识图的入口和出口。条件边根据State中的某个值动态决定下一个节点。这是实现分支的核心。普通边直接指向下一个节点。条件边通过一个路由函数来实现。这个函数读取State返回下一个要执行的节点的名称字符串。def decide_route(state: AgentState) - str: 决策路由根据上下文是否为空决定是生成回答还是告知无答案 if not state[“context”]: # 没有检索到相关信息进入“无答案”处理节点 return “no_answer_node” else: # 有相关信息进入“生成回答”节点 return “generate_answer_node”通过组合节点和边你就能描绘出Agent完整的决策逻辑图。例如“接收用户输入” - “判断意图” - (如果是查询) - “检索知识库” - “生成回答”(如果是闲聊) - “直接调用LLM闲聊” - “结束”。3. 实战构建一个多专家协作的客服Agent理论说得再多不如亲手构建一个。我们来设计一个稍微复杂的场景一个智能客服Agent它内部包含三个“专家”子Agent——产品咨询专家、技术支持专家和投诉处理专家。主Agent调度员需要先理解用户问题然后将其路由给最合适的专家专家处理后再将结果汇总返回。3.1 定义状态与节点首先定义我们的工作流状态from typing import TypedDict, List, Optional from langchain_core.messages import BaseMessage, HumanMessage, AIMessage import operator class CustomerServiceState(TypedDict): # 完整的对话历史 messages: List[BaseMessage] # 用户的最新输入 latest_input: str # 分类结果product, tech, complaint, unknown problem_category: Optional[str] # 被选中的专家Agent名称 selected_expert: Optional[str] # 专家处理后的初步答复 expert_response: Optional[str] # 最终面向用户的答复 final_response: Optional[str]接着我们创建各个节点函数from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate llm ChatOpenAI(model“gpt-4”, temperature0) def classify_problem_node(state: CustomerServiceState) - dict: 分类节点判断用户问题类型 user_input state[“latest_input”] prompt ChatPromptTemplate.from_messages([ (“system”, “你是一个客服问题分类器。请将用户问题严格分类为 ‘product’产品咨询、‘tech’技术问题、‘complaint’投诉或 ‘unknown’无法识别。只输出分类标签。”), (“human”, “{input}”) ]) chain prompt | llm category chain.invoke({“input”: user_input}).content.strip().lower() # 简单清理确保输出是预设类别之一 if category not in [‘product’, ‘tech’, ‘complaint’]: category ‘unknown’ return {“problem_category”: category} def route_to_expert_node(state: CustomerServiceState) - dict: 路由节点根据分类结果选择专家 category state[“problem_category”] routing_map { ‘product’: ‘product_expert_node’, ‘tech’: ‘tech_expert_node’, ‘complaint’: ‘complaint_expert_node’, ‘unknown’: ‘general_agent_node’ } next_node routing_map.get(category, ‘general_agent_node’) return {“selected_expert”: next_node} def product_expert_node(state: CustomerServiceState) - dict: 产品专家节点 # 这里可以集成产品知识库查询等逻辑 response “产品专家关于您咨询的XX产品特性它主要具备A、B、C三大功能非常适合您的使用场景。” return {“expert_response”: response} def tech_expert_node(state: CustomerServiceState) - dict: 技术专家节点 # 这里可以集成故障排查树、技术文档查询等 response “技术专家您遇到的报错问题通常可以通过重启服务、检查配置文件第X行来解决。这是详细步骤...” return {“expert_response”: response} def complaint_expert_node(state: CustomerServiceState) - dict: 投诉处理专家节点 # 这里可以集成工单系统、安抚话术库等 response “投诉专家非常抱歉给您带来了不好的体验。我们已经记录您的问题工单号是XYZ我们将在24小时内由专人跟进处理。” return {“expert_response”: response} def general_agent_node(state: CustomerServiceState) - dict: 通用Agent节点处理未知或简单问题 user_input state[“latest_input”] response f“通用助理我理解您的问题是‘{user_input}’。目前我主要处理产品、技术和投诉问题您可以尝试将问题描述得更具体一些吗” return {“expert_response”: response} def synthesize_response_node(state: CustomerServiceState) - dict: 综合响应节点将专家回复整合成最终答复并更新对话历史 expert_resp state[“expert_response”] final_resp f“客服助理为您服务\n{expert_resp}” # 更新消息历史 old_messages state[“messages”] new_messages old_messages [HumanMessage(contentstate[“latest_input”]), AIMessage(contentfinal_resp)] return {“final_response”: final_resp, “messages”: new_messages}3.2 构图与条件路由现在我们用LangGraph的StateGraph把这些节点和边组装起来。from langgraph.graph import StateGraph, END # 1. 创建图构建器并指定状态结构 workflow StateGraph(CustomerServiceState) # 2. 添加所有节点 workflow.add_node(“classify”, classify_problem_node) workflow.add_node(“router”, route_to_expert_node) # 这个节点只做路由映射不调用专家 workflow.add_node(“product_expert”, product_expert_node) workflow.add_node(“tech_expert”, tech_expert_node) workflow.add_node(“complaint_expert”, complaint_expert_node) workflow.add_node(“general_agent”, general_agent_node) workflow.add_node(“synthesize”, synthesize_response_node) # 3. 设置起始边从分类开始 workflow.set_entry_point(“classify”) # 4. 添加普通边分类完成后进入路由节点 workflow.add_edge(“classify”, “router”) # 5. 添加条件边从路由节点根据 selected_expert 的值动态跳转到不同的专家节点或通用节点 # 这里的关键是route_to_expert_node 返回的 selected_expert 值正好是下一个目标节点的名字。 workflow.add_conditional_edges( “router”, # 源节点 # 这是一个路由函数它从state中取出 selected_expert 字段的值这个值就是下一个节点的名字。 lambda state: state[“selected_expert”], # 可能的目的地节点映射 { “product_expert_node”: “product_expert”, “tech_expert_node”: “tech_expert”, “complaint_expert_node”: “complaint_expert”, “general_agent_node”: “general_agent”, } ) # 6. 添加普通边所有专家节点处理完后都汇聚到综合响应节点 workflow.add_edge(“product_expert”, “synthesize”) workflow.add_edge(“tech_expert”, “synthesize”) workflow.add_edge(“complaint_expert”, “synthesize”) workflow.add_edge(“general_agent”, “synthesize”) # 7. 设置结束边综合响应后工作流结束 workflow.add_edge(“synthesize”, END) # 8. 编译图 app workflow.compile()3.3 运行与可视化现在我们可以运行这个客服Agent了。# 初始化状态 initial_state {“messages”: [], “latest_input”: “你们的产品经常闪退怎么解决”} # 运行图 final_state app.invoke(initial_state) print(final_state[“final_response”]) # 输出可能类似客服助理为您服务技术专家您遇到的报错问题...更强大的是LangGraph内置了可视化功能让你能直观地看到你构建的“思维图”from IPython.display import Image, display try: display(Image(app.get_graph().draw_mermaid_png())) except: # 如果环境不支持可以输出文本表示 print(app.get_graph().draw_ascii())这张图会清晰地展示出从classify到router再根据条件分叉到四个不同的专家节点最后汇聚到synthesize的完整流程。这种可视化对于调试复杂工作流和向团队成员解释逻辑至关重要。4. 高级特性与实战避坑指南当你掌握了基础构建方法后以下这些高级特性和实战中的“坑”能帮助你构建更稳健、更强大的系统。4.1 中断、暂停与长期记忆让Agent拥有“上下文”基础的图是一次性执行的。但对于聊天机器人这样的应用我们需要它记住之前的对话。这就是Checkpointer和Thread的用武之地。Checkpointer是一个持久化存储用于保存和加载某个会话Thread的完整状态。Thread可以理解为一个独立的会话ID。通过它们你可以实现多轮对话每次调用都基于上一次的状态继续。暂停与恢复在某个节点如等待人工审核暂停流程稍后从该点继续。长期记忆将重要的历史信息持久化供后续对话使用。from langgraph.checkpoint.sqlite import SqliteSaver import sqlite3 # 创建一个基于SQLite的检查点存储器 conn sqlite3.connect(“checkpoints.db”) checkpointer SqliteSaver(conn) # 在编译图时传入检查点存储器 app workflow.compile(checkpointercheckpointer) # 使用 thread_id 来区分不同会话 config {“configurable”: {“thread_id”: “user_123_session_1”}} initial_state {“messages”: [], “latest_input”: “你好”} # 第一次调用会创建并保存状态 result1 app.invoke(initial_state, configconfig) print(result1[“messages”][-1].content) # 输出AI的第一次回复 # 第二次调用传入新的用户输入并复用之前的thread_id new_state_input {“latest_input”: “我刚刚问的是什么”} # 注意这里我们不需要再传完整的初始状态检查点会帮我们恢复 result2 app.invoke(new_state_input, configconfig) # 此时result2中的messages会包含完整的对话历史AI能基于上下文回答。注意使用检查点时invoke传入的state参数通常只是本次调用的“增量更新”如最新的用户消息而不是完整状态。完整的State由检查点管理器负责维护和恢复。这是新手最容易混淆的地方之一务必理解“增量更新”的概念。4.2 子图模块化与层次化设计当你的工作流变得非常庞大时把所有节点都放在一个主图里会难以管理。子图允许你将一部分功能例如一个完整的“文档处理流程”封装成一个独立的、内部有自己节点和边的图然后在主图中将其作为一个“超级节点”来引用。这极大地提升了代码的模块化和复用性。from langgraph.graph import StateGraph as SubStateGraph # 1. 定义一个文档处理的子图 doc_subgraph_builder SubStateGraph(AgentState) doc_subgraph_builder.add_node(“extract”, extract_text_node) doc_subgraph_builder.add_node(“summarize”, summarize_node) doc_subgraph_builder.set_entry_point(“extract”) doc_subgraph_builder.add_edge(“extract”, “summarize”) doc_subgraph_builder.add_edge(“summarize”, END) doc_subgraph doc_subgraph_builder.compile() # 2. 在主图中将这个子图作为一个节点添加 main_workflow.add_node(“process_document”, doc_subgraph)4.3 并行执行与异步优化默认情况下LangGraph的节点是顺序执行的。但对于那些彼此没有依赖关系的节点例如同时调用多个不同的API获取信息你可以利用asyncio实现并行大幅减少整体耗时。import asyncio async def parallel_api_call_node(state: AgentState) - dict: 并行调用多个外部API的节点 # 假设有三个不依赖的API任务 task1 asyncio.create_task(call_weather_api(state[“location”])) task2 asyncio.create_task(call_news_api(state[“topic”])) task3 asyncio.create_task(call_stock_api(state[“symbol”])) # 等待所有任务完成 weather, news, stock await asyncio.gather(task1, task2, task3) return {“weather_data”: weather, “news_data”: news, “stock_data”: stock}你需要使用app.ainvoke()来异步调用整个图。在设计节点时确保其中耗时的IO操作都是异步的才能充分发挥并行优势。4.4 常见“坑”与调试技巧状态更新冲突多个节点同时修改State中的同一个字段可能导致意外覆盖。务必清晰定义每个节点的读写域。对于列表类数据的追加善用Annotated和add_messages这类reducer函数来自动处理合并而不是直接赋值。条件边路由函数返回了不存在的节点名这是最常见的运行时错误之一。确保你的路由函数返回的字符串与add_conditional_edges中映射的键以及add_node时注册的节点名完全一致。建议将节点名定义为模块级常量字符串以减少拼写错误。图编译失败检查是否有节点没有连接到图中成了孤岛或者是否存在无法到达END的循环。使用app.get_graph().draw_ascii()可视化是排查此类结构问题的最佳手段。检查点状态混乱当State的结构TypedDict的字段发生变化后之前保存的检查点可能无法正确加载。在生产环境中对State模型的修改需要谨慎可能需要考虑数据迁移或版本化管理。LLM调用不稳定图中大量依赖LLM调用如在条件边中用LLM做路由判断会导致成本高、速度慢且不稳定。尽可能在关键决策点使用LLM在其他地方使用规则或更简单的分类器。为LLM节点设置明确的max_tokens和temperature并实现重试和回退机制。5. LangGraph vs LangChain如何选择与结合使用看到这里你可能会问有了LangGraph我还需要LangChain吗它们到底是什么关系这张表格清晰地说明了它们的定位和适用场景特性维度LangChainLangGraph核心范式链式思维图思维最佳场景线性、确定性的任务流程如RAG问答链检索 - 格式化 - 生成非线性、有状态、多分支的复杂工作流如多Agent协作、决策系统、复杂对话状态管理通常通过链的上下文传递或简单的内存对象。中心化的、强类型的State对象支持复杂更新和持久化检查点。控制流顺序执行可能通过RunnableBranch等实现简单分支。原生支持条件边、循环、并行、子图控制流表达能力极强。抽象层次提供大量现成的“链接”LCEL便于快速组装常见模式。提供底层的图构建原语更灵活但也需要更多代码来定义节点和边。关系基础与组件高级编排框架如何选择如果你的流程是“一条路走到黑”比如标准的文档加载-分割-向量化-检索-生成那么使用LangChain的LCEL来快速组装链是最佳选择代码简洁明了。如果你的流程是“可能走岔路还可能绕回来”比如一个Agent需要根据情况调用不同工具或者多个Agent需要协作并共享中间结果那么LangGraph是你的不二之选。如何结合在实践中它们完全可以协同工作形成一种“LangChain作砖LangGraph为蓝图”的模式使用LangChain来构建图中各个“节点”的内部逻辑。例如一个retrieve_node节点内部可能就是用一个LangChain的RAG链来实现的。使用LangGraph来定义这些“砖块”LangChain链或自定义函数如何连接、在什么条件下跳转、如何共享状态。from langchain.chains import create_retrieval_chain from langchain.chains.combine_documents import create_stuff_documents_chain # ... 其他LangChain导入 # 用LangChain快速构建一个RAG链 retrieval_qa_chain create_retrieval_chain(retriever, create_stuff_documents_chain(llm, prompt)) # 在LangGraph的节点函数中调用这个LangChain链 def rag_node(state: AgentState): question state[“query”] result retrieval_qa_chain.invoke({“input”: question}) # 调用LangChain链 return {“context”: result[“context”], “answer”: result[“answer”]}这种组合让你既能享受LangChain丰富的生态和组件化便利又能利用LangGraph强大的流程编排能力是开发现实世界复杂AI应用的高效路径。