1. 从“状态”说起为什么LangGraph的状态管理是Agent的基石如果你用过LangChain或者自己动手组装过基于LLM的Agent那你一定对“状态”这个词不陌生。简单来说状态就是你的Agent在运行过程中需要记住的所有东西。比如用户问了什么问题你调用了哪些工具工具返回了什么结果LLM思考的中间过程是什么等等。在早期的Agent开发中我们往往需要自己手动维护一个字典或者一个对象来承载这些信息然后在各个函数之间传来传去。这个过程不仅繁琐而且极易出错尤其是在处理多轮对话、复杂工作流或者并发请求时状态管理很快就会变成一团乱麻。LangGraph的出现很大程度上就是为了解决这个问题。它将Agent的执行过程抽象为一个有向图Graph图中的节点是执行单元比如调用LLM、执行工具边是状态流转的条件。而贯穿整个图、在各个节点间流动和更新的就是“状态”。因此理解LangGraph的状态管理是理解其工作原理和发挥其威力的关键。今天我们就来深入聊聊LangGraph中两个至关重要的预定义状态类MessagesState和AgentState。它们不是让你从零开始造轮子而是提供了两种经过深思熟虑的“标准答案”能覆盖绝大多数Agent开发场景。2. MessagesState为纯对话流而生的精简状态我们先从相对简单的MessagesState开始。顾名思义这个状态类是专门为以消息Message为核心驱动的对话流程设计的。如果你要构建的Agent其核心逻辑就是“用户输入 - LLM思考 - 返回回复”这样的循环中间可能穿插一些简单的工具调用那么MessagesState通常是你的首选。2.1 MessagesState的核心结构MessagesState本质上是一个TypedDict它只定义了一个必需的字段messages。这个字段是一个消息列表List[BaseMessage]它遵循LangChain的消息格式比如HumanMessage、AIMessage、ToolMessage、SystemMessage等。from typing import TypedDict, List, Annotated from langchain_core.messages import BaseMessage import operator class MessagesState(TypedDict): messages: Annotated[List[BaseMessage], operator.add]这里有一个关键细节Annotated[List[BaseMessage], operator.add]。这个注解是LangGraph状态管理的精髓之一。operator.add是一个“归约器”reducer它定义了当多个节点并发修改同一个状态字段时如何合并这些修改。对于列表来说operator.add就是列表的拼接extend。这意味着每个节点都可以向messages列表末尾追加新的消息而LangGraph的运行时会自动将这些追加操作合并起来形成最终的状态。这种设计非常契合对话场景。例如user_node节点接收用户输入追加一条HumanMessagellm_node节点读取最新的消息思考后追加一条AIMessagetool_node节点执行工具后追加一条ToolMessage。整个对话历史就自然地、有序地记录在了state[‘messages’]里。2.2 使用MessagesState构建一个简单聊天机器人让我们看一个具体的例子构建一个没有工具调用的基础聊天机器人。from langgraph.graph import StateGraph, END from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage, AIMessage, remove_image # 1. 定义状态直接使用MessagesState from langgraph.graph import MessagesState # 2. 定义图的工作流 workflow StateGraph(MessagesState) # 3. 定义节点函数 def call_model(state: MessagesState): 调用LLM生成回复 # 从状态中获取完整的对话历史 messages state[‘messages’] # 初始化模型 model ChatOpenAI(model“gpt-4”) # 调用模型传入所有历史消息 response model.invoke(messages) # 关键返回一个字典更新状态。这里会向messages列表追加AIMessage。 return {“messages”: [response]} # 4. 添加节点和边 workflow.add_node(“assistant”, call_model) workflow.set_entry_point(“assistant”) workflow.add_edge(“assistant”, END) # 5. 编译图 app workflow.compile() # 6. 运行 initial_state {“messages”: [HumanMessage(content“你好”)]} final_state app.invoke(initial_state) for msg in final_state[“messages”]: print(f“{msg.type}: {msg.content}”)这个例子虽然简单但揭示了使用预定义状态的核心模式你的节点函数只需要关心如何根据当前状态计算输出并以字典形式返回要更新的状态部分LangGraph会自动帮你完成状态的合并与传递。2.3 MessagesState的适用场景与局限适用场景纯对话机器人不需要外部工具仅进行多轮文本对话。简单工具调用工具调用的结果以ToolMessage形式返回并作为对话历史的一部分供LLM参考。这符合OpenAI的Function Calling或类似范式。对话历史记录你希望自动、完整地保存整个对话过程用于后续分析、检索或持久化。局限状态结构单一它强制你的整个Agent状态必须围绕messages列表来构建。如果你需要存储一些与消息序列无关的元数据比如当前对话的主题标签、用户的身份ID、本次会话的配置参数等MessagesState就无能为力了。归并策略固定messages字段强制使用operator.add进行归约。这在绝大多数对话场景下是合理的但如果你有更复杂的状态合并逻辑例如需要根据条件替换或删除某些中间消息就需要自定义状态。注意在使用MessagesState时一个常见的“坑”是忘记消息列表会不断增长。对于超长对话这可能导致上下文窗口溢出或处理速度下降。一个实用的技巧是在节点函数中可以只选取最近N条消息或进行摘要后再发送给LLM但更新状态时依然追加完整消息以保持历史记录。这需要你在节点逻辑内部做处理而不是依赖状态结构本身。3. AgentState为复杂智能体设计的全能状态当你的Agent需要超越简单的对话涉及更复杂的逻辑、多个工具协调、自定义中间数据存储时MessagesState就显得捉襟见肘了。这时就该AgentState登场了。AgentState是LangGraph中功能更全面、更灵活的一个预定义状态。3.1 AgentState的核心结构剖析AgentState也是一个TypedDict但它包含了多个预定义的字段旨在支持一个功能完整的ReActReasoning and Acting风格Agent。from typing import TypedDict, List, Annotated, Optional from langchain_core.messages import BaseMessage import operator class AgentState(TypedDict): # 核心输入信息 input: str # 核心对话历史消息 chat_history: List[BaseMessage] # 核心Agent运行过程中的中间消息如思考、工具调用、结果 messages: Annotated[List[BaseMessage], operator.add] # 核心下一个要执行的Action的名称 next: str # 辅助本次交互的唯一ID interaction_id: Optional[str]我们来逐一拆解这些字段的设计意图和用法input(str) 存储用户当前轮次的原始输入。这是一个普通的字符串字段没有设置归约器。这意味着它通常只在工作流开始时被设置一次后续节点可以读取它但一般不会去修改它。它清晰地分离了“本轮目标”和“历史对话”。chat_history(List[BaseMessage]) 存储历史对话记录。注意它和messages是分开的。一个常见的模式是在启动Agent时将过去的对话记录加载到chat_history中而messages则专门用于处理当前轮次交互中产生的新消息。这样可以避免历史对话和当前推理过程的消息混在一起便于管理。messages(Annotated[List[BaseMessage], operator.add]) 和MessagesState中的messages作用类似但它在这里特指当前交互周期内产生的消息序列。包括LLM的思考AIMessage、工具调用请求、工具返回结果等。它使用operator.add进行归约是工作流中各个节点通信和追加信息的主要通道。next(str) 这是实现条件路由的关键。节点函数在执行后可以通过返回{“next”: “tool_node_name”}来指定下一个要执行的节点。LangGraph的运行时根据这个字段的值来决定图的走向。这使得构建非线性的、动态的工作流成为可能例如根据工具执行结果决定是继续调用工具还是直接回复用户。interaction_id(Optional[str]) 一个辅助字段用于标识当前交互会话。在并发或持久化场景下非常有用可以将同一会话的状态关联起来。3.2 使用AgentState构建一个ReAct风格Agent下面我们构建一个经典的ReAct Agent它能够根据问题决定是否调用搜索工具。from langgraph.graph import StateGraph, END from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage, AIMessage, ToolMessage from langchain_community.tools.tavily_search import TavilySearchResults from typing import Literal # 1. 定义状态使用AgentState通常从langgraph.prebuilt导入 from langgraph.prebuilt import AgentState # 2. 初始化工具和模型 search_tool TavilySearchResults(max_results2) tools [search_tool] model ChatOpenAI(model“gpt-4”, temperature0).bind_tools(tools) # 3. 定义图和工作流 workflow StateGraph(AgentState) # 4. 定义节点函数 def agent_node(state: AgentState): Agent节点分析状态决定下一步行动。 # 准备给模型的上下文历史对话 当前轮次消息 all_messages state[‘chat_history’] state[‘messages’] # 调用模型 response model.invoke(all_messages) # 将模型的响应追加到messages中 new_messages state[‘messages’] [response] # 关键判断模型返回的是什么 if response.tool_calls: # 如果模型要求调用工具则设置下一个节点为‘tools’ next_node “tools” else: # 如果模型直接给出了最终回答则设置下一个节点为END结束流程 next_node END # 返回状态更新追加消息并指示下一步去向 return {“messages”: new_messages, “next”: next_node} def tools_node(state: AgentState): 工具执行节点执行模型请求的所有工具。 last_message state[‘messages’][-1] # 获取最后一条消息即包含tool_calls的AIMessage tool_calls last_message.tool_calls tool_messages [] for tool_call in tool_calls: tool_name tool_call[‘name’] tool_input tool_call[‘args’] # 找到对应的工具并执行 tool_to_use {t.name: t for t in tools}[tool_name] try: output tool_to_use.invoke(tool_input) except Exception as e: output f“调用工具{tool_name}时出错: {e}” # 为每个工具调用结果生成一条ToolMessage tool_messages.append(ToolMessage(contentstr(output), tool_call_idtool_call[‘id’])) # 将工具执行结果追加到messages中 new_messages state[‘messages’] tool_messages # 关键执行完工具后下一步永远是回到‘agent’节点让模型分析结果 return {“messages”: new_messages, “next”: “agent”} # 5. 添加节点和边 workflow.add_node(“agent”, agent_node) workflow.add_node(“tools”, tools_node) workflow.set_entry_point(“agent”) # 6. 定义条件边实际上通过‘next’字段实现这里用add_conditional_edges演示另一种方式 # 但更常见的ReAct模式是像上面节点函数那样通过返回的‘next’值来控制。 # 这里我们展示一个更简单的固定流程agent - (可能到tools可能到END) # 我们使用add_conditional_edges来实现基于‘next’的路由虽然上面在节点内设置了但图需要知道这个可能 # 实际上更简洁的做法是使用‘send’和‘next’字段我们调整一下设计。 # 让我们重构一下使用更经典的“工具调用后自动回到Agent”的模式通过条件边实现分支。 def route_after_agent(state: AgentState) - Literal[“tools”, END]: 路由函数检查最后一条消息是否有工具调用。 last_message state[‘messages’][-1] if hasattr(last_message, ‘tool_calls’) and last_message.tool_calls: return “tools” return END workflow.add_conditional_edges( “agent”, route_after_agent, {“tools”: “tools”, END: END} ) workflow.add_edge(“tools”, “agent”) # 工具执行完后无条件回到Agent节点 # 7. 编译图 app workflow.compile() # 8. 运行 initial_state AgentState( input“2024年巴黎奥运会的中国金牌数是多少”, chat_history[], # 没有历史对话 messages[HumanMessage(content“2024年巴黎奥运会的中国金牌数是多少”)], next“agent”, interaction_id“test_001” ) final_state app.invoke(initial_state) # 打印最终AI回复 for msg in final_state[‘messages’]: if isinstance(msg, AIMessage) and not msg.tool_calls: print(f“助理: {msg.content}”)这个例子比MessagesState的复杂得多它展示了状态分离input存放问题chat_history和messages分开管理。动态路由通过route_after_agent函数或节点返回的next字段实现条件判断决定是调用工具还是结束。循环执行agent - tools - agent - tools …的循环直到模型认为无需再调用工具给出最终答案。3.3 AgentState的威力与自定义扩展AgentState的强大在于它提供了一个即用型、功能完备的框架。但它的设计者也考虑到了灵活性。AgentState本身是一个TypedDict这意味着你可以很容易地继承并扩展它。假设你的Agent需要记录每次工具调用的耗时用于监控和优化。from typing import TypedDict, List, Annotated, Optional, Dict import operator from datetime import datetime from langchain_core.messages import BaseMessage class CustomAgentState(TypedDict): # 继承所有AgentState字段 input: str chat_history: List[BaseMessage] messages: Annotated[List[BaseMessage], operator.add] next: str interaction_id: Optional[str] # 自定义字段记录工具调用耗时 tool_metrics: Annotated[List[Dict], operator.add] # 在tools_node中你可以这样更新状态 def tools_node_with_metrics(state: CustomAgentState): last_message state[‘messages’][-1] tool_calls last_message.tool_calls tool_messages [] metrics [] for tool_call in tool_calls: tool_name tool_call[‘name’] start_time datetime.now() # … 执行工具 … end_time datetime.now() duration (end_time - start_time).total_seconds() metrics.append({“tool”: tool_name, “duration_seconds”: duration, “timestamp”: start_time.isoformat()}) # … 生成tool_messages … return { “messages”: tool_messages, “tool_metrics”: metrics, # 更新自定义字段 “next”: “agent” }通过这种方式你可以在不破坏原有工作流的基础上为状态添加任何需要的辅助信息。自定义字段的归约器如operator.add也需要根据你的合并逻辑来谨慎选择。4. MessagesState vs AgentState如何选择与实战心得看到这里你可能会问我到底该用哪一个选择的标准是什么选择MessagesState当你的工作流是线性的、以对话消息为唯一核心的。你不需要复杂的条件路由或者路由逻辑非常简单例如总是执行同一个节点序列。你希望代码尽可能简洁状态结构一目了然。你正在快速原型验证一个以对话为主的创意。选择AgentState当你正在构建一个标准的、功能完整的ReAct风格智能体。你的工作流需要基于LLM或工具的输出结果进行动态分支条件路由。你需要清晰地区分“历史背景”chat_history和“当前推理过程”messages。你预见到未来可能需要扩展状态存储更多元数据AgentState的结构为此预留了空间自定义也方便。我的实战心得与避坑指南状态初始化是关键无论是MessagesState还是AgentState在app.invoke(initial_state)时你必须提供所有必需字段的初始值。对于Annotated字段即使初始为空你也必须提供一个空列表[]否则会报错。一个常见的错误是只提供了messages却忘了AgentState还要求input和next。理解“归约器”的副作用operator.add对于列表是拼接。这意味着如果你在某个节点中错误地返回了{“messages”: [some_message]}它会被追加到列表末尾。但如果你返回{“messages”: some_new_list}它会被整个替换吗不会operator.add会尝试将some_new_list这个列表与原有列表相加拼接。如果你想“替换”列表需要使用不同的归约器比如lambda old, new: new。在预定义状态中messages字段被固定为追加模式这符合对话的直觉。如果你想修改中间某条消息就需要在节点函数内部处理整个列表逻辑。next字段的优先级在AgentState工作流中节点函数返回的next值会覆盖图结构中定义的边。这给了节点极大的控制权。但这也容易导致混乱。我的建议是明确约定一种路由控制方式。要么完全依靠节点返回的next字段此时图可以只用add_edge设置简单连接要么完全依靠add_conditional_edges定义的条件函数。混合使用容易让逻辑变得难以调试。调试可视化是神器LangGraph内置了可视化功能app.get_graph().draw_mermaid()。对于复杂的状态流转画出来看一眼比读代码要清晰十倍。特别是当你的next逻辑复杂时可视化能帮你快速发现路由错误。性能考量状态序列化LangGraph的状态在节点间传递时会被序列化和反序列化。如果你的状态对象非常庞大例如messages列表积累了上千条消息会影响性能。对于长对话应用考虑定期将messages归档到chat_history并清空messages或者使用外部的记忆存储如向量数据库来管理超长历史。5. 超越预定义何时需要自定义状态MessagesState和AgentState解决了80%的问题。那么剩下的20%是什么时候呢当你遇到以下情况时就需要考虑从头定义自己的状态类了领域特定状态你构建的不是一个通用对话Agent而是一个专门处理特定任务的“工作流引擎”。例如一个代码评审Agent的状态可能需要包含code_snippets代码片段列表、review_comments评审意见列表、severity_level当前问题严重等级等与代码评审强相关的字段。使用预定义状态反而会显得别扭。复杂的归并逻辑预定义状态使用简单的operator.add。但你的业务可能需要更复杂的合并策略。例如一个多人协作编辑的Agent状态中有一个document字段多个节点可能并发修改文档的不同部分你需要一个能合并差异diff的归约器而不是简单的追加或替换。极致的性能与精简如果你的Agent状态极其简单可能只有一个current_query字符串和一个results列表那么自定义一个只包含这两个字段的TypedDict比引入AgentState的所有字段更加轻量和高效。自定义状态并不复杂它就是定义一个TypedDict并为每个字段通过Annotated指定合适的归约器。这给了你最大的灵活性但也意味着你需要自己负责状态结构的设计和归并逻辑的正确性。6. 状态管理的最佳实践与模式总结经过多个项目的实践我总结出以下几点在LangGraph中使用状态的最佳实践模式一读写分离在节点函数中尽量遵循“读取所需返回所变”的原则。只读取状态中你真正需要的字段只返回你确实修改了的字段。这能让节点逻辑更清晰也避免了意外覆盖。模式二使用辅助函数处理复杂状态逻辑如果某个节点更新状态的逻辑非常复杂不要把所有代码都堆在节点函数里。将其抽离成独立的辅助函数节点函数只负责调用和返回。这大大提升了可测试性和可维护性。def complex_state_update(old_state): # … 一系列复杂的计算和逻辑 … return new_state_partial def my_node(state): update complex_state_update(state) return update模式三为状态变更添加日志在开发调试阶段可以在节点函数的开头和结尾打印状态的关键部分或者使用LangSmith等工具进行跟踪。理解状态是如何一步步演变的是调试LangGraph应用最有效的方法。模式四利用Pydantic进行状态验证进阶虽然TypedDict轻便但它缺乏运行时类型验证。对于生产级应用可以考虑使用Pydantic的BaseModel来定义状态并利用其强大的数据验证功能。不过这需要你更深入地理解LangGraph如何与Pydantic模型协作。回到我们最初的标题“预定义状态”的价值就在于它们不是束缚你的枷锁而是为你铺好的轨道。MessagesState和AgentState就像乐高积木中的标准件能让你快速搭建出坚固可靠的主体结构。当你需要建造更奇特、更个性化的部分时你也完全有能力自己铸造新的积木自定义状态。掌握它们你就掌握了LangGraph驱动智能体流畅运转的核心密码。下次当你开始一个新的LangGraph项目时不妨先停下来想一想我的Agent核心是什么是简单的对话还是复杂的推理与行动想清楚了这个问题MessagesState和AgentState之间的选择答案自然就清晰了。