LangGraph 中的 MessagesState

LangGraph 中的MessagesState（注意官方名称是复数MessagesState，不是MessageState）是整个框架专为对话和 Agent 场景预制的内置状态类型。它的核心价值就一句话：用一个带add_messagesreducer 的messages字段，把"多轮对话历史如何累积"这件最烦的事接管掉。

一、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. 追加（Append）：ID 不同 → 加到末尾

from langchain_core.messages import HumanMessage, AIMessage from langgraph.graph.message import add_messages old = [HumanMessage(content="Hello", id="1")] new = [AIMessage(content="Hi there!", id="2")] result = add_messages(old, new) # [HumanMessage(id='1'), AIMessage(id='2')] ← 两条都在

2. 更新（Replace）：ID 相同 → 原地替换

old = [AIMessage(content="你好", id="msg_001")] new = [AIMessage(content="你好，有什么可以帮你？", id="msg_001")] result = add_messages(old, new) # 长度仍是 1，内容被新消息替换 # [AIMessage(content="你好，有什么可以帮你？", id="msg_001")]

⚠️ 这个特性在流式输出场景中至关重要：AI 先返回一个占位消息（如id="stream_1"），token 逐步到达时不断用相同 ID回写add_messages，从而"更新"而非"堆叠"消息。

3. 删除（Delete）：使用RemoveMessage

from langchain_core.messages import RemoveMessage # 从状态中移除指定 ID 的消息 result = add_messages(old, [RemoveMessage(id="msg_001")])

这在对历史做裁剪、或者 human-in-the-loop 修正对话时非常有用。

附加能力：反序列化

add_messages还能把字典自动反序列化成 LangChain 消息对象。所以以下两种写法都合法：

{"messages": [HumanMessage(content="hi")]} # ✅ 对象形式 {"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(content=f"你说了：{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调用的多轮记忆，需要配合Checkpointer：

from 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(checkpointer=MemorySaver()) # 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, n=10)]

方案 2：用 RemoveMessage 主动删除

from langchain_core.messages import RemoveMessage def trim_node(state): # 删除最早的消息 oldest_id = state["messages"][0].id return {"messages": [RemoveMessage(id=oldest_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(id=m.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(checkpointer=MemorySaver()) # 使用 config = {"configurable": {"thread_id": "prod_session_001"}} result = app.invoke({ "messages": [("user", "帮我分析 Q3 销售数据")], "user_id": "u_12345" }, config)

🎯 核心要点回顾

MessagesState =TypedDict+messages: Annotated[list[AnyMessage], add_messages]

add_messages 是智能 reducer：ID 不同→追加，ID 相同→更新，RemoveMessage→删除

节点只返回增量，框架负责合并

多轮记忆必须配 Checkpointer + thread_id

生产环境必须考虑消息裁剪，否则上下文无限膨胀

多 Agent 协作可通过不同消息键实现通道隔离