langgraph的 MessagesState 解读

MessagesState是 LangGraph 预置的状态类专门为对话历史 工具调用这类场景设计。它的本质就是一个内置了messages字段、且该字段用add_messages作为 reducer 的TypedDict# MessagesState 在源码中大致等价于classMessagesState(TypedDict):messages:Annotated[list[AnyMessage],add_messages]文章目录一、最基本用法作为 StateGraph 的状态二、add_messages 的三个核心行为三、继承 MessagesState 扩展自定义字段四、多轮对话 记忆持久化五、⚠️ 三个最容易踩的坑这意味着你不用自己写Annotated[...]那串样板直接拿来用就行 。下面把关键用法拆开讲。一、最基本用法作为 StateGraph 的状态fromlanggraph.graphimportStateGraph,START,END,MessagesStatefromlangchain_core.messagesimportHumanMessage,AIMessagedefchatbot_node(state:MessagesState):# 节点内通过 state[messages] 读取完整历史last_user_msgstate[messages][-1]replyf你说了{last_user_msg.content}# 只返回新增的部分框架会用 add_messages 合并return{messages:[AIMessage(contentreply)]}builderStateGraph(MessagesState)builder.add_node(chatbot,chatbot_node)builder.add_edge(START,chatbot)builder.add_edge(chatbot,END)graphbuilder.compile()resultgraph.invoke({messages:[HumanMessage(content你好)]}) 注意MessagesState是状态 schema不是普通的 class 实例。不要写成MessagesState(messages[...])而是在调用图时传{messages: [...]}或继承它来扩展字段 。二、add_messages 的三个核心行为add_messages这个 reducer 决定了messages字段怎么合并它比简单的list list聪明得多 追加新消息没有 id 或 id 不匹配时# 旧: [msg1, msg2] 节点返回 [msg3]# 结果: [msg1, msg2, msg3]按 id 替换已有消息流式输出场景必备fromlangchain_core.messagesimportAIMessagefromlanggraph.graph.messageimportadd_messages existing[AIMessage(content它是5,idmsg-2)]correction[AIMessage(content它是4,idmsg-2)]# 同 idresultadd_messages(existing,correction)# 结果: [AIMessage(content它是4, idmsg-2)]# 长度还是 1没有重复如果换成operator.add错误答案和正确答案会并排在列表里 。用RemoveMessage删除消息裁剪历史以控制 tokenfromlangchain_core.messagesimportRemoveMessage# 节点返回这个id 对应的消息会从列表中移除return{messages:[RemoveMessage(idmsg-2)]}三、继承 MessagesState 扩展自定义字段如果你的 Agent 还需要追踪其他状态当前工具、迭代次数、用户信息等直接继承fromlanggraph.graphimportMessagesStateclassAgentState(MessagesState):current_tool:striteration:intuser_info:dictdefchatbot(state:AgentState):# messages 字段依然享受 add_messages 的全部能力replyAIMessage(content处理中...)return{messages:[reply],current_tool:search,iteration:state[iteration]1}这是多智能体协作、RAG、带业务上下文的对话 Agent 最常用的模式 。四、多轮对话 记忆持久化配合MemorySaver或SqliteSaver做 checkpoint用thread_id区分不同会话fromlanggraph.checkpoint.memoryimportMemorySaver checkpointerMemorySaver()graphbuilder.compile(checkpointercheckpointer)config{configurable:{thread_id:conversation-1}}# 第一轮graph.stream({messages:[{role:user,content:广州天气怎么样}]},config,stream_modevalues)# 第二轮不需要手动拼历史框架自动从 checkpoint 恢复graph.stream({messages:[{role:user,content:那明天呢}]},config,stream_modevalues)messages字段会自动累积thread_id不同的会话互不干扰 。五、⚠️ 三个最容易踩的坑坑 1原地修改state[messages]# ❌ 错误直接 append 会破坏 LangGraph 的不可变更新契约defbad_node(state):state[messages].append(AIMessage(contenthi))return{}# ✅ 正确返回增量让 reducer 合并defgood_node(state):return{messages:[AIMessage(contenthi)]}原地修改会导致浅拷贝下多节点共享引用引发竞态丢失 。坑 2自定义 TypedDict 忘了加add_messagesreducer# ❌ 错误没有 reducer新值会直接覆盖整个列表历史全丢classBadState(TypedDict):messages:list[AnyMessage]# ✅ 正确classGoodState(TypedDict):messages:Annotated[list[AnyMessage],add_messages]这也正是为什么直接用MessagesState更安全 。坑 3条件分支里有的路径返回messages、有的不返回如果route_to_tool分支里工具路径返回了{messages: [...]}而直接响应路径只返回{messages: [ai_msg]}会造成历史断裂。确保所有分支都按统一方式返回消息增量 。总的来说MessagesState就是 LangGraph 给你的一把对话状态瑞士军刀开箱即用管理消息历史、自动处理追加/更新/删除、还能继承扩展。只要是涉及聊天、工具调用、多轮对话的 Agent直接用它就对了只有当你需要非常特殊的合并逻辑比如滑动窗口只保留最近 N 条时才需要回到自定义的TypedDict 自定义 reducer 。如果你是在具体的 Agent 架构里用比如 ReAct Agent、多智能体协作、或者带 RAG 的对话告诉我场景我可以给到更贴合的代码示例。