Agent 多轮对话后,前面的关键信息被”挤”出 Context Window,开始答非所问。如何让 Agent 记住重要信息、忘记无关信息?

记忆管理通过短期记忆(STM)、长期记忆(LTM)、向量库等分层架构,突破 Context Window 限制。本篇解析如何设计”该记的记、该忘的忘”的记忆系统。

记忆管理 = 该记的记(重要信息存 LTM)、该忘的忘(压缩/剪枝)、该查的查(需要时从知识库检索)。Context Window 有限、成本高、全塞进去会拖慢推理——分层架构(STM + LTM + ExM)解决这三个问题。


🧠 一、Agent 的分层记忆架构

Agent 组成与记忆在整体架构中的位置详见第 6 篇。本篇深入解析记忆的分层架构。像人脑:短期记当前在做的事,长期存重要经历,外部需要时查资料。Agent 同理:

  • 短期记忆(STM):记住当前对话和任务的关键信息
  • 长期记忆(LTM):保存历史经验和用户偏好
  • 外部知识(ExM):需要时从知识库检索

1.1 三层记忆架构详解

层次 概念比喻 存储介质 内容与功能 核心挑战 简单理解
短期记忆 (STM) 大脑 RAM Context Window (Prompt) 当前会话的 Thought / Action / Observation 序列,确保任务连贯性 Token 长度限制:容易溢出 记住当前正在做的事情
长期记忆 (LTM) 硬盘 Hard Drive 向量数据库 / 知识图谱 历史经验、用户偏好、项目进度摘要 高效检索:如何召回相关信息 保存过去的重要经历
外部知识 (ExM) 百科全书 RAG 知识库 / API 数据 事实性、专业性、非个人化信息 知识时效性:需要定期更新 需要时查阅资料

1.2 生活化理解:三层记忆如何工作

场景:客服 Agent 处理多轮咨询——用户先问退换货政策,再问某订单状态,最后要求退款。Agent 需要记住对话上下文(STM)、用户历史偏好(LTM)、政策文档(ExM)。

短期记忆(STM)

记住当前任务的关键信息:

  • “我正在查询用户数据”
  • “查询成功,返回了 1000 条记录”
  • “现在需要生成报告”

就像你正在看书时,记住当前页的内容。

长期记忆(LTM)

保存重要的历史信息:

  • “用户偏好:喜欢 PDF 格式的报告”
  • “上次查询时遇到了数据库连接问题”
  • “项目进度:已完成 80%”

就像你记住过去学过的知识。

外部知识(ExM)

需要时从知识库检索:

  • “数据库查询的最佳实践”
  • “报告生成的模板”
  • “API 文档”

就像你查字典或查百科。

1.3 三层记忆如何协作

工作流程

1. 用户输入任务

2. 从 LTM 检索相关历史信息(用户偏好、历史经验)

3. 从 ExM 检索相关知识(API 文档、最佳实践)

4. 把检索到的信息 + 当前任务放入 STM(Context Window)

5. LLM 基于 STM 进行推理和决策

6. 把重要信息保存到 LTM(长期记忆)

7. 继续下一轮循环

💡 关键理解

  • STM:保证当前任务的即时连贯(就像工作记忆)
  • LTM:持久化经验,支持跨任务学习(就像长期记忆)
  • ExM:提供事实性与专业知识(就像查阅资料)

三者结合,确保 Agent 既能记住重要信息,又能智能遗忘不重要的细节。


💾 二、短期记忆 (STM) 的管理策略

Context Window 有限,写满了就要擦——擦哪些?留哪些? 三种策略:滑动窗口(只留最近 N 轮)、对话摘要(压缩成摘要)、重要性剪枝(只留关键信息)。

2.1 Sliding Window(滑动窗口)

只保留最近 N 轮,旧的自动丢弃。简单粗暴,但可能丢掉关键信息。

机制

保留最近 N 轮 Thought / Action / Observation,旧信息被丢弃或迁移到 LTM

生活例子

就像你的手机通知栏:

  • 只显示最近 10 条通知
  • 新的通知来了,最旧的通知被挤掉
  • 简单直接,但可能丢失重要信息

适用场景

  • 对话轮次有限
  • 任务步骤明确
  • 不需要长期上下文

代码示例

# 滑动窗口(伪代码)

class SlidingWindowMemory:
def __init__(self, max_size=10):
self.max_size = max_size # 最多保留 10 轮对话
self.memories = [] # 记忆列表

def add(self, thought, action, observation):
"""添加新的记忆"""
memory = {
"thought": thought,
"action": action,
"observation": observation
}
self.memories.append(memory)

# 如果超过最大长度,删除最旧的
if len(self.memories) > self.max_size:
old_memory = self.memories.pop(0) # 删除最旧的
# 可选:把旧记忆迁移到 LTM
# self.save_to_ltm(old_memory)

def get_context(self):
"""获取当前上下文"""
return self.memories

# 使用示例
memory = SlidingWindowMemory(max_size=5)

# 添加记忆
memory.add("思考1", "行动1", "观察1")
memory.add("思考2", "行动2", "观察2")
# ... 继续添加

# 当超过 5 轮时,最旧的会被自动删除

2.2 Conversational Summarization(对话摘要)

LLM 定期把最旧内容摘要成短文本,保留关键信息——像读书笔记,记结论不记全文。价值:

  • 节省 Token 空间
  • 保留关键主题和结论
  • 不会丢失重要信息

工程实现

每 K 轮循环调用 Summarizer LLM,生成更新后的记忆片段

代码示例

# 对话摘要(伪代码)

class SummarizedMemory:
def __init__(self, summarizer_llm, summary_interval=5):
self.summarizer_llm = summarizer_llm
self.summary_interval = summary_interval # 每 5 轮摘要一次
self.memories = []
self.summary = "" # 保存摘要

def add(self, thought, action, observation):
"""添加新的记忆"""
memory = {
"thought": thought,
"action": action,
"observation": observation
}
self.memories.append(memory)

# 每 N 轮进行一次摘要
if len(self.memories) >= self.summary_interval:
self.summarize()

def summarize(self):
"""对旧记忆进行摘要"""
# 获取需要摘要的记忆(最旧的部分)
old_memories = self.memories[:self.summary_interval]

# 构建摘要 Prompt
prompt = f"""
请对以下对话进行摘要,保留关键信息和结论:

{format_memories(old_memories)}

摘要:
"""

# 调用 LLM 生成摘要
new_summary = self.summarizer_llm.generate(prompt)

# 更新摘要(合并新旧摘要)
self.summary = self.summary + "\n" + new_summary

# 删除已摘要的记忆
self.memories = self.memories[self.summary_interval:]

def get_context(self):
"""获取当前上下文(摘要 + 最近记忆)"""
return {
"summary": self.summary,
"recent_memories": self.memories
}

# 使用示例
memory = SummarizedMemory(summarizer_llm=gpt4, summary_interval=5)

# 添加记忆
for i in range(10):
memory.add(f"思考{i}", f"行动{i}", f"观察{i}")
# 第 5 轮和第 10 轮会自动触发摘要

2.3 Importance-Based Pruning(基于重要性的剪枝)

为每段记忆打重要性得分,Context 满载时优先移除低分——像手机相册,空间不足时先删截图。评分方式:

通过小型分类模型或 LLM Prompt,根据与核心任务目标的关联性评分

代码示例

# 基于重要性的剪枝(伪代码)

class ImportanceBasedMemory:
def __init__(self, max_tokens=10000, importance_scorer=None):
self.max_tokens = max_tokens
self.importance_scorer = importance_scorer # 重要性评分器
self.memories = []

def add(self, thought, action, observation, task_goal=None):
"""添加新的记忆,并计算重要性"""
memory = {
"thought": thought,
"action": action,
"observation": observation,
"importance": self.score_importance(thought, action, observation, task_goal)
}
self.memories.append(memory)

# 如果超过 Token 限制,进行剪枝
if self.get_total_tokens() > self.max_tokens:
self.prune()

def score_importance(self, thought, action, observation, task_goal):
"""计算记忆的重要性得分(0-1)"""
if self.importance_scorer:
# 使用外部评分器
return self.importance_scorer.score(thought, action, observation, task_goal)
else:
# 简单的启发式评分(实际应该用 LLM 或模型)
# 例如:包含"错误"、"失败"的记忆重要性更高
if "错误" in observation or "失败" in observation:
return 0.9
elif "成功" in observation:
return 0.7
else:
return 0.5

def prune(self):
"""剪枝:移除低重要性记忆"""
# 按重要性排序
self.memories.sort(key=lambda x: x["importance"], reverse=True)

# 移除低重要性记忆,直到满足 Token 限制
while self.get_total_tokens() > self.max_tokens:
if len(self.memories) > 0:
removed = self.memories.pop() # 移除最低重要性记忆
# 可选:保存到 LTM
# self.save_to_ltm(removed)
else:
break

def get_total_tokens(self):
"""计算总 Token 数"""
total = 0
for memory in self.memories:
total += count_tokens(memory["thought"])
total += count_tokens(memory["action"])
total += count_tokens(memory["observation"])
return total

def get_context(self):
"""获取当前上下文(按重要性排序)"""
return sorted(self.memories, key=lambda x: x["importance"], reverse=True)

# 使用示例
memory = ImportanceBasedMemory(max_tokens=10000)

# 添加记忆
memory.add("思考1", "行动1", "观察1:查询成功", task_goal="查询数据")
memory.add("思考2", "行动2", "观察2:查询失败,数据库连接错误", task_goal="查询数据")
# 第二个记忆的重要性更高(包含"错误"),会被优先保留

2.4 三种策略对比

策略 优点 缺点 适用场景
滑动窗口 简单直接,实现容易 可能丢失重要信息 对话轮次有限,任务简单
对话摘要 保留关键信息,节省空间 需要额外的 LLM 调用,成本较高 需要长期上下文,任务复杂
重要性剪枝 智能选择,保留重要信息 评分准确性依赖模型 需要区分重要/不重要信息

选择指南

  • 简单任务:使用滑动窗口
  • 需要长期上下文:使用对话摘要
  • 需要智能选择:使用重要性剪枝
  • 最佳实践:结合使用(如:滑动窗口 + 重要性剪枝)

🗄️ 三、长期记忆 (LTM) 的存储与检索

LTM 持久化经验与用户画像,核心是高效检索——像硬盘,存得多但要能快速找到。

3.1 存储介质多样性

三种存储方式,各有优势

1. 向量数据库 (Vector Stores)

简单理解:就像搜索引擎,通过”语义相似性”找到相关内容。

存储内容

  • 历史对话
  • Observation(观察结果)
  • Thought(思考过程)

检索方式:语义相似性搜索(RAG 基础)

  • 把查询转换成向量
  • 在向量空间中找最相似的记忆
  • 就像”找相似的文章”

适用场景

  • 需要语义检索(如”找相关的历史对话”)
  • 非结构化数据(文本、对话)

常见工具

  • Pinecone、Weaviate、Qdrant、Chroma

2. 知识图谱 (Knowledge Graphs)

简单理解: 就像关系数据库,存储实体之间的关系。

存储内容

  • 结构化关系(如用户偏好与项目关联)
  • 实体和关系(如”用户A 喜欢 项目B”)

检索方式: 多跳推理(Cypher / SPARQL)

  • 通过关系链找到相关信息
  • 就像”找朋友的朋友”

适用场景

  • 需要关系推理(如”找用户喜欢的项目”)
  • 结构化数据(实体、关系)

常见工具

  • Neo4j、ArangoDB

3. Key-Value / 关系型数据库 (SQL/NoSQL)

简单理解: 就像传统的数据库,通过键值或 SQL 查询。

存储内容

  • 用户配置
  • 项目状态
  • 任务清单

检索方式: Function Calling 或 Text-to-SQL

  • 通过键值查询
  • 或通过 SQL 查询

适用场景

  • 需要精确查询(如”查询用户ID=123的配置”)
  • 结构化数据(表、字段)

常见工具

  • PostgreSQL、MySQL、MongoDB、Redis

3.2 存储介质选择指南

存储介质 检索方式 适用场景 简单理解
向量数据库 语义相似性搜索 非结构化文本、对话历史 像搜索引擎,找相似内容
知识图谱 关系推理 结构化关系、实体关联 像关系数据库,找关联信息
关系型数据库 SQL/键值查询 结构化数据、精确查询 像传统数据库,精确查找

选择指南

  • 文本、对话:使用向量数据库
  • 关系、实体:使用知识图谱
  • 配置、状态:使用关系型数据库
  • 最佳实践:结合使用(如:向量数据库 + 关系型数据库)

3.3 高级检索策略:Contextual Memory Retrieval

不只找”相似”,还要考虑相似性(是否相关)、时效性(是否近期)、重要性(是否关键)。检索公式:

$$\text{Recall Score} = \alpha \cdot \text{Contextual Similarity} + \beta \cdot \text{Recency} + \gamma \cdot \text{Importance}$$

三个维度

  1. Contextual Similarity(语义相似度)

    确保与当前任务相关

    • 例如:当前任务是”查询数据库”,优先召回”数据库相关”的记忆
  2. Recency(时效性)

    近期事件权重更高

    • 例如:昨天的记忆比去年的记忆更重要
  3. Importance(重要性)

    关键事件优先召回

    • 例如:包含”错误”、”失败”的记忆更重要

实现要点:相似度 + 时效性 + 重要性加权。以下为可选参考实现:

# 高级检索策略(伪代码)

class ContextualMemoryRetrieval:
def __init__(self, vector_db, alpha=0.5, beta=0.3, gamma=0.2):
self.vector_db = vector_db
self.alpha = alpha # 相似度权重
self.beta = beta # 时效性权重
self.gamma = gamma # 重要性权重

def retrieve(self, query, top_k=5):
"""检索相关记忆"""
# 1. 语义相似性搜索
similar_memories = self.vector_db.similarity_search(query, top_k=top_k*2)

# 2. 计算综合得分
scored_memories = []
for memory in similar_memories:
# 相似度得分(0-1)
similarity_score = memory["similarity"]

# 时效性得分(0-1):越新得分越高
recency_score = self.calculate_recency(memory["timestamp"])

# 重要性得分(0-1)
importance_score = memory.get("importance", 0.5)

# 综合得分
final_score = (
self.alpha * similarity_score +
self.beta * recency_score +
self.gamma * importance_score
)

scored_memories.append({
"memory": memory,
"score": final_score
})

# 3. 按得分排序,返回 Top K
scored_memories.sort(key=lambda x: x["score"], reverse=True)
return [m["memory"] for m in scored_memories[:top_k]]

def calculate_recency(self, timestamp):
"""计算时效性得分"""
import time
current_time = time.time()
age = current_time - timestamp # 年龄(秒)

# 越新得分越高(指数衰减)
# 例如:1 小时前 = 0.9,1 天前 = 0.5,1 周前 = 0.1
decay_rate = 0.0001 # 衰减率
return max(0, 1 - age * decay_rate)

# 使用示例
retrieval = ContextualMemoryRetrieval(vector_db=pinecone_db)

# 检索相关记忆
query = "查询数据库时遇到了连接错误"
memories = retrieval.retrieve(query, top_k=5)

# 返回的记忆会综合考虑相似度、时效性、重要性

💡 工程实践

  • 结合多策略排序,提高长期任务的记忆精度和效率
  • 根据任务类型调整权重(如:对话任务更重视时效性,知识任务更重视相似度)

🔄 四、记忆管理总线:信息流转机制

完整记忆系统需要一个 Memory Stream 来管理信息流入与流出。

简单理解

就像工厂的生产线:

  1. 接收:接收原材料(用户输入、观察结果)
  2. 处理:加工处理(分块、向量化)
  3. 存储:保存到仓库(LTM)
  4. 检索:需要时从仓库取出(检索相关记忆)
  5. 整合:组装成最终产品(Prompt)

4.1 记忆管理总线的四个组件

1. Perceiver(感知器)

简单理解

就像”信息接收器”,接收用户输入和观察结果,进行预处理。

功能

  • 预处理用户输入与 Observation
  • 进行分块(Chunking)
  • 提取关键信息

代码示例

# Perceiver(伪代码)

class Perceiver:
def process(self, user_input, observation):
"""处理用户输入和观察结果"""
# 1. 分块(如果内容太长)
chunks = self.chunk_text(user_input + observation)

# 2. 提取关键信息
key_info = self.extract_key_info(chunks)

return {
"chunks": chunks,
"key_info": key_info
}

def chunk_text(self, text, max_chunk_size=500):
"""文本分块"""
# 简单的按段落分块(实际应该用更智能的方法)
paragraphs = text.split('\n\n')
chunks = []
current_chunk = ""

for para in paragraphs:
if len(current_chunk) + len(para) <= max_chunk_size:
current_chunk += para + "\n\n"
else:
if current_chunk:
chunks.append(current_chunk.strip())
current_chunk = para + "\n\n"

if current_chunk:
chunks.append(current_chunk.strip())

return chunks

2. Embedding Generator(向量生成器)

简单理解

就像”翻译器”,把文本转换成向量(数字),方便存储和检索。

功能

  • 将文本转换为向量
  • 存入 LTM(向量数据库)

代码示例

# Embedding Generator(伪代码)

class EmbeddingGenerator:
def __init__(self, embedding_model):
self.embedding_model = embedding_model # 如 OpenAI Embeddings

def generate_and_store(self, chunks, metadata):
"""生成向量并存储到 LTM"""
vectors = []

for chunk in chunks:
# 生成向量
embedding = self.embedding_model.embed(chunk)

# 存储到向量数据库
vector_db.add(
vector=embedding,
text=chunk,
metadata=metadata
)

vectors.append(embedding)

return vectors

3. Retrieval Module(检索模块)

简单理解

就像”图书管理员”,需要时从仓库(LTM)找到相关书籍(记忆)。

功能

  • Planner 启动前,检索 Top K 记忆片段
  • 使用高级检索策略(相似度 + 时效性 + 重要性)

代码示例

# Retrieval Module(伪代码)

class RetrievalModule:
def __init__(self, retrieval_strategy):
self.retrieval_strategy = retrieval_strategy # 使用高级检索策略

def retrieve(self, query, top_k=5):
"""检索相关记忆"""
# 使用高级检索策略(见 3.3 节)
memories = self.retrieval_strategy.retrieve(query, top_k=top_k)
return memories

4. Context Refiner(上下文精炼器)

简单理解

就像”编辑”,把各种信息整合成一篇好文章(Prompt)。

功能

  • 整合 STM 摘要、LTM 检索结果及当前输入
  • 形成高效 Prompt 注入 LLM Planner
  • 防止”上下文污染”
  • 提升 LLM 推理效率

代码示例

# Context Refiner(伪代码)

class ContextRefiner:
def refine(self, stm_summary, ltm_memories, current_input):
"""整合上下文,生成高效 Prompt"""
# 1. 整合 STM 摘要
context_parts = []
if stm_summary:
context_parts.append(f"近期摘要:\n{stm_summary}")

# 2. 整合 LTM 检索结果
if ltm_memories:
memory_text = "\n".join([
f"- {m['text']}" for m in ltm_memories
])
context_parts.append(f"相关历史:\n{memory_text}")

# 3. 添加当前输入
context_parts.append(f"当前任务:\n{current_input}")

# 4. 组合成完整 Prompt
prompt = "\n\n".join(context_parts)

return prompt

4.2 完整的信息流转流程

流程图

用户输入

Perceiver(感知器)
↓ 分块、提取关键信息
Embedding Generator(向量生成器)
↓ 生成向量
LTM Storage(长期记忆存储)

Retrieval Module(检索模块)
↓ 检索相关记忆
Context Refiner(上下文精炼器)
↓ 整合上下文
LLM Planner(决策引擎)
↓ 生成 Thought 和 Action
执行工具
↓ 获取 Observation
回到 Perceiver(循环)

代码示例(完整流程)

# 完整的记忆管理总线(伪代码)

class MemoryBus:
def __init__(self, stm, ltm, perceiver, embedding_gen, retrieval, refiner):
self.stm = stm # 短期记忆
self.ltm = ltm # 长期记忆
self.perceiver = perceiver
self.embedding_gen = embedding_gen
self.retrieval = retrieval
self.refiner = refiner

def process(self, user_input, observation):
"""处理用户输入和观察结果"""
# 1. Perceiver:预处理
processed = self.perceiver.process(user_input, observation)

# 2. Embedding Generator:生成向量并存储
self.embedding_gen.generate_and_store(
processed["chunks"],
metadata={"timestamp": time.time()}
)

# 3. 添加到 STM
self.stm.add(
thought="",
action="",
observation=observation
)

# 4. Retrieval:检索相关记忆
query = user_input + observation
ltm_memories = self.retrieval.retrieve(query, top_k=5)

# 5. Context Refiner:整合上下文
stm_summary = self.stm.get_summary()
prompt = self.refiner.refine(stm_summary, ltm_memories, user_input)

return prompt

# 使用示例
memory_bus = MemoryBus(
stm=SummarizedMemory(),
ltm=vector_db,
perceiver=Perceiver(),
embedding_gen=EmbeddingGenerator(),
retrieval=RetrievalModule(),
refiner=ContextRefiner()
)

# 处理用户输入
prompt = memory_bus.process(
user_input="查询用户数据",
observation="查询成功,返回 1000 条记录"
)

# prompt 包含了 STM 摘要、LTM 检索结果、当前输入
# 可以发送给 LLM Planner 了

💡 关键理解

  • Perceiver:接收和预处理信息
  • Embedding Generator:把文本转换成向量
  • Retrieval Module:从 LTM 检索相关记忆
  • Context Refiner:整合所有信息,生成高效 Prompt

四个组件协作,实现完整的记忆管理流程。


🔍 总结:Agent 的记忆是自主性的载体

💡 快速回顾:你学到了什么?

  1. 分层记忆架构:STM(短期记忆)、LTM(长期记忆)、ExM(外部知识)
  2. 短期记忆管理:滑动窗口、对话摘要、重要性剪枝
  3. 长期记忆存储:向量数据库、知识图谱、关系型数据库
  4. 高级检索策略:综合考虑相似度、时效性、重要性
  5. 记忆管理总线:Perceiver → Embedding → Storage → Retrieval → Refiner

三层记忆的核心作用

层次 作用 简单理解
STM 保证当前任务的即时连贯 记住当前正在做的事情
LTM 持久化经验,支持跨任务学习 保存过去的重要经历
ExM 提供事实性与专业知识 需要时查阅资料

关键设计原则

  1. 智能遗忘:不是记住所有信息,而是记住重要的,忘记不重要的
  2. 高效检索:不是简单存储,而是能够快速找到相关信息
  3. 分层管理:不是单一存储,而是分层存储,各司其职
  4. 动态更新:不是静态存储,而是动态更新,保持时效性

实战建议

  1. 从简单开始:先实现滑动窗口,再逐步优化
  2. 选择合适的存储:根据数据类型选择存储介质
  3. 优化检索策略:根据任务类型调整检索权重
  4. 监控记忆质量:定期检查记忆的准确性和相关性

💡 核心理解
一个优秀的 Agent 工程师,不仅要让 Agent 记住重要信息,更要让它 智能遗忘,精准召回高价值记忆,实现长期自主性。

记忆管理不是简单的存储和检索,而是一套完整的工程化系统,需要综合考虑存储、检索、压缩、更新等多个维度。


📚 延伸阅读(含可直接访问链接)

以下资源按主题分类,每个资源都附有简要说明,帮助你选择合适的学习材料。

🧠 分层记忆架构

🔍 记忆检索策略

🗄️ 存储介质

🔄 记忆管理实践


🔔 系列说明

本文是《🧠 LLM/Agent 从入门到精通:告别浅尝辄止》系列第 9 篇。上一篇:复杂任务 Agent 怎么拆?任务规划与 Self-Correction。下一篇:LangChain、LlamaIndex、AutoGPT 怎么选?Agent 框架对比