情感陪伴应用的长期记忆架构：从会话缓存到持久化记忆网络-尧图网络科技

情感陪伴应用的长期记忆架构：从会话缓存到持久化记忆网络

一、对话结束后记忆即消失的体验断层

用户周一告诉 AI 陪伴助手"我最近在准备面试，很焦虑"。AI 给出了温暖的鼓励和实用的建议。周三用户再次打开对话，说"今天又是那种感觉"。AI 回复"听起来你今天过得不太顺利，能跟我聊聊吗"。这个回复在技术上没有错误——它识别了情绪，给出了共情回应。但它缺少一个关键信息：用户正在经历的是"面试焦虑"，而不是"今天不太顺利"的一般性情绪。

会话缓存在对话窗口关闭后即消失。这意味着每次新对话都是"初次见面"，AI 永远无法积累对用户的理解深度。情感陪伴应用的核心价值在于"了解你"，而"了解"的基础是长期记忆。

二、长期记忆的三层架构

按信息的时效性和结构化程度，将记忆分为三层：

graph TB subgraph 工作记忆（会话级） W1[当前对话历史] W2[本轮情绪状态] W3[最近 3 条用户提及] end subgraph 情景记忆（周-月级） E1[关键事件时间线<br/>"上周二提到面试"] E2[情绪变化趋势<br/>"本周焦虑程度上升"] E3[对话主题聚类<br/>"过去 7 天主要话题"] end subgraph 语义记忆（持久级） S1[用户画像<br/>年龄/职业/兴趣] S2[长期偏好<br/>喜欢的互动方式] S3[核心价值观<br/>对家人/工作的态度] end W1 -->|提取关键事件| E1 W2 -->|聚合趋势| E2 E1 -->|沉淀为持久知识| S1 E3 -->|提炼主题偏好| S2

工作记忆只在当前会话内有效——包括对话历史和实时情绪。情景记忆从工作记忆中提取关键事件和情绪趋势，保留数周到数月。语义记忆从情景记忆中沉淀出稳定的用户画像和偏好，理论上永久保留。

三、记忆提取与检索的实现

""" 长期记忆管理系统——三层记忆的提取、存储与检索。 设计意图：每轮对话后自动提取关键信息， 下次会话开始时检索相关记忆注入 Prompt。 """ from dataclasses import dataclass, field from typing import List, Optional from datetime import datetime, timedelta from enum import Enum import json class MemoryType(str, Enum): WORKING = "working" # 工作记忆 EPISODIC = "episodic" # 情景记忆 SEMANTIC = "semantic" # 语义记忆 @dataclass class Memory: """记忆条目""" id: str type: MemoryType content: str # 情绪标签——用于情绪趋势聚合 emotion: Optional[str] = None # 重要性 0-1——决定保留优先级 importance: float = 0.5 created_at: datetime = field(default_factory=datetime.now) last_accessed: datetime = field(default_factory=datetime.now) # 访问次数——高频访问的记忆不衰减 access_count: int = 0 @dataclass class MemoryContext: """注入 Prompt 的记忆上下文""" recent_events: List[str] # 近 7 天关键事件 emotion_trend: str # 情绪趋势描述 user_profile: str # 用户画像摘要 last_topic: str # 上次对话主题 class MemoryManager: """记忆管理器""" def __init__(self, user_id: str): self.user_id = user_id self.working_memory: List[Memory] = [] self.episodic_cache: List[Memory] = [] self.semantic_cache: List[Memory] = [] async def extract_from_conversation( self, user_message: str, ai_response: str, ) -> List[Memory]: """ 从对话中提取记忆——由 AI 辅助完成。 设计意图：不依赖规则匹配，让 AI 判断哪些信息值得记忆。 """ extraction_prompt = f"""分析以下对话，提取需要长期记忆的关键信息。 用户消息: {user_message} AI 回复: {ai_response} 提取规则： 1. 只提取用户相关的信息（事实、情绪、偏好、事件） 2. 每条信息赋予重要性评分（0-1）： - 0.8-1.0：核心身份信息（职业、家庭成员） - 0.5-0.7：重要事件（生病、面试、旅行计划） - 0.2-0.4：一般偏好（喜欢的颜色、音乐） 3. 标注情绪标签（happy/sad/anxious/angry/calm/neutral） 输出 JSON 数组： [{{"content": "记忆内容", "importance": 0.8, "emotion": "焦虑"}}]""" # 调用 AI 提取记忆（省略实际调用代码） raw_result = await self._call_ai_for_extraction(extraction_prompt) try: items = json.loads(raw_result) except json.JSONDecodeError: return [] memories = [] for item in items: memory = Memory( id=f"mem_{datetime.now().timestamp()}_{len(memories)}", type=MemoryType.EPISODIC, content=item['content'], emotion=item.get('emotion'), importance=item.get('importance', 0.5), ) memories.append(memory) # 重要性 > 0.7 的直接计入语义记忆候选 if memory.importance > 0.7: memory.type = MemoryType.SEMANTIC return memories async def get_context_for_session(self) -> MemoryContext: """为新一轮对话组装记忆上下文""" now = datetime.now() one_week_ago = now - timedelta(days=7) # 检索近 7 天的情景记忆 recent = [ m for m in self.episodic_cache if m.created_at > one_week_ago ] # 按重要性排序，取前 5 条 recent.sort(key=lambda m: m.importance, reverse=True) recent_events = [m.content for m in recent[:5]] # 情绪趋势聚合 emotion_trend = self._aggregate_emotion(recent) # 用户画像（语义记忆） profile_items = [m.content for m in self.semantic_cache[:3]] user_profile = '；'.join(profile_items) if profile_items else '暂无' # 上次对话主题 last_topic = self.episodic_cache[-1].content if self.episodic_cache else '暂无' return MemoryContext( recent_events=recent_events, emotion_trend=emotion_trend, user_profile=user_profile, last_topic=last_topic, ) def build_memory_prompt(self, context: MemoryContext) -> str: """将记忆上下文注入 Prompt""" parts = [] if context.user_profile != '暂无': parts.append(f"## 关于用户\n{context.user_profile}") if context.recent_events: events_text = '\n'.join( f"- {e}" for e in context.recent_events ) parts.append(f"## 近期关键事件\n{events_text}") if context.emotion_trend != '无显著情绪': parts.append(f"## 情绪趋势\n{context.emotion_trend}") if not parts: return "" # 注入指令——让 AI 在回复中参考但不复读记忆内容 parts.append("\n请在回复中自然地融入以上信息，不要直接复述记忆内容。") return '\n\n'.join(parts) def _aggregate_emotion(self, memories: List[Memory]) -> str: """聚合情绪趋势——简单多数判断""" if not memories: return '无显著情绪' emotions = [m.emotion for m in memories if m.emotion] if not emotions: return '无显著情绪' # 统计情绪分布 from collections import Counter counter = Counter(emotions) dominant = counter.most_common(1)[0] if dominant[1] / len(emotions) > 0.5: return f"近 7 天主要情绪：{dominant[0]}" return "情绪波动较大，无主导情绪" async def _call_ai_for_extraction(self, prompt: str) -> str: """调用 AI 做记忆提取（占位实现）""" return '[]' def decay_old_memories(self, days_threshold: int = 90) -> int: """记忆衰减——超过阈值的低重要性记忆自动清理""" threshold = datetime.now() - timedelta(days=days_threshold) before = len(self.episodic_cache) self.episodic_cache = [ m for m in self.episodic_cache if m.created_at > threshold or m.importance > 0.7 ] return before - len(self.episodic_cache)

三个关键设计：