引言

在人工智能领域，将大型语言模型（LLM）从简单的无状态问答机器转变为能够进行连贯、有上下文感知对话的复杂智能体，其核心关键在于内存。LLM 本身是无状态的，每次调用都是一次独立的计算，它们不会天生记得之前的交互。为了构建能够学习、适应和维护长期关系的智能体，开发者必须为其提供一个坚实的状态管理框架。LangGraph 正是为此而生，它提供了一个强大而灵活的解决方案。

本指南旨在深入剖析 LangGraph 的内存架构，这是一个精心设计的双系统，旨在模拟不同层次的人类记忆。我们将全面探讨其两大核心组件：

• 短期记忆（通过 Checkpointer 实现）：负责捕捉单次对话中逐轮变化的上下文和状态。这类似于人类的工作记忆，让我们能够流畅地进行一次对话，记住对方刚刚说了什么。
• 长期记忆（通过 Store 实现）：负责存储跨越多次对话的持久化知识、用户偏好和核心事实。这好比人类的长期记忆，储存着我们的身份、知识和经历。

本教程将引导您从基本概念出发，逐步深入到生产级的内存管理策略。我们将通过详尽的代码示例和深入的架构分析，揭示如何利用 LangGraph 的内存机制来构建真正智能、有状态的 AI 应用程序。无论您是希望构建一个简单的聊天机器人，还是一个复杂的多智能体系统，对 LangGraph 内存的深刻理解都将是您成功的基石。

第一部分：LangGraph 内存的架构基础

要精通 LangGraph 的内存管理，首先必须建立一个清晰的心理模型，理解其双轨制内存系统的设计哲学。这两个系统——Checkpointer 和 Store——并非相互替代，而是相辅相成，共同为智能体提供全面、多层次的记忆能力。

1.1 短期记忆：通过 Checkpointer 实现的对话线程

短期记忆在 LangGraph 中被定义为“线程级别的持久化”（thread-level persistence）。其核心目标是记录和恢复在单次连续交互（即一个“线程”）中图（Graph）的状态演变。

核心机制：Checkpointer

Checkpointer 是 LangGraph 短期记忆的引擎。它的工作模式是自动化的：每当图执行一步（例如，调用一个节点函数），checkpointer 就会自动将图的当前完整状态（State）保存为一个快照，即“检查点”（Checkpoint）。

关键标识符：thread_id

thread_id 是激活和区分不同对话线程的唯一钥匙。在调用图的 invoke 或 stream 方法时，通过 configurable 字典传入一个 thread_id，就等于告诉 LangGraph：“这次操作属于这个特定的对话”。在执行前，LangGraph 会使用这个 thread_id 从后端加载最新的检查点，恢复对话状态；执行完毕后，它会再次使用同一个 thread_id 保存更新后的状态。这种机制确保了对话的连续性 。

概念类比：浏览器会话 Cookie

可以将 checkpointer 和 thread_id 的关系类比为网站的会话（Session）Cookie。当您登录一个网站后，浏览器会保存一个会话 Cookie。在您浏览不同页面时，网站通过这个 Cookie 识别您的身份，从而保持您的登录状态。同样，thread_id 就像这个会话 Cookie，让 LangGraph 智能体在多轮对话中能够“记住”您是谁以及你们正在聊什么。

1.2 长期记忆：通过 Store 实现的持久化知识库

与专注于单次对话的短期记忆不同，长期记忆旨在存储“跨越不同对话或会话的用户特定或应用级别的数据”。这是智能体积累知识、形成对用户长期认知的地方。

核心机制：Store

Store 本质上是一个暴露给图节点和工具（Tools）的键值数据库。与 checkpointer 的自动化快照不同，对 Store 的操作是显式和主动的。开发者需要在节点或工具的逻辑中，通过调用 store.put()、store.get() 或 store.search() 等方法来精确地读写信息 。

关键标识符：user_id 和 namespace

Store 中的数据通常通过更持久的标识符来组织。user_id 是最常见的例子，用于关联特定用户的所有信息。此外，namespace（命名空间）提供了一种数据隔离机制，例如，使用 (“memories”, user_id) 这样的元组作为命名空间，可以将用户的记忆与其他类型的数据（如用户偏好 (“preferences”, user_id)）清晰地分离开来，避免数据冲突，保持知识库的整洁有序 。

概念类比：用户数据库中的个人档案

可以将 Store 想象成一个应用数据库中的用户个人档案表。用户的姓名、联系方式、购买历史等信息被永久存储，并且在用户每次访问时都可以被查询和使用。同样，LangGraph 的 Store 保存了那些不应随单次对话结束而消失的持久化信息，智能体可以在任何时候、任何对话中访问这些信息。

1.3 Checkpointer 与 Store 的协同作用

一个高级的智能体需要同时运用这两种记忆。Checkpointer 提供了流畅的对话体验，而 Store 则赋予了智能体深度和个性。

应用场景示例：AI 客户服务智能体

• Checkpointer (由 thread_id 标识) 记录了当前对话的上下文：
  • 用户：“你好，我的电脑开不了机了。”
  • 智能体：“您好，请问您检查过电源线是否插好了吗？”
• Store (由 user_id 标识) 存储了该用户的历史信息：
  • “用户 John Doe，于 YYYY-MM-DD 购买了 X 型号笔记本电脑，历史支持记录显示曾有电源适配器故障。”

在这个场景中，checkpointer 确保了智能体能够对用户当前的问题做出连贯的回应。而 store 则提供了深层次的历史背景，使智能体能够提供更具个性化和洞察力的服务（例如，“我看到您之前遇到过电源问题，这次的情况是否类似？”）。

LangGraph 的这种双内存架构，实际上是软件工程中“关注点分离”（Separation of Concerns）原则的体现。它将易变的、与会话紧密相关的对话状态（由 Checkpointer 管理）与稳定的、持久化的应用知识（由 Store 管理）分离开来。这种设计并非偶然，而是为了构建可扩展、可维护的复杂 AI 应用。它简化了开发流程，降低了数据管理的复杂性，并允许这两个系统根据应用需求独立扩展。这表明 LangGraph 的设计目标远不止于简单的聊天机器人，而是面向需要精细化状态管理的生产级企业应用。

为了更清晰地展示两者的区别，下表进行了详细对比：

第二部分：使用 Checkpointer 实现短期记忆

本部分将提供一个实践指南，指导您如何为 LangGraph 应用添加对话记忆功能，从简单的本地开发环境配置，到稳健的生产环境部署。

2.1 快速入门：用于开发的 InMemorySaver

对于开发、原型设计和测试阶段，最简单快捷的方式是使用 InMemorySaver。它将所有的对话状态存储在内存中的一个 Python 字典里。

让我们解构一下基础的实现代码 ：

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import StateGraph

# 1. 实例化一个内存检查点工具
checkpointer = InMemorySaver()

# 假设 builder 已经定义好了
builder = StateGraph(...) 
#... 添加节点和边...

# 2. 在编译图时传入 checkpointer
graph = builder.compile(checkpointer=checkpointer)

# 3. 调用图时，通过 configurable 传入 thread_id
graph.invoke(
    {"messages": [{"role": "user", "content": "hi! I'm bob"}]},
    {"configurable": {"thread_id": "1"}},
)

# 后续调用使用相同的 thread_id，图将能“记住”之前的状态
response = graph.invoke(
    {"messages": [{"role": "user", "content": "我的名字是什么？"}]},
    {"configurable": {"thread_id": "1"}},
)
# response 将会包含模型回答 "你的名字是 Bob" 的信息

代码分析：

• checkpointer = InMemorySaver(): 实例化过程非常简单，无需任何配置。
• graph.compile(checkpointer=checkpointer): 这是将持久化能力“注入”图的关键步骤。编译后的 graph 对象现在具备了状态管理的能力。
• {"configurable": {"thread_id": "1"}}: 这个字典是激活内存机制的核心。如果没有提供 thread_id，每次 invoke 调用都将是无状态的。而只要使用相同的 thread_id，LangGraph 就会在多次调用之间维持对话状态。

重要提示：
InMemorySaver 仅适用于开发和测试。由于所有状态都保存在内存中，一旦 Python 进程终止，所有对话历史都将丢失。

2.2 生产级持久化：数据库支持的 Checkpointer

当应用准备上线时，必须使用由数据库支持的持久化 Checkpointer，以确保数据的可靠性和服务的连续性。LangGraph 为主流数据库提供了开箱即用的支持，如 PostgreSQL、MongoDB 和 Redis。

以下是针对不同数据库的配置和使用详解。

2.2.1 PostgreSQL

PostgreSQL 是一个功能强大的开源对象关系数据库，非常适合需要事务完整性和结构化数据的应用。

安装：

pip install -U "psycopg[binary,pool]" langgraph langgraph-checkpoint-postgres

连接与设置：

from langgraph.checkpoint.postgres import PostgresSaver

DB_URI = "postgresql://user:password@host:port/dbname?sslmode=disable"

# 使用 with 上下文管理器确保连接被正确关闭
with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
    # 首次运行时，需要执行 setup() 来创建数据库表
    # checkpointer.setup() 

    builder = StateGraph(...)
    graph = builder.compile(checkpointer=checkpointer)

checkpointer.setup() 是一个关键的一次性操作，它会在您的 PostgreSQL 数据库中创建 LangGraph 所需的表结构，用于存储检查点数据 。

完整代码示例（同步与异步）：
下面的示例展示了一个完整的对话流程。用户首先介绍自己是 Bob，然后提问自己的名字。由于使用了 PostgresSaver 和相同的 thread_id，第二次调用能够成功恢复状态，让模型正确回答。

同步 (Sync) 示例:

from langchain.chat_models import init_chat_model
from langgraph.graph import StateGraph, MessagesState, START
from langgraph.checkpoint.postgres import PostgresSaver

model = init_chat_model(model="anthropic:claude-3-5-haiku-latest")
DB_URI = "postgresql://postgres:postgres@localhost:5442/postgres?sslmode=disable"

with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
    # checkpointer.setup()

    def call_model(state: MessagesState):
        response = model.invoke(state["messages"])
        return {"messages": response}

    builder = StateGraph(MessagesState)
    builder.add_node("call_model", call_model)
    builder.add_edge(START, "call_model")

    graph = builder.compile(checkpointer=checkpointer)

    config = {"configurable": {"thread_id": "1"}}

    # 第一次调用
    graph.invoke(
        {"messages": [{"role": "user", "content": "hi! I'm bob"}]},
        config
    )

    # 第二次调用
    final_state = graph.invoke(
        {"messages": [{"role": "user", "content": "what's my name?"}]},
        config
    )
    final_state["messages"][-1].pretty_print()

异步 (Async) 示例:

异步版本结构类似，但使用了 async/await 语法和异步库 AsyncPostgresSaver，适用于高并发场景。

from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver
#...
async with AsyncPostgresSaver.from_conn_string(DB_URI) as checkpointer:
    # await checkpointer.setup()
    async def call_model(state: MessagesState):
        response = await model.ainvoke(state["messages"])
        return {"messages": response}
    #...
    graph = builder.compile(checkpointer=checkpointer)
    #...
    await graph.ainvoke(...)

无论是同步还是异步，核心的业务逻辑 call_model 保持不变。LangGraph 的架构将持久化逻辑与业务逻辑解耦，开发者只需在编译时更换 checkpointer 即可切换持久化策略。

2.2.2 MongoDB 和 Redis

配置 MongoDB 和 Redis 的流程与 PostgreSQL 非常相似，只需替换为对应的 Saver 类即可。

• MongoDB：

  • 安装：pip install -U pymongo langgraph langgraph-checkpoint-mongodb
  • 使用：from langgraph.checkpoint.mongodb import MongoDBSaver

• Redis：

  • 安装：pip install -U langgraph langgraph-checkpoint-redis
  • 使用：from langgraph.checkpoint.redis import RedisSaver

选择哪种数据库取决于您的技术栈、性能需求和数据模型偏好。Redis 以其极高的读写速度著称，适合作为高速缓存和检查点存储。MongoDB 则以其灵活的文档模型见长，适合存储半结构化的状态数据。

2.3 高级模式：子图与工具中的内存交互

2.3.1 子图中的内存

当构建复杂的、由多个子图嵌套而成的应用时，LangGraph 提供了灵活的内存管理策略。

• 内存继承（默认行为）：
  默认情况下，子图会继承其父图的 checkpointer。这意味着整个嵌套图共享同一个对话状态，数据可以在父子图之间无缝流动。这对于将一个大型任务分解为多个模块化子任务非常有用 。

• 内存隔离：
  在某些场景下，例如构建多智能体系统，我们希望每个智能体（由一个子图表示）拥有自己独立的内存空间，互不干扰。此时，可以在编译子图时设置 checkpointer=True。

  subgraph_builder = StateGraph(...)
  #...
  # 这会为子图创建一个独立的内存空间
  subgraph = subgraph_builder.compile(checkpointer=True)

  这种模式赋予了每个智能体独立的“记忆”，是实现复杂协作和自治行为的基础 。

2.3.2 在工具中读取状态

工具（Tools）不再是无状态的外部函数。LangGraph 允许工具直接访问和读取当前的图状态，使其具备上下文感知能力。

from typing import Annotated
from langgraph.prebuilt import InjectedState, create_react_agent
from langgraph.prebuilt.chat_agent_executor import AgentState

class CustomState(AgentState):
    user_id: str

def get_user_info(
    # 关键在于此：使用 Annotated 和 InjectedState
    state: Annotated[CustomState, InjectedState]
) -> str:
    """根据状态中的 user_id 查询用户信息。"""
    user_id = state["user_id"]
    #... 查询数据库...
    return f"用户信息查询结果 for {user_id}"

核心机制：state: Annotated[CustomState, InjectedState] 。

InjectedState 是一个特殊的标记，它告诉 LangGraph 运行时：“在调用这个工具时，请将当前的完整状态对象作为第一个参数注入进来。” 这使得工具能够基于图的当前状态（例如，从状态中获取 user_id）来执行更智能的操作。

2.3.3 从工具中写入状态

更进一步，工具甚至可以直接修改图的状态，这使得智能体能够通过工具使用来“学习”和更新自己的内部“信念”。

from langgraph.types import Command
from langchain_core.messages import ToolMessage

def update_user_info(...) -> Command:
    """查找并更新用户信息到状态中。"""
    #... 业务逻辑...
    user_name = "John Smith"
    
    # 返回一个 Command 对象，而不是简单的字符串
    return Command(update={
        "user_name": user_name,
        "messages": [ToolMessage(content="Updated user info", tool_call_id="...")]
    })

核心机制：返回一个 Command 对象 。

当工具的返回类型被注解为 Command 时，LangGraph 会将其返回值解释为对状态的直接修改指令。Command(update={...}) 中的字典定义了要更新的状态字段及其新值。这允许工具在完成其主要任务（例如，调用 API）的同时，将结果写回智能体的短期记忆中，从而影响后续的决策。

InjectedState 和 Command 这两种模式，将工具从被动的外部执行单元提升为状态图中的一等公民。它们能够主动地感知和改变智能体的内部世界，这是构建能够通过与环境交互来动态调整自身状态的、更高级自主智能体的关键一步。

第三部分：使用 Store 构建长期记忆

如果说 Checkpointer 赋予了智能体流畅对话的能力，那么 Store 则赋予了它沉淀知识、形成持久记忆的能力。本部分将深入探讨如何使用 Store 为您的智能体构建一个跨越时间、跨越会话的知识库。

3.1 基础概念与 InMemoryStore

与 Checkpointer 类似，LangGraph 提供了一个 InMemoryStore 用于快速开发和原型验证。它将所有数据存储在内存中，API 与持久化 Store 保持一致。

from langgraph.store.memory import InMemoryStore
from langgraph.graph import StateGraph

# 1. 实例化一个内存 Store
store = InMemoryStore()

builder = StateGraph(...)
#...

# 2. 在编译图时传入 store
graph = builder.compile(store=store)

通过在 compile 方法中传入 store 参数，该 store 实例就在整个图的执行上下文中变得可用 。

3.2 用于生产应用的持久化 Store

为了让记忆真正“长期”，生产环境必须使用数据库支持的 Store。LangGraph 目前主要支持 PostgresStore 和 RedisStore。

3.2.1 使用 PostgresStore 和 RedisStore

配置过程与 Checkpointer 类似，但关键区别在于 Store 是如何在节点内部被访问和使用的。

安装：

• PostgreSQL: pip install -U "psycopg[binary,pool]" langgraph langgraph-checkpoint-postgres
• Redis: pip install -U langgraph langgraph-checkpoint-redis

完整代码示例（PostgreSQL）：
下面的示例展示了 Store 的核心用法。智能体被要求记住用户的名字 “Bob”。这个信息通过 store.put() 被写入数据库。随后，在一个全新的对话线程 (thread_id: “2”) 中，当被问及名字时，智能体通过 store.search() 从数据库中检索到了这个信息，并成功回答。这清晰地展示了 Store 跨会话记忆的能力。

同步 (Sync) 示例:

import uuid
from langchain_core.runnables import RunnableConfig
from langgraph.store.base import BaseStore
from langgraph.store.postgres import PostgresStore
from langgraph.checkpoint.postgres import PostgresSaver

DB_URI = "postgresql://postgres:postgres@localhost:5442/postgres?sslmode=disable"

with (
    PostgresStore.from_conn_string(DB_URI) as store,
    PostgresSaver.from_conn_string(DB_URI) as checkpointer,
):
    # store.setup()
    # checkpointer.setup()

    # 1. 节点函数签名中声明 store 参数
    def call_model(
        state: MessagesState,
        config: RunnableConfig,
        *,
        store: BaseStore,
    ):
        user_id = config["configurable"]["user_id"]
        namespace = ("memories", user_id)

        # 2. 从 Store 中读取（搜索）记忆
        memories = store.search(namespace, query=str(state["messages"][-1].content))
        info = "\n".join([d.value["data"] for d in memories])
        system_msg = f"用户信息: {info}"

        # 3. 向 Store 中写入记忆
        last_message = state["messages"][-1]
        if "remember" in last_message.content.lower():
            memory_to_save = "User name is Bob"
            # 使用 uuid 作为 key 确保唯一性
            store.put(namespace, str(uuid.uuid4()), {"data": memory_to_save})

        response = model.invoke(...)
        return {"messages": response}

    builder = StateGraph(MessagesState)
    builder.add_node("call_model", call_model)
    builder.add_edge(START, "call_model")

    # 4. 编译时同时传入 checkpointer 和 store
    graph = builder.compile(
        checkpointer=checkpointer,
        store=store,
    )

    # 第一次对话，要求记住名字
    config1 = {"configurable": {"thread_id": "1", "user_id": "user_001"}}
    graph.invoke(
        {"messages": [{"role": "user", "content": "remember my name is Bob"}]},
        config1
    )

    # 第二次对话（新线程），询问名字
    config2 = {"configurable": {"thread_id": "2", "user_id": "user_001"}}
    final_state = graph.invoke(
        {"messages": [{"role": "user", "content": "what is my name?"}]},
        config2
    )
    final_state["messages"][-1].pretty_print()

代码分析：

• def call_model(..., *, store: BaseStore): 这是在节点中访问 Store 的标准方式。通过在函数签名中添加一个关键字参数 store，LangGraph 运行时会自动将编译时传入的 store 实例注入进来。
• store.search(...): 用于从 Store 中检索信息。
• store.put(...): 用于向 Store 中写入信息。注意使用 namespace 和 user_id 来组织数据。
• graph.compile(checkpointer=checkpointer, store=store): 一个强大的智能体通常需要同时配置短期和长期记忆。

异步版本的逻辑完全相同，只需使用 AsyncPostgresStore 和 async/await 即可 。

3.3 用长期知识赋能工具

与短期记忆类似，工具也可以与长期记忆 Store 进行交互，从而执行依赖于持久化知识的复杂任务。

3.3.1 从 Store 中读取

from langgraph.config import get_store

def get_user_info(config: RunnableConfig) -> str:
    """从 Store 中查找用户信息。"""
    # 1. 获取在上下文中可用的 store 实例
    store = get_store()
    
    user_id = config["configurable"].get("user_id")
    
    # 2. 使用 store 实例进行查询
    user_info_doc = store.get(("users",), user_id)
    
    return str(user_info_doc.value) if user_info_doc else "Unknown user"

# 创建 Agent 时传入 store
agent = create_react_agent(..., store=store)

核心机制：store = get_store() 。

这个函数是一个上下文感知（context-aware）的辅助函数。它能够在工具执行时，自动获取并返回在创建智能体时 compile 或 create_react_agent 中传入的那个 store 实例。

3.3.2 向 Store 写入

from typing_extensions import TypedDict

class UserInfo(TypedDict):
    name: str
    language: str

def save_user_info(user_info: UserInfo, config: RunnableConfig) -> str:
    """将用户信息保存到 Store。"""
    store = get_store()
    user_id = config["configurable"].get("user_id")
    
    # 将结构化数据写入 Store
    store.put(("users",), user_id, user_info)
    
    return "Successfully saved user info."

最佳实践：使用 TypedDict 或 Pydantic 模型来定义数据模式（schema）。这不仅能提供代码级别的类型提示，更重要的是，它为 LLM 提供了一个清晰的结构化输出格式。当 LLM 决定调用 save_user_info 工具时，它会知道需要生成一个包含 name 和 language 字段的字典，从而大大提高了工具调用的成功率和稳定性。

3.4 语义记忆：智能体知识的未来

LangGraph 的 Store 最强大的功能之一是支持语义搜索。这能将 Store 从一个简单的键值数据库，转变为一个功能完备的向量数据库。智能体不再只能通过精确的关键词来检索记忆，而是能够根据概念的相似性来查找相关信息。这实际上是在智能体的记忆系统内部，原生集成了一套检索增强生成（RAG）流程。

设置：
要启用语义搜索，需要在实例化 Store 时配置一个 index。

from langchain.embeddings import init_embeddings
from langgraph.store.memory import InMemoryStore

# 1. 初始化一个嵌入模型
embeddings = init_embeddings("openai:text-embedding-3-small")

# 2. 在创建 Store 时配置 index
store = InMemoryStore(
    index={
        "embed": embeddings,       # 指定用于向量化的模型
        "dims": 1536,              # 指定向量维度
    }
)

使用示例：
下面的示例中，我们向 Store 中存入了两个事实：“我喜欢披萨”和“我是一个水管工”。当用户说“我饿了”时，store.search 会进行语义相似度搜索。尽管查询中没有“披萨”这个词，但嵌入模型知道“饿”和“披萨”在语义上是高度相关的，因此成功地检索出了“我喜欢披萨”这条记忆，并将其提供给 LLM 作为上下文。

# 存入记忆
store.put(("user_123", "memories"), "1", {"text": "I love pizza"})
store.put(("user_123", "memories"), "2", {"text": "I am a plumber"})

# 进行语义搜索
# "I'm hungry" 在语义上与 "I love pizza" 更接近
items = store.search(
    ("user_123", "memories"), 
    query="I'm hungry", 
    limit=1
)
# items.value 将会是 {"text": "I love pizza"}

这种将 RAG 流程原生集成到记忆系统中的架构模式极其强大。智能体的“长期记忆”不再是一个被动存储事实的仓库，而是一个可以实时查询、用以增强自身推理和生成能力的动态知识库。这模糊了“记忆”和“知识库”之间的界限，为构建能够持续学习和成长的智能体铺平了道路。随着智能体与世界交互，它可以通过工具不断地将新知识写入其语义 Store，从而动态地扩展和优化其知识储备，使其在未来的交互中变得越来越智能。

第四部分：主动内存管理策略

随着对话的进行，短期记忆（即消息历史）会不断增长。如果不加以管理，很快就会超出 LLM 的上下文窗口限制（context window），导致 API 调用失败或更糟——模型在无提示的情况下丢失早期上下文。因此，主动的内存管理不是一个可选项，而是构建任何非平凡、可长期运行智能体的必需品。

LangGraph 提供了多种策略来应对这一挑战。

4.1 上下文窗口挑战

所有 LLM 都有一个输入token数量的上限。例如，一个模型的上下文窗口可能是 4096 或 32768 个token。一次对话，包括历史消息、系统提示、工具定义等，其总长度不能超过这个限制。本节将介绍三种核心策略来解决这个问题：修剪、删除和总结。

4.2 策略一：修剪消息 (Trimming)

这是最简单直接的策略：当消息历史过长时，从开头或结尾丢弃一部分消息，以确保总长度符合限制。

在智能体 (Agent) 中实现：

使用 create_react_agent 等预构建的智能体时，可以通过 pre_model_hook 参数轻松实现。这个钩子函数会在每次调用 LLM 之前执行，是修改输入的理想位置。

from langchain_core.messages.utils import trim_messages, count_tokens_approximately

def pre_model_hook(state):
    trimmed_messages = trim_messages(
        state["messages"],
        strategy="last",  # 保留最新的消息
        token_counter=count_tokens_approximately,
        max_tokens=4096,
        start_on="human", # 确保修剪后以人类消息开始
    )
    # 将修剪后的消息放入一个新键，供 LLM 使用
    return {"llm_input_messages": trimmed_messages}

agent = create_react_agent(
    model,
    tools,
    pre_model_hook=pre_model_hook,
    checkpointer=checkpointer,
)

trim_messages 是一个非常方便的工具函数，strategy="last" 表示保留最近的消息，丢弃最老的消息，这是最常见的修剪策略 。

在工作流 (Workflow) 中实现：

如果想获得更精细的控制，可以直接在调用模型的节点内部实现修剪逻辑。

def call_model(state: MessagesState):
    messages = trim_messages(
        state["messages"],
        strategy="last",
        max_tokens=4096,
    )
    response = model.invoke(messages)
    return {"messages": [response]}

4.3 策略二：删除消息 (Deleting)

这是一种更具针对性的方法，允许从状态中永久移除特定的消息。这对于清理不再需要的系统指令、冗余的工具输出或错误信息非常有用。

实现方式：

要删除消息，不能直接从状态的 messages 列表中移除。正确的做法是，在节点中返回一个包含 RemoveMessage 对象的列表。LangGraph 的状态更新机制会识别这些对象，并执行删除操作。

删除特定消息:

from langchain_core.messages import RemoveMessage

def delete_messages_node(state):
    messages = state["messages"]
    if len(messages) > 10:
        # 假设我们要删除最早的两条消息
        messages_to_delete = messages[:2]
        return {
            "messages": [RemoveMessage(id=m.id) for m in messages_to_delete]
        }
    return {}

删除所有消息:

LangGraph 提供了一个特殊的常量 REMOVE_ALL_MESSAGES 来清空整个消息历史。

from langgraph.graph.message import REMOVE_ALL_MESSAGES

def clear_history_node(state):
    return {"messages": REMOVE_ALL_MESSAGES}

4.4 策略三：总结对话 (Summarizing)

这是最智能但也是最复杂的策略。它不是简单地丢弃旧消息，而是使用一个 LLM 将对话的早期部分压缩成一个摘要。这样既减小了token占用，又最大限度地保留了核心信息。

在智能体 (Agent) 中实现（推荐）：

langmem 库提供了一个预构建的 SummarizationNode，可以极大地简化实现过程。

from langmem.short_term import SummarizationNode, RunningSummary

# 1. 配置总结节点
summarization_node = SummarizationNode(
    token_counter=count_tokens_approximately,
    model=summarization_model, # 用于生成摘要的模型
    max_tokens=4096,           # 输入给主模型的最大 token 数
    max_summary_tokens=512,    # 摘要本身的最大 token 数
    output_messages_key="llm_input_messages",
)

# 2. 扩展状态以包含总结上下文
class State(AgentState):
    context: dict

# 3. 将总结节点用作 pre_model_hook
agent = create_react_agent(
   ...,
    pre_model_hook=summarization_node,
    state_schema=State,
    checkpointer=checkpointer,
)

这里，SummarizationNode 被用作 pre_model_hook。它会自动检查消息历史的长度，当超过阈值时，触发一次总结，然后将“摘要 + 最近消息”的组合传递给主模型。State 中的 context 字段用于存储运行中的摘要信息，避免在每次调用时都重复总结 。

在工作流 (Workflow) 中自定义实现：

开发者也可以从零开始构建自己的总结逻辑，这提供了最大的灵活性。

class StateWithSummary(MessagesState):
    summary: str

def summarize_conversation_node(state: StateWithSummary):
    # 获取已有的摘要
    current_summary = state.get("summary", "")

    # 构建用于生成新摘要的提示
    prompt = f"当前摘要: {current_summary}\n\n请结合以下新消息扩展摘要:..."

    # 调用 LLM 生成新摘要
    new_summary = model.invoke(prompt).content

    # 找出已经被总结的消息并准备删除它们
    messages_to_delete = [...] 

    return {
        "summary": new_summary,
        "messages": [RemoveMessage(id=m.id) for m in messages_to_delete]
    }

这个自定义节点负责维护状态中的 summary 字段，并定期清理已被总结的旧消息。

选择哪种内存管理策略，本质上是在计算成本、信息保真度和实现复杂度之间做出权衡。

• 修剪：成本最低，实现最简单，但信息丢失最严重。它像一把钝刀，粗暴地切掉历史。
• 删除：成本低，复杂度中等，可以精确控制信息移除，但需要自定义逻辑。
• 总结：成本最高（需要额外的 LLM 调用），实现最复杂，但信息保真度最高。它保留了对话的语义精髓。

没有一种策略是普适的“最佳”选择。开发者应根据应用的具体需求来决定。一个简单的问答机器人可能用修剪就足够了，而一个需要建立长期关系的心理辅导或私人教练智能体，则必须使用总结策略来维持对用户历史的深刻理解。

第五部分：管理控制与调试

构建和部署了有状态的智能体后，对其内存进行检查、管理和重置的能力变得至关重要，这对于调试、监控和用户体验管理都必不可少。

5.1 检查智能体状态

LangGraph 提供了简单的 API 来查看任何给定对话线程的当前状态和历史状态。

查看线程的当前状态（最新的检查点）：

使用 graph.get_state(config) 方法可以获取指定 thread_id 的最新状态快照。

config = {"configurable": {"thread_id": "1"}}
latest_state = graph.get_state(config)
print(latest_state.values)

也可以通过 checkpointer 的底层 API 实现：checkpointer.get_tuple(config) 。

查看线程的状态历史：

使用 graph.get_state_history(config) 可以获取一个迭代器，它会按时间顺序返回该线程的所有历史检查点。这对于追溯状态是如何一步步演变的、排查问题非常有帮助。

config = {"configurable": {"thread_id": "1"}}
for state_snapshot in graph.get_state_history(config):
    print(f"Checkpoint ID: {state_snapshot.checkpoint_id}")
    print(state_snapshot.values)

对应的底层 API 是 checkpointer.list(config) 。

5.2 重置对话

在很多应用中，用户需要能够“开始新的聊天”，这在后端就对应着重置或删除当前的对话历史。此外，出于数据隐私合规（如 GDPR）的考虑，也需要有能力永久删除用户数据。

删除线程的所有检查点：

checkpointer 对象提供了一个直接的方法 delete_thread 来完成这个任务。

thread_id_to_delete = "1"
checkpointer.delete_thread(thread_id_to_delete)

执行此命令后，与 thread_id="1" 相关的所有短期记忆（检查点）都将被从后端数据库中永久删除 。

结论：构建一致的内存策略

本指南全面探讨了 LangGraph 的双轨内存架构，从基础概念到高级应用，再到生产环境中的管理策略。我们看到，LangGraph 不仅仅提供了一个简单的记忆功能，而是提供了一套完整、强大且灵活的状态管理工具集，让开发者能够构建出真正具有深度和持久性的智能体。

为了成功地构建您自己的有状态应用，以下是一些关键的架构性建议：

• 从规划开始：在编写代码之前，清晰地规划您的内存策略。明确哪些信息属于短暂的对话上下文（应由 Checkpointer 管理），哪些是需要永久保存的核心知识（应存入 Store）。
• 选择合适的后端：在开发阶段使用 InMemory 版本进行快速迭代，但在部署生产环境时，根据团队的技术栈、性能需求和数据一致性要求，谨慎选择一个持久化的数据库（如 PostgreSQL, Redis）。
• 让工具具备状态感知能力：充分利用 InjectedState 和 get_store()，将您的工具从无状态的函数转变为能够感知和利用智能体内部状态的强大组件。
• 主动管理上下文：不要等到上下文窗口溢出时才去补救。在开发初期就实施一种内存管理策略（修剪、删除或总结），确保您的智能体能够稳健地处理长时间的交互。

最后，LangGraph 的生态系统还在不断发展。LangMem 库 提供了更多预构建的高级内存工具和策略，值得进一步探索。通过结合 LangGraph 强大的核心框架和社区贡献的工具，您将能够构建出下一代的人工智能应用，它们不仅能言善辩，更能记忆、学习和成长。