1. 从LangChain到LangGraph为什么我们需要一个新的“图”如果你在过去一年里接触过大语言模型应用开发那么“LangChain”这个名字对你来说一定不陌生。它像是一套乐高积木把LLM、工具调用、记忆、向量检索这些组件标准化、模块化让我们能快速拼装出一个可用的AI应用。但用久了尤其是在构建稍微复杂一点的、有状态、多步骤的流程时你可能会感到一丝掣肘。比如你想实现一个根据用户问题动态决定调用哪个工具并且工具调用的结果可能影响后续流程走向的客服机器人。用LangChain的Chain和Agent来写代码往往会变得嵌套很深状态管理分散调试起来像在迷宫里找路。这就是LangGraph诞生的背景。它不是要取代LangChain而是作为LangChain生态系统中的一个新成员专门解决复杂、有状态、可能带循环的工作流的编排问题。你可以把它理解为LangChain在“工作流引擎”这个维度上的深度补充和增强。如果说LangChain提供了构建AI应用所需的“砖块”Components那么LangGraph则提供了设计并建造复杂“建筑结构”Stateful Workflows的蓝图和脚手架。它的核心思想非常直观用“图”Graph来建模你的应用逻辑。图中的节点Node代表一个可执行的操作单元比如调用一次LLM、执行一个工具函数、做一个条件判断边Edge则定义了这些操作之间的流转关系。这种建模方式天然适合描述那些非线性的、有分支、有循环的流程。当你面对“根据情况可能走A分支也可能走B分支”、“某个步骤需要循环执行直到满足条件”这类需求时用LangGraph会比用传统的顺序链Sequential Chain清晰得多。从技术栈上看LangGraph与LangChain高度集成共享许多概念如Runnable, Messages等但它引入了几个关键的新抽象StateGraph定义图的结构和状态、Node节点、Edge边以及编译后得到的CompiledStateGraph可执行的工作流对象。而最常用的入口点就是那个stream方法它允许你以流式的方式执行整个图并观察每一步的状态变化这对于调试和理解流程至关重要。2. LangGraph的核心三要素State, Node, Edge要理解LangGraph的执行流程必须先吃透它的三个核心构建块状态State、节点Node和边Edge。这三者共同定义了一个有向图驱动着整个工作流的运转。2.1 状态State工作流的“记忆体”在LangGraph中状态是一个贯穿始终的核心概念。它不是一个简单的变量而是一个类型化的字典TypedDict定义了在整个图执行过程中哪些信息会被传递和修改。你可以把它想象成游戏里角色的属性面板或者一个工单在处理过程中不断被填写的表单。定义一个状态通常使用Pydantic的BaseModel或者Python的TypedDict。例如一个简单的聊天代理的状态可能包含from typing import TypedDict, List, Annotated from langgraph.graph.message import add_messages import operator class AgentState(TypedDict): # 消息历史使用LangGraph提供的注解实现自动累加 messages: Annotated[List[BaseMessage], add_messages] # 用户当前查询 query: str # 代理的思考过程或中间结果 reasoning: str # 决定下一步要做什么 next_step: str这里的关键是Annotated的使用。Annotated[List[BaseMessage], add_messages]是一个“缩减器”Reducer。它告诉LangGraphmessages这个字段在节点间传递时如果后一个节点返回了新的消息列表不要直接覆盖而是通过add_messages函数通常是追加来更新。这是实现对话历史累积的优雅方式。状态中的其他字段如query则默认是“覆盖”模式即新值直接替换旧值。为什么状态设计如此重要因为它直接决定了数据的生命周期和共享范围。一个设计良好的状态结构能让节点之间的协作清晰明了避免出现数据污染或丢失。在规划工作流时我的经验是先别急着写节点逻辑而是花时间想清楚“在整个流程中我需要记录和传递哪些信息”。2.2 节点Node执行具体任务的“工作单元”节点是图中实际干活的部分。在LangGraph中一个节点就是一个可调用对象Callable它接收当前状态State作为输入并返回一个对该状态的更新字典。节点的函数签名非常简单def node_function(state: StateType) - dict:。它从state中读取需要的信息执行逻辑调用LLM、查询数据库、运行计算然后返回一个字典这个字典中的键值对指明了要更新状态的哪些部分。def call_llm(state: AgentState): 节点调用大语言模型生成回复 # 1. 从状态中获取所需信息 history state[“messages”] user_query state[“query”] # 2. 构造提示词调用LLM这里简化表示 prompt construct_prompt(history, user_query) llm_response llm.invoke(prompt) # 3. 返回要更新的状态部分 return {“messages”: [AIMessage(contentllm_response)], “reasoning”: “Generated response via LLM.”}节点设计的一个最佳实践是保持单一职责。一个节点最好只做一件事要么专门调用LLM要么专门查询工具要么专门做条件判断。这样不仅易于测试和调试也使得图的結構更加清晰。当某个节点逻辑过于复杂时就是考虑将其拆分为多个节点或引入子图Subgraph的信号。2.3 边Edge控制流程的“导航线”边定义了节点之间的执行顺序。LangGraph提供了几种类型的边让你能灵活地控制流程起始边Start Edge通过graph.set_entry_point(“node_name”)设置指定工作流从哪个节点开始。普通边Linear Edgegraph.add_edge(“node_a”, “node_b”)。这表示当node_a执行完后无条件地执行node_b。这是顺序执行的基础。条件边Conditional Edge这是实现分支逻辑的关键。它不是一个简单的指向而是由一个路由函数Router Function来决定下一个节点是谁。def route_after_tool(state: AgentState) - str: 根据工具调用结果决定下一步是继续调用工具还是结束 last_message state[“messages”][-1] if “需要更多信息” in last_message.content: return “call_tool” # 返回下一个节点的名字 else: return “__end__” # 特殊关键字表示结束 # 添加条件边 graph.add_conditional_edges( “call_llm”, # 源节点 route_after_tool, # 路由函数 {“call_tool”: “tool_node”, “__end__”: END} # 映射路由返回值 - 目标节点 )条件边赋予了工作流“智能”。它允许流程根据中间结果动态变化这是构建复杂Agent如ReAct模式的核心。这里容易踩的一个坑是路由函数必须返回一个字符串并且这个字符串必须在你提供的映射字典path_map的键中否则LangGraph会抛出异常不知道下一步该去哪。__end__和END是LangGraph预定义的结束标识。3. 从蓝图到引擎compile与CompiledStateGraph当你定义好StateGraph添加了节点和边之后得到的还是一个“蓝图”或“设计图”。它描述了工作流的结构但还不能直接运行。要让这个图“活”起来需要经过编译compile这一步。graph.compile()是LangGraph中一个至关重要的方法。它的作用类似于高级语言中的编译器会对你定义的图进行一系列检查、优化和打包生成一个CompiledStateGraph对象。这个对象才是真正可执行、可调用的工作流引擎。3.1 编译过程中发生了什么图结构验证编译器会检查你的图是否连通是否存在无法到达的节点是否有节点没有出口死节点以及起始点设置是否正确。这能在早期避免许多运行时错误。状态模式推断编译器会分析所有节点函数对状态的读写确保状态类型的一致性。如果某个节点尝试更新一个未在状态类型中定义的字段或者返回了错误类型的值在编译时或首次运行时就可能报错。优化在某些情况下编译器可能会对图结构进行一些内部优化比如合并简单的线性路径以提高执行效率。生成可调用对象最终CompiledStateGraph实现了Runnable接口。这意味着它可以像LangChain中的任何其他Runnable如LLMChain一样使用invoke、batch、stream等方法被调用。这提供了极大的统一性和便利性。# 定义图 graph StateGraph(AgentState) graph.add_node(“call_llm”, call_llm) graph.add_node(“call_tool”, call_tool) graph.set_entry_point(“call_llm”) graph.add_conditional_edges(“call_llm”, route_after_tool, {“more”: “call_tool”, “end”: END}) graph.add_edge(“call_tool”, “call_llm”) # 工具调用后回到LLM # 编译得到可执行的工作流 app graph.compile() # app 现在是一个 CompiledStateGraph 实例一个重要的实操心得养成在开发阶段频繁编译的习惯。不要等到把所有节点和边都加完才编译。每添加几个关键节点和边就compile一下然后用一个简单的初始状态invoke一下看看是否能跑通基本路径。这能帮你快速定位是节点逻辑错误还是图结构设计错误。3.2 CompiledStateGraph你的工作流执行器CompiledStateGraph通常我们称其为app是你与工作流交互的主要对象。它有几个核心方法invoke(input_state, config): 同步执行整个工作流直到结束返回最终状态。适用于不需要中间过程的快速调用。batch(inputs): 批量处理多个输入状态。stream(input_state, config):这是最强大、最常用的方法。它以流式生成器的方式执行工作流每执行完一个节点就yield一次返回该节点执行后的状态快照。这对于调试、实时观察Agent的“思考过程”、以及构建交互式前端至关重要。4. 深入stream一步步“可视化”执行过程stream方法是理解LangGraph执行流程的钥匙。它让你能像看一场电影的逐帧画面一样观察工作流是如何一步步推进的。4.1 stream的输出是什么当你调用app.stream(initial_state)时它返回的是一个异步生成器在同步上下文中stream方法也返回一个生成器。每次迭代yield会产生一个元组(node_name, new_state)。node_name: 刚刚执行完毕的节点的名称。new_state: 该节点执行后的完整状态。initial_state {“messages”: [HumanMessage(content“你好”)], “query”: “你好”, “reasoning”: “”, “next_step”: “”} for step in app.stream(initial_state): node, state step # 解包 print(f“节点 [{node}] 执行完毕。”) print(f“当前消息历史: {state[‘messages’]}”) print(“---”)假设我们有一个简单的“LLM - 判断 - 结束”的图输出可能如下节点 [call_llm] 执行完毕。 当前消息历史: [HumanMessage(‘你好’), AIMessage(‘你好有什么可以帮您’)] --- 节点 [route_after_tool] 执行完毕。 # 注意条件判断节点本身也会被“执行”并产出状态 当前消息历史: [HumanMessage(‘你好’), AIMessage(‘你好有什么可以帮您’)] --- 节点 [__end__] 执行完毕。 当前消息历史: [HumanMessage(‘你好’), AIMessage(‘你好有什么可以帮您’)] ---你会看到甚至连条件路由节点route_after_tool和特殊的结束节点__end__都会出现在流中。这提供了无与伦比的透明性。4.2 如何利用stream进行调试这是stream方法最大的价值所在。当你的Agent行为不符合预期时不再需要漫无目的地打印日志。你可以定位问题节点观察流输出看是在执行到哪个节点后状态出现了异常。是LLM的回复不对还是工具调用返回了错误数据或者是路由函数做出了错误判断检查状态演变对比每一步前后的状态差异。是不是某个节点错误地覆盖了不该覆盖的状态字段messages的累积是否符合预期理解循环对于ReAct这类循环执行“思考-行动-观察”的Agentstream能清晰展示每一次循环的完整过程帮助你判断循环是否卡住或退出条件是否合理。一个常见的坑是“状态污染”。由于每个节点都接收完整状态并返回更新字典如果某个节点不小心返回了一个包含未更改字段的大字典可能会无意中覆盖其他节点设置的临时值。在调试时仔细检查每个节点返回的字典确保它只包含真正需要更新的字段。4.3 关于stream的“断开”错误在相关热搜词中我们看到了诸如stream disconnected before completion这类错误。这通常不是LangGraph的stream方法本身的问题而是发生在与上游API如OpenAI ChatGPT、Codex等通信的过程中。网络问题transport error: network error指向了不稳定的网络连接。配额耗尽you have no credits remaining明确是API调用额度用尽。客户端/服务器端中断websocket closed by server可能是服务端主动断开了连接。当你在LangGraph的节点中调用外部LLM API并使用stream模式指LLM API的流式响应时如果这个网络连接中断就会导致整个LangGraph工作流的stream迭代器提前抛出异常。处理这类问题的关键不在LangGraph层面而在你的节点内部增加重试机制在调用LLM的节点函数中使用tenacity等库为API调用添加指数退避的重试逻辑。使用更稳定的客户端配置调整HTTP客户端如httpx的超时设置、连接池大小。做好错误处理与状态回滚在节点函数中使用try...except捕获API异常并返回一个指示错误的状态更新例如{“error”: “API调用失败”, “next_step”: “handle_error”}然后通过图的条件边引导到一个专门的错误处理节点。这样即使外部服务失败你的工作流也能优雅降级而不是彻底崩溃。5. 高级模式与实战技巧子图、记忆与中断掌握了基础执行流程后我们可以看看LangGraph如何应对更复杂的场景。5.1 子图Subgraph管理复杂性的利器当单个工作流变得过于庞大和复杂时你可以使用子图进行模块化。子图允许你将一个功能集群封装成一个独立的、内部有完整逻辑的图然后将其作为单个节点嵌入到主图中。# 定义一个处理用户查询的子图 query_processing_graph StateGraph(...) # ... 构建子图内部逻辑 query_processing_app query_processing_graph.compile() # 在主图中将子图作为一个节点添加 main_graph.add_node(“process_query”, query_processing_app)这样做的好处是关注点分离主图结构保持清晰只需关心高层级的流程如接收请求 - 处理查询 - 生成报告。复用性同一个子图如“安全检查”、“信息格式化”可以在多个主图中使用。独立测试与调试子图可以单独编译、测试确保其内部逻辑正确。在流式执行时子图节点内部的执行步骤默认是聚合的。也就是说对于主图的stream你只会看到(“process_query”, updated_state)这一条输出而不会看到子图内部每个节点的步骤。如果你需要调试子图内部可以单独对子图调用stream。5.2 长期记忆Long-term Memory的实现热搜词中提到了“langgraph 长期记忆”。LangGraph本身不提供开箱即用的长期记忆存储如向量数据库但它提供了完美的集成点。长期记忆通常通过以下方式实现在状态中设计记忆字段在State中定义一个字段如long_term_memory: List[RelevantMemory]。创建专用的记忆节点检索节点在流程开始时根据用户查询从外部向量数据库检索相关记忆并写入状态。更新节点在流程结束时将本次交互中有价值的信息经过LLM总结后存储回外部数据库。将节点插入工作流在图的适当位置如开始和结束添加这些记忆节点。def retrieve_memory(state: State): query state[“query”] # 调用向量数据库检索 relevant_memories vectorstore.similarity_search(query) return {“long_term_memory”: relevant_memories} def update_memory(state: State): new_memory llm.invoke(f“总结对话要点{state[‘messages’]}”) # 存储到向量数据库 vectorstore.add_texts([new_memory]) return {} # 可能不需要更新状态关键在于LangGraph的状态管理和图编排能力使得在复杂工作流中穿插记忆读写操作变得非常规整和可控。5.3 工作流的暂停、取消与持久化这是一个高级话题。CompiledStateGraph.stream()本身运行在当前的Python进程中要“取消”它最直接的方法就是中断Python进程如CtrlC。但对于一个部署为服务的Agent我们需要更精细的控制。检查点CheckpointingLangGraph内置了检查点机制。这允许你在执行过程中将完整状态包括图的结构位置保存下来。之后你可以从某个检查点恢复执行。这对于需要长时间运行、可能被中断的工作流如多轮复杂任务至关重要。这通常通过配置checkpointer来实现。异步与外部信号如果你在异步框架如FastAPI中使用LangGraph可以将app.astream()包装在一个可取消的异步任务中。通过一个外部的标志位或消息队列来通知工作流停止。在节点函数中定期检查这个标志位如果被设置则返回一个特殊状态引导至结束节点。超时控制可以为整个invoke或stream操作设置超时防止单个工作流运行过久。import asyncio from langgraph.checkpoint import MemorySaver checkpointer MemorySaver() app graph.compile(checkpointercheckpointer) # 第一次执行保存检查点 config {“configurable”: {“thread_id”: “user_123”}} async for step in app.astream(initial_state, configconfig): ... # 假设此时流程中断... # 稍后根据thread_id恢复到最后一次检查点并继续 saved_state checkpointer.get(config) async for step in app.astream(saved_state, configconfig): ...这实现了简单的“暂停/继续”功能。对于更复杂的取消逻辑需要在业务层面设计状态机来管理工作流的生命周期。6. 常见问题排查与性能考量结合热搜词中的错误信息我们来系统梳理一下使用LangGraph时可能遇到的坑。6.1 编译与执行时错误KeyError或字段类型错误这通常是因为状态State定义与节点返回值不匹配。确保节点返回的字典中的每个键都在State的TypedDict或BaseModel中有定义且类型一致。编译器的类型检查有时不能捕获所有运行时错误所以需要仔细编写节点函数。节点找不到或边指向错误“Node ‘xxx’ not found.”错误。检查add_edge或add_conditional_edges中引用的节点名称是否拼写正确。特别注意条件边路由函数返回的字符串必须完全匹配path_map中的键。循环依赖与无限循环如果你添加了类似graph.add_edge(“node_a”, “node_b”)和graph.add_edge(“node_b”, “node_a”)的边就会形成无限循环。LangGraph本身不会阻止你这样做。你需要通过条件边和明确的终止条件来避免。在调试时使用stream并设置一个最大迭代次数来观察是否陷入循环。6.2 性能优化建议节点粒度节点不是越细越好。虽然单一职责是好事但过多的微小节点会增加图遍历的开销。将紧密相关、顺序执行且无分支的多个操作合并到一个节点中是常见的优化手段。异步支持LangGraph完全支持异步。如果你的节点涉及大量I/O操作如网络请求、数据库查询使用async def定义节点函数并在其中使用await可以显著提高并发性能。使用app.astream()进行异步流式调用。状态大小状态对象会在每个节点间传递。避免在状态中存储过大的、不必要的数据如巨大的文件内容。尽量只存储引用如文件ID、数据库主键在需要时再按需加载。条件边的开销条件边需要执行一个额外的路由函数。如果路由逻辑非常复杂可能会成为瓶颈。尽量保持路由函数轻量级只做简单的判断。6.3 与LangChain的协作模式最后澄清一个常见疑问LangGraph和LangChain是什么关系如何选择LangChain是你的AI应用工具箱。它提供了与上百种LLM、向量库、工具集成的标准化接口Runnable以及一些基础的链Chain和代理Agent模板。它适合快速构建概念验证POC和相对线性的应用。LangGraph是你的复杂工作流编排引擎。当你需要清晰定义带有循环、条件分支、复杂状态管理的多步骤AI流程时就应该选择LangGraph。它通常与LangChain一起使用——用LangChain的组件LLM、工具、检索器作为节点的实现用LangGraph来编排这些组件的执行顺序。在实践中我的项目架构往往是用LangGraph定义主干流程和状态用LangChain的Runnable系列组件作为每个节点的具体实现。这样既能享受LangChain生态的丰富性又能获得LangGraph在复杂流程控制上的强大能力。
1. 从LangChain到LangGraph为什么我们需要一个新的“图”如果你在过去一年里接触过大语言模型应用开发那么“LangChain”这个名字对你来说一定不陌生。它像是一套乐高积木把LLM、工具调用、记忆、向量检索这些组件标准化、模块化让我们能快速拼装出一个可用的AI应用。但用久了尤其是在构建稍微复杂一点的、有状态、多步骤的流程时你可能会感到一丝掣肘。比如你想实现一个根据用户问题动态决定调用哪个工具并且工具调用的结果可能影响后续流程走向的客服机器人。用LangChain的Chain和Agent来写代码往往会变得嵌套很深状态管理分散调试起来像在迷宫里找路。这就是LangGraph诞生的背景。它不是要取代LangChain而是作为LangChain生态系统中的一个新成员专门解决复杂、有状态、可能带循环的工作流的编排问题。你可以把它理解为LangChain在“工作流引擎”这个维度上的深度补充和增强。如果说LangChain提供了构建AI应用所需的“砖块”Components那么LangGraph则提供了设计并建造复杂“建筑结构”Stateful Workflows的蓝图和脚手架。它的核心思想非常直观用“图”Graph来建模你的应用逻辑。图中的节点Node代表一个可执行的操作单元比如调用一次LLM、执行一个工具函数、做一个条件判断边Edge则定义了这些操作之间的流转关系。这种建模方式天然适合描述那些非线性的、有分支、有循环的流程。当你面对“根据情况可能走A分支也可能走B分支”、“某个步骤需要循环执行直到满足条件”这类需求时用LangGraph会比用传统的顺序链Sequential Chain清晰得多。从技术栈上看LangGraph与LangChain高度集成共享许多概念如Runnable, Messages等但它引入了几个关键的新抽象StateGraph定义图的结构和状态、Node节点、Edge边以及编译后得到的CompiledStateGraph可执行的工作流对象。而最常用的入口点就是那个stream方法它允许你以流式的方式执行整个图并观察每一步的状态变化这对于调试和理解流程至关重要。2. LangGraph的核心三要素State, Node, Edge要理解LangGraph的执行流程必须先吃透它的三个核心构建块状态State、节点Node和边Edge。这三者共同定义了一个有向图驱动着整个工作流的运转。2.1 状态State工作流的“记忆体”在LangGraph中状态是一个贯穿始终的核心概念。它不是一个简单的变量而是一个类型化的字典TypedDict定义了在整个图执行过程中哪些信息会被传递和修改。你可以把它想象成游戏里角色的属性面板或者一个工单在处理过程中不断被填写的表单。定义一个状态通常使用Pydantic的BaseModel或者Python的TypedDict。例如一个简单的聊天代理的状态可能包含from typing import TypedDict, List, Annotated from langgraph.graph.message import add_messages import operator class AgentState(TypedDict): # 消息历史使用LangGraph提供的注解实现自动累加 messages: Annotated[List[BaseMessage], add_messages] # 用户当前查询 query: str # 代理的思考过程或中间结果 reasoning: str # 决定下一步要做什么 next_step: str这里的关键是Annotated的使用。Annotated[List[BaseMessage], add_messages]是一个“缩减器”Reducer。它告诉LangGraphmessages这个字段在节点间传递时如果后一个节点返回了新的消息列表不要直接覆盖而是通过add_messages函数通常是追加来更新。这是实现对话历史累积的优雅方式。状态中的其他字段如query则默认是“覆盖”模式即新值直接替换旧值。为什么状态设计如此重要因为它直接决定了数据的生命周期和共享范围。一个设计良好的状态结构能让节点之间的协作清晰明了避免出现数据污染或丢失。在规划工作流时我的经验是先别急着写节点逻辑而是花时间想清楚“在整个流程中我需要记录和传递哪些信息”。2.2 节点Node执行具体任务的“工作单元”节点是图中实际干活的部分。在LangGraph中一个节点就是一个可调用对象Callable它接收当前状态State作为输入并返回一个对该状态的更新字典。节点的函数签名非常简单def node_function(state: StateType) - dict:。它从state中读取需要的信息执行逻辑调用LLM、查询数据库、运行计算然后返回一个字典这个字典中的键值对指明了要更新状态的哪些部分。def call_llm(state: AgentState): 节点调用大语言模型生成回复 # 1. 从状态中获取所需信息 history state[“messages”] user_query state[“query”] # 2. 构造提示词调用LLM这里简化表示 prompt construct_prompt(history, user_query) llm_response llm.invoke(prompt) # 3. 返回要更新的状态部分 return {“messages”: [AIMessage(contentllm_response)], “reasoning”: “Generated response via LLM.”}节点设计的一个最佳实践是保持单一职责。一个节点最好只做一件事要么专门调用LLM要么专门查询工具要么专门做条件判断。这样不仅易于测试和调试也使得图的結構更加清晰。当某个节点逻辑过于复杂时就是考虑将其拆分为多个节点或引入子图Subgraph的信号。2.3 边Edge控制流程的“导航线”边定义了节点之间的执行顺序。LangGraph提供了几种类型的边让你能灵活地控制流程起始边Start Edge通过graph.set_entry_point(“node_name”)设置指定工作流从哪个节点开始。普通边Linear Edgegraph.add_edge(“node_a”, “node_b”)。这表示当node_a执行完后无条件地执行node_b。这是顺序执行的基础。条件边Conditional Edge这是实现分支逻辑的关键。它不是一个简单的指向而是由一个路由函数Router Function来决定下一个节点是谁。def route_after_tool(state: AgentState) - str: 根据工具调用结果决定下一步是继续调用工具还是结束 last_message state[“messages”][-1] if “需要更多信息” in last_message.content: return “call_tool” # 返回下一个节点的名字 else: return “__end__” # 特殊关键字表示结束 # 添加条件边 graph.add_conditional_edges( “call_llm”, # 源节点 route_after_tool, # 路由函数 {“call_tool”: “tool_node”, “__end__”: END} # 映射路由返回值 - 目标节点 )条件边赋予了工作流“智能”。它允许流程根据中间结果动态变化这是构建复杂Agent如ReAct模式的核心。这里容易踩的一个坑是路由函数必须返回一个字符串并且这个字符串必须在你提供的映射字典path_map的键中否则LangGraph会抛出异常不知道下一步该去哪。__end__和END是LangGraph预定义的结束标识。3. 从蓝图到引擎compile与CompiledStateGraph当你定义好StateGraph添加了节点和边之后得到的还是一个“蓝图”或“设计图”。它描述了工作流的结构但还不能直接运行。要让这个图“活”起来需要经过编译compile这一步。graph.compile()是LangGraph中一个至关重要的方法。它的作用类似于高级语言中的编译器会对你定义的图进行一系列检查、优化和打包生成一个CompiledStateGraph对象。这个对象才是真正可执行、可调用的工作流引擎。3.1 编译过程中发生了什么图结构验证编译器会检查你的图是否连通是否存在无法到达的节点是否有节点没有出口死节点以及起始点设置是否正确。这能在早期避免许多运行时错误。状态模式推断编译器会分析所有节点函数对状态的读写确保状态类型的一致性。如果某个节点尝试更新一个未在状态类型中定义的字段或者返回了错误类型的值在编译时或首次运行时就可能报错。优化在某些情况下编译器可能会对图结构进行一些内部优化比如合并简单的线性路径以提高执行效率。生成可调用对象最终CompiledStateGraph实现了Runnable接口。这意味着它可以像LangChain中的任何其他Runnable如LLMChain一样使用invoke、batch、stream等方法被调用。这提供了极大的统一性和便利性。# 定义图 graph StateGraph(AgentState) graph.add_node(“call_llm”, call_llm) graph.add_node(“call_tool”, call_tool) graph.set_entry_point(“call_llm”) graph.add_conditional_edges(“call_llm”, route_after_tool, {“more”: “call_tool”, “end”: END}) graph.add_edge(“call_tool”, “call_llm”) # 工具调用后回到LLM # 编译得到可执行的工作流 app graph.compile() # app 现在是一个 CompiledStateGraph 实例一个重要的实操心得养成在开发阶段频繁编译的习惯。不要等到把所有节点和边都加完才编译。每添加几个关键节点和边就compile一下然后用一个简单的初始状态invoke一下看看是否能跑通基本路径。这能帮你快速定位是节点逻辑错误还是图结构设计错误。3.2 CompiledStateGraph你的工作流执行器CompiledStateGraph通常我们称其为app是你与工作流交互的主要对象。它有几个核心方法invoke(input_state, config): 同步执行整个工作流直到结束返回最终状态。适用于不需要中间过程的快速调用。batch(inputs): 批量处理多个输入状态。stream(input_state, config):这是最强大、最常用的方法。它以流式生成器的方式执行工作流每执行完一个节点就yield一次返回该节点执行后的状态快照。这对于调试、实时观察Agent的“思考过程”、以及构建交互式前端至关重要。4. 深入stream一步步“可视化”执行过程stream方法是理解LangGraph执行流程的钥匙。它让你能像看一场电影的逐帧画面一样观察工作流是如何一步步推进的。4.1 stream的输出是什么当你调用app.stream(initial_state)时它返回的是一个异步生成器在同步上下文中stream方法也返回一个生成器。每次迭代yield会产生一个元组(node_name, new_state)。node_name: 刚刚执行完毕的节点的名称。new_state: 该节点执行后的完整状态。initial_state {“messages”: [HumanMessage(content“你好”)], “query”: “你好”, “reasoning”: “”, “next_step”: “”} for step in app.stream(initial_state): node, state step # 解包 print(f“节点 [{node}] 执行完毕。”) print(f“当前消息历史: {state[‘messages’]}”) print(“---”)假设我们有一个简单的“LLM - 判断 - 结束”的图输出可能如下节点 [call_llm] 执行完毕。 当前消息历史: [HumanMessage(‘你好’), AIMessage(‘你好有什么可以帮您’)] --- 节点 [route_after_tool] 执行完毕。 # 注意条件判断节点本身也会被“执行”并产出状态 当前消息历史: [HumanMessage(‘你好’), AIMessage(‘你好有什么可以帮您’)] --- 节点 [__end__] 执行完毕。 当前消息历史: [HumanMessage(‘你好’), AIMessage(‘你好有什么可以帮您’)] ---你会看到甚至连条件路由节点route_after_tool和特殊的结束节点__end__都会出现在流中。这提供了无与伦比的透明性。4.2 如何利用stream进行调试这是stream方法最大的价值所在。当你的Agent行为不符合预期时不再需要漫无目的地打印日志。你可以定位问题节点观察流输出看是在执行到哪个节点后状态出现了异常。是LLM的回复不对还是工具调用返回了错误数据或者是路由函数做出了错误判断检查状态演变对比每一步前后的状态差异。是不是某个节点错误地覆盖了不该覆盖的状态字段messages的累积是否符合预期理解循环对于ReAct这类循环执行“思考-行动-观察”的Agentstream能清晰展示每一次循环的完整过程帮助你判断循环是否卡住或退出条件是否合理。一个常见的坑是“状态污染”。由于每个节点都接收完整状态并返回更新字典如果某个节点不小心返回了一个包含未更改字段的大字典可能会无意中覆盖其他节点设置的临时值。在调试时仔细检查每个节点返回的字典确保它只包含真正需要更新的字段。4.3 关于stream的“断开”错误在相关热搜词中我们看到了诸如stream disconnected before completion这类错误。这通常不是LangGraph的stream方法本身的问题而是发生在与上游API如OpenAI ChatGPT、Codex等通信的过程中。网络问题transport error: network error指向了不稳定的网络连接。配额耗尽you have no credits remaining明确是API调用额度用尽。客户端/服务器端中断websocket closed by server可能是服务端主动断开了连接。当你在LangGraph的节点中调用外部LLM API并使用stream模式指LLM API的流式响应时如果这个网络连接中断就会导致整个LangGraph工作流的stream迭代器提前抛出异常。处理这类问题的关键不在LangGraph层面而在你的节点内部增加重试机制在调用LLM的节点函数中使用tenacity等库为API调用添加指数退避的重试逻辑。使用更稳定的客户端配置调整HTTP客户端如httpx的超时设置、连接池大小。做好错误处理与状态回滚在节点函数中使用try...except捕获API异常并返回一个指示错误的状态更新例如{“error”: “API调用失败”, “next_step”: “handle_error”}然后通过图的条件边引导到一个专门的错误处理节点。这样即使外部服务失败你的工作流也能优雅降级而不是彻底崩溃。5. 高级模式与实战技巧子图、记忆与中断掌握了基础执行流程后我们可以看看LangGraph如何应对更复杂的场景。5.1 子图Subgraph管理复杂性的利器当单个工作流变得过于庞大和复杂时你可以使用子图进行模块化。子图允许你将一个功能集群封装成一个独立的、内部有完整逻辑的图然后将其作为单个节点嵌入到主图中。# 定义一个处理用户查询的子图 query_processing_graph StateGraph(...) # ... 构建子图内部逻辑 query_processing_app query_processing_graph.compile() # 在主图中将子图作为一个节点添加 main_graph.add_node(“process_query”, query_processing_app)这样做的好处是关注点分离主图结构保持清晰只需关心高层级的流程如接收请求 - 处理查询 - 生成报告。复用性同一个子图如“安全检查”、“信息格式化”可以在多个主图中使用。独立测试与调试子图可以单独编译、测试确保其内部逻辑正确。在流式执行时子图节点内部的执行步骤默认是聚合的。也就是说对于主图的stream你只会看到(“process_query”, updated_state)这一条输出而不会看到子图内部每个节点的步骤。如果你需要调试子图内部可以单独对子图调用stream。5.2 长期记忆Long-term Memory的实现热搜词中提到了“langgraph 长期记忆”。LangGraph本身不提供开箱即用的长期记忆存储如向量数据库但它提供了完美的集成点。长期记忆通常通过以下方式实现在状态中设计记忆字段在State中定义一个字段如long_term_memory: List[RelevantMemory]。创建专用的记忆节点检索节点在流程开始时根据用户查询从外部向量数据库检索相关记忆并写入状态。更新节点在流程结束时将本次交互中有价值的信息经过LLM总结后存储回外部数据库。将节点插入工作流在图的适当位置如开始和结束添加这些记忆节点。def retrieve_memory(state: State): query state[“query”] # 调用向量数据库检索 relevant_memories vectorstore.similarity_search(query) return {“long_term_memory”: relevant_memories} def update_memory(state: State): new_memory llm.invoke(f“总结对话要点{state[‘messages’]}”) # 存储到向量数据库 vectorstore.add_texts([new_memory]) return {} # 可能不需要更新状态关键在于LangGraph的状态管理和图编排能力使得在复杂工作流中穿插记忆读写操作变得非常规整和可控。5.3 工作流的暂停、取消与持久化这是一个高级话题。CompiledStateGraph.stream()本身运行在当前的Python进程中要“取消”它最直接的方法就是中断Python进程如CtrlC。但对于一个部署为服务的Agent我们需要更精细的控制。检查点CheckpointingLangGraph内置了检查点机制。这允许你在执行过程中将完整状态包括图的结构位置保存下来。之后你可以从某个检查点恢复执行。这对于需要长时间运行、可能被中断的工作流如多轮复杂任务至关重要。这通常通过配置checkpointer来实现。异步与外部信号如果你在异步框架如FastAPI中使用LangGraph可以将app.astream()包装在一个可取消的异步任务中。通过一个外部的标志位或消息队列来通知工作流停止。在节点函数中定期检查这个标志位如果被设置则返回一个特殊状态引导至结束节点。超时控制可以为整个invoke或stream操作设置超时防止单个工作流运行过久。import asyncio from langgraph.checkpoint import MemorySaver checkpointer MemorySaver() app graph.compile(checkpointercheckpointer) # 第一次执行保存检查点 config {“configurable”: {“thread_id”: “user_123”}} async for step in app.astream(initial_state, configconfig): ... # 假设此时流程中断... # 稍后根据thread_id恢复到最后一次检查点并继续 saved_state checkpointer.get(config) async for step in app.astream(saved_state, configconfig): ...这实现了简单的“暂停/继续”功能。对于更复杂的取消逻辑需要在业务层面设计状态机来管理工作流的生命周期。6. 常见问题排查与性能考量结合热搜词中的错误信息我们来系统梳理一下使用LangGraph时可能遇到的坑。6.1 编译与执行时错误KeyError或字段类型错误这通常是因为状态State定义与节点返回值不匹配。确保节点返回的字典中的每个键都在State的TypedDict或BaseModel中有定义且类型一致。编译器的类型检查有时不能捕获所有运行时错误所以需要仔细编写节点函数。节点找不到或边指向错误“Node ‘xxx’ not found.”错误。检查add_edge或add_conditional_edges中引用的节点名称是否拼写正确。特别注意条件边路由函数返回的字符串必须完全匹配path_map中的键。循环依赖与无限循环如果你添加了类似graph.add_edge(“node_a”, “node_b”)和graph.add_edge(“node_b”, “node_a”)的边就会形成无限循环。LangGraph本身不会阻止你这样做。你需要通过条件边和明确的终止条件来避免。在调试时使用stream并设置一个最大迭代次数来观察是否陷入循环。6.2 性能优化建议节点粒度节点不是越细越好。虽然单一职责是好事但过多的微小节点会增加图遍历的开销。将紧密相关、顺序执行且无分支的多个操作合并到一个节点中是常见的优化手段。异步支持LangGraph完全支持异步。如果你的节点涉及大量I/O操作如网络请求、数据库查询使用async def定义节点函数并在其中使用await可以显著提高并发性能。使用app.astream()进行异步流式调用。状态大小状态对象会在每个节点间传递。避免在状态中存储过大的、不必要的数据如巨大的文件内容。尽量只存储引用如文件ID、数据库主键在需要时再按需加载。条件边的开销条件边需要执行一个额外的路由函数。如果路由逻辑非常复杂可能会成为瓶颈。尽量保持路由函数轻量级只做简单的判断。6.3 与LangChain的协作模式最后澄清一个常见疑问LangGraph和LangChain是什么关系如何选择LangChain是你的AI应用工具箱。它提供了与上百种LLM、向量库、工具集成的标准化接口Runnable以及一些基础的链Chain和代理Agent模板。它适合快速构建概念验证POC和相对线性的应用。LangGraph是你的复杂工作流编排引擎。当你需要清晰定义带有循环、条件分支、复杂状态管理的多步骤AI流程时就应该选择LangGraph。它通常与LangChain一起使用——用LangChain的组件LLM、工具、检索器作为节点的实现用LangGraph来编排这些组件的执行顺序。在实践中我的项目架构往往是用LangGraph定义主干流程和状态用LangChain的Runnable系列组件作为每个节点的具体实现。这样既能享受LangChain生态的丰富性又能获得LangGraph在复杂流程控制上的强大能力。
