摘要:短期记忆(Short-term memory)是构建智能对话系统的核心组件,它让智能体能够记住同一会话线程内的历史交互,实现连贯的多轮对话。本文将深入解析 LangChain 框架中的短期记忆机制,从基础概念到生产实践,系统阐述检查点(Checkpoint)持久化、自定义状态扩展、长对话处理模式(精简/删除/摘要/滑动窗口)以及多维度内存访问策略。通过丰富的代码示例与架构图解,帮助开发者构建具备高效上下文管理能力的 AI 应用。

关键字:LangChain;短期记忆;Short-term Memory;Agent;Checkpoint;上下文窗口;消息管理;LangGraph


文章目录


1. Short-term memory介绍

1.1 什么是短期记忆

记忆(Memory)是一种能够留存过往交互信息的系统。对于智能体(Agent)而言,记忆至关重要——它帮助智能体记住历史交互内容、从反馈中学习,并适配用户偏好。当智能体需要处理包含大量用户交互的复杂任务时,这项能力对提升运行效率与用户体验都不可或缺。

在 LangChain 生态中,短期记忆以 会话线程(Thread) 为单位组织。每个线程通过唯一的 thread_id 标识,同一线程内的多轮对话共享一份状态,不同线程之间完全隔离。这意味着:

  • 用户 A 在 thread_1 中的对话历史不会泄露到用户 B 的 thread_2;
  • 同一用户可以在不同线程中开启独立话题,互不干扰;
  • 线程状态可随时恢复,支持长会话的断点续聊。

1.2 对话历史与上下文窗口挑战

对话历史是短期记忆最常见的形式。对话模型通过 messages 接收上下文,消息列表包含系统提示(SystemMessage)与用户输入(HumanMessage),模型回复(AIMessage)交替追加,消息列表随时间不断增长。

然而,长对话对大语言模型构成实质性挑战:

挑战具体表现
上下文窗口溢出完整历史可能超出模型的 token 上限,导致截断或报错
注意力稀释过时或偏离主题的内容使模型 “注意力分散”,回复质量下降
性能与成本长上下文导致推理延迟增加、API 调用成本攀升

💡 本质问题:LLM 的上下文窗口并非 “越大越好”。即便 GPT-4-128K 能容纳整本书,模型在长上下文中准确定位关键信息的能力仍有衰减。因此,主动管理消息队列比被动依赖大窗口更为重要。

1.3 短期记忆 vs 长期记忆

一个常见的认知误区是将 “短期记忆” 与 “长期记忆” 混为一谈。两者的核心区别如下:

维度短期记忆(Short-term Memory)长期记忆(Long-term Memory)
作用域单个会话线程内跨会话、跨线程
生命周期会话结束即归档/清除持久化存储,用户级别
典型数据对话历史、临时变量用户偏好、知识库、历史摘要
实现方式Checkpoint + StateVectorStore / Database + Embedding

如果需要在不同对话间留存信息(如用户档案、偏好设置),则需要借助长期记忆功能,在不同对话线程与会话中存储、调取用户专属数据或应用级数据。本文聚焦短期记忆,长期记忆将在后续文章中单独讨论。


2. 使用短期记忆

2.1 检查点与线程隔离机制

LangChain 智能体将短期记忆作为 图状态(Graph State) 的组成部分进行管理。其核心机制如下:

用户请求
thread_id='abc'

检查点器
是否存在该线程?

加载已有状态
恢复 messages 历史

初始化新状态
空 messages 列表

Agent 执行步骤

工具调用 / 模型推理

更新状态
追加新消息

持久化到检查点

返回响应

用户继续对话?

线程挂起
状态保留

关键设计要点:

  • 每步自动保存:每当智能体完成一个步骤(如工具调用),短期记忆就会自动更新并持久化;
  • 每步自动读取:每个步骤开始时,系统从检查点加载最新状态;
  • 线程隔离:不同 thread_id 的状态在物理上隔离,确保数据安全。

2.2 基础示例:内存检查点

以下示例使用 InMemorySaver 作为检查点器,适合开发调试:

from langchain.agents import create_agent
from langgraph.checkpoint.memory import InMemorySaver  

def get_user_info() -> str:
    """Look up information about the current user."""
    return "No user profile on file."

agent = create_agent(
    model="google_genai:gemini-3.5-flash",
    tools=[get_user_info],
    checkpointer=InMemorySaver(),
)

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

# 第一轮对话:告知名字
response = agent.invoke(
    {"messages": [{"role": "user", "content": "Hi! My name is Bob."}]},
    thread_config,
)["messages"][-1].content
print(response)  # "Hi Bob! Nice to see you here. How are you doing?"

# 第二轮对话:验证记忆
response = agent.invoke(
    {"messages": [{"role": "user", "content": "What's my name?"}]},
    thread_config,
)["messages"][-1].content
print(response)  # "You are Bob!"

发生了什么?

  1. 第一轮 invoke 时,thread_id="1" 的状态为空,Agent 收到 “Hi! My name is Bob.” 并回复;
  2. 回复后,系统将用户消息和 AI 回复一同保存到 thread_id="1" 的状态中;
  3. 第二轮 invoke 时,Agent 加载 thread_id="1" 的已有状态,从历史消息中得知用户名为 Bob,正确回答。

⚠️ 注意InMemorySaver 将数据保存在内存中,进程重启即丢失。仅用于开发和测试。

2.3 生产环境:数据库持久化

在生产环境下,必须使用基于数据库的检查点器以保证数据持久性:

from langchain.agents import create_agent
from langgraph.checkpoint.postgres import PostgresSaver  

def get_user_info() -> str:
    """Look up information about the current user."""
    return "No user profile on file."

DB_URI = "postgresql://postgres:postgres@localhost:5432/postgres?sslmode=disable"
with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
    checkpointer.setup()  # 自动创建所需表结构
    agent = create_agent(
        "gpt-5.5",
        tools=[get_user_info],
        checkpointer=checkpointer,
    )

LangGraph 支持多种持久化后端:

后端适用场景特点
InMemorySaver开发/测试内存存储,进程重启后丢失
PostgresSaver生产环境关系型数据库,支持事务和查询
SqliteSaver轻量级应用本地文件存储,无需额外服务
RedisSaver高并发场景内存数据库,极低延迟

3. 自定义Agent记忆

3.1 扩展 AgentState

默认情况下,Agent 使用 AgentState 管理短期记忆,核心是 messages 键保存对话历史。但在实际业务中,我们往往需要额外的状态字段——比如用户 ID、会话偏好、业务中间结果等。

LangChain 支持通过 state_schema 参数传入自定义状态类,扩展 AgentState

继承扩展

AgentState

+list messages

CustomAgentState

+str user_id

+dict preferences

+list messages

3.2 自定义状态实战

from langchain.agents import create_agent, AgentState
from langgraph.checkpoint.memory import InMemorySaver


class CustomAgentState(AgentState):
    user_id: str
    preferences: dict

agent = create_agent(
    "gpt-5.5",
    tools=[get_user_info],
    state_schema=CustomAgentState,
    checkpointer=InMemorySaver(),
)

# 自定义状态在 invoke 时传入
result = agent.invoke(
    {
        "messages": [{"role": "user", "content": "Hello"}],
        "user_id": "user_123",
        "preferences": {"theme": "dark"}
    },
    {"configurable": {"thread_id": "1"}}
)

💡 设计思路:短期记忆不仅是消息队列的存储,更是一个 有状态的计算上下文。通过扩展 AgentState,你可以将业务数据与对话历史绑定,让 Agent 在工具调用、提示词构建等环节直接读取这些状态字段。


4. 长对话常见处理模式

开启短期记忆后,消息历史无限增长,必然触及上下文窗口上限。LangChain 提供了四种核心处理模式:
在这里插入图片描述

消息队列
无限增长

处理策略选择

精简消息
保留最近 N 条

删除消息
移除特定/全部消息

摘要消息
LLM 压缩历史

滑动窗口
固定大小的最近消息

4.1 精简消息(Trim Messages)

核心思路:统计 token 数量,当接近上限时截断,保留最近的消息。

判断何时截断的关键指标是 token 计数。LangChain 提供的内置工具可以指定保留的 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
def trim_messages(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
    """保留最近几条消息以适应上下文窗口。"""
    messages = state["messages"]

    if len(messages) <= 3:
        return None  # 消息较少,无需处理

    # 始终保留首条消息(通常包含系统指令或初始上下文)
    first_msg = messages[0]
    # 保留最近 3-4 条消息,确保保留完整对话轮次
    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(
    "gpt-5.5",
    tools=[...],
    middleware=[trim_messages],
    checkpointer=InMemorySaver(),
)

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. You told me that earlier.
If you'd like me to call you a nickname or use a different name, just say the word.
"""

💡 关键设计:注意代码中 first_msg = messages[0] 的设计——即使精简消息,也保留首条消息。因为首条消息通常包含重要的上下文锚点(如系统指令),完全丢弃可能导致 Agent “失忆”。

4.2 删除消息(Delete Messages)

核心思路:从图谱状态中精确删除特定消息或清空全部历史。

RemoveMessage 是 LangGraph 提供的消息删除原语。要让其生效,状态键需要使用 add_messages 归约器——幸运的是,默认的 AgentState 已提供该配置。

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)]}
方法适用场景
删除指定消息移除过时的工具调用结果、错误的用户输入
清空全部消息 (REMOVE_ALL_MESSAGES)用户显式重置会话、话题完全切换

4.3 摘要消息(Summarization)

核心思路:使用 LLM 将历史消息压缩为摘要,既释放上下文空间,又保留关键信息。

裁剪和删除的问题在于信息永久丢失。摘要策略通过 “压缩” 替代 “丢弃”——让对话模型对历史消息进行汇总,用简洁的摘要替代冗长的原始对话。LangChain 提供了内置的 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.5",
    tools=[...],
    middleware=[
        SummarizationMiddleware(
            model="gpt-5.4-mini",           # 使用更便宜的模型做摘要
            trigger=("tokens", 4000),        # 当 token 数超过 4000 时触发
            keep=("messages", 20)             # 保留最近 20 条原始消息
        )
    ],
    checkpointer=checkpointer,
)

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 核心参数

参数说明建议值
model执行摘要的模型(建议用轻量模型降本)gpt-4o-mini / claude-3-haiku
trigger触发摘要的条件(token 数或消息数)("tokens", 4000)("messages", 40)
keep保留的原始消息数量("messages", 20) 确保最近对话完整

4.4 滑动窗口策略

除了上述三种官方模式外,实践中常见的第四种策略是 滑动窗口(Sliding Window)

消息队列: [msg1, msg2, msg3, msg4, msg5, msg6, msg7, msg8, msg9, msg10]
                                         |___________滑动窗口___________|
                                                 (保留最近 N 条)

滑动窗口与 “精简消息” 的区别在于:滑动窗口每次只保留固定数量的最近消息(如最近 20 条),而不是基于 token 计数动态截断。这种方式更简单、性能更好,适合消息长度相对均匀的场景。

def sliding_window(state: AgentState, max_messages: int = 20):
    """保留最近 N 条消息的滑动窗口策略。"""
    messages = state["messages"]
    if len(messages) <= max_messages:
        return None
    # 保留首条消息(系统指令锚点)+ 最近消息
    return {
        "messages": [messages[0]] + messages[-(max_messages - 1):]
    }

5. 访问内存

LangChain 提供了多个维度的内存访问入口,让开发者能够在 Agent 生命周期的不同阶段读写短期记忆:

短期记忆(State)

Agent 生命周期

读取/修改

读取/修改

读取/写入

1. 模型调用前
before_model

2. LLM 推理

3. 模型调用后
after_model

4. 工具调用
ToolRuntime

messages / 自定义字段

5.1 工具层访问

5.1.1 在工具中读取短期记忆

借助 ToolRuntime 运行时参数,工具可在执行时读取当前短期记忆状态。关键特性:该参数不会出现在工具的签名中,模型无法感知它的存在,但工具内部能通过它获取完整的状态。

from langchain.agents import create_agent, AgentState
from langchain.tools import tool, ToolRuntime

class CustomState(AgentState):
    user_id: str

@tool
def get_user_info(
    runtime: ToolRuntime
) -> str:
    """Look up user info."""
    user_id = runtime.state["user_id"]
    return "User is John Smith" if user_id == "user_123" else "Unknown user"

agent = create_agent(
    model="gpt-5-nano",
    tools=[get_user_info],
    state_schema=CustomState,
)

result = agent.invoke({
    "messages": "look up user information",
    "user_id": "user_123"
})
print(result["messages"][-1].content)
# > User is John Smith.

💡 设计亮点ToolRuntime 对模型透明,这意味着模型不需要知道 “状态” 的存在。工具作为状态和模型之间的桥梁,在需要时从状态中提取信息,让模型聚焦于推理本身。

5.1.2 从工具写入短期记忆

工具不仅可以读取状态,还可以返回状态更新。这适用于留存中间结果,或让后续工具、提示词获取工具执行产生的信息。

from langchain.tools import tool, ToolRuntime
from langchain_core.runnables import RunnableConfig
from langchain.messages import ToolMessage
from langchain.agents import create_agent, AgentState
from langgraph.types import Command
from pydantic import BaseModel

class CustomState(AgentState):
    user_name: str

class CustomContext(BaseModel):
    user_id: str

@tool
def update_user_info(
    runtime: ToolRuntime[CustomContext, CustomState],
) -> Command:
    """Look up and update user info."""
    user_id = runtime.context.user_id
    name = "John Smith" if user_id == "user_123" else "Unknown user"
    return Command(update={
        "user_name": name,
        "messages": [
            ToolMessage(
                "Successfully looked up user information",
                tool_call_id=runtime.tool_call_id
            )
        ]
    })

@tool
def greet(
    runtime: ToolRuntime[CustomContext, CustomState]
) -> str | Command:
    """Use this to greet the user once you found their info."""
    user_name = runtime.state.get("user_name", None)
    if user_name is None:
       return Command(update={
            "messages": [
                ToolMessage(
                    "Please call the 'update_user_info' tool it will get and update the user's name.",
                    tool_call_id=runtime.tool_call_id
                )
            ]
        })
    return f"Hello {user_name}!"

agent = create_agent(
    model="gpt-5-nano",
    tools=[update_user_info, greet],
    state_schema=CustomState,
    context_schema=CustomContext,
)

agent.invoke(
    {"messages": [{"role": "user", "content": "greet the user"}]},
    context=CustomContext(user_id="user_123"),
)

💡 设计亮点:上述代码展示了 工具间协作 的经典模式——greet 工具检测到状态中缺少 user_name,不直接报错,而是引导模型调用 update_user_info 工具完成信息获取。这是一种优雅的 “渐进式信息填充” 策略。

5.2 提示词注入

通过 @dynamic_prompt 装饰器,可以在构建系统提示词时读取短期记忆,实现动态提示词注入

from langchain.agents import create_agent
from typing import TypedDict
from langchain.agents.middleware import dynamic_prompt, ModelRequest


class CustomContext(TypedDict):
    user_name: str


def get_weather(city: str) -> str:
    """Get the weather in a city."""
    return f"The weather in {city} is always sunny!"


@dynamic_prompt
def dynamic_system_prompt(request: ModelRequest) -> str:
    user_name = request.runtime.context["user_name"]
    system_prompt = f"You are a helpful assistant. Address the user as {user_name}."
    return system_prompt


agent = create_agent(
    model="gpt-5-nano",
    tools=[get_weather],
    middleware=[dynamic_system_prompt],
    context_schema=CustomContext,
)

result = agent.invoke(
    {"messages": [{"role": "user", "content": "What is the weather in SF?"}]},
    context=CustomContext(user_name="John Smith"),
)
for msg in result["messages"]:
    msg.pretty_print()

这个模式的核心价值在于:提示词不再是一成不变的静态文本,而是可以根据用户身份、历史行为、业务上下文实时生成。例如:

  • 根据用户角色切换不同的系统指令;
  • 根据对话阶段调整助手的语气和风格;
  • 注入用户专属的功能说明或限制。

5.3 中间件拦截

5.3.1 模型调用后处理

使用 @after_model 中间件在模型响应生成后进行拦截,可用于内容审核、敏感词过滤、响应格式化等:

from langchain.messages import RemoveMessage
from langgraph.checkpoint.memory import InMemorySaver
from langchain.agents import create_agent, AgentState
from langchain.agents.middleware import after_model
from langgraph.runtime import Runtime


@after_model
def validate_response(state: AgentState, runtime: Runtime) -> dict | None:
    """移除包含敏感词的消息。"""
    STOP_WORDS = ["password", "secret"]
    last_message = state["messages"][-1]
    if any(word in last_message.content for word in STOP_WORDS):
        return {"messages": [RemoveMessage(id=last_message.id)]}
    return None

agent = create_agent(
    model="gpt-5-nano",
    tools=[],
    middleware=[validate_response],
    checkpointer=InMemorySaver(),
)
5.3.2 模型调用前处理

使用 @before_model 中间件在模型调用前处理消息,常见用途包括消息裁剪、上下文注入等。该模式已在上文 4.1 精简消息 中详细演示,此处不再重复。


6. 总结与最佳实践

短期记忆是构建生产级 AI Agent 的基石。通过本文的深入解析,我们梳理了 LangChain 短期记忆的完整知识体系:

  1. 理解核心机制:短期记忆以线程(thread_id)为单位组织,通过检查点(Checkpoint)实现状态持久化,每步自动保存和加载;
  2. 选择合适的持久化后端:开发用 InMemorySaver,生产用 PostgresSaverRedisSaver
  3. 按需扩展状态:通过继承 AgentState 添加自定义字段,将业务数据与对话状态绑定;
  4. 主动管理消息队列:综合运用精简、删除、摘要和滑动窗口策略,避免上下文溢出;
  5. 多维度访问内存:根据场景选择工具层(ToolRuntime)、提示词层(@dynamic_prompt)或中间件层(@before_model / @after_model)读写状态。

推荐最佳实践清单

  • ✅ 生产环境务必使用数据库持久化检查点,避免内存数据丢失;
  • ✅ 长对话场景优先使用 SummarizationMiddleware,兼顾信息保留和 token 控制;
  • ✅ 敏感场景在 @after_model 中增加内容审核中间件;
  • ✅ 自定义状态字段保持最小化原则,避免状态膨胀;
  • ✅ 使用 ToolRuntime 而非全局变量在工具间传递状态。

短期记忆并非孤立的技术组件,它与长期记忆、上下文工程、Agent 架构设计深度交织。掌握短期记忆的原理与实践,是构建智能、可靠、可扩展的 AI 应用的关键一步。


参考文档LangChain Short-term Memory - Official Docs

Logo

小龙虾开发者社区是 CSDN 旗下专注 OpenClaw 生态的官方阵地,聚焦技能开发、插件实践与部署教程,为开发者提供可直接落地的方案、工具与交流平台,助力高效构建与落地 AI 应用

更多推荐