LangGraph 中的 MessagesState

LangGraph 中的MessagesState注意官方名称是复数MessagesState不是MessageState是整个框架专为对话和 Agent 场景预制的内置状态类型。它的核心价值就一句话用一个带add_messagesreducer 的messages字段把多轮对话历史如何累积这件最烦的事接管掉。文章目录一、MessagesState 到底是什么二、add_messages 的三种合并语义1. 追加AppendID 不同 → 加到末尾2. 更新ReplaceID 相同 → 原地替换3. 删除Delete使用 RemoveMessage附加能力反序列化三、在图中使用 MessagesState基础用法直接使用进阶用法继承 MessagesState 扩展字段四、多轮对话 持久化五、生产级必考虑的上下文裁剪方案 1自定义 Reducer 做滑动窗口方案 2用 RemoveMessage 主动删除方案 3配合 LangGraph 官方的消息裁剪工具六、多智能体场景消息通道隔离七、常见坑与排查坑 1以为 invoke 自动记忆坑 2用了 operator.add 而非 add_messages坑 3消息 ID 重复坑 4忘记消息是累积的八、完整生产级模板 核心要点回顾下面从原理到实战完整讲透。一、MessagesState 到底是什么它本质上是一个TypedDict只有一个字段messages类型注解为Annotated[list[AnyMessage], add_messages]。# 等价于以下定义来自官方源码 class MessagesState(TypedDict): messages: Annotated[list[AnyMessage], add_messages]这里有两个关键点list[AnyMessage]消息列表可以装HumanMessage、AIMessage、SystemMessage、ToolMessage等所有 LangChain 消息对象add_messagesreducer告诉 LangGraph 当节点返回新的messages时如何合并到现有状态而不是简单替换 如果你不写 reducer框架默认是覆盖语义——节点返回{messages: [msg3]}历史里的 msg1、msg2 就全丢了。这正是多轮对话的致命坑。add_messages就是为解决这个而生。二、add_messages 的三种合并语义这是MessagesState的灵魂。它不只是追加而是按消息 ID 智能合并1. 追加AppendID 不同 → 加到末尾from langchain_core.messages import HumanMessage, AIMessage from langgraph.graph.message import add_messages old [HumanMessage(contentHello, id1)] new [AIMessage(contentHi there!, id2)] result add_messages(old, new) # [HumanMessage(id1), AIMessage(id2)] ← 两条都在2. 更新ReplaceID 相同 → 原地替换old [AIMessage(content你好, idmsg_001)] new [AIMessage(content你好有什么可以帮你, idmsg_001)] result add_messages(old, new) # 长度仍是 1内容被新消息替换 # [AIMessage(content你好有什么可以帮你, idmsg_001)]⚠️ 这个特性在流式输出场景中至关重要AI 先返回一个占位消息如idstream_1token 逐步到达时不断用相同 ID回写add_messages从而更新而非堆叠消息。3. 删除Delete使用RemoveMessagefrom langchain_core.messages import RemoveMessage # 从状态中移除指定 ID 的消息 result add_messages(old, [RemoveMessage(idmsg_001)])这在对历史做裁剪、或者 human-in-the-loop 修正对话时非常有用。附加能力反序列化add_messages还能把字典自动反序列化成 LangChain 消息对象。所以以下两种写法都合法{messages: [HumanMessage(contenthi)]} # ✅ 对象形式 {messages: [{type: human, content: hi}]} # ✅ 字典形式自动转成 HumanMessage三、在图中使用 MessagesState基础用法直接使用from langgraph.graph import StateGraph, MessagesState, START, END from langchain_core.messages import HumanMessage, AIMessage def chat_node(state: MessagesState): # state[messages] 是整个历史 last_user_msg state[messages][-1].content # 节点只需要返回增量 return {messages: [AIMessage(contentf你说了{last_user_msg})]} graph StateGraph(MessagesState) graph.add_node(chat, chat_node) graph.add_edge(START, chat) graph.add_edge(chat, END) app graph.compile() result app.invoke({messages: [(user, 今天天气怎么样)]}) print(result[messages][-1].content) # 你说了今天天气怎么样核心规律节点函数只需要返回变化的部分这里是新增的 AI 消息LangGraph 会通过add_messages自动把它追加到state[messages]历史末尾。进阶用法继承 MessagesState 扩展字段实际项目中你往往需要更多状态字段用户信息、迭代次数、检索到的文档等from langgraph.graph import MessagesState from typing import Optional class AgentState(MessagesState): 继承 MessagesState自动拥有 messages 字段 user_id: Optional[str] None current_tool: str iteration: int 0 documents: list[str] []继承后messages字段原样继承reducer 仍是add_messages新字段如果没有标注 reducer默认是覆盖语义新字段也可以自己加 reducer比如iteration: Annotated[int, operator.add]实现累加四、多轮对话 持久化MessagesState本身只是内存里的记事本要实现跨invoke调用的多轮记忆需要配合Checkpointerfrom langgraph.checkpoint.memory import MemorySaver # 1. 定义状态 class ChatState(MessagesState): pass # 2. 构建图时挂载检查点 graph StateGraph(ChatState) graph.add_node(chat, chat_node) graph.add_edge(START, chat) graph.add_edge(chat, END) app graph.compile(checkpointerMemorySaver()) # 3. 用 thread_id 区分不同会话 config {configurable: {thread_id: user_123_session_abc}} # 第一轮 app.invoke({messages: [(user, 我叫小明)]}, config) # 第二轮同一个 thread_id历史自动从 Checkpointer 恢复 app.invoke({messages: [(user, 我刚才说我叫什么)]}, config) # AI 能正确回答你叫小明State 是单次运行的临时记事本Checkpointer 是跨调用的存档柜靠thread_id隔离不同会话。五、生产级必考虑的上下文裁剪MessagesState的add_messages永远追加意味着消息列表会无限增长。生产环境中必须主动裁剪方案 1自定义 Reducer 做滑动窗口from typing import Annotated, TypedDict def keep_last_n(current: list, new: list, n: int 10) - list: 只保留最近 N 条消息 combined current new return combined[-n:] class BoundedState(MessagesState): # 重写 messages 字段的 reducer messages: Annotated[list, lambda c, n: keep_last_n(c, n, n10)]方案 2用 RemoveMessage 主动删除from langchain_core.messages import RemoveMessage def trim_node(state): # 删除最早的消息 oldest_id state[messages][0].id return {messages: [RemoveMessage(idoldest_id)]}方案 3配合 LangGraph 官方的消息裁剪工具对于超长对话可以定期把旧消息摘要化只保留最近的 K 条原文 一个历史摘要消息。六、多智能体场景消息通道隔离在多 Agent 系统中默认所有 Agent 共享同一个messages列表。如果你不希望子 Agent 的内部历史污染主对话可以用不同的消息键from typing_extensions import TypedDict, Annotated from langchain.messages import AnyMessage from langgraph.graph import StateGraph, add_messages class AliceState(TypedDict): alice_messages: Annotated[list[AnyMessage], add_messages] # Alice 用自己的 alice_messages 通道 alice ( StateGraph(AliceState) .add_node(model, ...) .compile() ) # 父图调用时用 wrapper 转换 def call_alice(state: SwarmState): response alice.invoke({alice_messages: state[messages]}) return {messages: response[alice_messages]}这是 LangGraph 官方 Multi-Agent Swarm 推荐的做法每个 Agent 使用独立消息通道通过 wrapper 函数做父子图状态转换。七、常见坑与排查坑 1以为invoke自动记忆# ❌ 错误每次 invoke 都是全新状态 app.invoke({messages: [(user, 你好)]}) app.invoke({messages: [(user, 还记得我吗)]}) # AI 不记得上一句 # ✅ 正确必须配合 Checkpointer thread_id app.invoke({messages: [...]}, config)坑 2用了operator.add而非add_messages# ❌ 错误operator.add 只会盲目追加无法更新相同 ID 的消息 messages: Annotated[list, operator.add] # ✅ 正确add_messages 能正确处理更新语义 messages: Annotated[list[AnyMessage], add_messages]流式输出场景下operator.add会导致同一个流式消息被叠加成多条。坑 3消息 ID 重复如果手动构造消息时复用 ID会导致意外覆盖。让 LangChain 自动生成 ID 通常是更安全的选择。坑 4忘记消息是累积的新手常写出这样的代码def node(state): # ❌ 错误把历史和新消息拼在一起返回 return {messages: state[messages] [new_msg]} # 这样 add_messages 会把整个历史再追加一次 → 消息翻倍 # ✅ 正确只返回新增的消息 def node(state): return {messages: [new_msg]}八、完整生产级模板from typing import Optional, Annotated from langgraph.graph import StateGraph, MessagesState, START, END from langgraph.checkpoint.memory import MemorySaver from langchain_core.messages import HumanMessage, AIMessage, RemoveMessage from operator import add class ProductionChatState(MessagesState): user_id: Optional[str] None iteration: Annotated[int, add] 0 # 累加器 tool_calls: list[str] [] def chatbot(state: ProductionChatState): # 1. 读取历史 messages state[messages] # 2. 调用 LLM伪代码 response llm.invoke(messages) # 3. 返回增量 return { messages: [response], iteration: 1, # 通过 reducer 累加 tool_calls: extract_tool_names(response) } def trim_history(state: ProductionChatState): 保持消息列表不超过 20 条 if len(state[messages]) 20: to_remove state[messages][:-20] return { messages: [RemoveMessage(idm.id) for m in to_remove] } return {} # 构建图 graph StateGraph(ProductionChatState) graph.add_node(chat, chatbot) graph.add_node(trim, trim_history) graph.add_edge(START, chat) graph.add_edge(chat, trim) graph.add_edge(trim, END) app graph.compile(checkpointerMemorySaver()) # 使用 config {configurable: {thread_id: prod_session_001}} result app.invoke({ messages: [(user, 帮我分析 Q3 销售数据)], user_id: u_12345 }, config) 核心要点回顾MessagesState TypedDictmessages: Annotated[list[AnyMessage], add_messages]add_messages 是智能 reducerID 不同→追加ID 相同→更新RemoveMessage→删除节点只返回增量框架负责合并多轮记忆必须配 Checkpointer thread_id生产环境必须考虑消息裁剪否则上下文无限膨胀多 Agent 协作可通过不同消息键实现通道隔离