长期记忆
长期记忆是 Agent 的"知识库"，实现跨会话的信息持久化存储。它使 Agent 能够记住用户偏好、积累领域知识、复用历史经验，是智能体持续进化的基础。
一、什么是长期记忆？
1.1 定义
长期记忆（Long-Term Memory） 是 Agent 将信息持久化存储到外部存储系统中的机制，生命周期跨越多个会话，支持按需检索和更新。
1.2 核心特征
┌─────────────────────────────────────────────────────────────┐
│ 长期记忆核心特征 │
├─────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ 存储位置：外部存储（向量数据库、关系数据库） │ │
│ │ 生命周期：永久 │ │
│ │ 容量限制：理论上无限制 │ │
│ │ 访问方式：需要检索后注入到短期记忆 │ │
│ │ 访问速度：毫秒到秒级（取决于存储和检索方式） │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
│ 典型内容： │
│ • 用户画像和偏好设置 │
│ • 历史对话记录和关键信息 │
│ • 领域知识库和文档 │
│ • 任务状态和执行记录 │
│ • 学习到的经验和规则 │
│ │
└─────────────────────────────────────────────────────────────┘1.3 与短期记忆的关系
┌─────────────────────────────────────────────────────────────┐
│ 短期记忆与长期记忆的协作 │
├─────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────┐ │
│ │ 用户输入 │ │
│ └──────┬──────┘ │
│ │ │
│ ↓ │
│ ┌──────────────────────────────────────────────────────┐ │
│ │ 检索长期记忆 │ │
│ │ • 根据用户ID获取偏好 │ │
│ │ • 语义检索相关历史知识 │ │
│ └──────────────────────┬───────────────────────────────┘ │
│ │ │
│ ↓ │
│ ┌──────────────────────────────────────────────────────┐ │
│ │ 注入短期记忆 │ │
│ │ • 将检索结果加入 Context Window │ │
│ │ • 与当前对话上下文合并 │ │
│ └──────────────────────┬───────────────────────────────┘ │
│ │ │
│ ↓ │
│ ┌──────────────────────────────────────────────────────┐ │
│ │ LLM 推理 │ │
│ └──────────────────────┬───────────────────────────────┘ │
│ │ │
│ ↓ │
│ ┌──────────────────────────────────────────────────────┐ │
│ │ 写入长期记忆 │ │
│ │ • 提取重要信息持久化 │ │
│ │ • 更新用户画像 │ │
│ └──────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────┘二、长期记忆的分类
2.1 按内容类型分类
| 类型 | 说明 | 存储方式 | 示例 |
|---|---|---|---|
| 用户画像 | 用户的属性和偏好 | 关系型数据库 | 语言偏好、沟通风格、兴趣领域 |
| 对话记忆 | 历史对话记录 | 向量数据库 | 重要对话、用户说过的话 |
| 知识记忆 | 领域知识和文档 | 向量数据库 | 产品文档、技术手册、FAQ |
| 经验记忆 | 学习到的经验规则 | 关系型数据库 | 成功案例、失败教训、最佳实践 |
| 任务记忆 | 任务执行状态 | 关系型数据库 | 进行中的任务、中间结果 |
2.2 按存储结构分类
┌─────────────────────────────────────────────────────────────┐
│ 长期记忆存储分类 │
├─────────────────────────────────────────────────────────────┤
│ │
│ ┌───────────────────────────────────────────────────────┐ │
│ │ 向量数据库 (Vector Database) │ │
│ │ │ │
│ │ 特点： │ │
│ │ • 存储文本的向量表示（Embedding） │ │
│ │ • 支持语义相似度检索 │ │
│ │ • 适合非结构化文本记忆 │ │
│ │ │ │
│ │ 适用场景： │ │
│ │ • 对话历史存储 │ │
│ │ • 知识库问答 │ │
│ │ • 语义搜索 │ │
│ │ │ │
│ │ 代表产品：Chroma, Milvus, Pinecone, Weaviate │ │
│ └───────────────────────────────────────────────────────┘ │
│ │
│ ┌───────────────────────────────────────────────────────┐ │
│ │ 关系型数据库 (Relational Database) │ │
│ │ │ │
│ │ 特点： │ │
│ │ • 存储结构化数据 │ │
│ │ • 支持复杂 SQL 查询 │ │
│ │ • 支持 ACID 事务 │ │
│ │ │ │
│ │ 适用场景： │ │
│ │ • 用户画像管理 │ │
│ │ • 任务状态跟踪 │ │
│ │ • 精确数据查询 │ │
│ │ │ │
│ │ 代表产品：PostgreSQL, MySQL, SQLite │ │
│ └───────────────────────────────────────────────────────┘ │
│ │
│ ┌───────────────────────────────────────────────────────┐ │
│ │ 图数据库 (Graph Database) │ │
│ │ │ │
│ │ 特点： │ │
│ │ • 存储实体和关系 │ │
│ │ • 支持图遍历查询 │ │
│ │ • 适合复杂关系网络 │ │
│ │ │ │
│ │ 适用场景： │ │
│ │ • 知识图谱 │ │
│ │ • 实体关系记忆 │ │
│ │ │ │
│ │ 代表产品：Neo4j, NebulaGraph │ │
│ └───────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────┘三、向量数据库详解
3.1 向量数据库原理
┌─────────────────────────────────────────────────────────────┐
│ 向量数据库工作原理 │
├─────────────────────────────────────────────────────────────┤
│ │
│ 1. 文本向量化 │
│ ┌─────────────────────────────────────────────────┐ │
│ │ "用户喜欢简洁的回答" │ │
│ │ ↓ │ │
│ │ Embedding Model (如 text-embedding-3-small) │ │
│ │ ↓ │ │
│ │ [0.023, -0.145, 0.678, ..., 0.234] │ │
│ │ 1536 维向量 │ │
│ └─────────────────────────────────────────────────┘ │
│ │
│ 2. 向量存储 │
│ ┌─────────────────────────────────────────────────┐ │
│ │ 向量数据库存储： │ │
│ │ ID: "mem_001" │ │
│ │ Vector: [0.023, -0.145, ...] │ │
│ │ Metadata: {"type": "preference", "user": "u1"}│ │
│ │ Document: "用户喜欢简洁的回答" │ │
│ └─────────────────────────────────────────────────┘ │
│ │
│ 3. 向量检索 │
│ ┌─────────────────────────────────────────────────┐ │
│ │ Query: "用户的沟通风格" │ │
│ │ ↓ │ │
│ │ Query Vector: [0.034, -0.123, ...] │ │
│ │ ↓ │ │
│ │ 计算相似度 (余弦相似度/欧氏距离) │ │
│ │ ↓ │ │
│ │ 返回最相似的 Top-K 个结果 │ │
│ └─────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────┘3.2 Chroma 实战示例
基础用法
import chromadb
from chromadb.utils import embedding_functions
# 初始化客户端（内存模式）
client = chromadb.Client()
# 或持久化到磁盘
# client = chromadb.PersistentClient(path="./chroma_db")
# 使用 OpenAI Embedding
openai_ef = embedding_functions.OpenAIEmbeddingFunction(
api_key="your-api-key",
model_name="text-embedding-3-small"
)
# 创建集合
collection = client.create_collection(
name="agent_memory",
embedding_function=openai_ef,
metadata={"description": "Agent 长期记忆存储"}
)存储记忆
# 存储用户偏好
collection.add(
documents=[
"用户偏好使用中文进行交流",
"用户喜欢简洁的技术回答",
"用户关注 Python 和机器学习领域"
],
metadatas=[
{"type": "preference", "category": "language", "user_id": "user_123"},
{"type": "preference", "category": "style", "user_id": "user_123"},
{"type": "preference", "category": "interest", "user_id": "user_123"}
],
ids=["pref_1", "pref_2", "pref_3"]
)
# 存储对话记忆
collection.add(
documents=[
"用户询问了关于 RAG 技术的问题，表示对知识库检索很感兴趣",
"用户正在开发一个 AI 助手项目，使用 LangChain 框架"
],
metadatas=[
{"type": "conversation", "date": "2024-01-15", "user_id": "user_123"},
{"type": "context", "date": "2024-01-15", "user_id": "user_123"}
],
ids=["conv_1", "ctx_1"]
)检索记忆
# 语义检索
results = collection.query(
query_texts=["用户的语言偏好是什么？"],
n_results=3,
where={"user_id": "user_123"} # 元数据过滤
)
print(results)
# {
# 'documents': [['用户偏好使用中文进行交流', ...]],
# 'metadatas': [[{'type': 'preference', ...}, ...]],
# 'distances': [[0.12, 0.34, 0.45]]
# }
# 更新记忆
collection.update(
ids=["pref_1"],
documents=["用户偏好使用英文进行技术讨论，中文进行日常交流"],
metadatas=[{"type": "preference", "category": "language", "user_id": "user_123", "updated": True}]
)
# 删除记忆
collection.delete(ids=["conv_1"])3.3 LangChain 集成
from langchain_openai import OpenAIEmbeddings
from langchain_chroma import Chroma
from langchain.memory import VectorStoreRetrieverMemory
from langchain.chains import ConversationChain
from langchain_openai import OpenAI
# 创建向量存储
embeddings = OpenAIEmbeddings()
vectorstore = Chroma(
embedding_function=embeddings,
persist_directory="./chroma_memory"
)
# 创建记忆检索器
retriever = vectorstore.as_retriever(
search_kwargs={"k": 3} # 返回最相似的3条记忆
)
# 创建向量记忆
memory = VectorStoreRetrieverMemory(retriever=retriever)
# 保存记忆
memory.save_context(
{"input": "我叫小明，是一名 Python 开发者"},
{"output": "你好小明！很高兴认识你。作为 Python 开发者，有什么我可以帮助你的吗？"}
)
memory.save_context(
{"input": "我对机器学习很感兴趣"},
{"output": "很好的兴趣方向！机器学习是 Python 最热门的应用领域之一..."}
)
# 在对话中使用
llm = OpenAI(temperature=0)
conversation = ConversationChain(
llm=llm,
memory=memory,
verbose=True
)
# 提问会自动检索相关记忆
response = conversation.predict(input="你还记得我的职业吗？")
# 会检索到之前保存的"小明是一名 Python 开发者"四、关系型数据库详解
4.1 适用场景
关系型数据库适合存储结构化的长期记忆：
┌─────────────────────────────────────────────────────────────┐
│ 关系型数据库存储的记忆类型 │
├─────────────────────────────────────────────────────────────┤
│ │
│ 1. 用户画像表 (user_profiles) │
│ ┌─────────────────────────────────────────────────┐ │
│ │ user_id (PK) | preferences | created_at | ... │ │
│ │ user_123 | {"lang":"zh"} | 2024-01-01 | ... │ │
│ └─────────────────────────────────────────────────┘ │
│ │
│ 2. 对话记录表 (conversation_history) │
│ ┌─────────────────────────────────────────────────┐ │
│ │ id | user_id | role | content | timestamp | ... │ │
│ │ 1 | user_123| user | 你好 | 2024-01-15 | ...│ │
│ └─────────────────────────────────────────────────┘ │
│ │
│ 3. 任务状态表 (task_states) │
│ ┌─────────────────────────────────────────────────┐ │
│ │ task_id | user_id | status | result | ... │ │
│ │ task_1 | user_123| running| null | ... │ │
│ └─────────────────────────────────────────────────┘ │
│ │
│ 4. 经验知识表 (learned_rules) │
│ ┌─────────────────────────────────────────────────┐ │
│ │ id | rule | confidence | last_used | ... │ │
│ │ 1 | "用户喜欢简洁回答"| 0.9 | 2024-01-15 | ... │ │
│ └─────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────┘4.2 SQLite 实战示例
import sqlite3
import json
from datetime import datetime
from typing import Optional, Dict, List
class RelationalMemoryStore:
"""关系型长期记忆存储"""
def __init__(self, db_path: str = "agent_memory.db"):
self.conn = sqlite3.connect(db_path)
self.conn.row_factory = sqlite3.Row
self._init_tables()
def _init_tables(self):
"""初始化数据表"""
cursor = self.conn.cursor()
# 用户画像表
cursor.execute('''
CREATE TABLE IF NOT EXISTS user_profiles (
user_id TEXT PRIMARY KEY,
preferences JSON,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
''')
# 重要记忆表
cursor.execute('''
CREATE TABLE IF NOT EXISTS important_memories (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id TEXT NOT NULL,
memory_type TEXT NOT NULL,
content TEXT NOT NULL,
importance REAL DEFAULT 0.5,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
last_accessed TIMESTAMP,
access_count INTEGER DEFAULT 0,
FOREIGN KEY (user_id) REFERENCES user_profiles(user_id)
)
''')
# 创建索引
cursor.execute('''
CREATE INDEX IF NOT EXISTS idx_memory_user
ON important_memories(user_id)
''')
cursor.execute('''
CREATE INDEX IF NOT EXISTS idx_memory_type
ON important_memories(memory_type)
''')
self.conn.commit()
def save_user_preference(self, user_id: str, preferences: Dict):
"""保存用户偏好"""
cursor = self.conn.cursor()
cursor.execute('''
INSERT OR REPLACE INTO user_profiles (user_id, preferences, updated_at)
VALUES (?, ?, CURRENT_TIMESTAMP)
''', (user_id, json.dumps(preferences, ensure_ascii=False)))
self.conn.commit()
def get_user_preference(self, user_id: str) -> Optional[Dict]:
"""获取用户偏好"""
cursor = self.conn.cursor()
cursor.execute('''
SELECT preferences FROM user_profiles WHERE user_id = ?
''', (user_id,))
row = cursor.fetchone()
if row:
return json.loads(row['preferences'])
return None
def save_memory(self, user_id: str, memory_type: str, content: str, importance: float = 0.5):
"""保存重要记忆"""
cursor = self.conn.cursor()
cursor.execute('''
INSERT INTO important_memories (user_id, memory_type, content, importance)
VALUES (?, ?, ?, ?)
''', (user_id, memory_type, content, importance))
self.conn.commit()
return cursor.lastrowid
def get_memories(self, user_id: str, memory_type: Optional[str] = None, limit: int = 10) -> List[Dict]:
"""获取记忆列表"""
cursor = self.conn.cursor()
if memory_type:
cursor.execute('''
SELECT * FROM important_memories
WHERE user_id = ? AND memory_type = ?
ORDER BY importance DESC, created_at DESC
LIMIT ?
''', (user_id, memory_type, limit))
else:
cursor.execute('''
SELECT * FROM important_memories
WHERE user_id = ?
ORDER BY importance DESC, created_at DESC
LIMIT ?
''', (user_id, limit))
return [dict(row) for row in cursor.fetchall()]
def update_memory_access(self, memory_id: int):
"""更新记忆访问记录"""
cursor = self.conn.cursor()
cursor.execute('''
UPDATE important_memories
SET last_accessed = CURRENT_TIMESTAMP,
access_count = access_count + 1
WHERE id = ?
''', (memory_id,))
self.conn.commit()
def cleanup_old_memories(self, days: int = 90, min_importance: float = 0.3):
"""清理过时的低重要性记忆"""
cursor = self.conn.cursor()
cursor.execute('''
DELETE FROM important_memories
WHERE importance < ?
AND datetime(created_at) < datetime('now', ?)
''', (min_importance, f'-{days} days'))
deleted = cursor.rowcount
self.conn.commit()
return deleted
def close(self):
"""关闭连接"""
self.conn.close()五、混合存储架构
5.1 架构设计
┌─────────────────────────────────────────────────────────────────────┐
│ 混合存储架构 │
├─────────────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────────┐ │
│ │ 记忆写入请求 │ │
│ └────────┬────────┘ │
│ │ │
│ ↓ │
│ ┌─────────────────┐ │
│ │ 记忆分类器 │ │
│ └────────┬────────┘ │
│ │ │
│ ┌───────────────────┼───────────────────┐ │
│ │ │ │ │
│ ↓ ↓ ↓ │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │
│ │ 结构化记忆 │ │ 非结构化记忆│ │ 关系记忆 │ │
│ │ (用户画像) │ │ (对话内容) │ │ (实体关系) │ │
│ └──────┬──────┘ └──────┬──────┘ └──────┬──────┘ │
│ │ │ │ │
│ ↓ ↓ ↓ │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │
│ │ 关系型 │ │ 向量库 │ │ 图数据库 │ │
│ │ Database │ │ Vector DB │ │ Graph DB │ │
│ │ (SQLite) │ │ (Chroma) │ │ (Neo4j) │ │
│ └─────────────┘ └─────────────┘ └─────────────┘ │
│ │
│ │ │
│ ↓ │
│ ┌─────────────────┐ │
│ │ 统一检索接口 │ │
│ └─────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────────┘5.2 实现示例
from typing import List, Dict, Any, Optional
import chromadb
import sqlite3
import json
class HybridMemoryStore:
"""混合存储的长期记忆系统"""
def __init__(self, sqlite_path: str = "memory.db", chroma_path: str = "./chroma"):
# 初始化向量数据库
self.vector_store = chromadb.PersistentClient(path=chroma_path)
self.vector_collection = self.vector_store.get_or_create_collection(
name="agent_memory"
)
# 初始化关系数据库
self.rel_conn = sqlite3.connect(sqlite_path)
self._init_rel_db()
def _init_rel_db(self):
"""初始化关系数据库"""
cursor = self.rel_conn.cursor()
cursor.execute('''
CREATE TABLE IF NOT EXISTS user_profiles (
user_id TEXT PRIMARY KEY,
preferences JSON,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
''')
cursor.execute('''
CREATE TABLE IF NOT EXISTS entity_memories (
id INTEGER PRIMARY KEY,
user_id TEXT,
entity_type TEXT,
entity_name TEXT,
attributes JSON,
UNIQUE(user_id, entity_type, entity_name)
)
''')
self.rel_conn.commit()
def save_memory(self, user_id: str, content: str, memory_type: str, metadata: Dict = None):
"""智能保存记忆到合适的存储"""
metadata = metadata or {}
metadata["user_id"] = user_id
metadata["memory_type"] = memory_type
# 结构化记忆存入关系数据库
if memory_type in ["preference", "profile", "entity"]:
self._save_to_rel_db(user_id, content, memory_type, metadata)
# 所有记忆都存入向量数据库（支持语义检索）
self._save_to_vector_db(content, metadata)
def _save_to_rel_db(self, user_id: str, content: str, memory_type: str, metadata: Dict):
"""保存到关系数据库"""
cursor = self.rel_conn.cursor()
if memory_type == "preference":
# 更新用户偏好
cursor.execute('''
INSERT OR REPLACE INTO user_profiles (user_id, preferences)
VALUES (?, ?)
''', (user_id, content))
elif memory_type == "entity":
# 保存实体记忆
cursor.execute('''
INSERT OR REPLACE INTO entity_memories
(user_id, entity_type, entity_name, attributes)
VALUES (?, ?, ?, ?)
''', (
user_id,
metadata.get("entity_type", "unknown"),
metadata.get("entity_name", ""),
json.dumps(metadata, ensure_ascii=False)
))
self.rel_conn.commit()
def _save_to_vector_db(self, content: str, metadata: Dict):
"""保存到向量数据库"""
doc_id = f"{metadata['user_id']}_{metadata['memory_type']}_{hash(content)}"
self.vector_collection.add(
documents=[content],
metadatas=[metadata],
ids=[doc_id]
)
def retrieve_memories(self, user_id: str, query: str, limit: int = 5) -> List[Dict]:
"""检索记忆（混合检索）"""
results = []
# 1. 从向量数据库检索（语义相似）
vector_results = self.vector_collection.query(
query_texts=[query],
n_results=limit,
where={"user_id": user_id}
)
if vector_results["documents"]:
for i, doc in enumerate(vector_results["documents"][0]):
results.append({
"content": doc,
"metadata": vector_results["metadatas"][0][i],
"score": 1 - vector_results["distances"][0][i], # 转换为相似度
"source": "vector"
})
# 2. 从关系数据库检索（精确匹配）
# 这里可以添加精确查询逻辑
return results
def get_user_profile(self, user_id: str) -> Optional[Dict]:
"""获取用户画像（从关系数据库）"""
cursor = self.rel_conn.cursor()
cursor.execute('''
SELECT preferences FROM user_profiles WHERE user_id = ?
''', (user_id,))
row = cursor.fetchone()
if row:
return json.loads(row[0])
return None六、长期记忆的管理策略
6.1 记忆生命周期管理
┌─────────────────────────────────────────────────────────────┐
│ 记忆生命周期 │
├─────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────┐ ┌─────────┐ ┌─────────┐ ┌─────────┐ │
│ │ 创建 │ ─→ │ 存储 │ ─→ │ 检索 │ ─→ │ 更新 │ │
│ │ Create │ │ Store │ │Retrieve │ │ Update │ │
│ └─────────┘ └─────────┘ └─────────┘ └─────────┘ │
│ │ │ │ │ │
│ │ │ │ │ │
│ ↓ ↓ ↓ ↓ │
│ 评估重要性 选择存储层 检索策略 更新策略 │
│ 提取关键信息 元数据标注 排序过滤 冲突处理 │
│ │
│ │ │
│ ↓ │
│ ┌─────────────┐ │
│ │ 遗忘 │ │
│ │ Forget │ │
│ └─────────────┘ │
│ │ │
│ ↓ │
│ TTL过期删除 │
│ LRU淘汰清理 │
│ 低重要性清理 │
│ │
└─────────────────────────────────────────────────────────────┘6.2 记忆更新策略
class MemoryUpdateStrategy:
"""记忆更新策略"""
@staticmethod
def should_update(new_info: str, old_memory: str) -> bool:
"""判断是否需要更新记忆"""
# 简化示例：实际应用中可用 LLM 判断
# 如果新信息与旧记忆冲突，则需要更新
return new_info != old_memory
@staticmethod
def merge_memories(old_memory: str, new_info: str, llm_client) -> str:
"""合并新旧记忆"""
prompt = f"""
旧记忆：{old_memory}
新信息：{new_info}
请将新旧信息合并为一条更新后的记忆。
如果有冲突，以新信息为准。
只输出合并后的记忆内容，不要解释。
"""
# return llm_client.generate(prompt)
return f"{old_memory}；{new_info}" # 简化实现
@staticmethod
def detect_conflict(new_info: str, existing_memories: List[str], llm_client) -> bool:
"""检测新信息是否与现有记忆冲突"""
prompt = f"""
现有记忆：
{chr(10).join(f'- {m}' for m in existing_memories)}
新信息：{new_info}
新信息是否与现有记忆冲突？回答"是"或"否"。
"""
# response = llm_client.generate(prompt)
# return "是" in response
return False # 简化实现6.3 记忆遗忘策略
import time
from typing import List, Dict
class MemoryForgettingStrategy:
"""记忆遗忘策略"""
def __init__(self, ttl_days: int = 90, max_access_age_days: int = 30):
self.ttl_days = ttl_days
self.max_access_age_days = max_access_age_days
def calculate_forget_score(self, memory: Dict) -> float:
"""
计算遗忘分数 (0-1，越高越应该遗忘)
考虑因素：
- 创建时间（越久越应该遗忘）
- 最后访问时间（越久未访问越应该遗忘）
- 访问次数（越多访问越不应该遗忘）
- 重要性评分（越重要越不应该遗忘）
"""
now = time.time()
# 时间衰减
created_days = (now - memory.get("created_at", now)) / 86400
last_access_days = (now - memory.get("last_accessed", now)) / 86400
# 访问频率因子
access_count = memory.get("access_count", 0)
access_factor = 1 / (1 + access_count * 0.1)
# 重要性因子
importance = memory.get("importance", 0.5)
importance_factor = 1 - importance
# 综合遗忘分数
forget_score = (
created_days / self.ttl_days * 0.3 +
last_access_days / self.max_access_age_days * 0.3 +
access_factor * 0.2 +
importance_factor * 0.2
)
return min(forget_score, 1.0)
def should_forget(self, memory: Dict) -> bool:
"""判断是否应该遗忘"""
forget_score = self.calculate_forget_score(memory)
return forget_score > 0.7 # 遗忘阈值七、面试高频问题
Q1: 向量数据库检索不准怎么办？
答案要点：
解决方案：
1. 数据层优化
• 数据清洗：去除噪声、重复数据
• Chunking 策略：合理切分文档
• 元数据丰富：添加时间、类型等标签
2. 检索层优化
• 混合检索：语义检索 + 关键词检索
• Rerank 模型：对检索结果重排序
• Query 改写：优化用户查询语句
3. 应用层优化
• Prompt 指示：告诉 LLM "如果检索内容无关，请回答不知道"
• 多轮检索：根据初步结果扩展查询
• 用户反馈：收集用户反馈优化检索Q2: 为什么需要混合存储（向量+关系）？
答案要点：
| 存储类型 | 优势 | 劣势 |
|---|---|---|
| 向量数据库 | 语义检索、理解同义 | 精确查询差、无事务 |
| 关系数据库 | 精确查询、事务支持 | 无语义理解 |
| 混合存储 | 综合优势 | 架构复杂、维护成本 |
混合存储价值：
- 用户偏好用关系库（精确查询）
- 对话内容用向量库（语义检索）
- 统一检索接口封装复杂性
Q3: 如何设计一个支持"用户偏好记忆"的 Agent？
答案要点：
设计步骤：
1. 提取
- 识别用户偏好表述（如"我喜欢简洁的回答"）
- 可用规则或 LLM 进行提取
2. 存储
- 结构化存入关系数据库
- 同时存入向量库支持语义检索
3. 检索
- 会话开始时根据用户 ID 检索偏好
- 将偏好注入 System Prompt
4. 更新
- 当用户表达新偏好时触发更新
- 冲突检测：新偏好覆盖旧偏好
5. 应用
- 在推理时自动应用偏好设置Q4: 长期记忆如何与短期记忆协作？
答案要点：
协作流程：
1. 会话开始
- 检索用户画像和偏好
- 注入到 System Prompt
2. 对话进行中
- 检索相关历史知识
- 注入到 Retrieved Context
3. 会话结束
- 提取重要信息
- 写入长期记忆
4. 定期维护
- 压缩旧记忆
- 清理过时信息Q5: Chroma、Milvus、Pinecone 有什么区别？如何选择？
答案要点：
| 特性 | Chroma | Milvus | Pinecone |
|---|---|---|---|
| 部署方式 | 本地/嵌入式 | 自托管/云服务 | 纯云服务 |
| 适用规模 | 小型项目 | 中大型项目 | 各种规模 |
| 学习曲线 | 低 | 中 | 低 |
| 成本 | 免费 | 开源免费/云付费 | 按使用付费 |
| 性能 | 中等 | 高 | 高 |
| 推荐场景 | 开发测试、小项目 | 企业自建、大规模 | 快速上线、免运维 |
选择建议：
- 开发测试：Chroma（简单快速）
- 企业自建：Milvus（开源可控）
- 快速上线：Pinecone（免运维）
八、总结
核心概念回顾
| 概念 | 定义 | 关键要点 |
|---|---|---|
| 长期记忆 | 持久化存储的外部记忆 | 跨会话、需检索 |
| 向量数据库 | 存储文本向量，支持语义检索 | Chroma, Milvus, Pinecone |
| 关系数据库 | 存储结构化数据，支持精确查询 | 用户画像、任务状态 |
| 混合存储 | 向量+关系数据库结合 | 综合优势、架构复杂 |
| 记忆管理 | 更新、遗忘、压缩策略 | 生命周期管理 |
一句话总结
长期记忆通过向量数据库实现语义检索，通过关系数据库实现精确查询，混合存储架构综合两者优势，配合合理的管理策略，实现 Agent 的知识积累和跨会话连续性。
最后更新：2026年3月18日