langgraph_State状态模式+流程图渲染+reducer函数+Annotated特殊标注

一、 定义

前面我们通过比较 LangChain 中构建 AI Agent 的底层设计解释了 LangGraph 在应对复杂 AI Agent 开发场景中的明显优势。在 LangGraph 框架中，不论构建的代理简单或复杂，其本质都是通过节点（Node）和边（Edge）的有机组合来形成一个完整的图（Graph）。这种构建方式所形成的工作流逻辑十分清晰：每个节点在完成其任务后，都会通过边来指示下一个工作步骤，从而赋予整个应用系统更高的灵活性和可扩展性。掌握三个核心组件 —— 状态（State）、节点（Node）和边（Edge），并了解如何使用这些基本元素构建完整的图形结构是学习和应用 LangGraph 框架重要且最基础的一步。通过深入理解这些关键组件足够支撑我们去设计并实现更加复杂且功能丰富的基于 LangGraph 的 AI Agent 应用程序。

在 AI Agent 应用程序的设计中，场景的复杂性直接决定了构建图的复杂度。例如，最简单的场景可能仅涉及一个大模型的问答流程，形式为：START -> Node -> END（其中大模型的交互逻辑被封装在 Node 中）。而更复杂的场景则可能涉及多个 AI Agent 的协同工作，包括多个分支和循环的构成。无论是简单还是复杂的图，LangGraph 的价值永远不在于如何去定义节点，如何去定义边，而是在于如何有效管理各个节点的输入和输出，以保持图的持续运行状态。LangGraph 的底层算法采用消息传递机制来定义和执行这些图中的交互流程，其中状态（State）组件扮演着关键的载体角色，负责在图的各个节点之间传递信息。这也就意味着，LangGraph 框架的核心在于 State 的有效使用和掌握。在复杂的应用中，State 组件需要存储和管理的信息量会显著增加。核心功能如工具使用、记忆能力和人机交互等，都依赖 State 来实现和维护。所以，接下来我们对 LangGraph 框架的探索，都将紧密围绕 State 的实现和应用机制展开，这包括 LangGraph 内置封装好的工具 / 方法的使用，以及我们自定义构建功能时的实现方法。

因此，对于 LangGraph 的底层算法是利用消息传递机制来定义程序的运行这一结论，接下来我们将详细探讨消息（Messages）是如何通过 State 进行传递的，其中包含了什么传递模式和内容。

LangGraph 构建的图中的每个节点都具备访问、读取和写入状态的权限。当某一个节点去修改状态时，它会将此信息广播到图中的所有其他节点。这种广播机制允许其他节点响应状态的变化并相应地调整其行为。如上图所示，从初始状态（Initial State）开始，其中包含了一条消息 {"x": "10"}，随着消息在节点间通过边传递，每个节点根据其逻辑对状态进行更新。Node 1 和 Node 2 分别对状态进行了处理和变更，结果是在图的末端，我们得到了一个包含三条消息的最终状态 {"x": "10"}, {"x": "11"}, {"y": "9"}。从开发的角度来看，State 实际上是一个共享的数据结构。如上图所示，状态表现为一个简单的字典。通过对这个字典进行读写操作，可以实现自主而有序的数据流动，从而构建一个可运行的图结构。那么根据前面学习的内容，我们可以利用这个流程来复现并理解图中的动态数据交换，整体的设计如下：

二、代码演示

from langgraph.graph import StateGraph

# 构建图
builder = StateGraph(dict)
# print(type(builder))
# print(type(builder.schemas))

# 接下来，定义两个节点。addition 节点是一个加法逻辑，接收当前状态，将字典中 x 的值增加 1，并返回新的状态。
# 而 subtraction 节点是一个减法逻辑，接收从 addition 节点传来的状态，从字典中的 x 值减去 2，创建并返回一个新的键y。

def addition(state):
    print(state)
    return {"x": state["x"] + 1}

def subtraction(state):
    print(state)
    return {"y": state["x"] - 2}

# 然后，进行图结构的设计。具体来看，我们添加名为 addition 和 subtraction 的节点，并关联到上面定义的函数。
# 设定图的起始节点为 addition，并从 addition 到 subtraction 设置一条边，最后从 subtraction 到结束节点设置另一条边。代码如下：

from langgraph.graph import START, END

# 向图中添加两个节点
builder.add_node("addition", addition)
builder.add_node("subtraction", subtraction)

# 构建节点之间的边
builder.add_edge(START, "addition")
builder.add_edge("addition", "subtraction")
builder.add_edge("subtraction", END)

# 最后，执行图的编译。需要通过调用 compile() 方法将这些设置编译成一个可执行的图。代码如下所示：
# compile() 作用
# builder 是图构造器，仅用于定义节点、边、状态规则，不能直接运行；
# 调用 .compile() 会做三件事：
# 校验图合法性：检查是否存在孤立节点、非法循环、无入口 / 出口；
# 内部构建执行调度器、状态合并逻辑；
# 返回可运行图实例 graph，拥有 invoke()、stream() 等执行方法。
# 使用区分
# builder：构建阶段专用，只增节点、连边；
# graph：运行阶段专用，传入初始状态执行流程。
graph = builder.compile()

"""
除了上述通过打印的方式查看构建图的结构，LangGraph 还提供了多种内置的图形可视化方法，能够将任何 Graph 以图形的形式展示出来，帮助我们更好地理解节点之间的关系和流程的动态变化。可视化最大的好处是：直接从代码中生成图形化的表示，可以检查图的执行逻辑是否符合构建的预期。LangGraph 提供的三种图形可视化方法如下：
Mermaid.Ink：一个开源服务，可以根据 Mermaid 代码生成图表的 URL。它通过 API 提供多种输出格式，包括 PNG、JPEG、SVG 和 PDF，并可以自定义尺寸、主题和背景颜色等选项。开源仓库：mermaid
Mermaid + Pyppeteer：使用 Mermaid 结合 Pyppeteer 的主要区别在于如何将 Mermaid 图表转换成图像或其他格式。Mermaid 本身是一个轻量级的工具，用于通过文本描述生成图表的图形表示。而 Pyppeteer 是一个 Python 库，它提供了一个接口来控制 Chrome，自动打开包含 Mermaid 图表的网页，然后通过浏览器自动截图功能捕获这些图表，生成图像文件。
Graphviz：Graphviz 是一个图形可视化软件，主要用于自动图形布局。它非常适合于复杂图形的生成，如有向图和无向图，而且它支持多种格式的图像输出，如 PNG、SVG、PDF 等，有更精细的布局控制。
官方可视化文档地址：
https://langchain-ai.github.io/langgraph/how-tos/visualization/
"""

# 如果是 Linux 操作系统，建议使用 Graphviz 工具。而 Windows 系统建议使用 Mermaid + Pyppeteer 方法，
# 因为在 Windows 中 Graphviz 并不能直接通过 pip install 的形式安装，编译安装的方法较为复杂。
# 这里我们就使用 Mermaid + Pyppeteer 来进行图的可视化操作。首先，在当前的虚拟环境中安装依赖包，
# 执行如下代码：

# 生成图结构的可视化非常直接，只需一行代码即可完成。具体代码如下：
import os
from IPython.display import Image, display
# jupyter环境：
display(Image(graph.get_graph(xray=True).draw_mermaid_png()))
# windows环境：
image_data = graph.get_graph().draw_mermaid_png()
# 写入文件
with open("state-desc.png", "wb") as f:
    f.write(image_data)
print(f"✅ 图片已成功生成: {os.path.abspath("state-desc.png")}")

"""
draw_mermaid_png() 返回的始终是 PNG 的二进制字节流（bytes）。两种环境下区别只在于"这些字节往哪里送"：
1. Jupyter / IPython 是"富文本输出"环境
Jupyter Notebook 的内核和前端之间有一条 display 协议。当你调用：
display(Image(...))
Image 对象会被序列化成 image/png 类型的富文本数据，推送到前端 notebook 界面的 当前单元格输出区 直接渲染。也就是说，Jupyter 前端本身就是一个"图像显示载体"，字节不必落盘。

2. VSCode 运行 .py 是"纯终端脚本"环境
在 VSCode 里按运行按钮，本质上是在终端里执行 python State-Desc.py。标准终端只认文本字符，没有富文本渲染区域。所以：
display(Image(...)) 这行代码其实不报错（IPython.display 可以正常导入），但它"无处显示"——你调用了一个要渲染图像的 API，却没有前端接收，于是什么也看不到。
没办法，只能把 bytes 写进磁盘文件，再手动打开图片看。

一句话：不是 LangGraph 的 API 在 Windows 上变复杂了，而是纯脚本运行方式天生无法显示图像，必须落盘。
"""

# 当通过 builder.compile() 方法编译图后，编译后的 graph 对象提供了 invoke 方法，该方法用于启动图的执行。
# 我们可以通过 invoke 方法传递一个初始状态（如 initial_state = {"x": 10}），这个状态将作为图执行的起始输入。代码如下：

result = graph.invoke(input={"x":1})
print(f"最后返回的答案：\n{result}")

"""
在图的执行过程中，每个节点的函数会被调用，并且接收到前一个节点返回的状态作为输入。每个函数处理完状态后，会输出一个新的状态，传递给下一个节点。
这里需要注意的一个关键信息是：节点函数不需要返回整个状态，而是仅返回它们更新的部分。也就是说：在每个节点的函数内部逻辑中，需要使用和更新哪些 State 中的参数，只需要在 return 的时候指定即可，不必担心未在当前节点处理的 State 中的其他值会丢失，因为 LangGraph 的内部机制已经自动处理了状态的合并和维护。
"""

三、 reducer 函数

I 描述

LangGraph 内部原理：State 中的每个 key 都有自己独立的 Reducer 函数，通过指定的 reducer 函数应用状态值的更新。
Reducer 函数用来根据当前的状态（state）和一个操作（action）来计算并返回新的状态。它是一种设计模式，用于将业务逻辑与状态变更解耦，使得状态的变更预测性更强并且容易追踪。这样的函数通常接收两个参数：当前状态（state）和一个描述应用了什么操作的对象（action），根据 action 类型来决定如何修改状态。比如，在一个购物车应用中，可能会有添加商品、删除商品、修改商品数量等操作。返回一个新的状态对象，而不是修改原始状态对象。简单理解，Reducer 函数做的就是根据给定的输入（当前状态和操作）生成新的状态。
LangGraph Update State 源码：https://langchain-ai.github.io/langgraph/reference/graphs/#langgraph.graph.graph.CompiledGraph.update_state
LangGraph 中，如果没有显示的指定，则对该键的所有更新都执行的是覆盖操作。我们看一下下面这段代码：

II 代码演示

from typing_extensions import TypedDict
from langgraph.graph import START, StateGraph, END

def addition(state):
    print(state)
    return {"x": state["x"] + 1}

def subtraction(state):
    print(state)
    return {"x": state["x"] - 2}

# 指定了state的数据类型
class State(TypedDict):
    x: int
    y: int

# 构建图
builder = StateGraph(State)

# 向图中添加两个节点
builder.add_node("addition", addition)
builder.add_node("subtraction", subtraction)

# 构建节点之间的边
builder.add_edge(START, "addition")
builder.add_edge("addition", "subtraction")
builder.add_edge("subtraction", END)

graph = builder.compile()

# 定义一个初始化的状态
initial_state = {"x":10}

graph.invoke(initial_state)
"""
在这里，我们通过使用 TypedDict 来定义 State 的模式，从而精确控制图结构中状态信息的格式和类型。与上面所使用的传统字典类型相比，
TypedDict 允许我们明确指定每个键的类型，有助于防止在状态管理过程中出现类型错误。我们强烈建议在开发过程中采用这种方式来定义和管理状态，
特别是在涉及复杂状态逻辑和多个状态依赖的应用中。

"""

四、Annotated

接下来我们再看另一种情况：在下面的图中，State 还是一个结构化字典，其中包含一个名为 messages 的键，该键保存一个字符串列表。
我们用这个状态管理节点在执行期间将处理的数据。状态的模式是使用 TypedDict 定义，它指定消息是带注释的字符串列表。
该注释包括 operator.add，表示可以通过使用添加操作将新消息与现有消息组合来更新列表。代码如下：

messages: 这是字典中的一个键，它表示某个字段（比如消息列表）。List [str]: 这是这个字段值的数据类型，
它表示一个字符串类型的列表（即一个包含多个字符串的集合）。Annotated [List [str], operator.add]: 这是一个类型注解的特殊用法，
它使用 Annotated 来附加额外的元数据（在这里是 operator.add）。Annotated 是 Python 3.9 引入的，它允许你在指定类型的同时附加额外的信息
（例如，方法、操作符等）。operator.add 是 Python 中的一个加法运算符，通常用来对数字进行加法操作。在这里，它作为元数据附加在 List [str] 类型后面，
表明这个字段 messages 可能会涉及到某些与加法相关的操作，虽然这里并没有直接用到它。
当定义状态模式的结构发生了变化以后，在节点函数中的读取和存储逻辑也要发生相应的变化。流程图如下：

import operator
from typing import Annotated, TypedDict, List
from typing_extensions import TypedDict
from langgraph.graph import START, StateGraph, END

class State(TypedDict):
    messages: Annotated[List[str], operator.add]

def addition(state):
    print(state)
    msg = state['messages'][-1]
    response = {"x": msg["x"] + 1}
    return {"messages": [response]}

def subtraction(state):
    print(state)
    msg = state['messages'][-1]
    response = {"x": msg["x"] - 2}
    return {"messages": [response]}

# 构建图
builder = StateGraph(State)

# 向图中添加两个节点
builder.add_node("node1", addition)
builder.add_node("node2", subtraction)

# 构建节点之间的边
builder.add_edge(START, "node1")
builder.add_edge("node1", "node2")
builder.add_edge("node2", END)

graph = builder.compile()

data = graph.get_graph(xray=True).draw_mermaid_png()
with open("reducer.png", "wb") as f:
    f.write(data)

result = graph.invoke(input={"messages": [{"x": 10}]})

print(f"最后返回的答案：\n{result}")

通过这样的例子我们可以理解：Reducers 的目的是在 LangGraph 框架中的状态管理系统中，允许更灵活地定义状态如何根据各种操作更新。通过指定不同的 reducer 函数，我们可以控制状态的每个部分应如何响应特定的更新。

四、 在图状态中处理消息的思路（operator.add）

I 定义

Reducer 机制的一个现实意义是：我们可以基于这种方式去构建历史对话记录。因为目前大多数大模型应用都是接受消息列表作为输入，就像 LangChain 中的 Chat Model，需要接收 Message 对象列表作为输入。这些消息有多种形式，例如 HumanMessage（用户输入）或 AIMessage（大模型响应）。

下面这个示例，我们进一步将大模型接入到 LangGraph 工作流程中，并允许动态消息处理以及与模型的交互。其余组件与先前定义的图中的组件相同。在这里，第一个节点调用大模型并生成一个输出，该输出是一个 AIMessage 对象类型，然后，第二个节点直接将前一个节点的 AIMessage 提取为具体的 JSON 格式，完整代码如下：

II 代码演示

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

import getpass
import os

import getpass
import os
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
import operator
from typing import Annotated, TypedDict, List
from langgraph.graph import StateGraph, END
from IPython.display import Image, display
from langchain_core.messages import SystemMessage, HumanMessage, AIMessage

# 1、配置模型
from dotenv import load_dotenv
load_dotenv()
import os
key = os.environ["OPENAI_API_KEY"]
base_url = os.environ["OPENAI_BASE_URL"]
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4o", api_key=key,base_url=base_url,temperature=0,)

# 定义图的状态模式
class State(TypedDict):
    messages: Annotated[List[str], operator.add]

# 创建图的实例
builder = StateGraph(State)

def chat_with_model(state):
    print(state)
    print("------------------")
    messages = state['messages']
    response = llm.invoke(messages)
    return {"messages": [response]}

def convert_messages(state):
    # "您是一位数据提取专家，负责从文本中检索关键信息。请为所提供的文本提取相关信息，并以 JSON 格式输出。概述所提取的关键数据点。"
    EXTRACTION_PROMPT = """
    You are a data extraction specialist tasked with retrieving key information from a text.
    Extract such information for the provided text and output it in JSON format. Outline the key data points extracted.
    """
    print(state)
    print("------------------")

    messages = [
        SystemMessage(content=EXTRACTION_PROMPT),
        HumanMessage(content=state['messages'][-1].content)
    ]

    response = llm.invoke(messages)
    return {"messages": [response]}

# 添加节点
builder.add_node("chat_with_model", chat_with_model)
builder.add_node("convert_messages", convert_messages)

# 设置启动点
builder.set_entry_point("chat_with_model")

# 添加边
builder.add_edge("chat_with_model", "convert_messages")
builder.add_edge("convert_messages", END)

# 编译图
graph = builder.compile()

data = graph.get_graph(xray=True).draw_mermaid_png()
with open("process.png","wb") as f:
    f.write(data)
print(f"✅ 图片已成功生成: {os.path.abspath("process.png")}")   

query="你好，请你介绍一下你自己"
input_message = {"messages": [HumanMessage(content=query)]}

result = graph.invoke(input_message)

print(result)

print("*"*80)

print(result["messages"][-1].content)

"""

{
  "name": "MiniMax-M2.5",
  "developer": "MiniMax公司",
  "type": "AI助手",
  "capabilities": [
    "回答问题和提供信息",
    "撰写和润色文本内容",
    "编程和代码调试",
    "翻译和语言处理",
    "头脑风暴和创意建议",
    "学习和教育辅导"
  ],
  "language": "中文"
}

"""

"""
如上所示，在许多现实开发情况下，将先前的对话历史记录存储为图状态中的消息列表是必须要做的。而实现这一功能，
在 LangGraph 框架中可以向存储 Message 对象列表的图状态添加一个键（通道），并使用 Reducer 函数对其进行注释，
从而告诉图在每次状态更新时（例如，当节点发送更新时）更新状态中的 Message 对象列表。
如上所演示的案例中，简单地将消息追加到现有列表中，就是通过operator.add实现。
"""

五、MessageGraph 源码功能解析（add_messages函数）

I 概念

operator.add 的局限性
operator.add 仅支持列表追加，无法实现消息覆盖 / 修改：
新增消息：直接拼接到列表末尾；
已有消息更新：无法根据消息 ID 匹配旧消息并替换，只能追加一条全新消息，会造成历史冗余。

add_messages 高级 Reducer 优势
LangGraph 内置专用消息合并工具，兼顾两种场景：
全新消息：追加至消息列表；
带 ID 的更新消息：匹配相同 ID 的历史消息并覆盖，完美适配人机交互、消息修正、多轮对话重写场景。

更复杂一点的，如果我们还想手动更新图形状态中的消息（例如人机交互），使用 operator.add 能做到的功能极限是：发送到图表的手动状态更新将被附加到现有的消息列表中，而不是更新现有的消息。为了避免这种情况，我们则需要一个可以跟踪消息 ID 并覆盖现有消息（如果更新）的 Reducer 函数。为此，就引出了 LangGraph 预构建的 add_messages 函数，这个更高级的 Reducer 所实现的是：对于全新的消息，它会附加到现有列表，但它也会正确处理现有消息的更新。如何理解这句话呢？我们接下来就从源码角度进行详细解析。
上节课程中我们分析过 LangGraph 中的 StateGraph 类，这个类允许我们创建图，其节点通过读取和写入共享状态进行通信。StateGraph 类由开发者定义的 State 对象进行参数化，该对象表示图中的节点将通过其进行通信的共享数据结构。
MessageGraph 是 StateGraph 的一个子类，使用了 Annotated [list [AnyMessage], add_messages] 来初始化其基类 StateGraph。这里的 list [AnyMessage] 指明了 MessageGraph 的状态由消息列表组成，而这个列表类型是一个可以不断添加消息的结构（因为列表是可变的数据类型），MessageGraph 中的每个节点都将消息列表作为输入，并返回零个或多个消息作为输出。add_messages 函数用于将每个节点的输出消息合并进图的状态中已存在的消息列表。其源码定义如下：

MessageGraph 源码：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])

Annotated[list[AnyMessage], add_messages]作用：存储任何类型的message，并且给每一个message添加一个独特的编号。两种功能：不同编号的message往集合里面添加，同编号的message直接覆盖修改。

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

# 1、配置模型
from dotenv import load_dotenv
load_dotenv()
import os
key = os.environ["OPENAI_API_KEY"]
base_url = os.environ["OPENAI_BASE_URL"]
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4o", api_key=key,base_url=base_url,temperature=0,)
import operator
from typing import Annotated, TypedDict, List
# 定义图的状态模式
class State(TypedDict):
    messages: Annotated[List[str], operator.add]

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()

result = graph.invoke([("user", "你好，请你介绍一下你自己.")])

print(f"{'*' * 80} \n {result} \n {'*' * 80} ")

"""
HumanMessage这个消息，有id='f7c3f7ac-06a8-4b37-a219-79af6f892276'
AIMessage这个消息，也有id='841a914b-e85a-47aa-9d28-332c01174c78'
********************************************************************************
[HumanMessage(content='你好，请你介绍一下你自己.', additional_kwargs={}, response_metadata={}, id='f7c3f7ac-06a8-4b37-a219-79af6f892276'), 
 AIMessage(content='你好，最帅气的人！', additional_kwargs={}, response_metadata={}, id='841a914b-e85a-47aa-9d28-332c01174c78', tool_calls=[], invalid_tool_calls=[])]
********************************************************************************
"""

MessageGraph 通过使用单个仅附加消息列表作为其整个状态来管理状态，其中每个节点处理该列表并可以返回其他消息。这种设计特别适合对话式应用程序，可以轻松跟踪对话历史和交互。
继而进一步来看 add_messages 这个 Reducer 函数的定义情况：

类的源码：

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

def add_messages(left: Messages, right: Messages) -> Messages:
    """Merges two lists of messages, updating existing messages by ID.

    By default, this ensures the state is "append-only", unless the
    new message has the same ID as an existing message.

    Args:
        left: The base list of messages.
        right: The list of messages (or single message) to merge
            into the base list.

    Returns:
        A new list of messages with the messages from <code>right</code> merged into <code>left</code>.
        If a message in <code>right</code> has the same ID as a message in <code>left</code>, the
        message from <code>right</code> will replace the message(content='Hello', id=...)]]
    """

# coerce to list
if not isinstance(left, list):
    left = [left]  # type: ignore[assignment]
if not isinstance(right, list):
    right = [right]  # type: ignore[assignment]

# coerce to message
left = [
    message_chunk_to_message(cast(BaseMessageChunk, m))
    for m in convert_to_messages(left)
]
right = [
    message_chunk_to_message(cast(BaseMessageChunk, m))
    for m in convert_to_messages(right)
]

# assign missing ids
for m in left:
    if m.id is None:
        m.id = str(uuid.uuid4())
for m in right:
    if m.id is None:
        m.id = str(uuid.uuid4())

# merge
merged = left.copy()
merged_by_id = {m.id: i for i, m in enumerate(merged)}
ids_to_remove = set()
for m in right:
    if (existing_idx := merged_by_id.get(m.id)) is not None:
        if isinstance(m, RemoveMessage):
            ids_to_remove.add(m.id)
        else:
            ids_to_remove.discard(m.id)
            merged[existing_idx] = m          # ← 按 ID 覆盖已有消息
    else:
        if isinstance(m, RemoveMessage):
            raise ValueError(
                f"Attempting to delete a message with an ID that doesn't exist ('{m.id}')"
            )

        merged_by_id[m.id] = len(merged)     # ← 更新索引
        merged.append(m)                      # ← 追加新消息
merged = [m for m in merged if m.id not in ids_to_remove]

II add_messages

LangGraph 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 的消息具有相同的 ID，则 right 的消息将替换 left 的消息，否则作为一条新的消息进行追加。

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")]
add_messages(msgs1, msgs2)
result = add_messages(msgs1, msgs2)
print(f"{'*' * 80} \n {result} \n {'*' * 80} ")
"""
# 输出结果
[
    HumanMessage(content='你好。', additional_kwargs={}, response_metadata={}, id='1'),
    AIMessage(content='你好，很高兴认识你。', additional_kwargs={}, response_metadata={}, id='2')
]
"""

# 示例2：ID重复，新消息覆盖旧消息
msgs1 = [HumanMessage(content="你好。", id="1")]
msgs2 = [HumanMessage(content="你好呀。", id="1")]
result = add_messages(msgs1, msgs2)
print(f"{'*' * 80} \n {result} \n {'*' * 80} ")
"""
# 输出结果
[HumanMessage(content='你好呀。', additional_kwargs={}, response_metadata={}, id='1')]
"""

消息状态管理不仅可以在 MessageGraph 中定义，还可以在 StateGraph 中定义。相比之下，StateGraph 允许更复杂的状态结构，其中状态可以是任何 Python 类型（如 TypedDict 或 Pydantic 模型），并且可以通过各种方式更新。StateGraph 中的每个节点都会接收当前状态并返回更新后的状态，从而实现除消息处理之外的更复杂的数据操作和工作流程。因此，MessageGraph 专门用于以消息为中心的工作流程，而 StateGraph 则更通用，适用于更广泛的应用程序。
所以，对于人机交互的程序，在 LangGraph 的开发流程中，我们一般会首先创建一个 StateGraph，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
from langchain_core.prompts import ChatPromptTemplate

# 1、配置模型
import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
load_dotenv()
key = os.environ["OPENAI_API_KEY"]
base_url = os.environ["OPENAI_BASE_URL"]
llm = ChatOpenAI(model="gpt-4o", api_key=key,base_url=base_url,temperature=0,)

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

# 接下来，添加一个 chatbot 节点，将当前 State 作为输入并返回一个字典，该字典中更新了 messages 中的状态信息。

graph_builder.add_node("chatbot", chatbot)

# 输出对象：<langgraph.graph.state.StateGraph at 0x7fa9eb7db9e0>

# 配置流程连线
graph_builder.add_edge(START, "chatbot")
graph_builder.add_edge("chatbot", END)

# 编译流程图
graph = graph_builder.compile()

# 渲染流程图可视化
# image_data = graph.get_graph().draw_mermaid_png()
# # 写入文件
# with open("Add_Messages_G.png", "wb") as f:
#     f.write(image_data)
# print(f"✅ 图片已成功生成: {os.path.abspath("Add_Messages_G.png")}")

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

# 掌握 State 的定义模式和消息传递是 LangGraph 中最关键，也是构建应用最核心的部分，
# 所有的高阶功能，如工具调用、上下文记忆，人机交互等依赖 State 的管理和使用，所以务必理解并掌握上述相关内容。