初级
智能体记忆：详细解析
智能体记忆：详细解析
想象一下，有一天你雇佣了一位才华横溢的自由职业者。第一天，她表现惊人，抓住了每一个漏洞，编写了清晰的文档，甚至提出了你没想到的改进建议。你印象深刻。
第二天，你走进来问：“嘿，还记得我们昨天讨论的那个问题吗？”
她停顿了一下。看着你。微微一笑。
“抱歉……什么问题？”
没有记忆。没有上下文。完全消失。你会像我写到这里时一样震惊。
这正是大多数大语言模型的行为方式。每一次新对话都是全新的开始。模型不知道你是谁，不知道你们一起构建了什么，也不知道几分钟前在另一个聊天窗口中讨论了什么。
对于简单的聊天机器人来说，这没问题。但对于一个执行任务、做出决策并随时间改进的智能体来说，这种健忘症是致命的。
因为真正的智能不仅仅是做出好的回应。它关乎记忆、学习，并在已有的基础上构建。
记忆是将无状态系统转变为能够真正进化的东西的关键。
智能体记忆到底是什么？#
智能体记忆并非单一事物。它更像是一个在幕后工作的系统——不同类型的存储、检索信息的方式，以及管理这一切的智能策略，以便智能体能够随时间携带上下文。
关键思想很简单：记忆不是只做一项工作；它同时做三项截然不同的工作。
➜ 连续性关乎身份。它是智能体如何知道你是谁、你的偏好以及你们已经共同构建了什么。没有它，每次互动都感觉是从零开始。
➜ 上下文关乎手头的任务。刚刚发生了什么，使用了哪个工具，返回了什么结果，以及接下来需要做什么。它是防止多步骤工作流崩溃的关键。
➜ 学习关乎变得更好。理解什么有效、什么无效，并随时间慢慢改进决策，而不是重复同样的错误。
综合起来，它使智能体在每次互动中显得一致、可靠，并更智能一些。
一个设计良好的智能体记忆系统处理所有这三项，为每一项使用不同的存储后端。
4种记忆类型#
该领域已收敛于四种不同的记忆类型。将它们视为大脑的四个不同部分，每个部分都针对特定任务进化而来。
1. 上下文内记忆#
上下文窗口是智能体的工作台。上面的所有内容都可以立即访问。模型可以在单次前向传递中对其进行推理。无需检索步骤。
但工作台有大小限制。每个令牌都花费金钱和时间。当会话结束时，工作台会被清空。
上下文中有哪些内容？
- 系统提示：智能体角色、规则、能力、当前日期/用户信息
- 对话历史：本次会话中到目前为止的来回交流
- 工具调用结果：智能体刚刚调用的工具的输出
- 检索到的记忆：从外部存储中拉入的块
- 草稿板：中间推理（逐步思考输出）
滑动窗口问题
在长对话中，历史会累积并最终溢出上下文限制。截断最旧消息的简单解决方案会丢失重要的早期上下文。更好的策略：
- 摘要：定期将旧的轮次压缩成简短摘要，并用摘要替换它们。
- 选择性保留：保留包含关键事实、决策或工具结果的轮次；丢弃闲聊。
- 卸载到外部记忆：将重要事实提取到向量存储中，然后根据需要检索它们。
2. 外部记忆#
外部记忆是任何持久化在模型之外的东西——数据库、向量存储、键值存储和文件。它跨越会话边界。如果你存储得当，你的智能体可以记住六个月前的事情。
外部存储有两种形式：
结构化存储（精确查找）：PostgreSQL、Redis、SQLite。你通过键、ID或SQL查询。快速、可预测，非常适合用户配置文件、偏好和结构化数据。
向量存储（语义搜索）：Pinecone、Chroma、pgvector。你通过含义查询，“找到类似于这个概念的记忆。”对于非结构化笔记和情景回忆至关重要。
检索步骤是一个瓶颈。如果你没有检索到正确的记忆，智能体的行为就好像它们不存在一样。好的记忆架构是20%的存储和80%的检索设计。
3. 情景记忆#
情景记忆是最被低估的类型。外部记忆存储事实，而情景记忆存储事件，特别是过去行动的结果。
最简单的形式是结构化日志：每次智能体完成任务时，它都会记录发生了什么。随着时间的推移，这个日志成为丰富的自我知识来源，智能体可以在做出决策前查阅。
一个情景的样子：
{
"episode_id": "ep_20240315_003",
"timestamp": "2024-03-15T14:23:11Z",
"task": "将50页PDF总结为3个要点",
"approach": "顺序分块，每块2000个令牌",
"outcome": "成功",
"duration_ms": 4820,
"token_cost": 12400,
"quality_score": 0.91,
"notes": "效果不错。分层分块会更快。",
"embedding": [0.023, -0.441, 0.182, /* ... 1536维 */]
}当新任务到来时，智能体会检索语义上最相似的过去情景，并使用它们来选择策略。这本质上是基于个人历史而非手工制作数据集的少样本学习。
反思循环👇
语义/参数记忆#
这是模型天生就有的记忆。一切都在训练期间编码到权重中——关于世界的事实、语言模式、推理策略、编码约定和文化知识。
它始终存在。智能体永远不必检索它。但它有硬性限制：
- 训练时冻结：模型不知道其截止日期之后发生的事情。
- 运行时无法更新：不重新训练或微调，就无法注入新的永久事实。
- 不透明：你无法检查模型具体“知道”或不知道什么。
- 容易产生幻觉：模型用看似合理但错误的补全来填补空白。
对于任何时间敏感、领域特定或私有的内容，不要依赖参数记忆。使用外部检索。参数记忆是你在没有更好来源时用于通用世界知识的后备方案。
正确的思维模型：参数记忆是智能体的通识教育。外部记忆、情景记忆和上下文内记忆是智能体的在职经验。最好的智能体结合两者。
记忆如何在智能体循环中流动？#
让我们把所有内容整合起来。以下是智能体每次处理请求时发生的情况——展示每个记忆系统的作用。
注意，记忆操作包围了LLM调用：先检索，后写入。模型本身是无状态的；记忆系统是赋予智能体有状态、有意识错觉的东西。
构建记忆层#
让我们来构建它。我们将使用 Python 搭配 OpenAI 进行嵌入，并使用 ChromaDB 作为本地向量存储。同样的概念适用于任何其他技术栈——只需替换相应的库即可。
pip install chromadb openai anthropic python-dotenvMemoryStore 类#
该类负责处理记忆的写入（包含嵌入）和语义检索。它是其他所有功能的基础。
import chromadb
from openai import OpenAI
from datetime import datetime
import json, uuid
class MemoryStore:
"""AI 代理的持久化向量记忆。"""
def __init__(self, agent_id: str, persist_dir: str = "./memory_db"):
self.agent_id = agent_id
self.openai = OpenAI()
# ChromaDB 将向量存储在磁盘上，重启后数据持久化
self.client = chromadb.PersistentClient(path=persist_dir)
self.collection = self.client.get_or_create_collection(
name=f"agent_{agent_id}_memories",
metadata={"hnsw:space": "cosine"} # 余弦相似度
)
def _embed(self, text: str) -> list[float]:
"""使用 OpenAI 将文本转换为嵌入向量。"""
response = self.openai.embeddings.create(
model="text-embedding-3-small",
input=text
)
return response.data[0].embedding
def remember(
self,
content: str,
memory_type: str = "general",
metadata: dict = None
) -> str:
"""存储一条记忆。返回记忆 ID。"""
memory_id = str(uuid.uuid4())
embedding = self._embed(content)
meta = {
"type": memory_type,
"timestamp": datetime.utcnow().isoformat(),
"agent_id": self.agent_id,
**(metadata or {})
}
self.collection.add(
ids=[memory_id],
embeddings=[embedding],
documents=[content],
metadatas=[meta]
)
return memory_id
def recall(
self,
query: str,
k: int = 5,
memory_type: str = None,
min_relevance: float = 0.6
) -> list[dict]:
"""检索与查询最相关的 k 条记忆。"""
query_embedding = self._embed(query)
where = {"type": memory_type} if memory_type else None
results = self.collection.query(
query_embeddings=[query_embedding],
n_results=k,
where=where,
include=["documents", "metadatas", "distances"]
)
memories = []
for doc, meta, dist in zip(
results["documents"][0],
results["metadatas"][0],
results["distances"][0]
):
relevance = 1 - dist # 余弦距离 → 相似度
if relevance >= min_relevance:
memories.append({
"content": doc,
"metadata": meta,
"relevance": round(relevance, 3)
})
return sorted(memories, key=lambda x: x["relevance"], reverse=True)
def forget(self, memory_id: str):
"""删除特定记忆（GDPR 合规、过期数据等）"""
self.collection.delete(ids=[memory_id])EpisodicLogger 类#
现在让我们在上面添加事件日志记录层。
from .store import MemoryStore
from dataclasses import dataclass, asdict
from typing import Optional
import time
\@dataclass
class Episode:
task: str
approach: str
outcome: str # "success" | "partial" | "failure"
duration_ms: int
token_cost: int
quality_score: float # 0.0 – 1.0，由评估器或用户设置
notes: str = ""
error: Optional[str] = None
class EpisodicLogger:
def __init__(self, memory_store: MemoryStore):
self.store = memory_store
def log(self, episode: Episode):
"""将事件保存为可搜索的文档存入记忆。"""
# 构建用于语义搜索的富文本表示
doc = (
f"Task: {episode.task}\n"
f"Approach: {episode.approach}\n"
f"Outcome: {episode.outcome}\n"
f"Notes: {episode.notes}"
)
self.store.remember(
content=doc,
memory_type="episode",
metadata={
"outcome": episode.outcome,
"quality_score": episode.quality_score,
"duration_ms": episode.duration_ms,
"token_cost": episode.token_cost,
}
)
def recall_similar(self, task: str, k: int = 3) -> list[dict]:
"""查找与当前任务相似的过去事件。"""
return self.store.recall(
query=task,
k=k,
memory_type="episode",
min_relevance=0.65
)整合起来：一个记忆增强的代理#
import anthropic
from memory.store import MemoryStore
from memory.episodic import EpisodicLogger, Episode
import time
class MemoryAugmentedAgent:
def __init__(self, agent_id: str):
self.client = anthropic.Anthropic()
self.memory = MemoryStore(agent_id)
self.episodes = EpisodicLogger(self.memory)
def _build_memory_context(self, user_message: str) -> str:
"""检索相关记忆并格式化以便注入。"""
# 语义搜索相关事实
memories = self.memory.recall(user_message, k=4)
# 相似的过去任务方法
episodes = self.episodes.recall_similar(user_message, k=2)
context_parts = []
if memories:
context_parts.append("## 相关记忆\n" +
"\n".join([
f"- [{m['metadata']['type']}] {m['content']}"
f" (相关性: {m['relevance']})"
for m in memories
])
)
if episodes:
context_parts.append("## 过去相似任务\n" +
"\n".join([
f"- {e['content'][:200]}..."
for e in episodes
])
)
return "\n\n".join(context_parts) if context_parts else ""
def run(self, user_message: str) -> str:
start = time.time()
# 1. 检索相关记忆
memory_context = self._build_memory_context(user_message)
# 2. 构建包含注入记忆的系统提示
system = """你是一个拥有记忆的智能助手。
你可以访问过去交互中的相关上下文。
利用这些上下文来提供更好、更个性化的回复。
"""
if memory_context:
system += f"\n\n{memory_context}"
# 3. 调用模型
response = self.client.messages.create(
model="claude-opus-4-6",
max_tokens=1024,
system=system,
messages=[{"role": "user", "content": user_message}]
)
answer = response.content[0].text
duration = int((time.time() - start) * 1000)
# 4. 将有用信息保存到记忆中以备下次使用
self.memory.remember(
content=f"用户询问: {user_message[:200]}",
memory_type="interaction"
)
# 5. 记录事件
self.episodes.log(Episode(
task=user_message[:200],
approach="单轮对话配合记忆检索",
outcome="success",
duration_ms=duration,
token_cost=response.usage.input_tokens + response.usage.output_tokens,
quality_score=1.0, # 生产环境中将由评估系统提供
))
return answer向量数据库#
它是任何严肃记忆系统的核心。与通过精确匹配查询（如 SQL）不同，它能在高维空间中查找向量的最近邻。这正是实现语义搜索的关键——即使没有共享任何词语，也能找到概念上相关的记忆。
相似性搜索的工作原理#
每条记忆都会被转换为一个向量（使用 OpenAI 的嵌入模型，得到一个包含 1,536 个浮点数的数组）。概念上相似的文本会产生相似的向量。当你查询时，你会嵌入查询内容，并使用余弦相似度找到最接近的向量。
import numpy as np
def cosine_similarity(a: list, b: list) -> float:
"""
1.0 = 含义相同
0.0 = 不相关
-1.0 = 含义相反
"""
a, b = np.array(a), np.array(b)
return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))
# 示例：这两个句子将具有很高的相似度
embedding_a = embed("用户偏好深色模式")
embedding_b = embed("他们喜欢深色的界面主题")
score = cosine_similarity(embedding_a, embedding_b)
# → ~0.91（非常相似）本地开发时，从 ChromaDB 开始。准备部署时，如果你已经在使用 Postgres 且无需额外基础设施，请评估 pgvector。当需要大规模扩展时，使用 Pinecone 或 Qdrant。
记忆管理#
真正的记忆系统不仅仅是积累，还需要精心管理。一个不断增长、缺乏重点的存储会随着时间的推移而退化——检索会变得嘈杂，延迟会增加，矛盾的记忆会混淆智能体。
你需要一个遗忘策略。以下是三种主要方法：
1. 基于时间的衰减#
较旧的记忆相关性较低。通过结合时效性和语义相关性来对记忆进行评分。研究中使用的公式：
import math
from datetime import datetime
def memory_score(
relevance: float, # 余弦相似度 0–1
importance: float, # 写入时存储 0–1
created_at: datetime, # 记忆形成的时间
recency_weight: float = 0.3,
decay_factor: float = 0.995
) -> float:
"""
灵感来源于《生成式智能体》论文（Park 等人，2023 年）。
平衡因素：相关性、重要性、时效性。
"""
hours_old = (datetime.utcnow() - created_at).total_seconds() / 3600
recency = math.pow(decay_factor, hours_old)
return (
relevance * 0.4 +
importance * 0.3 +
recency * recency_weight
)2. 写入时的重要性评分#
在存储记忆时，让模型对其自身输出进行重要性评分。只存储高分项。这可以从源头过滤噪声。
import re
async def score_importance(client, content: str) -> float:
"""询问 LLM 该信息是否值得保存（0.0 到 1.0）。"""
prompt = f"""评估保存此信息以供未来交互的重要性。
0.0 = 琐碎（问候语）
0.5 = 中等有用
1.0 = 关键（偏好、错误、决策）
信息：{content}
仅回复数字。"""
try:
response = await client.messages.create(
model="claude-3-haiku-20240307", # 使用当前可用的 Haiku 模型
max_tokens=10,
messages=[{"role": "user", "content": prompt}]
)
# 提取第一个看起来像浮点数/整数的字符串
text = response.content[0].text.strip()
match = re.search(r"[-+]?\d*\.\d+|\d+", text)
if match:
score = float(match.group())
return max(0.0, min(1.0, score))
except Exception:
pass
return 0.5 # 默认回退值3. 定期整合#
运行一个夜间任务，将重复或高度相似的记忆合并成一个单一的规范摘要。这类似于人类睡眠如何巩固记忆。
async def consolidate_memories(store: MemoryStore, similarity_threshold: float = 0.92):
"""使用向量搜索高效合并近乎重复的记忆。"""
all_mems = store.collection.get(include=["documents", "embeddings", "ids"])
if not all_mems["ids"]:
return
visited = set()
consolidated_docs = []
for i, (mem_id, doc, emb) in enumerate(zip(
all_mems["ids"], all_mems["documents"], all_mems["embeddings"]
)):
if mem_id in visited:
continue
# 使用向量存储的内置搜索查找邻居
# 这比手动嵌套循环快得多
results = store.collection.query(
query_embeddings=[emb],
n_results=10, # 根据预期密度调整
include=["documents", "distances"]
)
# 识别组成员（1.0 - 距离 = 余弦相似度）
group = [doc]
visited.add(mem_id)
for res_id, res_doc, dist in zip(results["ids"][0], results["documents"][0], results["distances"][0]):
sim = 1.0 - dist
if res_id != mem_id and res_id not in visited and sim >= similarity_threshold:
group.append(res_doc)
visited.add(res_id)
# 处理该组
if len(group) > 1:
summary = await summarize_group(group) # 可能需要异步处理
consolidated_docs.append(summary)
else:
consolidated_docs.append(doc)
# 原子性替换：清空并重新填充
store.collection.delete(where={})
for doc in consolidated_docs:
await store.remember(doc)最后想法#
归根结底，记忆是让 AI 感觉更像伙伴而非工具的关键。没有记忆，每次交互都从零开始。有了记忆，智能体才能理解、适应并随着时间的推移不断改进。
真正的力量不仅仅在于模型本身，而在于你如何设计模型记住什么、忘记什么，以及如何使用这些信息。
构建好记忆层，其他一切都会变得更智能。
代码由 AI 生成。
关注 @techwith_ram 获取更多此类内容。