LangChain：短记忆之长对话的处理方式

记忆（Memory）是一个能够记住先前交互信息的系统。对于 AI 代理（Agent）而言，记忆至关重要，因为它使得代理能够记住之前的交互、从反馈中学习，并适应用户的偏好。

随着代理处理的任务日益复杂、与用户交互次数的增多，这种能力对于提升效率和用户满意度都变得不可或缺。

短期记忆（Short-term Memory） 让应用程序能够记住单个线程（thread）或对话中的先前交互。最常见的短期记忆形式就是对话历史（Conversation history）。

概述

记忆（Memory）是一个能够记住先前交互信息的系统。对于 AI 代理（Agent）而言，记忆至关重要，因为它使得代理能够记住之前的交互、从反馈中学习，并适应用户的偏好。

随着代理处理的任务日益复杂、与用户交互次数的增多，这种能力对于提升效率和用户满意度都变得不可或缺。

短期记忆（Short-term Memory） 让应用程序能够记住单个线程（thread）或对话中的先前交互。最常见的短期记忆形式就是对话历史。

长期对话的挑战

长对话对当今的大语言模型（LLM）构成了挑战：

• 上下文窗口限制：完整的对话历史可能无法放入 LLM 的上下文窗口（context window）中，导致上下文丢失或出错。
• 性能下降：即使模型支持完整的上下文长度，大多数 LLM 在长上下文上的表现仍然不佳——它们会被陈旧或偏离主题的内容“分散注意力”，同时响应变慢、成本更高。
• 消息列表增长：聊天模型通过消息（messages）接收上下文，包括指令（系统消息）和输入（用户消息）。在聊天应用中，消息在用户输入和模型响应之间交替，导致消息列表随着时间的推移越来越长。

由于上下文窗口是有限的，许多应用程序可以受益于使用技术来移除或“遗忘”陈旧信息。

一个线程（thread）将会话中的多次交互组织在一起，类似于电子邮件往来0将消息分组到同一个对话中。

区分短期与长期记忆：如需跨对话记住信息，应使用长期记忆（long-term memory）在不同线程和会话之间存储和检索用户特定或应用级别的数据。简单来说：短期记忆关乎“当前会话的连续性”，长期记忆关乎“跨会话的个性化”。

使用方法

要为智能体添加短期记忆（线程级持久化），需要在创建智能体时指定一个检查点器（checkpointer）。LangChain的智能体将短期记忆作为智能体状态（state）的一部分进行管理

通过将这些信息存储在图的状态中，智能体可以访问特定对话的完整上下文，同时保持不同线程之间的隔离。状态通过检查点器持久化到数据库（或内存）中，因此线程可以随时恢复。短期记忆在智能体被调用或完成一个步骤（如工具调用）时更新，状态则会在每个步骤开始时被读取。

核心机制说明：这里的“检查点器”（checkpointer）是一个负责保存和恢复智能体状态的组件。当智能体执行对话时，检查点器会自动将每一轮的对话状态（包括消息历史、用户ID、偏好设置等）保存下来。这样即使程序重启，只要提供相同的thread_id，智能体就能恢复到之前的对话上下文。

基础示例

基础示例：使用内存检查点器

# 导入创建代理的函数
from langchain.agents import create_agent
# 导入内存检查点器（仅用于开发和测试）
from langgraph.checkpoint.memory import InMemorySaver

# 创建代理：指定模型、工具列表和检查点器
# InMemorySaver 将状态保存在内存中，重启后丢失
agent = create_agent(
    "gpt-5.4",                     # 使用的 LLM 模型
    tools=[get_user_info],         # 代理可使用的工具列表
    checkpointer=InMemorySaver(),  # 检查点器：负责持久化状态
)

# 调用代理：发送用户消息，并指定线程 ID
agent.invoke(
    {"messages": [{"role": "user", "content": "Hi! My name is Bob."}]},
    {"configurable": {"thread_id": "1"}},  # thread_id 用于区分不同对话线程
)

代码说明：此示例展示了如何为智能体添加短期记忆。

• create_agent() 是 LangChain 提供的代理工厂函数，用于快速创建具备记忆能力的 Agent。
• InMemorySaver 将对话状态暂存在内存中，适合开发调试，但不适用于生产环境。
• thread_id: "1" 为该对话线程分配唯一标识，同一个 thread_id 的多次调用会共享对话历史。

生产环境

在生产环境中，使用数据库检查点器，应该使用由数据库支持的检查点器，以确保状态的持久化和可靠性：

# 安装 PostgreSQL 支持的检查点器
pip install langgraph-checkpoint-postgres

# 导入创建代理的函数
from langchain.agents import create_agent
# 导入 PostgreSQL 检查点器
from langgraph.checkpoint.postgres import PostgresSaver

# 数据库连接字符串（请根据实际配置修改）
DB_URI = "postgresql://postgres:postgres@localhost:5432/postgres?sslmode=disable"

# 使用上下文管理器确保连接正确关闭
with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
    checkpointer.setup()  # 自动在 PostgreSQL 中创建所需的数据表
    # 创建代理，使用 PostgreSQL 作为持久化后端
    agent = create_agent(
        "gpt-5.4",               # 使用的 LLM 模型
        tools=[get_user_info],   # 代理可使用的工具列表
        checkpointer=checkpointer,  # 使用数据库检查点器
    )

代码说明：

• PostgresSaver 将对话状态持久化到 PostgreSQL 数据库，代理重启后对话历史不会丢失。
• checkpointer.setup() 会自动创建所需的数据库表结构，无需手动建表。
• 生产环境建议使用数据库（PostgreSQL、SQLite）或云服务（Azure Cosmos DB）作为检查点后端。

更多检查点器选项（包括 SQLite、Postgres、Azure Cosmos DB），请参阅持久化文档中的检查点器库列表。

自定义智能体记忆

默认情况下，代理使用 AgentState 来管理短期记忆，特别是通过 messages 键来维护对话历史。开发者可以通过扩展 AgentState 来添加自定义字段。

为什么需要自定义状态：默认的AgentState只包含messages字段。如果智能体需要记住用户ID、主题偏好、会话标签等额外信息，就需要扩展自定义状态。这些自定义字段会被检查点器自动保存，并在后续的对话步骤中可用。

# 导入创建智能体和AgentState基类
from langchain.agents import create_agent, AgentState
# 导入内存检查点器
from langgraph.checkpoint.memory import InMemorySaver

# 定义自定义状态类，继承AgentState
class CustomAgentState(AgentState):
    user_id: str          # 添加用户ID字段
    preferences: dict     # 添加偏好设置字段

# 创建智能体，通过state_schema参数指定自定义状态
agent = create_agent(
    "gpt-5.4",                           # 指定的语言模型
    tools=[get_user_info],               # 工具列表
    state_schema=CustomAgentState,       # 使用自定义状态结构
    checkpointer=InMemorySaver(),        # 内存检查点器
)

# 调用智能体时，可以传入自定义状态字段
result = agent.invoke(
    {
        "messages": [{"role": "user", "content": "Hello"}],  # 对话消息
        "user_id": "user_123",                              # 自定义字段：用户ID
        "preferences": {"theme": "dark"}                     # 自定义字段：偏好设置
    },
    {"configurable": {"thread_id": "1"}}                     # 线程配置
)

代码说明：通过继承AgentState，可以为智能体添加任意自定义字段。这些字段会随对话状态一起被检查点器保存，从而在智能体执行过程中保持可用。

• AgentState 是代理状态的基类，包含 messages 字段用于存储对话历史。
• 通过继承 AgentState 并添加自定义字段，可以让代理记住更多上下文信息。
• 自定义字段的值在每次 invoke 调用时传入，并随对话状态一起持久化。

长对话的记忆处理

启用了短期记忆后，长对话仍然可能超出 LLM 的上下文窗口。以下是几种常见的解决方案：

策略	原理	适用场景
修剪消息（Trim ）	只保留最近的消息，移除旧消息	对话内容较长，但旧消息相关性低
删除消息（Delete）	手动移除指定消息或清空对话	需要删除敏感内容或重置对话
摘要总结（Summarize）	用 LLM 对对话历史进行摘要压缩	需要保留语义但压缩 token 用量
自定义策略 （Custom strategies）		可用于关键词过滤、敏感信息删除等

几种策略的适用场景：修剪和删除适用于对历史细节要求不高的场景（如任务型机器人）；总结适用于需要保留语义要点但不需要逐字历史的长对话场景；自定义策略则可用于关键词过滤、敏感信息删除等特殊需求。

修剪消息（Trim Messages）

大多数 LLM 有最大上下文窗口限制（以 token 数量计）。一种常见的处理方式是统计消息历史中的 token 数量，并在接近限制时进行截断。如果你正在使用 LangChain，可以使用 trim_messages 工具来指定从列表中保留的 token 数量以及处理边界的策略。

在代理中修剪消息历史，使用 @before_model 中间件装饰器：

# 导入必要的类和函数
from langchain.messages import RemoveMessage
from langgraph.graph.message import REMOVE_ALL_MESSAGES
from langgraph.checkpoint.memory import InMemorySaver
from langchain.agents import create_agent, AgentState
from langchain.agents.middleware import before_model
from langgraph.runtime import Runtime
from langchain_core.runnables import RunnableConfig
from typing import Any

# 使用before_model装饰器定义消息修剪中间件
# 该装饰器确保函数在模型被调用之前执行
@before_model
def trim_messages(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
    """
    仅保留最后几条消息以适应上下文窗口。
    
    参数说明:
        state: 智能体的当前状态，包含消息列表
        runtime: 运行时上下文，提供执行环境信息
    
    返回:
        更新后的消息字典，如果无需修改则返回None
    """
    # 从状态中获取消息列表
    messages = state["messages"]
    
    # 如果消息数量不超过3条，无需修改
    if len(messages) <= 3:
        return None
    
    # 保留第一条消息（通常是系统提示或最早的用户消息）
    first_msg = messages[0]
    # 保留最后3条消息
    recent_messages = messages[-3:] if len(messages) % 2 == 0 else messages[-4:]
    # 组合新的消息列表：第一条消息 + 最新的消息
    new_messages = [first_msg] + recent_messages
    
    return {
        "messages": [
            RemoveMessage(id=REMOVE_ALL_MESSAGES),  # 移除所有现有消息
            *new_messages                           # 添加新消息列表
        ]
    }

# 创建智能体并配置修剪中间件
agent = create_agent(
    your_model_here,        # 替换为实际的语言模型
    tools=your_tools_here,  # 替换为实际的工具列表
    middleware=[trim_messages],   # 添加修剪中间件
    checkpointer=InMemorySaver(), # 使用内存检查点器
)

# 配置线程ID
config: RunnableConfig = {"configurable": {"thread_id": "1"}}

# 多轮对话示例：智能体将记住用户的名字
agent.invoke({"messages": "hi, my name is bob"}, config)
agent.invoke({"messages": "write a short poem about cats"}, config)
agent.invoke({"messages": "now do the same but for dogs"}, config)

# 验证智能体是否还记得用户的名字
final_response = agent.invoke({"messages": "what's my name?"}, config)
final_response["messages"][-1].pretty_print()

代码说明：

• @before_model 装饰器标记的函数会在每次模型调用之前执行，用于预处理消息。
• RemoveMessage(id=REMOVE_ALL_MESSAGES) 会清空当前消息列表，然后 new_messages 添加保留的消息。
• 示例中的修剪策略保留了第一条消息（通常是系统提示）和最近的三条消息，既能控制上下文长度，又能维持对话的连贯性。
• 注意：修剪消息会丢失被移除消息中的信息，适用于旧消息相关性较低的对话场景。

删除消息（Delete Messages）

可以从 Graph 状态中删除消息来管理消息历史。这在需要移除特定消息或清空整个消息历史时非常有用。

删除特定消息

from langchain.messages import RemoveMessage

def delete_messages(state):
    messages = state["messages"]
    if len(messages) > 2:
        # 移除最早的两条消息
        return {"messages": [RemoveMessage(id=m.id) for m in messages[:2]]}

删除所有消息

from langgraph.graph.message import REMOVE_ALL_MESSAGES

def delete_messages(state):
    return {"messages": [RemoveMessage(id=REMOVE_ALL_MESSAGES)]}

完整示例

完整示例：在模型响应后自动删除旧消息

# 导入所需模块
from langchain.messages import RemoveMessage
from langchain.agents import create_agent, AgentState
from langchain.agents.middleware import after_model
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.runtime import Runtime
from langchain_core.runnables import RunnableConfig

# 使用after_model装饰器，在模型响应后执行消息删除
@after_model
def delete_old_messages(state: AgentState, runtime: Runtime) -> dict | None:
    """
    移除旧消息以保持对话可控。
    该函数在模型生成响应后执行。
    """
    messages = state["messages"]
    # 如果消息超过2条，移除最早的两条消息
    if len(messages) > 2:
        return {"messages": [RemoveMessage(id=m.id) for m in messages[:2]]}
    return None

# 创建智能体，配置删除中间件
agent = create_agent(
    "gpt-5-nano",                                    # 指定的模型
    tools=[],                                         # 工具列表（此处为空）
    system_prompt="Please be concise and to the point.",  # 系统提示
    middleware=[delete_old_messages],                # 添加删除中间件
    checkpointer=InMemorySaver(),                    # 内存检查点器
)

# 配置线程ID
config: RunnableConfig = {"configurable": {"thread_id": "1"}}

# 第一轮对话：介绍名字
for event in agent.stream(
    {"messages": [{"role": "user", "content": "hi! I'm bob"}]},
    config,
    stream_mode="values",
):
    print([(message.type, message.content) for message in event["messages"]])

# 第二轮对话：询问名字
for event in agent.stream(
    {"messages": [{"role": "user", "content": "what's my name?"}]},
    config,
    stream_mode="values",
):
    print([(message.type, message.content) for message in event["messages"]])

预期输出：

[('human', "hi! I'm bob")]
[('human', "hi! I'm bob"), ('ai', 'Hi Bob! Nice to meet you. How can I help you today?')]
[('human', "hi! I'm bob"), ('ai', 'Hi Bob! Nice to meet you.'), ('human', "what's my name?")]

代码说明：

• @after_model 装饰器标记的函数会在每次模型调用之后执行，可用于清理或验证响应。示例中：删除后，较早的消息（如"hi! I'm bob"）会从状态中被移除，因此后续对话中智能体可能无法获取这些信息——这既是效果也是提醒：删除消息是永久性的。
• stream 方法以流式方式返回每次状态更新，便于观察消息变化的中间过程。

消息删除的有效性要求：删除消息时，需确保生成的消息历史是有效的。不同LLM提供商的限制有所不同：有些要求消息历史必须以用户消息开头；大多数要求带有工具调用的助手消息后必须跟有对应的工具结果消息。

摘要消息（Summarize Messages）

修剪或删除消息的问题在于可能会丢失被移除消息中的信息。因此，一些应用程序受益于更复杂的方法——使用聊天模型对消息历史进行摘要总结。

使用内置的 SummarizationMiddleware 可以方便地实现这一功能：

# 导入创建智能体函数和总结中间件
from langchain.agents import create_agent
from langchain.agents.middleware import SummarizationMiddleware
from langgraph.checkpoint.memory import InMemorySaver
from langchain_core.runnables import RunnableConfig

# 创建检查点器
checkpointer = InMemorySaver()

# 创建智能体，配置总结中间件
agent = create_agent(
    model="gpt-5.4",                       # 主模型
    tools=[],                              # 工具列表
    middleware=[
        SummarizationMiddleware(
            model="gpt-5.4-mini",          # 用于生成摘要的模型（可使用更经济的模型）
            trigger=("tokens", 4000),      # 触发条件：token数达到4000时触发总结
            keep=("messages", 20)          # 总结后保留的最新消息条数
        )
    ],
    checkpointer=checkpointer,
)

# 配置线程ID
config: RunnableConfig = {"configurable": {"thread_id": "1"}}

# 多轮对话示例
agent.invoke({"messages": "hi, my name is bob"}, config)
agent.invoke({"messages": "write a short poem about cats"}, config)
agent.invoke({"messages": "now do the same but for dogs"}, config)

# 验证智能体是否还记得名字
final_response = agent.invoke({"messages": "what's my name?"}, config)
final_response["messages"][-1].pretty_print()

预期输出：

================================== Ai Message ==================================
Your name is Bob!

代码说明：

• SummarizationMiddleware 会在消息 token 数超过 trigger 阈值时，自动使用指定的模型对历史消息生成摘要。
• 生成摘要后，历史消息会被压缩成摘要保留，同时保留最近 keep 条原始消息，以维持对话的连贯性。
• 这种方式比单纯删除消息更智能，因为它保留了关键信息，只是以更高效的方式存储。
• 建议使用轻量级模型（如示例中的 gpt-5.4-mini）来生成摘要，以节省成本。
• 配置详解：trigger=("tokens", 4000)表示当总token数达到4000时触发总结；keep=("messages", 20)表示总结后保留最新20条原始消息，其余被总结的消息将被摘要替换。