主题
原文：https://www.mstx.cn/harness/428.html · 创建：2026-09-06 · 更新：2026-09-06 · 阅读：1091
短期记忆的边界
在5.1节中，你构建了短期记忆系统——它管理着当前会话的上下文窗口。但短期记忆有一个根本性的局限：会话结束后，一切都被清空。
用户昨天告诉Agent"我喜欢在下午开会"，今天Agent完全不记得。用户上周让Agent查过某个项目的文档，这周再问相关问题时，Agent需要从头查起。
这不是因为Agent"不够聪明"，而是因为短期记忆的物理边界就是当前会话的上下文窗口。要突破这个边界，你需要一个不同的记忆机制——长期记忆。
长期记忆解决的核心问题是：如何让Agent跨会话记住信息，并在需要时高效地检索出来。
长期记忆与短期记忆的本质区别
| 维度 | 短期记忆 | 长期记忆 |
|---|---|---|
| 生命周期 | 当前会话 | 跨会话持久 |
| 存储位置 | 上下文窗口 | 外部存储（向量库、数据库） |
| 访问方式 | 直接可见 | 需要主动检索 |
| 容量 | 有限（窗口大小） | 理论上无限 |
| 速度 | 即时 | 有检索延迟 |
| 精度 | 完整保留 | 可能检索到不相关内容 |
关键的架构差异：短期记忆是推送给LLM的（LLM被动看到所有内容），长期记忆是按需检索的（LLM主动查询相关内容）。这意味着长期记忆系统需要解决一个额外的问题：如何知道什么信息与当前任务相关？
向量数据库的基本原理
向量数据库是长期记忆最常用的存储后端。它的核心思想是：将文本转换为向量（一组数字），然后用向量相似度来衡量"语义相关性"。
文本 → 嵌入模型 → 向量 → 存储
↓
查询 → 嵌入模型 → 查询向量 → 相似度检索 → 相关文本嵌入（Embedding）是将文本映射到高维向量空间的过程。语义相近的文本，其向量在空间中距离也相近。
例如：
- "用户喜欢在下午开会" → [0.12, -0.45, 0.78, ...]
- "用户的会议时间偏好是下午" → [0.11, -0.43, 0.79, ...]（向量相近）
这两句话虽然措辞不同，但语义相近，它们的向量在空间中距离很近。向量检索就是利用这个特性，找到与查询"语义相关"而不仅仅是"字面匹配"的内容。
构建一个长期记忆系统
第一步：定义记忆条目的数据结构
python
from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional
from datetime import datetime
import json
import hashlib
@dataclass
class MemoryEntry:
"""一条长期记忆"""
memory_id: str
content: str # 记忆的文本内容
category: str = "general" # 分类：user_preference, fact, event, knowledge
source: str = "conversation" # 来源：哪次对话、哪个工具
created_at: str = field(default_factory=lambda: datetime.now().isoformat())
last_accessed_at: Optional[str] = None
access_count: int = 0
importance: int = 3 # 1-5，重要性评分
metadata: Dict[str, Any] = field(default_factory=dict)
def to_text(self) -> str:
"""转换为可嵌入的文本"""
return self.content
def to_dict(self) -> Dict[str, Any]:
return {
"memory_id": self.memory_id,
"content": self.content,
"category": self.category,
"source": self.source,
"created_at": self.created_at,
"importance": self.importance,
"metadata": self.metadata,
}
def generate_memory_id(content: str, category: str) -> str:
"""生成记忆ID（基于内容的哈希，用于去重）"""
hash_input = f"{category}:{content}"
return hashlib.md5(hash_input.encode('utf-8')).hexdigest()[:16]第二步：实现嵌入层
嵌入是长期记忆系统的基础。你可以使用商业API（如OpenAI的Embeddings）或本地模型：
python
class EmbeddingProvider:
"""嵌入提供者：将文本转换为向量"""
def __init__(self, client=None, model: str = "text-embedding-3-small"):
self.client = client
self.model = model
def embed(self, text: str) -> List[float]:
"""将文本转换为向量"""
if self.client is None:
return self._fallback_embed(text)
try:
response = self.client.embeddings.create(
model=self.model,
input=text[:8000], # 限制输入长度
)
return response.data[0].embedding
except Exception as e:
print(f"嵌入失败，使用降级方案：{e}")
return self._fallback_embed(text)
def _fallback_embed(self, text: str) -> List[float]:
"""降级方案：基于字符的简单哈希嵌入"""
# 这不是语义嵌入，但作为降级方案可以提供基本的相似度
vector = [0.0] * 64
for char in text.lower():
idx = ord(char) % 64
vector[idx] += 1.0
# 归一化
norm = sum(v * v for v in vector) ** 0.5
if norm > 0:
vector = [v / norm for v in vector]
return vector第三步：实现向量存储与检索
不依赖外部向量数据库，用纯Python实现一个简单的向量存储：
python
import math
from typing import List, Tuple
class VectorStore:
"""简单的向量存储与检索"""
def __init__(self):
self.vectors: Dict[str, List[float]] = {}
self.entries: Dict[str, MemoryEntry] = {}
def add(self, memory_id: str, vector: List[float], entry: MemoryEntry):
"""添加一个向量和对应的记忆条目"""
self.vectors[memory_id] = vector
self.entries[memory_id] = entry
def search(self, query_vector: List[float], top_k: int = 5,
min_similarity: float = 0.3) -> List[Tuple[MemoryEntry, float]]:
"""检索与查询向量最相似的记忆"""
results = []
for memory_id, vector in self.vectors.items():
similarity = self._cosine_similarity(query_vector, vector)
if similarity >= min_similarity:
results.append((self.entries[memory_id], similarity))
# 按相似度排序
results.sort(key=lambda x: x[1], reverse=True)
return results[:top_k]
def _cosine_similarity(self, vec_a: List[float], vec_b: List[float]) -> float:
"""计算余弦相似度"""
dot_product = sum(a * b for a, b in zip(vec_a, vec_b))
norm_a = math.sqrt(sum(a * a for a in vec_a))
norm_b = math.sqrt(sum(b * b for b in vec_b))
if norm_a == 0 or norm_b == 0:
return 0.0
return dot_product / (norm_a * norm_b)
def remove(self, memory_id: str):
"""删除一条记忆"""
self.vectors.pop(memory_id, None)
self.entries.pop(memory_id, None)
def size(self) -> int:
return len(self.entries)第四步：组装完整的长期记忆系统
python
class LongTermMemory:
"""长期记忆系统"""
def __init__(self, embedding_provider: EmbeddingProvider,
vector_store: VectorStore = None):
self.embedding = embedding_provider
self.store = vector_store or VectorStore()
def store(self, content: str, category: str = "general",
importance: int = 3, metadata: Dict[str, Any] = None) -> str:
"""存储一条记忆"""
# 生成记忆ID（用于去重）
memory_id = generate_memory_id(content, category)
# 检查是否已存在
if memory_id in self.store.entries:
# 已存在：更新重要性（累加）
existing = self.store.entries[memory_id]
existing.importance = min(5, existing.importance + 1)
existing.last_accessed_at = datetime.now().isoformat()
return memory_id
# 创建新记忆
entry = MemoryEntry(
memory_id=memory_id,
content=content,
category=category,
importance=importance,
metadata=metadata or {},
)
# 嵌入并存储
vector = self.embedding.embed(content)
self.store.add(memory_id, vector, entry)
return memory_id
def retrieve(self, query: str, top_k: int = 5,
category: str = None) -> List[Tuple[MemoryEntry, float]]:
"""检索与查询相关的记忆"""
query_vector = self.embedding.embed(query)
results = self.store.search(query_vector, top_k=top_k)
# 按分类过滤
if category:
results = [(entry, score) for entry, score in results
if entry.category == category]
# 更新访问统计
for entry, _ in results:
entry.access_count += 1
entry.last_accessed_at = datetime.now().isoformat()
return results
def retrieve_as_context(self, query: str, top_k: int = 5) -> str:
"""检索并将结果格式化为上下文字符串"""
results = self.retrieve(query, top_k=top_k)
if not results:
return ""
lines = ["[以下是相关的历史记忆]"]
for i, (entry, similarity) in enumerate(results, 1):
lines.append(f"{i}. {entry.content}（相关度：{similarity:.2f}，分类：{entry.category}）")
lines.append("[历史记忆结束]")
return "\n".join(lines)
def get_stats(self) -> Dict[str, Any]:
"""获取记忆系统的统计信息"""
total = self.store.size()
categories = {}
total_access = 0
for entry in self.store.entries.values():
categories[entry.category] = categories.get(entry.category, 0) + 1
total_access += entry.access_count
return {
"total_memories": total,
"categories": categories,
"total_access_count": total_access,
"avg_access_per_memory": total_access / total if total > 0 else 0,
}何时写入长期记忆
一个关键的设计问题是：什么信息值得被存入长期记忆？ 不是所有对话内容都值得长期保存。
以下信息值得存储：
python
class MemoryExtractor:
"""从对话中提取值得长期记忆的内容"""
def extract_from_user_message(self, content: str) -> List[Dict[str, Any]]:
"""从用户消息中提取候选记忆"""
candidates = []
# 偏好类信息
preference_patterns = [
r"我(?:喜欢|偏好|习惯|通常|一般)(?:在|用|选择|是)?(.{2,50})",
r"不要(.{2,50})",
r"我(?:总是|从不|经常)(.{2,50})",
]
import re
for pattern in preference_patterns:
matches = re.findall(pattern, content)
for match in matches:
candidates.append({
"content": f"用户偏好：{match.strip()}",
"category": "user_preference",
"importance": 4,
})
# 事实类信息
fact_patterns = [
r"我(?:是|在|住|工作)(?:在|于)?(.{2,50})",
r"我(?:的电话|邮箱|地址)(?:是|为)?(.{2,50})",
]
for pattern in fact_patterns:
matches = re.findall(pattern, content)
for match in matches:
candidates.append({
"content": f"用户信息：{match.strip()}",
"category": "fact",
"importance": 4,
})
return candidates
def extract_from_tool_result(self, tool_name: str, result: str) -> List[Dict[str, Any]]:
"""从工具结果中提取候选记忆"""
candidates = []
# 工具结果的摘要信息
if len(result) > 100:
candidates.append({
"content": f"[{tool_name}] {result[:200]}",
"category": "knowledge",
"importance": 2,
"metadata": {"tool": tool_name},
})
return candidates
def should_store(self, content: str, category: str,
existing_memories: List[MemoryEntry]) -> bool:
"""判断是否应该存储这条记忆"""
# 太短的记忆不值得存储
if len(content) < 10:
return False
# 与已有记忆高度重复的不存储
for existing in existing_memories:
if existing.category == category:
# 简单的相似度判断
overlap = self._text_overlap(content, existing.content)
if overlap > 0.7:
return False
return True
def _text_overlap(self, text_a: str, text_b: str) -> float:
"""计算两段文本的重叠度"""
words_a = set(text_a)
words_b = set(text_b)
if not words_a or not words_b:
return 0.0
intersection = len(words_a & words_b)
union = len(words_a | words_b)
return intersection / union if union > 0 else 0.0将长期记忆集成到Agent循环中
更新MemorySystem，让它同时管理短期和长期记忆：
python
class MemorySystem:
"""完整的记忆系统：短期 + 长期"""
def __init__(self, system_prompt: str,
long_term_memory: Optional[LongTermMemory] = None,
embedding_provider: Optional[EmbeddingProvider] = None):
self.system_prompt = system_prompt
self.short_term = WorkingMemory(max_tokens=8000)
self.short_term.set_system_prompt(system_prompt)
self.long_term = long_term_memory or (
LongTermMemory(embedding_provider) if embedding_provider else None
)
self.extractor = MemoryExtractor()
self.messages: List[Dict[str, Any]] = []
def initialize(self, user_input: str):
"""初始化一次新的会话"""
# 清空短期记忆
self.short_term = WorkingMemory(max_tokens=8000)
self.short_term.set_system_prompt(self.system_prompt)
# 添加用户输入
self.short_term.add_user_message(user_input)
# 检查长期记忆中是否有相关信息
if self.long_term:
relevant = self.long_term.retrieve(user_input, top_k=3)
if relevant:
context_text = self.long_term.retrieve_as_context(user_input, top_k=3)
# 将相关记忆注入上下文
self.short_term.add_user_message(
f"[系统注入了相关的历史记忆]\n{context_text}\n"
f"请在有帮助的情况下使用这些历史信息。"
)
self.messages = self.short_term.get_context()
# 从用户输入中提取并存储长期记忆
if self.long_term:
candidates = self.extractor.extract_from_user_message(user_input)
for candidate in candidates:
if self.extractor.should_store(
candidate["content"],
candidate["category"],
[entry for entry, _ in self.long_term.retrieve(candidate["content"], top_k=3)]
):
self.long_term.store(
content=candidate["content"],
category=candidate["category"],
importance=candidate["importance"],
)
def add_tool_result(self, result):
"""添加工具结果到短期记忆，并提取长期记忆"""
self.short_term.add_tool_result(
result.tool_call_id if hasattr(result, 'tool_call_id') else "",
result.content
)
# 从工具结果中提取长期记忆
if self.long_term:
candidates = self.extractor.extract_from_tool_result(
getattr(result, 'tool_name', 'unknown'),
result.content
)
for candidate in candidates:
if self.extractor.should_store(
candidate["content"],
candidate["category"],
[entry for entry, _ in self.long_term.retrieve(candidate["content"], top_k=3)]
):
self.long_term.store(
content=candidate["content"],
category=candidate["category"],
importance=candidate["importance"],
metadata=candidate.get("metadata", {}),
)
self.messages = self.short_term.get_context()
def get_context(self) -> List[Dict[str, Any]]:
"""获取当前上下文"""
return self.short_term.get_context()
def get_stats(self) -> Dict[str, Any]:
"""获取完整的记忆统计"""
stats = {
"short_term": self.short_term.get_stats(),
}
if self.long_term:
stats["long_term"] = self.long_term.get_stats()
return stats一个完整的示例
python
# examples/long_term_memory_demo.py
from openai import OpenAI
# 创建嵌入提供者
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY", ""))
embedding_provider = EmbeddingProvider(client=client)
# 创建长期记忆
ltm = LongTermMemory(embedding_provider)
# 存储一些记忆
ltm.store("用户偏好：喜欢在下午开会", category="user_preference", importance=5)
ltm.store("用户信息：产品经理，负责搜索功能", category="fact", importance=4)
ltm.store("项目：Q3的重点是提升搜索相关性和用户体验", category="knowledge", importance=3)
# 检索测试
print("检索'会议时间'：")
results = ltm.retrieve("什么时候开会比较好", top_k=3)
for entry, score in results:
print(f" [{score:.2f}] {entry.content}")
print("\n检索'用户的工作'：")
results = ltm.retrieve("用户是做什么的", top_k=3)
for entry, score in results:
print(f" [{score:.2f}] {entry.content}")
# 集成到MemorySystem
memory_system = MemorySystem(
system_prompt="你是一个智能助手。",
long_term_memory=ltm,
)
# 模拟一次新会话
memory_system.initialize("帮我安排一个会议")
context = memory_system.get_context()
print(f"\n会话上下文包含{len(context)}条消息")
for msg in context:
print(f" [{msg['role']}] {msg['content'][:80]}...")长期记忆的挑战与权衡
挑战一：检索的相关性
向量检索可能返回语义相关但实际无用的信息。用户问"帮我查一下订单"，检索可能返回"用户上次问过订单状态"的记忆——这条记忆对当前任务没有帮助，但占据了宝贵的上下文空间。
应对：设置相似度阈值，只注入高于阈值的记忆。在实践中，阈值需要根据场景调优。
挑战二：记忆的时效性
用户说"我喜欢用Python"，但一年后可能已经转向了其他语言。长期记忆中过时的信息可能误导Agent的判断。
应对：在记忆条目中记录last_accessed_at和access_count，对长期未被访问的记忆降低重要性。在检索时，可以给新近的记忆更高的权重。
挑战三：记忆的膨胀
随着时间推移，长期记忆会不断增长。如果不去管理，存储量会越来越大，检索也越来越慢。
应对：
- 定期清理：删除长期未被访问且重要性低的记忆
- 记忆合并：将相似的记忆合并为一条
- 分类归档：将不常用的记忆移到"冷存储"
挑战四：存储与检索的成本
每次嵌入和检索都有计算成本（API调用或本地计算）。高频的检索会显著增加Agent的执行成本。
应对：
- 只在会话开始时检索一次，而非每轮循环都检索
- 缓存查询向量，避免重复嵌入
- 设置检索触发条件（如仅当用户消息包含某些关键词时）
本节小结
长期记忆让Agent突破会话的边界，跨时间记住信息。本节构建了一个基于向量检索的长期记忆系统：
- 存储：将文本嵌入为向量，存入向量存储
- 检索：将查询嵌入为向量，通过余弦相似度找到相关记忆
- 集成：在会话开始时注入相关记忆，在对话过程中提取并存储新记忆
- 管理：去重、重要性评分、访问统计
核心原则：长期记忆不是"存得越多越好"，而是"在需要时能检索到最相关的信息"。存储什么、何时检索、检索多少——这些决策比存储技术本身更影响Agent的表现。
在下一节中，我们将讨论会话状态管理——如何管理多轮对话的连续性和一致性。