LangGraph状态管理：MessagesState与AgentState的实战解析

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节点接收用户输入，追加一条HumanMessage；llm_node节点读取最新的消息，思考后追加一条AIMessage；tool_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的适用场景与局限

适用场景：

1. 纯对话机器人：不需要外部工具，仅进行多轮文本对话。
2. 简单工具调用：工具调用的结果以ToolMessage形式返回，并作为对话历史的一部分供LLM参考。这符合OpenAI的Function Calling或类似范式。
3. 对话历史记录：你希望自动、完整地保存整个对话过程，用于后续分析、检索或持久化。

局限：

1. 状态结构单一：它强制你的整个Agent状态必须围绕messages列表来构建。如果你需要存储一些与消息序列无关的元数据，比如当前对话的主题标签、用户的身份ID、本次会话的配置参数等，MessagesState就无能为力了。
2. 归并策略固定：messages字段强制使用operator.add进行归约。这在绝大多数对话场景下是合理的，但如果你有更复杂的状态合并逻辑（例如，需要根据条件替换或删除某些中间消息），就需要自定义状态。

注意：在使用MessagesState时，一个常见的“坑”是忘记消息列表会不断增长。对于超长对话，这可能导致上下文窗口溢出或处理速度下降。一个实用的技巧是在节点函数中，可以只选取最近N条消息或进行摘要后再发送给LLM，但更新状态时依然追加完整消息以保持历史记录。这需要你在节点逻辑内部做处理，而不是依赖状态结构本身。

3. AgentState：为复杂智能体设计的全能状态

当你的Agent需要超越简单的对话，涉及更复杂的逻辑、多个工具协调、自定义中间数据存储时，MessagesState就显得捉襟见肘了。这时，就该AgentState登场了。AgentState是LangGraph中功能更全面、更灵活的一个预定义状态。

3.1 AgentState的核心结构剖析

AgentState也是一个TypedDict，但它包含了多个预定义的字段，旨在支持一个功能完整的ReAct（Reasoning 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]

我们来逐一拆解这些字段的设计意图和用法：

1. input(str)： 存储用户当前轮次的原始输入。这是一个普通的字符串字段，没有设置归约器。这意味着它通常只在工作流开始时被设置一次，后续节点可以读取它，但一般不会去修改它。它清晰地分离了“本轮目标”和“历史对话”。

2. chat_history(List[BaseMessage])： 存储历史对话记录。注意，它和messages是分开的。一个常见的模式是，在启动Agent时，将过去的对话记录加载到chat_history中，而messages则专门用于处理当前轮次交互中产生的新消息。这样可以避免历史对话和当前推理过程的消息混在一起，便于管理。

3. messages(Annotated[List[BaseMessage], operator.add])： 和MessagesState中的messages作用类似，但它在这里特指当前交互周期内产生的消息序列。包括LLM的思考（AIMessage）、工具调用请求、工具返回结果等。它使用operator.add进行归约，是工作流中各个节点通信和追加信息的主要通道。

4. next(str)： 这是实现条件路由的关键。节点函数在执行后，可以通过返回{“next”: “tool_node_name”}来指定下一个要执行的节点。LangGraph的运行时根据这个字段的值来决定图的走向。这使得构建非线性的、动态的工作流成为可能，例如，根据工具执行结果决定是继续调用工具还是直接回复用户。

5. 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_results=2) tools = [search_tool] model = ChatOpenAI(model=“gpt-4”, temperature=0).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(content=str(output), tool_call_id=tool_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的结构为此预留了空间，自定义也方便）。

我的实战心得与避坑指南：

1. 状态初始化是关键：无论是MessagesState还是AgentState，在app.invoke(initial_state)时，你必须提供所有必需字段的初始值。对于Annotated字段，即使初始为空，你也必须提供一个空列表[]，否则会报错。一个常见的错误是只提供了messages，却忘了AgentState还要求input和next。

2. 理解“归约器”的副作用：operator.add对于列表是拼接。这意味着如果你在某个节点中错误地返回了{“messages”: [some_message]}，它会被追加到列表末尾。但如果你返回{“messages”: some_new_list}，它会被整个替换吗？不会，operator.add会尝试将some_new_list这个列表与原有列表相加（拼接）。如果你想“替换”列表，需要使用不同的归约器，比如lambda old, new: new。在预定义状态中，messages字段被固定为追加模式，这符合对话的直觉。如果你想修改中间某条消息，就需要在节点函数内部处理整个列表逻辑。

3. next字段的优先级：在AgentState工作流中，节点函数返回的next值会覆盖图结构中定义的边。这给了节点极大的控制权。但这也容易导致混乱。我的建议是：明确约定一种路由控制方式。要么完全依靠节点返回的next字段（此时图可以只用add_edge设置简单连接），要么完全依靠add_conditional_edges定义的条件函数。混合使用容易让逻辑变得难以调试。

4. 调试可视化是神器：LangGraph内置了可视化功能（app.get_graph().draw_mermaid()）。对于复杂的状态流转，画出来看一眼，比读代码要清晰十倍。特别是当你的next逻辑复杂时，可视化能帮你快速发现路由错误。

5. 性能考量：状态序列化：LangGraph的状态在节点间传递时会被序列化和反序列化。如果你的状态对象非常庞大（例如messages列表积累了上千条消息），会影响性能。对于长对话应用，考虑定期将messages归档到chat_history并清空messages，或者使用外部的记忆存储（如向量数据库）来管理超长历史。

5. 超越预定义：何时需要自定义状态

MessagesState和AgentState解决了80%的问题。那么剩下的20%是什么时候呢？当你遇到以下情况时，就需要考虑从头定义自己的状态类了：

1. 领域特定状态：你构建的不是一个通用对话Agent，而是一个专门处理特定任务的“工作流引擎”。例如，一个代码评审Agent的状态可能需要包含code_snippets（代码片段列表）、review_comments（评审意见列表）、severity_level（当前问题严重等级）等与代码评审强相关的字段。使用预定义状态反而会显得别扭。

2. 复杂的归并逻辑：预定义状态使用简单的operator.add。但你的业务可能需要更复杂的合并策略。例如，一个多人协作编辑的Agent，状态中有一个document字段，多个节点可能并发修改文档的不同部分，你需要一个能合并差异（diff）的归约器，而不是简单的追加或替换。

3. 极致的性能与精简：如果你的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之间的选择，答案自然就清晰了。