1. 从“状态”说起LangGraph为何需要它如果你用过LangChain可能会觉得它像是一个功能强大的工具箱帮你把各种LLM大语言模型工具、检索器、记忆模块组装起来。但当你尝试构建一个需要多步骤、有状态、甚至能根据中间结果动态调整流程的复杂应用时比如一个能和你进行多轮对话并记住上下文的客服机器人或者一个能根据用户反馈逐步优化代码的编程助手你会发现LangChain的链式结构有点“力不从心”。它擅长线性流程但对于循环、分支、并行这些更复杂的控制流编排起来就变得异常繁琐。这就是LangGraph诞生的背景。它不是一个替代品而是一个在LangChain生态之上的“编排引擎”。你可以把它想象成一个专门为构建有状态、可循环、可分支的智能体Agent或工作流而设计的框架。而这一切的核心都围绕着“状态”这个概念展开。那么LangGraph里的“状态”到底是什么简单说它就是你的工作流在运行过程中的“记忆体”和“数据黑板”。它不是一个简单的变量而是一个结构化的、可被工作流中所有节点Node读取和更新的共享数据空间。比如在一个对话机器人中状态里可能存储着当前的用户问题、历史对话记录、从知识库检索到的文档片段、以及LLM生成的中间思考过程。工作流中的每个步骤节点都可以查看当前状态并决定如何修改它从而影响后续步骤的走向。理解状态管理是掌握LangGraph的钥匙。而graph.invoke则是你启动这个有状态工作流的“点火开关”。今天我们就来彻底拆解这两者尤其是graph.invoke的入参这往往是新手构建第一个可运行LangGraph应用时遇到的第一个拦路虎。2. LangGraph状态管理的核心StateGraph与Reducer要管理状态首先得定义状态的结构。在LangGraph中这是通过定义一个State类通常继承自TypedDict来完成的。这个类定义了状态中有哪些字段以及每个字段的类型。但光有结构还不够我们还需要定义当多个节点同时尝试更新同一个字段时应该遵循什么规则。这就是Reducer归约器的用武之地。Reducer决定了如何将“旧的字段值”和“节点试图写入的新值”合并成“最终的新值”。这是LangGraph状态管理中最精妙也最容易出错的部分。2.1 定义状态结构你的数据黑板蓝图我们来看一个经典的智能体状态定义它通常包含用户输入、AI的思考、执行工具调用的历史等。from typing import TypedDict, List, Annotated from langgraph.graph.message import add_messages import operator class AgentState(TypedDict): # 用户的最新输入 input: str # 对话历史这是一个特殊的列表使用add_messages这个reducer messages: Annotated[List, add_messages] # AI的“内心独白”或思考链使用列表追加的reducer thoughts: Annotated[List[str], operator.add] # 已经调用过的工具及其结果历史 tool_calls: Annotated[List[dict], operator.add] # 一个标志位表示AI是否应该继续思考继续循环还是停止 continue_loop: bool这里有几个关键点TypedDict 这只是一个类型提示帮助IDE和类型检查器理解状态的结构运行时并不强制。但它对于代码的可读性和可维护性至关重要。Annotated类型 这是LangGraph定义Reducer的方式。Annotated[字段类型, reducer函数]。它告诉LangGraph“这个messages字段是一个列表当有节点要更新它时请使用add_messages这个函数来决定如何合并新旧值。”2.2 理解Reducer合并策略的艺术Reducer是一个函数它接收两个参数current_value当前状态中的值和update_value节点想要设置的值然后返回合并后的新值。为什么需要Reducer想象一下工作流中的两个节点几乎同时运行或者在并行分支中它们都可能去修改messages列表。如果没有一个明确的合并规则就会发生数据竞争最终状态是不确定的。Reducer提供了这个确定性规则。LangGraph内置了一些常用的Reducer你也可以自定义operator.add(或lambda a, b: a b) 用于列表。将新列表追加到旧列表之后。这是最常用的Reducer之一适用于记录历史、收集结果等场景。上面例子中的thoughts和tool_calls字段就使用了它。add_messages这是为聊天消息列表特制的Reducer极其重要它不仅仅做追加还能智能地处理HumanMessage,AIMessage,ToolMessage等LangChain消息对象确保消息线程的正确性。对于任何存储对话历史的字段都应该使用add_messages而不是简单的operator.add。lambda current, update: update 这是“覆盖”Reducer。新值直接覆盖旧值。适用于那些每次只需要最新值的字段比如input最新的用户输入会覆盖上一次的或者continue_loop只需要最新的判断结果。自定义Reducer 你可以实现任何合并逻辑。例如对于一个数字类型的retry_count重试次数字段你可以定义Reducer为lambda current, update: current 1表示每次更新都加1。一个常见的坑 错误地为列表字段使用覆盖Reducerlambda c, u: u。这会导致历史记录被完全清空每次只有最后一个节点的修改生效。你的智能体就会患上“失忆症”。2.3 构建状态图StateGraph的创建与节点绑定定义了状态结构后我们就可以创建StateGraph并将节点函数绑定到图上。每个节点函数都必须接收一个状态字典作为参数并返回一个字典其中包含它想要更新的状态字段。from langgraph.graph import StateGraph, END # 1. 创建图并指定状态结构 graph_builder StateGraph(AgentState) # 2. 定义节点函数 def call_llm(state: AgentState): # 从状态中获取所需信息 user_input state[“input”] history state[“messages”] # 调用LLM生成思考和回复... ai_thought “用户想了解天气我需要调用天气查询工具。” ai_response “我将为您查询天气。” # 返回要更新的状态部分 return {“thoughts”: [ai_thought], “messages”: [AIMessage(contentai_response)]} def call_tool(state: AgentState): # 根据thoughts或messages决定调用哪个工具... tool_result “北京晴25℃” return {“tool_calls”: [{“tool”: “weather”, “result”: tool_result}]} # 3. 将节点添加到图中 graph_builder.add_node(“llm_node”, call_llm) graph_builder.add_node(“tool_node”, call_tool) # 4. 定义边流程走向 graph_builder.set_entry_point(“llm_node”) # 设置入口节点 graph_builder.add_edge(“llm_node”, “tool_node”) # llm_node执行完后无条件跳到tool_node graph_builder.add_edge(“tool_node”, END) # tool_node执行完后结束 # 5. 编译图 graph graph_builder.compile()至此一个最简单的、有状态的工作流图就构建好了。但它是静态的从LLM节点直接到工具节点再到结束。一个真正的智能体通常需要根据LLM的输出比如是否调用了工具来决定下一步是继续循环思考还是结束。这需要引入条件边我们稍后会谈到。现在我们先来看看如何让这个图“跑”起来。3.graph.invoke入参深度解析启动引擎的钥匙graph.compile()之后得到的graph对象其invoke方法就是执行工作流的入口。它的入参决定了工作流的初始状态理解它至关重要。3.1 基本调用传递初始状态最直接的方式是传入一个字典这个字典的内容必须与你定义的State结构兼容。# 定义初始状态 initial_state { “input”: “北京今天天气怎么样”, “messages”: [], # 初始对话历史为空 “thoughts”: [], “tool_calls”: [], “continue_loop”: True, } # 执行图 final_state graph.invoke(initial_state) print(final_state[“messages”]) # 查看最终的对话消息 print(final_state[“tool_calls”]) # 查看工具调用记录这里发生了什么invoke接收initial_state字典。LangGraph会将这个字典与你StateGraph(AgentState)中定义的AgentState结构进行“对齐”。它不会做严格的类型检查但会依据你定义的Reducer来处理字段。图从入口节点llm_node开始执行将该节点的初始状态即你传入的initial_state传递给call_llm函数。节点函数返回更新字典如{“thoughts”: [“...”], “messages”: [...]}LangGraph会调用对应字段的Reducer将更新合并到当前状态中。状态更新后根据边目前是固定边跳转到下一个节点tool_node并将更新后的状态传递给它。重复此过程直到到达END节点。invoke返回最终的状态字典。3.2 入参的“魔法”自动包装与消息处理invoke的入参设计得非常灵活它内部会尝试将你的输入“包装”成完整的状态。这是为了方便起见但也是困惑的来源。场景一只传入input和messages这是最常见、最推荐的调用方式尤其对于聊天应用。result graph.invoke({“input”: “你好”, “messages”: [HumanMessage(content“你好”)]}) # 或者更常见的直接传入用户输入字符串并让LangGraph帮你构造HumanMessage result graph.invoke({“input”: “你好”})背后的逻辑 当你没有显式提供所有状态字段时LangGraph会使用该字段类型默认值来填充缺失字段。对于AgentStateinput: 你提供了“你好”所以用它。messages: 如果你提供了列表就用它。如果没提供如第二个例子LangGraph会自动将input的值包装成一个HumanMessage对象并放入messages列表。这是一个非常贴心的特性。thoughts,tool_calls,continue_loop: 你没提供LangGraph会使用空列表[]和True对于bool类型作为默认初始值。这意味着对于聊天你通常只需要关心input即可messages的历史维护可以交给add_messages这个Reducer和invoke的自动包装逻辑。场景二传入config字典invoke还可以接受一个config参数用于控制运行时行为比如设置LLM的API Key、温度系数或者启用调试。from langchain_core.runnables import RunnableConfig config RunnableConfig(configurable{“thread_id”: “user_123”}) # 例如用于区分不同会话线程 result graph.invoke( {“input”: “查询余额”}, configconfig )config是一个强大的工具它可以用来传递会话ID 在多用户场景下通过configurable字典传递thread_id结合LangGraph的检查点Checkpoint功能可以实现对话状态的持久化和恢复。传递API密钥等元数据 避免在节点函数中硬编码。控制递归深度 对于可能循环的图设置最大循环次数防止无限循环。3.3 常见错误与排查指南错误1TypeError: State.... got an unexpected keyword argument ‘...‘原因 你传入invoke的初始状态字典里包含了一个你的State类中没有定义的键。解决 检查拼写确保所有键名都与TypedDict中定义的字段名完全一致。LangGraph对初始状态是宽容的但如果你使用了Pydantic等更严格的模型或者字段名不匹配就会报错。错误2节点更新后状态字段没有按预期累积比如messages历史丢了原因 几乎可以肯定是Reducer配置错误。你很可能为列表字段如messages,thoughts错误地设置了覆盖型的Reducer或者忘记设置默认为覆盖。排查检查状态类定义。确保列表字段使用了Annotated[List, operator.add]或Annotated[List, add_messages]。在节点函数中确保你返回的是更新字典。例如如果你想在thoughts列表中添加一项应该返回{“thoughts”: [“新的思考”]}而不是{“thoughts”: “新的思考”}。Reducer会处理这个列表与现有列表的合并。错误3消息顺序混乱或格式错误原因 对于对话消息混用了add_messages和operator.add或者手动构造消息对象时格式不对。解决始终坚持使用add_messages作为messages字段的Reducer。在节点函数中返回消息时使用LangChain提供的消息类如AIMessage,ToolMessage,SystemMessage等。避免直接操作state[“messages”]列表而是通过返回更新字典让Reducer去合并。错误4invoke后图没有执行或者只执行了部分节点原因 图的边Edge没有正确设置特别是条件边。可能你的逻辑导致流程提前进入了END。排查打印graph.get_graph().draw_mermaid()生成的Mermaid图在Jupyter等环境中可视化检查你的流程逻辑。在关键节点函数中添加print语句或使用LangGraph的调试模式通过config设置查看执行轨迹。检查条件边add_conditional_edges的判断函数确保其返回值能正确映射到下一个节点名或END。4. 进阶动态流程与条件边下的状态流转一个静态的、顺序执行的图意义有限。LangGraph的强大在于它能根据状态动态决定下一步。这通过add_conditional_edges方法实现。让我们改造之前的图让LLM节点之后根据其输出决定是调用工具还是直接结束。from langchain_core.messages import AIMessage from langgraph.graph import StateGraph, END # ... 假设StateGraph和节点函数已定义如上 ... # 移除之前的固定边 # graph_builder.add_edge(“llm_node”, “tool_node”) # graph_builder.add_edge(“tool_node”, END) # 定义条件路由函数 def route_after_llm(state: AgentState) - str: “”“根据LLM节点的输出存储在state中决定下一步。”“” # 假设我们通过一个简单的规则判断如果thoughts里包含‘工具’二字就去调用工具 latest_thoughts state.get(“thoughts”, []) if latest_thoughts and any(“工具” in t for t in latest_thoughts[-1:]): return “tool_node” # 跳转到工具节点 else: return END # 直接结束 # 添加条件边 graph_builder.add_conditional_edges( “llm_node”, # 源节点 route_after_llm, # 路由函数它接收当前状态返回下一个节点的名字字符串或END # 可选映射路由函数返回值到节点名这里返回值本身就是节点名所以不需要 # path_map{“use_tool”: “tool_node”, “end”: END} ) # 工具节点执行完后我们可能还需要让LLM节点来总结工具结果所以再加一条边回到LLM节点 # 但这可能导致无限循环所以我们需要在状态中用一个字段如continue_loop来控制 def route_after_tool(state: AgentState) - str: if state.get(“continue_loop”, True): return “llm_node” # 继续循环 else: return END graph_builder.add_conditional_edges(“tool_node”, route_after_tool) # 重新编译图 graph graph_builder.compile()在这个动态图中invoke的入参初始状态就更加关键了因为它决定了第一次条件判断的走向。同时节点函数现在需要负责更新那些影响路由的状态字段比如thoughts和continue_loop。执行流程示例invoke({“input”: “今天天气如何”})启动。llm_node执行可能将thoughts更新为[“用户问天气需要调用工具”]。条件边函数route_after_llm检查thoughts发现包含“工具”返回“tool_node”。跳转到tool_node执行查询天气更新tool_calls并可能将continue_loop设为False表示工具调用完成不需要再循环。条件边函数route_after_tool检查continue_loop为False返回END。工作流结束返回最终状态。5. 实战心得与性能考量心得1状态设计要“恰到好处”不要把所有东西都塞进状态。状态应该只包含影响流程控制和需要在节点间共享的数据。一些中间计算结果如果只被一个节点使用最好作为该节点函数的局部变量。过大的状态会增加序列化/反序列化的开销如果使用持久化检查点也会让Reducer的逻辑变得复杂。心得2善用config管理上下文将不频繁变化、但与运行时相关的信息如用户ID、API密钥、实验性开关放在config中而不是状态里。状态关注“发生了什么”config关注“在什么环境下发生”。心得3Reducer的副作用记住Reducer函数应该是纯函数给定相同输入产生相同输出无副作用。不要在Reducer里执行网络请求、读写文件等IO操作。所有业务逻辑都应该在节点函数中完成。心得4invoke与streamgraph.invoke()是同步调用会阻塞直到整个工作流完成。对于长时间运行的工作流或者需要实时看到中间结果的场景如聊天流式输出可以使用graph.stream(initial_state)。它会返回一个生成器每执行完一个节点就yield一次当前状态非常适合前端展示。心得5调试是门艺术当图变得复杂时调试可能很棘手。除了加print强烈建议使用graph.get_graph().draw_mermaid()生成流程图宏观把握逻辑。在invoke时传入configRunnableConfig(recursion_limit50, callbacks[ConsoleCallbackHandler()])来启用回调打印详细日志。将复杂的条件判断逻辑单独写成函数并进行单元测试。理解StateGraph的状态管理和graph.invoke的入参是构建可靠、可维护LangGraph应用的基础。它要求你从“链式思维”转向“图状态思维”仔细设计数据的流动和合并规则。一旦掌握你将能构建出远比简单链式调用更强大、更灵活的AI应用。