四种记忆类型详解
1. 工作记忆（Working Memory）
1.1 概述
工作记忆是记忆系统中的”短期记忆”，负责存储当前对话会话中的临时信息。它模拟了人类大脑的工作记忆机制，具有容量有限、访问快速、自动清理的特点。
1.2 核心特性
| 特性 | 说明 | 配置 |
|---|---|---|
| 存储介质 | 纯内存 | - |
| 容量限制 | 默认50条 | working_memory_capacity |
| TTL（生存时间） | 默认60分钟 | working_memory_ttl |
| 检索方式 | 混合检索 | TF-IDF + 关键词 |
| 访问速度 | 极快 | O(1) |
1.3 存储策略
class WorkingMemory:
def __init__(self, config: MemoryConfig):
self.max_capacity = config.working_memory_capacity or 50
self.max_age_minutes = config.working_memory_ttl or 60
self.memories = []
def add(self, memory_item: MemoryItem) -> str:
"""添加工作记忆"""
# 1. 过期清理
self._expire_old_memories()
# 2. 容量管理
if len(self.memories) >= self.max_capacity:
self._remove_lowest_priority_memory()
# 3. 添加记忆
self.memories.append(memory_item)
return memory_item.id1.4 检索机制
混合检索策略
工作记忆采用混合检索策略，结合语义检索和关键词匹配：
def retrieve(self, query: str, limit: int = 5, **kwargs) -> List[MemoryItem]:
"""混合检索：TF-IDF向量化 + 关键词匹配"""
# 1. 过期清理
self._expire_old_memories()
# 2. TF-IDF向量检索
vector_scores = self._try_tfidf_search(query)
# 3. 计算综合分数
scored_memories = []
for memory in self.memories:
vector_score = vector_scores.get(memory.id, 0.0)
keyword_score = self._calculate_keyword_score(query, memory.content)
# 混合评分
base_relevance = vector_score * 0.7 + keyword_score * 0.3
time_decay = self._calculate_time_decay(memory.timestamp)
importance_weight = 0.8 + (memory.importance * 0.4)
final_score = base_relevance * time_decay * importance_weight
if final_score > 0:
scored_memories.append((final_score, memory))
# 4. 排序返回
scored_memories.sort(key=lambda x: x[0], reverse=True)
return [memory for _, memory in scored_memories[:limit]]评分公式
final_score = (向量相似度 × 0.7 + 关键词分数 × 0.3) × 时间衰减 × (0.8 + 重要性 × 0.4)
1.5 使用示例
# 添加工作记忆
memory_tool.execute("add",
content="用户刚才询问了Python列表推导式的用法",
memory_type="working",
importance=0.6
)
# 添加多个工作记忆
for item in recent_conversations:
memory_tool.execute("add",
content=item["text"],
memory_type="working",
importance=item.get("importance", 0.5)
)
# 搜索工作记忆
result = memory_tool.execute("search",
query="列表推导式",
memory_type="working",
limit=3
)1.6 应用场景
- 当前对话上下文：保存最近的对话内容
- 临时状态跟踪：记录任务进度和状态
- 会话级信息：单个会话内的临时数据
- 快速访问场景：需要极快响应速度的场景
1.7 注意事项
⚠️ 重要提醒：
- 工作记忆在系统重启后会丢失
- 不适合存储需要长期保存的信息
- 容量达到上限时会自动删除最不重要的记忆
- TTL过期后自动清理
2. 情景记忆（Episodic Memory）
2.1 概述
情景记忆负责存储具体的事件和经历，类似于人类的”情景记忆”。它保存了完整的上下文信息，支持按时间序列和会话进行检索，是智能体进行”复盘”和学习的基础。
2.2 核心特性
| 特性 | 说明 | 配置 |
|---|---|---|
| 存储架构 | 混合存储 | SQLite + Qdrant |
| 检索能力 | 多维度 | 时间序列 + 语义 + 结构化 |
| 数据持久化 | 长期 | SQLite持久化 |
| 会话管理 | 支持 | session_id索引 |
| 上下文完整性 | 高 | 保存完整元数据 |
2.3 存储架构
class EpisodicMemory:
def __init__(self, config: MemoryConfig):
# SQLite：结构化数据存储
self.doc_store = SQLiteDocumentStore(config.database_path)
# Qdrant：向量检索
self.vector_store = QdrantVectorStore(
config.qdrant_url,
config.qdrant_api_key
)
# 嵌入模型
self.embedder = create_embedding_model_with_fallback()
# 会话索引：session_id -> [episode_ids]
self.sessions = {}数据结构
@dataclass
class Episode:
episode_id: str
session_id: str
timestamp: datetime
content: str
context: Dict[str, Any]
importance: float2.4 存储流程
def add(self, memory_item: MemoryItem) -> str:
"""添加情景记忆"""
# 1. 创建情景对象
episode = Episode(
episode_id=memory_item.id,
session_id=memory_item.metadata.get("session_id", "default"),
timestamp=memory_item.timestamp,
content=memory_item.content,
context=memory_item.metadata
)
# 2. 更新会话索引
session_id = episode.session_id
if session_id not in self.sessions:
self.sessions[session_id] = []
self.sessions[session_id].append(episode.episode_id)
# 3. 持久化存储
self._persist_episode(episode)
return memory_item.id2.5 检索机制
混合检索流程
def retrieve(self, query: str, limit: int = 5, **kwargs) -> List[MemoryItem]:
"""混合检索：结构化过滤 + 语义向量检索"""
# 1. 结构化预过滤
candidate_ids = self._structured_filter(
time_range=kwargs.get("time_range"),
session_id=kwargs.get("session_id"),
min_importance=kwargs.get("min_importance")
)
# 2. 向量语义检索
hits = self._vector_search(
query=query,
limit=limit * 5,
user_id=kwargs.get("user_id")
)
# 3. 综合评分与排序
results = []
for hit in hits:
if self._should_include(hit, candidate_ids, kwargs):
score = self._calculate_episode_score(hit)
memory_item = self._create_memory_item(hit)
results.append((score, memory_item))
results.sort(key=lambda x: x[0], reverse=True)
return [item for _, item in results[:limit]]评分算法
def _calculate_episode_score(self, hit) -> float:
"""情景记忆评分算法"""
vec_score = float(hit.get("score", 0.0))
recency_score = self._calculate_recency(hit["metadata"]["timestamp"])
importance = hit["metadata"].get("importance", 0.5)
# 评分公式
base_relevance = vec_score * 0.8 + recency_score * 0.2
importance_weight = 0.8 + (importance * 0.4)
return base_relevance * importance_weight评分公式：
score = (向量相似度 × 0.8 + 时间近因性 × 0.2) × (0.8 + 重要性 × 0.4)
2.6 时间近因性计算
def _calculate_recency(self, timestamp: str) -> float:
"""计算时间近因性得分"""
memory_time = datetime.fromisoformat(timestamp)
current_time = datetime.now()
age_hours = (current_time - memory_time).total_seconds() / 3600
# 指数衰减模型
decay_factor = 0.05
recency_score = math.exp(-decay_factor * age_hours / 24)
return max(0.1, recency_score) # 最低保持0.12.7 使用示例
# 添加情景记忆 - 学习里程碑
memory_tool.execute("add",
content="2024年3月15日，用户张三完成了第一个Python机器学习项目",
memory_type="episodic",
importance=0.9,
event_type="milestone",
project_name="MNIST分类器",
location="在线学习平台",
session_id="learning_session_20240315"
)
# 添加情景记忆 - 重要对话
memory_tool.execute("add",
content="用户询问了关于神经网络的优化技巧",
memory_type="episodic",
importance=0.7,
event_type="question",
topic="neural_networks",
session_id="daily_chat_20240320"
)
# 按时间范围检索
result = memory_tool.execute("search",
query="机器学习项目",
memory_type="episodic",
time_range=("2024-03-01", "2024-03-31"),
limit=10
)
# 按会话检索
result = memory_tool.execute("search",
query="神经网络",
memory_type="episodic",
session_id="daily_chat_20240320",
limit=5
)2.8 应用场景
- 历史事件记录：保存重要对话和事件
- 学习经历追踪：记录学习里程碑和进展
- 用户行为分析：追踪用户交互历史
- 经验复盘：回顾过去的事件以学习经验
2.9 高级特性
会话管理
# 获取特定会话的所有记忆
def get_session_memories(session_id: str) -> List[MemoryItem]:
"""获取会话内所有情景记忆"""
if session_id not in self.sessions:
return []
episode_ids = self.sessions[session_id]
return [self._get_episode_by_id(eid) for eid in episode_ids]时间序列分析
# 获取时间趋势
def get_time_series_trend(days: int = 30) -> Dict:
"""获取时间趋势分析"""
memories = self.retrieve("", limit=1000)
# 按天统计
daily_counts = {}
for memory in memories:
date = memory.timestamp.date()
daily_counts[date] = daily_counts.get(date, 0) + 1
return daily_counts3. 语义记忆（Semantic Memory）
3.1 概述
语义记忆是记忆系统中最复杂的部分，负责存储抽象的概念、规则和知识。它采用知识图谱技术，能够进行复杂的关联推理，是智能体形成”知识体系”的核心。
3.2 核心特性
| 特性 | 说明 | 配置 |
|---|---|---|
| 存储架构 | 混合存储 | Qdrant + Neo4j |
| 知识表示 | 图结构 | 实体 + 关系 |
| 检索能力 | 智能推理 | 向量 + 图 + 语义 |
| 推理能力 | 强 | 关系遍历和推断 |
| 持久性 | 长期 | 专业的图数据库 |
3.3 存储架构
class SemanticMemory(BaseMemory):
def __init__(self, config: MemoryConfig, storage_backend=None):
super().__init__(config, storage_backend)
# 嵌入模型
self.embedding_model = get_text_embedder()
# 专业数据库存储
self.vector_store = QdrantConnectionManager.get_instance(**qdrant_config)
self.graph_store = Neo4jGraphStore(**neo4j_config)
# 实体和关系缓存
self.entities: Dict[str, Entity] = {}
self.relations: List[Relation] = []
# NLP处理器（支持中英文）
self.nlp = self._init_nlp()知识图谱数据结构
@dataclass
class Entity:
entity_id: str
entity_type: str # 人物、地点、概念等
name: str
properties: Dict[str, Any]
@dataclass
class Relation:
relation_id: str
source_entity: str
target_entity: str
relation_type: str # 是、属于、包含等
properties: Dict[str, Any]3.4 知识图谱构建
def add(self, memory_item: MemoryItem) -> str:
"""添加语义记忆并构建知识图谱"""
# 1. 生成文本嵌入
embedding = self.embedding_model.encode(memory_item.content)
# 2. 提取实体
entities = self._extract_entities(memory_item.content)
print(f"📝 提取到 {len(entities)} 个实体: {[e.name for e in entities]}")
# 3. 提取关系
relations = self._extract_relations(memory_item.content, entities)
print(f"🔗 提取到 {len(relations)} 个关系: {[r.relation_type for r in relations]}")
# 4. 存储到Neo4j图数据库
for entity in entities:
self._add_entity_to_graph(entity, memory_item)
for relation in relations:
self._add_relation_to_graph(relation, memory_item)
# 5. 存储到Qdrant向量数据库
metadata = {
"memory_id": memory_item.id,
"entities": [e.entity_id for e in entities],
"entity_count": len(entities),
"relation_count": len(relations)
}
self.vector_store.add_vectors(
vectors=[embedding.tolist()],
metadata=[metadata],
ids=[memory_item.id]
)
return memory_item.id3.5 实体和关系提取
def _extract_entities(self, text: str) -> List[Entity]:
"""使用NLP提取实体"""
doc = self.nlp(text)
entities = []
for ent in doc.ents:
entity = Entity(
entity_id=f"ent_{hash(ent.text + str(ent.label_))}",
entity_type=ent.label_,
name=ent.text,
properties={
"start": ent.start_char,
"end": ent.end_char
}
)
entities.append(entity)
return entities
def _extract_relations(self, text: str, entities: List[Entity]) -> List[Relation]:
"""提取实体间的关系"""
relations = []
# 基于规则的关系提取
for i, entity1 in enumerate(entities):
for entity2 in entities[i+1:]:
# 使用依存句法分析提取关系
relation_type = self._infer_relation_type(text, entity1, entity2)
if relation_type:
relation = Relation(
relation_id=f"rel_{hash(entity1.entity_id + entity2.entity_id)}",
source_entity=entity1.entity_id,
target_entity=entity2.entity_id,
relation_type=relation_type,
properties={}
)
relations.append(relation)
return relations3.6 检索机制
混合检索策略
def retrieve(self, query: str, limit: int = 5, **kwargs) -> List[MemoryItem]:
"""检索语义记忆：向量检索 + 图检索"""
# 1. 向量检索
vector_results = self._vector_search(
query=query,
limit=limit * 2,
user_id=kwargs.get("user_id")
)
# 2. 图检索
graph_results = self._graph_search(
query=query,
limit=limit * 2,
user_id=kwargs.get("user_id")
)
# 3. 混合排序
combined_results = self._combine_and_rank_results(
vector_results,
graph_results,
query,
limit
)
return combined_results[:limit]图检索实现
def _graph_search(self, query: str, limit: int, user_id: str) -> List[Dict]:
"""基于知识图谱的检索"""
# 1. 提取查询中的实体
query_entities = self._extract_entities(query)
if not query_entities:
return []
# 2. 在Neo4j中查询相关节点和关系
results = []
for entity in query_entities:
# 查找同类型实体
cypher = f"""
MATCH (e:{entity.entity_type})
WHERE e.name CONTAINS $name
RETURN e, [(e)-[r]->(n) | {{rel: r, node: n}}] as relations
LIMIT {limit}
"""
hits = self.graph_store.execute_query(cypher, name=entity.name)
results.extend(hits)
# 3. 计算相似度分数
for result in results:
result["similarity"] = self._calculate_graph_similarity(
query_entities,
result
)
return results混合排序算法
def _combine_and_rank_results(
self,
vector_results,
graph_results,
query,
limit
) -> List[Dict]:
"""混合排序结果"""
combined = {}
# 合并向量和图检索结果
for result in vector_results:
combined[result["memory_id"]] = {
**result,
"vector_score": result.get("score", 0.0),
"graph_score": 0.0
}
for result in graph_results:
memory_id = result["memory_id"]
if memory_id in combined:
combined[memory_id]["graph_score"] = result.get("similarity", 0.0)
else:
combined[memory_id] = {
**result,
"vector_score": 0.0,
"graph_score": result.get("similarity", 0.0)
}
# 计算混合分数
for memory_id, result in combined.items():
vector_score = result["vector_score"]
graph_score = result["graph_score"]
importance = result.get("importance", 0.5)
# 基础相似度得分
base_relevance = vector_score * 0.7 + graph_score * 0.3
# 重要性权重 [0.8, 1.2]
importance_weight = 0.8 + (importance * 0.4)
# 最终得分
combined_score = base_relevance * importance_weight
result["combined_score"] = combined_score
# 排序并返回
sorted_results = sorted(
combined.values(),
key=lambda x: x["combined_score"],
reverse=True
)
return sorted_results[:limit]评分公式：
score = (向量相似度 × 0.7 + 图相似度 × 0.3) × (0.8 + 重要性 × 0.4)
3.7 知识推理
def infer_knowledge(self, entity: str, relation_type: str) -> List[Dict]:
"""基于知识图谱进行推理"""
# Neo4j查询：查找通过多跳关系连接的实体
cypher = f"""
MATCH (start:Entity {{name: $name}})-[r1:{relation_type}*1..3]->(end)
RETURN end, r1, length(r1) as hops
ORDER BY hops ASC, end.name ASC
LIMIT 10
"""
results = self.graph_store.execute_query(cypher, name=entity)
# 格式化结果
inferred = []
for result in results:
inferred.append({
"entity": result["end"]["name"],
"type": result["end"]["type"],
"path_length": result["hops"],
"confidence": 1.0 / result["hops"]
})
return inferred3.8 使用示例
# 添加语义记忆 - 概念知识
memory_tool.execute("add",
content="深度学习是机器学习的一个子领域，使用神经网络进行学习",
memory_type="semantic",
importance=0.9,
knowledge_type="concept",
domain="AI"
)
# 添加语义记忆 - 用户偏好
memory_tool.execute("add",
content="张三喜欢使用Python进行数据分析，特别是pandas和numpy库",
memory_type="semantic",
importance=0.85,
preference_type="programming",
domain="data_science"
)
# 语义检索
result = memory_tool.execute("search",
query="Python数据分析",
memory_type="semantic",
limit=5
)
# 知识推理
knowledge = memory_tool.execute("infer",
entity="深度学习",
relation_type="属于"
)3.9 应用场景
- 知识库构建：存储领域知识和概念
- 用户偏好管理：记录用户长期偏好和习惯
- 关联推理：基于知识图谱进行推理
- 规则和约束：存储需要遵守的规则
4. 感知记忆（Perceptual Memory）
4.1 概述
感知记忆专门处理多模态数据（图像、音频、视频等），支持跨模态检索和语义理解。随着多模态交互的普及，感知记忆变得越来越重要。
4.2 核心特性
| 特性 | 说明 | 配置 |
|---|---|---|
| 支持模态 | 多模态 | 文本、图像、音频、视频 |
| 编码器 | 专用 | CLIP（图像）、CLAP（音频） |
| 检索方式 | 跨模态 | 同模态 + 跨模态检索 |
| 存储策略 | 分离存储 | 按模态分集合 |
| 语义对齐 | 支持 | 统一向量空间 |
4.3 存储架构
class PerceptualMemory(BaseMemory):
def __init__(self, config: MemoryConfig, storage_backend=None):
super().__init__(config, storage_backend)
# 多模态编码器
self.text_embedder = get_text_embedder()
self._clip_model = self._init_clip_model() # 图像编码
self._clap_model = self._init_clap_model() # 音频编码
# 按模态分离的向量存储
self.vector_stores = {
"text": QdrantConnectionManager.get_instance(
collection_name="perceptual_text",
vector_size=self.vector_dim
),
"image": QdrantConnectionManager.get_instance(
collection_name="perceptual_image",
vector_size=self._image_dim
),
"audio": QdrantConnectionManager.get_instance(
collection_name="perceptual_audio",
vector_size=self._audio_dim
)
}4.4 多模态编码
def _encode_data(self, data: Any, modality: str) -> np.ndarray:
"""根据模态编码数据"""
if modality == "text":
return self.text_embedder.encode(data)
elif modality == "image":
image = self._load_image(data)
return self._clip_model.encode_image(image)
elif modality == "audio":
audio = self._load_audio(data)
return self._clap_model.encode_audio(audio)
else:
raise ValueError(f"不支持的模态: {modality}")4.5 存储流程
def add(self, memory_item: MemoryItem) -> str:
"""添加感知记忆"""
modality = memory_item.metadata.get("modality", "text")
# 1. 根据模态编码数据
if modality == "image":
embedding = self._encode_data(
memory_item.metadata["raw_data"],
"image"
)
elif modality == "audio":
embedding = self._encode_data(
memory_item.metadata["raw_data"],
"audio"
)
else:
embedding = self._encode_data(
memory_item.content,
"text"
)
# 2. 存储到对应的向量集合
store = self.vector_stores[modality]
metadata = {
"memory_id": memory_item.id,
"modality": modality,
"timestamp": memory_item.timestamp.isoformat(),
"importance": memory_item.importance,
**memory_item.metadata
}
store.add_vectors(
vectors=[embedding.tolist()],
metadata=[metadata],
ids=[memory_item.id]
)
return memory_item.id4.6 检索机制
同模态检索
def retrieve(self, query: str, limit: int = 5, **kwargs) -> List[MemoryItem]:
"""检索感知记忆（支持同模态和跨模态）"""
user_id = kwargs.get("user_id")
target_modality = kwargs.get("target_modality")
query_modality = kwargs.get("query_modality", target_modality or "text")
# 同模态向量检索
try:
query_vector = self._encode_data(query, query_modality)
store = self._get_vector_store_for_modality(
target_modality or query_modality
)
# 构建过滤条件
where = {"memory_type": "perceptual"}
if user_id:
where["user_id"] = user_id
if target_modality:
where["modality"] = target_modality
hits = store.search_similar(
query_vector=query_vector,
limit=max(limit * 5, 20),
where=where
)
except Exception:
hits = []
# 融合排序
results = []
for hit in hits:
vector_score = float(hit.get("score", 0.0))
recency_score = self._calculate_recency_score(
hit["metadata"]["timestamp"]
)
importance = hit["metadata"].get("importance", 0.5)
# 评分算法
base_relevance = vector_score * 0.8 + recency_score * 0.2
importance_weight = 0.8 + (importance * 0.4)
combined_score = base_relevance * importance_weight
results.append((combined_score, self._create_memory_item(hit)))
results.sort(key=lambda x: x[0], reverse=True)
return [item for _, item in results[:limit]]跨模态检索
def cross_modal_retrieve(
self,
query: str,
query_modality: str,
target_modality: str,
limit: int = 5
) -> List[MemoryItem]:
"""跨模态检索"""
# 1. 编码查询
query_vector = self._encode_data(query, query_modality)
# 2. 检索目标模态
target_store = self.vector_stores[target_modality]
hits = target_store.search_similar(
query_vector=query_vector,
limit=limit * 5
)
# 3. 排序并返回
results = []
for hit in hits:
score = float(hit.get("score", 0.0))
results.append((score, self._create_memory_item(hit)))
results.sort(key=lambda x: x[0], reverse=True)
return [item for _, item in results[:limit]]4.7 时间衰减模型
def _calculate_recency_score(self, timestamp: str) -> float:
"""计算时间近因性得分（指数衰减）"""
try:
memory_time = datetime.fromisoformat(timestamp)
current_time = datetime.now()
age_hours = (current_time - memory_time).total_seconds() / 3600
# 指数衰减：24小时内保持高分，之后逐渐衰减
decay_factor = 0.1
recency_score = math.exp(-decay_factor * age_hours / 24)
return max(0.1, recency_score)
except Exception:
return 0.54.8 使用示例
# 添加图像记忆
memory_tool.execute("add",
content="用户上传了一张代码截图，展示了Python函数定义",
memory_type="perceptual",
importance=0.7,
modality="image",
file_path="./uploads/code_screenshot.png",
raw_data="./uploads/code_screenshot.png"
)
# 添加音频记忆
memory_tool.execute("add",
content="用户录制了一段语音，解释了机器学习的基本概念",
memory_type="perceptual",
importance=0.8,
modality="audio",
file_path="./uploads/ml_explanation.wav",
raw_data="./uploads/ml_explanation.wav"
)
# 文本搜索图像（跨模态）
result = memory_tool.execute("search",
query="Python函数定义",
target_modality="image",
limit=5
)
# 图像搜索图像（同模态）
result = memory_tool.execute("search",
query="./uploads/query_image.jpg",
query_modality="image",
target_modality="image",
limit=5
)4.9 应用场景
- 图像检索：根据文本描述检索相关图像
- 音频分析：存储和检索语音信息
- 多模态对话：支持图像、语音等交互
- 内容审核：检测和过滤不当的多模态内容
5. 记忆类型对比总结
| 特性 | 工作记忆 | 情景记忆 | 语义记忆 | 感知记忆 |
|---|---|---|---|---|
| 存储介质 | 纯内存 | SQLite+Qdrant | Qdrant+Neo4j | Qdrant |
| 容量限制 | 有 | 无 | 无 | 无 |
| TTL | 有 | 无 | 无 | 无 |
| 检索速度 | 极快 | 快 | 中等 | 快 |
| 语义理解 | 基础 | 较强 | 很强 | 强 |
| 推理能力 | 无 | 无 | 有 | 无 |
| 多模态支持 | 否 | 否 | 否 | 是 |
| 适用场景 | 临时信息 | 历史事件 | 知识概念 | 多媒体 |
选择建议：
- 需要快速访问 → 工作记忆
- 需要时间序列 → 情景记忆
- 需要关系推理 → 语义记忆
- 需要多模态 → 感知记忆