前言

在复杂的人机交互场景中，对于图形状态中消息的手动更新需求日益凸显。在 LangGraph 框架下，operator.add在处理手动状态更新时存在一定局限性，其仅能将手动发送到图表的状态更新附加到现有消息列表，而无法对现有消息进行直接更新。为突破这一限制，实现更灵活高效的消息管理，LangGraph引入了预构建的add_messages函数。该函数作为一个功能强大的Reducer，不仅能将全新消息追加到列表，还能智能地处理现有消息的更新操作。接下来，我们将深入源码层面，一探究竟。

MessageGraph 的类结构与核心定位

看本节内容前，你已经对LangGraph中的StateGraph类有了深入了解。StateGraph类允许开发者创建图形，其节点通过读取和写入共享状态实现通信，该类通过开发者自定义的State对象进行参数化，这个State对象代表了图中节点通信所依赖的共享数据结构。

MessageGraph作为StateGraph的子类，在初始化基类StateGraph时，使用了Annotated[list[AnyMessage], add_messages]。其中，list[AnyMessage]明确指出MessageGraph的状态由消息列表构成，利用列表可变的特性，实现消息的动态添加。在MessageGraph中，每个节点均以消息列表作为输入，并输出零个或多个消息。而add_messages函数则承担着关键职责，它负责将各个节点输出的消息，精准地合并到图状态中已有的消息列表内 。其源码定义如下（具体路径：https://github.com/langchain-ai/langgraph/blob/e3ef9adac7395e5c0943c22bbc8a4a856b103aa3/libs/langgraph/langgraph/graph/message.py#L150）：

class MessageGraph(StateGraph):
    def __init__(self) -> None:
        super().__init__(Annotated[list[AnyMessage], add_messages])

MessageGraph可以独立构建，以下代码展示了一个简单的构建示例：

import getpass
import os
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

if not os.environ.get("OPENAI_API_KEY"):
    os.environ["OPENAI_API_KEY"] = getpass.getpass("Enter your OpenAI API key: ")

llm = ChatOpenAI(model="gpt-4o")

def chatbot(state: State):
    print(state)
    return {"messages": [llm.invoke(state["messages"])]}

from langgraph.graph.message import MessageGraph

builder = MessageGraph()
builder.add_node("chatbot", lambda state: [("assistant", "你好，最帅气的人！")])
builder.set_entry_point("chatbot")
builder.set_finish_point("chatbot")
graph = builder.compile()
graph.invoke([("user", "你好，请你介绍一下你自己.")])

通过上述代码可知，MessageGraph以单个仅附加消息列表作为其全部状态进行管理，各节点对该列表进行处理并返回新消息。这种设计模式高度契合对话式应用场景，能够轻松实现对话历史记录与交互过程的有效跟踪。

add_messages 函数的核心逻辑剖析

接下来，我们深入探究add_messages这个Reducer函数的具体实现细节：

class MessagesState(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]

您暂时无权查看此隐藏内容！

（函数文档链接：https://langchain-ai.github.io/langgraph/reference/graphs/#langgraph.graph.message.add_messages ）

add_messages函数的核心功能是合并两个消息列表，并依据消息ID对现有消息进行更新操作。在默认情况下，状态呈现“仅附加”特性，只有当新消息与现有消息的ID一致时，才会触发更新逻辑。其参数与返回值具体说明如下：

• 参数：
  • left（Messages类型）：作为消息的基础列表，承载着已有的消息数据。
  • right（Messages类型）：包含了需要合并到基础列表中的消息列表，也可以是单个消息。
• 返回值：返回一个全新的消息列表，在该列表中，right中的消息已按照既定规则合并到left中。即若right中的消息与left中的消息ID相同，则right的消息会替换left中的对应消息；若ID不同，则作为新消息追加到列表末尾。

以下通过示例代码进一步说明：

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

# 示例1：ID不同的消息合并
msgs1 = [HumanMessage(content="你好。", id="1")]
msgs2 = [AIMessage(content="你好，很高兴认识你。", id="2")]
result1 = add_messages(msgs1, msgs2)

# 示例2：ID相同的消息合并
msgs3 = [HumanMessage(content="你好。", id="1")]
msgs4 = [HumanMessage(content="你好呀。", id="1")]
result2 = add_messages(msgs3, msgs4)

MessageGraph 与 StateGraph 的对比与应用场景

消息状态管理在MessageGraph和StateGraph中均有涉及，但二者存在明显差异。StateGraph具备更高的灵活性，它允许使用任何 Python 类型（如 TypedDict 或 Pydantic 模型）来定义复杂的状态结构，并且支持通过多种方式实现状态更新。在StateGraph中，每个节点接收当前状态并返回更新后的状态，这种机制能够支持除消息处理之外的各类复杂数据操作与工作流程。

而MessageGraph则是专为以消息为核心的工作流程设计，相比之下，它的应用场景更为聚焦。因此，在实际开发中，MessageGraph适用于处理对话、聊天等场景，而StateGraph则因其通用性，可广泛应用于更多样化的应用程序开发。

在 LangGraph 中构建交互式应用的实践

在基于LangGraph开发人机交互程序时，通常会先创建StateGraph，将应用程序结构定义为 “状态机”。以下是一个接入大模型的聊天机器人示例：

from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages

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

graph_builder = StateGraph(State)

上述代码中，State是一个TypedDict，其包含单个键messages。messages键通过使用add_messages作为Reducer函数，告知LangGraph将新消息追加到现有列表，而非覆盖。

接着，引入大模型并定义节点处理逻辑：

import getpass
import os
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

if not os.environ.get("OPENAI_API_KEY"):
    os.environ["OPENAI_API_KEY"] = getpass.getpass("Enter your OpenAI API key: ")

llm = ChatOpenAI(model="gpt-4o")

def chatbot(state: State):
    # print(state)
    return {"messages": [llm.invoke(state["messages"])]}

随后，添加chatbot节点并构建图形连接：

graph_builder.add_node("chatbot", chatbot)
graph_builder.add_edge(START, "chatbot")
graph_builder.add_edge("chatbot", END)
graph = graph_builder.compile()

最后，通过以下代码实现交互式聊天功能：

def stream_graph_updates(user_input: str):  
    for event in graph.stream({"messages": [("user", user_input)]}):
        for value in event.values():
            print("模型回复:", value["messages"][-1].content)

while True:
    try:
        user_input = input("用户提问: ")
        if user_input.lower() in ["退出"]:
            print("下次再见！")
            break
        stream_graph_updates(user_input)
    except:
        break

在LangGraph的开发过程中，精准掌握State的定义模式与消息传递机制至关重要。这些核心内容是实现工具调用、上下文记忆、人机交互等高阶功能的基础，只有深入理解并熟练运用，才能充分发挥LangGraph的强大潜力，构建出功能丰富、交互流畅的应用程序。