本文是对 LangGraph 中 langgraph.graph.add_messages 函数的详细介绍，涵盖其定义、功能、参数、使用场景、代码示例以及注意事项。

---

1. add_messages 函数的概述

langgraph.graph.add_messages 是 LangGraph 提供的一个实用函数，位于 langgraph.graph 模块中，用于处理和更新消息列表（List[BaseMessage]），特别适用于基于消息的工作流，例如对话系统或 MessageGraph。它的主要作用是：

• 合并消息：将新的消息（来自节点输出或用户输入）追加或合并到现有的消息列表中。
• 处理消息更新：支持消息的覆盖、追加或条件更新，基于消息的 id 或其他逻辑。
• 简化状态管理：在 StateGraph 或 MessageGraph 中，作为状态更新函数，确保消息列表的一致性和正确性。

add_messages 通常用作 StateGraph 中状态注解的一部分，指定如何更新状态中的消息字段，或者直接在节点函数中调用以处理消息。

核心特点

• 消息类型：处理 langchain_core.messages.BaseMessage 的子类（如 HumanMessage、AIMessage、ToolMessage）。
• 灵活性：支持追加新消息、覆盖同 ID 消息或自定义更新逻辑。
• 与 LangGraph 集成：专为 MessageGraph 或 StateGraph 的消息字段设计，与对话工作流无缝配合。

典型应用场景

• 在 MessageGraph 中，管理对话历史，追加用户输入和模型回复。
• 在 StateGraph 中，更新状态中的消息字段（如 messages: List[BaseMessage]）。
• 处理工具调用结果，合并 ToolMessage 到消息列表。

---

2. add_messages 函数的定义

2.1 函数签名

根据 LangGraph 文档，add_messages 的定义如下：

from langgraph.graph import add_messages
from typing import Union, List, Optional
from langchain_core.messages import BaseMessage
from langchain_core.runnables import RunnableConfig

def add_messages(
    left: Union[List[BaseMessage], BaseMessage, str, None],
    right: Union[List[BaseMessage], BaseMessage, str, None],
    config: Optional[RunnableConfig] = None
) -> List[BaseMessage]:
    ...

2.2 参数

• left：

  • 类型：Union[List[BaseMessage], BaseMessage, str, None]
  • 描述：当前状态中的消息，通常是状态中的消息列表（List[BaseMessage]）或单个消息。
  • 用途：表示现有的消息历史或初始状态。
  • 示例：[HumanMessage(content="Hello")] 或 None（空状态）。

• right：

  • 类型：Union[List[BaseMessage], BaseMessage, str, None]
  • 描述：新传入的消息，通常是节点函数的输出或用户输入。
  • 用途：表示需要追加或合并到 left 的消息。
  • 示例：AIMessage(content="Hi!") 或 "New message"（自动转换为 HumanMessage）。

• config：

  • 类型：Optional[RunnableConfig]
  • 描述：运行时配置，包含 configurable 字段（如 thread_id、user_id），通常由 LangGraph 自动传递。
  • 用途：支持上下文相关的消息处理（目前主要用于扩展，未广泛使用）。
  • 示例：{"configurable": {"user_id": "user123"}}。

2.3 返回值

• 类型：List[BaseMessage]
• 描述：更新后的消息列表，包含 left 和 right 合并后的结果。
• 行为：
  • 如果 left 是 List[BaseMessage]，right 会被追加或合并。
  • 如果 right 是 str，自动转换为 HumanMessage。
  • 如果 left 或 right 是 None，返回适当的默认值（空列表或单个消息的列表）。

2.4 合并逻辑

add_messages 的核心逻辑是基于消息的 id 和类型进行合并：

• 追加：如果 right 是一个新消息（无重复 id），将其追加到 left。
• 覆盖：如果 right 的消息 id 与 left 中的某个消息相同，right 会覆盖对应的消息。
• 类型转换：如果 right 是字符串，转换为 HumanMessage。
• 空值处理：
  • 如果 left 是 None 或空列表，返回 right（转换为列表）。
  • 如果 right 是 None，返回 left（保持不变）。

---

3. add_messages 的功能和行为

3.1 基本功能

• 追加消息：将新消息添加到现有消息列表，保持对话历史。
• 覆盖消息：支持基于 id 的消息更新，例如更新工具调用的结果。
• 类型规范化：自动处理 str、单个 BaseMessage 或 List[BaseMessage]，确保输出是 List[BaseMessage]。

3.2 特殊行为

• 字符串处理：如果 right 是字符串，转换为 HumanMessage 并追加。
  • 示例：add_messages([], "Hello") 返回 [HumanMessage(content="Hello")].
• 消息 ID：如果消息有 id（例如 ToolMessage 或 LLM 生成的消息），add_messages 会检查 id 是否匹配，决定覆盖还是追加。
• 空输入：
  • add_messages(None, None) 返回 []。
  • add_messages([], AIMessage(content="Hi")) 返回 [AIMessage(content="Hi")]。

3.3 与状态注解的结合

在 StateGraph 中，add_messages 常用于状态注解，指定某个字段（如 messages）的更新方式。通过 Annotated 类型，告诉 LangGraph 如何合并节点输出的消息。

示例：

from typing import TypedDict, Annotated
from langgraph.graph import add_messages

class State(TypedDict):
    messages: Annotated[List[BaseMessage], add_messages]

这里，messages 字段使用 add_messages 作为更新函数，节点返回的消息会自动通过 add_messages 合并到 messages 列表。

---

4. 使用场景

add_messages 在以下场景中非常有用：

1. 对话系统：

   • 在 MessageGraph 或 StateGraph 中，管理用户输入和模型回复的对话历史。
   • 示例：追加用户消息和 LLM 回复。

2. 工具调用：

   • 处理工具调用的结果（如 ToolMessage），合并到消息列表。
   • 示例：LLM 调用搜索工具后，追加 ToolMessage。

3. 状态管理：

   • 在 StateGraph 中，更新状态中的 messages 字段，确保消息列表的正确性。
   • 示例：多轮对话中维护消息历史。

4. 消息更新：

   • 覆盖或更新特定消息（基于 id），例如修正 LLM 的输出。
   • 示例：更新工具调用的临时消息。

---

5. 代码示例

以下是 add_messages 的典型使用场景和代码示例，展示其在不同上下文中的应用。

5.1 简单消息追加（MessageGraph）

使用 MessageGraph 构建对话系统，add_messages 由 LangGraph 内部自动处理。

from langgraph.graph import MessageGraph, START, END
from langchain_core.messages import HumanMessage, AIMessage

# 定义节点
def respond(messages):
    return AIMessage(content=f"你说: {messages[-1].content}")

# 创建 MessageGraph
workflow = MessageGraph()
workflow.add_node("respond", respond)
workflow.add_edge(START, "respond")
workflow.add_edge("respond", END)

# 编译和运行
graph = workflow.compile()
result = graph.invoke([HumanMessage(content="Hello!")])
print(result)  # [HumanMessage(content='Hello!'), AIMessage(content='你说: Hello!')]

说明：MessageGraph 内部使用 add_messages 将节点返回的 AIMessage 追加到状态（消息列表）。

5.2 使用 add_messages 在 StateGraph 中

在 StateGraph 中，使用 Annotated 和 add_messages 更新状态的 messages 字段。

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END, add_messages
from langchain_core.messages import HumanMessage, AIMessage
from langchain_openai import ChatOpenAI

# 定义状态
class State(TypedDict):
    messages: Annotated[List[BaseMessage], add_messages]

# 定义节点
def agent(state: State):
    llm = ChatOpenAI(model="gpt-4o-mini")
    response = llm.invoke(state["messages"])
    return {"messages": response}

# 创建 StateGraph
workflow = StateGraph(State)
workflow.add_node("agent", agent)
workflow.add_edge(START, "agent")
workflow.add_edge("agent", END)

# 编译和运行
graph = workflow.compile()
result = graph.invoke({"messages": [HumanMessage(content="Hello!")]})
print(result["messages"][-1].content)  # LLM 的回复，例如：Hi there!

说明：

• messages 字段使用 Annotated[..., add_messages]，表示节点返回的 messages 会通过 add_messages 合并。
• 节点返回 {"messages": AIMessage(...)}，add_messages 将其追加到状态的 messages 列表。

5.3 工具调用和消息覆盖

处理工具调用，合并 ToolMessage 并覆盖临时消息。

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END, add_messages
from langchain_core.messages import HumanMessage, AIMessage, ToolMessage
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI

# 定义工具
@tool
def search(query: str):
    return f"搜索结果: {query}"

# 定义状态
class State(TypedDict):
    messages: Annotated[List[BaseMessage], add_messages]

# 定义节点
def agent(state: State):
    llm = ChatOpenAI(model="gpt-4o-mini").bind_tools([search])
    response = llm.invoke(state["messages"])
    return {"messages": response}

def tool_node(state: State):
    tool_call = state["messages"][-1].tool_calls[0]
    result = search.invoke(tool_call["args"]["query"])
    return {"messages": ToolMessage(content=result, tool_call_id=tool_call["id"])}

# 条件函数
def route(state: State):
    return "tool" if state["messages"][-1].tool_calls else END

# 创建 StateGraph
workflow = StateGraph(State)
workflow.add_node("agent", agent)
workflow.add_node("tool", tool_node)
workflow.add_edge(START, "agent")
workflow.add_conditional_edges("agent", route, {"tool": "tool", END: END})
workflow.add_edge("tool", "agent")  # 返回 agent 继续处理

# 编译和运行
graph = workflow.compile()
result = graph.invoke({"messages": [HumanMessage(content="搜索 LangGraph")]})
print(result["messages"][-1].content)  # 搜索结果: LangGraph

说明：

• agent 节点生成 AIMessage 包含工具调用。
• tool_node 返回 ToolMessage，通过 add_messages 追加到 messages。
• add_messages 确保 ToolMessage 的 tool_call_id 与工具调用匹配，正确合并。

5.4 手动调用 add_messages

直接使用 add_messages 处理消息列表。

from langgraph.graph import add_messages
from langchain_core.messages import HumanMessage, AIMessage

# 现有消息
messages = [HumanMessage(content="Hello!")]
# 新消息
new_message = AIMessage(content="Hi there!")

# 使用 add_messages
updated_messages = add_messages(messages, new_message)
print(updated_messages)
# 输出: [HumanMessage(content='Hello!'), AIMessage(content='Hi there!')]

说明：add_messages 将 new_message 追加到 messages，返回更新后的列表。

---

6. 注意事项

1. 消息类型：

   • left 和 right 必须是 BaseMessage、其列表、字符串或 None，否则可能抛出类型错误。
   • 字符串会自动转换为 HumanMessage，确保节点输出符合预期。

2. 消息 ID：

   • 如果消息有 id（如 ToolMessage 或 LLM 生成的消息），add_messages 会检查 id 匹配以决定覆盖。
   • 确保工具调用返回的 ToolMessage 设置了正确的 tool_call_id。

3. 状态注解：

   • 在 StateGraph 中，使用 Annotated[..., add_messages] 是推荐做法，简化状态更新。
   • 避免直接修改状态的 messages 字段，交给 add_messages 处理。

4. 异步支持：

   • add_messages 是同步函数，但可以在异步节点中调用。
   • 如果需要异步消息处理，确保节点函数使用 async def 并配合 graph.astream。

5. 版本兼容性：

   • 确保使用最新版本的 LangGraph（如 0.2.x +）和 langchain-core（如 0.2.33 +）。
   • 旧版本可能对 add_messages 的行为有细微差异（如字符串处理）。

6. 性能考虑：

   • 对于长对话历史，消息列表可能变大，考虑截断或使用检查点（checkpointer）优化内存。
   • 避免重复调用 add_messages 处理相同消息，节点应返回最小更新。

---

7. 总结

langgraph.graph.add_messages 是一个核心函数，用于在 LangGraph 中管理消息列表，特别适合对话系统和消息驱动的工作流。其主要功能包括：

• 消息合并：追加或覆盖消息，基于 id 和类型。
• 类型规范化：处理字符串、单个消息或消息列表，输出 List[BaseMessage]。
• 状态集成：通过 Annotated[..., add_messages] 简化 StateGraph 的状态更新。

add_messages 在 MessageGraph 中由 LangGraph 自动调用，在 StateGraph 中通过状态注解或手动调用使用。结合 LLM 和工具调用，它是构建对话系统和多轮交互的关键组件。

参考文献：

• LangGraph 官方文档：https://langchain-ai.github.io/langgraph/
• GitHub 仓库：https://github.com/langchain-ai/langgraph
• LangChain 文档：https://python.langchain.com/docs/