LangGraph 中的 add_messages

有两种导入方式：

• from operator import add as add_messages
• from langgraph.graph.message import add_messages

from operator import add as add_messages

这是 Python 标准库operator模块中的add函数，它只是一个通用的加法运算。

from operator import add # 普通加法 result = add(3, 5) # 结果是 8 # 列表拼接 list1 = [1, 2] list2 = [3, 4] result = add(list1, list2) # 结果是 [1, 2, 3, 4] # 字符串拼接 result = add("Hello ", "World!") # 结果是 "Hello World!"

在 LangGraph 中的使用场景：

• 通常用于简单状态更新，比如在State中定义 reducer

from typing import List, TypedDict from operator import add as add_messages class State(TypedDict): messages: List[str] # 使用 operator.add 来合并列表 # 当多个节点返回 {"messages": ["新消息"]} 时，它们会被简单地拼接在一起

from langgraph.graph.message import add_messages

这是 LangGraph 专门为消息管理设计的reducer 函数，它处理的是BaseMessage对象（如HumanMessage,AIMessage等）。

补充：什么是 reducer 函数？简单来说就是状态合并规则

• 在 LangGraph 里，多个不同的节点（node）都可能会修改同一个状态字段（比如往messages列表里加新消息）。当这些修改同时发生时，reducer 函数的作用就是决定以什么方式把这些修改合并到一起，形成最终的新状态
• LangGraph 的状态是跨节点共享的。节点 A 返回{"messages": [msg1]}，节点 B 返回{"messages": [msg2]}。如果没有 reducer，后返回的msg2会把msg1覆盖，导致消息丢失
• 通过Annotated为状态字段指定一个 reducer，告诉 LangGraph：当多个节点都返回这个字段的新值时，用这个指定的函数来合并它们

from langgraph.graph.message import add_messages from langchain_core.messages import HumanMessage, AIMessage # 专门处理消息列表 msg1 = HumanMessage(content="Hello") msg2 = AIMessage(content="Hi there!") # 智能合并消息 result = add_messages([msg1], [msg2]) # 结果是一个包含两个消息对象的列表

在 LangGraph 中的使用场景：

• 用于消息状态管理，能智能处理消息的添加、更新和去重

from typing import Annotated, Sequence, TypedDict from langgraph.graph.message import add_messages from langchain_core.messages import BaseMessage class State(TypedDict): messages: Annotated[Sequence[BaseMessage], add_messages] # 使用 add_messages 作为 reducer

对比

一个简单的示例：

• 使用operator.add（简单拼接）

from operator import add as add_messages from typing import List, TypedDict class State(TypedDict): messages: List[str] # 简单字符串列表 # 节点1 返回 node1_output = {"messages": ["用户说: 你好"]} # 节点2 返回 node2_output = {"messages": ["助手说: 你好！有什么可以帮助你？"]} # 状态合并后： # messages = ["用户说: 你好", "助手说: 你好！有什么可以帮助你？"] # 只是简单拼接

• 使用langgraph.graph.message.add_messages（智能合并）

from langgraph.graph.message import add_messages from typing import Annotated, Sequence, TypedDict from langchain_core.messages import BaseMessage, HumanMessage, AIMessage class State(TypedDict): messages: Annotated[Sequence[BaseMessage], add_messages] # 节点1 返回 node1_output = { "messages": [ HumanMessage(content="你好", id="msg1"), AIMessage(content="你好！", id="msg2") ] } # 节点2 尝试更新 msg2 node2_output = { "messages": [ AIMessage(content="你好！有什么可以帮助你？", id="msg2") # 相同 ID ] } # 合并后，msg2 会被替换为新的内容，而不是追加 # 最终 messages = [HumanMessage("你好", id="msg1"), # AIMessage("你好！有什么可以帮助你？", id="msg2")]