有两种导入方式from operator import add as add_messagesfrom langgraph.graph.message import add_messagesfrom 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中定义 reducerfrom 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(contentHello) msg2 AIMessage(contentHi 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你好, idmsg1), AIMessage(content你好, idmsg2) ] } # 节点2 尝试更新 msg2 node2_output { messages: [ AIMessage(content你好有什么可以帮助你, idmsg2) # 相同 ID ] } # 合并后msg2 会被替换为新的内容而不是追加 # 最终 messages [HumanMessage(你好, idmsg1), # AIMessage(你好有什么可以帮助你, idmsg2)]