LangChain核心组件深入理解第五篇 -- Short-term Memory Agent里线程隔离的上下文管理
摘要:短期记忆(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 + State | VectorStore / Database + Embedding |
如果需要在不同对话间留存信息(如用户档案、偏好设置),则需要借助长期记忆功能,在不同对话线程与会话中存储、调取用户专属数据或应用级数据。本文聚焦短期记忆,长期记忆将在后续文章中单独讨论。
2. 使用短期记忆
2.1 检查点与线程隔离机制
LangChain 智能体将短期记忆作为 图状态(Graph State) 的组成部分进行管理。其核心机制如下:
关键设计要点:
- 每步自动保存:每当智能体完成一个步骤(如工具调用),短期记忆就会自动更新并持久化;
- 每步自动读取:每个步骤开始时,系统从检查点加载最新状态;
- 线程隔离:不同
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!"
发生了什么?
- 第一轮
invoke时,thread_id="1"的状态为空,Agent 收到 “Hi! My name is Bob.” 并回复; - 回复后,系统将用户消息和 AI 回复一同保存到
thread_id="1"的状态中; - 第二轮
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:
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 提供了四种核心处理模式:

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 生命周期的不同阶段读写短期记忆:
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 短期记忆的完整知识体系:
- 理解核心机制:短期记忆以线程(
thread_id)为单位组织,通过检查点(Checkpoint)实现状态持久化,每步自动保存和加载; - 选择合适的持久化后端:开发用
InMemorySaver,生产用PostgresSaver或RedisSaver; - 按需扩展状态:通过继承
AgentState添加自定义字段,将业务数据与对话状态绑定; - 主动管理消息队列:综合运用精简、删除、摘要和滑动窗口策略,避免上下文溢出;
- 多维度访问内存:根据场景选择工具层(
ToolRuntime)、提示词层(@dynamic_prompt)或中间件层(@before_model/@after_model)读写状态。
推荐最佳实践清单:
- ✅ 生产环境务必使用数据库持久化检查点,避免内存数据丢失;
- ✅ 长对话场景优先使用
SummarizationMiddleware,兼顾信息保留和 token 控制; - ✅ 敏感场景在
@after_model中增加内容审核中间件; - ✅ 自定义状态字段保持最小化原则,避免状态膨胀;
- ✅ 使用
ToolRuntime而非全局变量在工具间传递状态。
短期记忆并非孤立的技术组件,它与长期记忆、上下文工程、Agent 架构设计深度交织。掌握短期记忆的原理与实践,是构建智能、可靠、可扩展的 AI 应用的关键一步。
更多推荐



所有评论(0)