背景：记忆系统的进化压力

2025年之前，AI Agent 的记忆系统基本等同于"对话历史 + 文本向量库"。进入2026年，随着多模态 LLM 的普及，用户与 Agent 的交互介质已扩展至图像、语音、视频，单一文本记忆架构的局限性愈发明显：- 用户上传的图表、截图无法被后续对话有效检索- 语音会议记录以文字形式存储，丢失大量语调、情绪上下文- 短视频内容被截成孤立帧，无法捕捉时序信息- 多模态内容之间缺乏语义关联，形成"孤岛记忆"本文深入探讨如何将图像、语音、视频融入 Agent 长期记忆系统的完整工程方案。—## 一、多模态记忆的分类体系### 1.1 记忆类型矩阵text按时间维度： ├─ 工作记忆（Working Memory）：当前对话上下文，< 128K tokens ├─ 情节记忆（Episodic Memory）：历史对话事件，可检索 └─ 语义记忆（Semantic Memory）：提炼的长期知识，高度压缩按模态维度： ├─ 文本记忆：对话历史、文档摘要、知识条目 ├─ 图像记忆：截图、图表、照片的视觉嵌入 ├─ 语音记忆：转录文本 + 说话人特征 + 情感标注 └─ 视频记忆：关键帧嵌入 + 时序摘要 + 动作描述### 1.2 统一记忆条目模型pythonfrom dataclasses import dataclass, fieldfrom datetime import datetimefrom typing import Optional, Literalimport numpy as npModalityType = Literal["text", "image", "audio", "video", "multimodal"]@dataclassclass MemoryEntry: """统一多模态记忆条目""" entry_id: str user_id: str session_id: str created_at: datetime # 模态标识 modality: ModalityType # 内容字段（按模态填充） text_content: Optional[str] = None # 文本/转录内容 text_embedding: Optional[np.ndarray] = None # 文本语义向量 visual_embedding: Optional[np.ndarray] = None # 图像/视频帧视觉向量 visual_caption: Optional[str] = None # 视觉内容的自然语言描述 audio_transcript: Optional[str] = None # 语音转文字 speaker_embedding: Optional[np.ndarray] = None # 说话人特征向量 emotion_labels: Optional[dict] = None # {"sentiment": "positive", "arousal": 0.7} video_keyframes: Optional[list[np.ndarray]] = None # 关键帧嵌入列表 video_temporal_summary: Optional[str] = None # 时序摘要 # 元数据 importance_score: float = 0.5 # 记忆重要性 0-1 access_count: int = 0 # 被检索次数 last_accessed: Optional[datetime] = None tags: list[str] = field(default_factory=list) source_url: Optional[str] = None—## 二、图像记忆：视觉内容的编码与检索### 2.1 图像编码流水线pythonimport torchfrom PIL import Imageimport base64from io import BytesIOclass ImageMemoryEncoder: def __init__(self, vision_model, caption_model): self.vision_model = vision_model # 如 CLIP/SigLIP self.caption_model = caption_model # 如 LLaVA/Qwen-VL async def encode_image(self, image_path: str, context: str = "") -> dict: """ 图像 → 记忆条目编码 输出：视觉向量 + 语义描述 + 结构化标签 """ image = Image.open(image_path).convert("RGB") # 1. 提取视觉嵌入（用于相似度检索） visual_emb = self._extract_visual_embedding(image) # 2. 生成详细描述（用于文本检索） caption = await self._generate_caption(image, context) # 3. 提取结构化信息 tags = await self._extract_tags(image, caption) # 4. 生成融合嵌入（视觉+文本） text_emb = self._extract_text_embedding(caption) fused_emb = self._fuse_embeddings(visual_emb, text_emb) return { "visual_embedding": visual_emb.tolist(), "text_embedding": text_emb.tolist(), "fused_embedding": fused_emb.tolist(), "caption": caption, "tags": tags, } def _extract_visual_embedding(self, image: Image.Image) -> np.ndarray: with torch.no_grad(): inputs = self.vision_model.processor(images=image, return_tensors="pt") outputs = self.vision_model.encode_image(**inputs) return outputs.pooler_output.squeeze().numpy() async def _generate_caption(self, image: Image.Image, context: str) -> str: """使用多模态 LLM 生成图像的详细文字描述""" prompt = f"""请详细描述这张图片的内容，包括：1. 主要内容是什么2. 包含哪些关键信息（如图表数据、文字内容）3. 图片的类型（截图/图表/照片等）4. 与用户问题"{context}"的关联性输出格式：详细描述（200字以内）""" # 调用多模态模型 img_b64 = self._image_to_base64(image) response = await self.caption_model.chat( messages=[{ "role": "user", "content": [ {"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{img_b64}"}}, {"type": "text", "text": prompt} ] }] ) return response.content def _fuse_embeddings(self, visual: np.ndarray, text: np.ndarray, alpha: float = 0.4) -> np.ndarray: """加权融合视觉和文本嵌入""" # 归一化 visual_norm = visual / (np.linalg.norm(visual) + 1e-8) text_norm = text / (np.linalg.norm(text) + 1e-8) # 加权融合 fused = alpha * visual_norm + (1 - alpha) * text_norm return fused / (np.linalg.norm(fused) + 1e-8)### 2.2 图像记忆检索pythonclass ImageMemoryRetriever: def __init__(self, vector_store, encoder: ImageMemoryEncoder): self.store = vector_store self.encoder = encoder async def search_by_text(self, query: str, top_k: int = 5) -> list[MemoryEntry]: """文本查询 → 检索相关图像记忆""" query_emb = self.encoder._extract_text_embedding(query) results = await self.store.similarity_search( vector=query_emb, collection="image_memories", top_k=top_k, search_field="fused_embedding" ) return results async def search_by_image(self, query_image: Image.Image, top_k: int = 5) -> list[MemoryEntry]: """图像查询 → 检索相似图像记忆""" visual_emb = self.encoder._extract_visual_embedding(query_image) results = await self.store.similarity_search( vector=visual_emb, collection="image_memories", top_k=top_k, search_field="visual_embedding" ) return results—## 三、语音记忆：超越文字的音频信息保留### 3.1 语音记忆编码流水线pythonclass AudioMemoryEncoder: def __init__(self, asr_model, speaker_model, emotion_model): self.asr = asr_model # 语音转文字（如 Whisper） self.speaker = speaker_model # 说话人识别 self.emotion = emotion_model # 情感分析 async def encode_audio(self, audio_path: str) -> dict: """ 音频文件 → 多维度记忆条目 """ # 1. 语音转文字（带时间戳） transcript_result = await self.asr.transcribe( audio_path, word_timestamps=True, language="zh" ) # 2. 说话人分离（多人场景） speaker_segments = await self.speaker.diarize(audio_path) # 3. 情感分析 emotion_result = await self.emotion.analyze( audio_path, transcript_result["text"] ) # 4. 生成结构化摘要 structured = self._build_structured_transcript( transcript_result, speaker_segments, emotion_result ) # 5. 提取说话人嵌入（用于身份检索） speaker_embs = await self.speaker.extract_embeddings( audio_path, speaker_segments ) return { "full_transcript": transcript_result["text"], "structured_transcript": structured, "speaker_embeddings": speaker_embs, "emotion_timeline": emotion_result["timeline"], "key_moments": self._extract_key_moments( transcript_result, emotion_result ), } def _build_structured_transcript(self, transcript, speakers, emotions) -> list[dict]: """构建带说话人和情感标注的结构化转录""" segments = [] for seg in transcript["segments"]: speaker_id = self._find_speaker(seg["start"], speakers) emotion = self._find_emotion(seg["start"], seg["end"], emotions) segments.append({ "start": seg["start"], "end": seg["end"], "speaker": speaker_id, "text": seg["text"], "emotion": emotion, "confidence": seg.get("avg_logprob", 0) }) return segments def _extract_key_moments(self, transcript, emotions) -> list[dict]: """提取高重要性时刻：情感峰值、关键词密集区""" key_moments = [] for seg in transcript["segments"]: # 检测情感强度峰值 arousal = self._get_arousal_at(seg["start"], emotions) if arousal > 0.8: key_moments.append({ "timestamp": seg["start"], "type": "emotional_peak", "text": seg["text"], "arousal": arousal }) return key_moments—## 四、视频记忆：时序信息的压缩与检索### 4.1 视频记忆编码pythonclass VideoMemoryEncoder: def __init__(self, frame_encoder, caption_model, asr_model): self.frame_enc = frame_encoder self.captioner = caption_model self.asr = asr_model async def encode_video(self, video_path: str, sample_fps: float = 1.0) -> dict: """ 视频 → 关键帧嵌入 + 时序摘要 + 音轨记忆 """ # 1. 提取关键帧 keyframes = self._extract_keyframes(video_path, sample_fps) # 2. 对每帧生成描述和嵌入 frame_memories = [] for ts, frame in keyframes: emb = self.frame_enc.encode(frame) caption = await self.captioner.describe(frame) frame_memories.append({ "timestamp": ts, "embedding": emb.tolist(), "caption": caption, }) # 3. 音轨处理（同语音记忆流水线） audio_memory = await self.asr.transcribe(video_path) # 4. 生成时序摘要 temporal_summary = await self._generate_temporal_summary( frame_memories, audio_memory ) # 5. 场景分割 scenes = self._detect_scene_changes(keyframes) return { "keyframes": frame_memories, "audio_memory": audio_memory, "temporal_summary": temporal_summary, "scenes": scenes, "duration": keyframes[-1][0] if keyframes else 0, } def _extract_keyframes(self, video_path: str, fps: float) -> list[tuple]: """使用场景变化检测提取关键帧""" import cv2 cap = cv2.VideoCapture(video_path) video_fps = cap.get(cv2.CAP_PROP_FPS) interval = int(video_fps / fps) frames = [] frame_idx = 0 prev_frame = None while True: ret, frame = cap.read() if not ret: break if frame_idx % interval == 0: # 场景变化检测 if prev_frame is None or self._is_scene_change(prev_frame, frame): timestamp = frame_idx / video_fps frames.append((timestamp, Image.fromarray(cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)))) prev_frame = frame frame_idx += 1 cap.release() return frames—## 五、统一检索层：多模态记忆融合查询pythonclass MultiModalMemoryRetriever: """统一多模态记忆检索器""" def __init__(self, vector_store, encoders: dict): self.store = vector_store self.encoders = encoders # {"text": ..., "image": ..., "audio": ...} async def search(self, query: str, query_image: Optional[Image.Image] = None, modalities: list[str] = None, top_k: int = 10) -> list[MemoryEntry]: """ 多模态联合检索：文本+图像查询，返回所有模态的相关记忆 """ if modalities is None: modalities = ["text", "image", "audio", "video"] # 编码查询 text_emb = self.encoders["text"].encode(query) visual_emb = None if query_image: visual_emb = self.encoders["image"].encode(query_image) # 并发检索各模态 import asyncio tasks = [] for modality in modalities: tasks.append(self._search_modality( modality, text_emb, visual_emb, top_k )) results_per_modality = await asyncio.gather(*tasks) # 合并去重，按相关性重排 all_results = [] for results in results_per_modality: all_results.extend(results) # 去重 seen_ids = set() unique_results = [] for r in all_results: if r.entry_id not in seen_ids: seen_ids.add(r.entry_id) unique_results.append(r) # 重排：综合文本相关性 + 时间衰减 + 访问频率 return self._rerank(unique_results, query, top_k) def _rerank(self, results: list[MemoryEntry], query: str, top_k: int) -> list[MemoryEntry]: """综合多维度信号重排记忆条目""" from datetime import datetime, timezone now = datetime.now(timezone.utc) scored = [] for entry in results: # 时间衰减（半衰期7天） age_days = (now - entry.created_at.replace(tzinfo=timezone.utc)).days time_score = 0.5 ** (age_days / 7) # 访问频率奖励 access_score = min(entry.access_count / 10, 1.0) * 0.2 # 重要性权重 importance = entry.importance_score final_score = 0.6 * importance + 0.2 * time_score + 0.2 * access_score scored.append((final_score, entry)) scored.sort(key=lambda x: x[0], reverse=True) return [e for _, e in scored[:top_k]]—## 六、存储架构选型| 存储需求 | 推荐方案 | 备注 ||---------|---------|------|| 向量检索（高维嵌入） | Milvus / Qdrant | 支持多向量字段 || 原始媒体文件 | 对象存储（OSS/S3） | 冷热分层存储 || 结构化元数据 | PostgreSQL + pgvector | 混合查询 || 实时工作记忆 | Redis | TTL 自动过期 || 时序数据（情感轨迹） | InfluxDB | 时间序列查询 |—## 总结2026年 AI Agent 多模态记忆系统的工程核心是统一记忆条目模型 + 模态专用编码流水线 + 融合检索层。图像需要视觉嵌入与文字描述双轨存储；语音需要保留超越文字的情感与说话人信息；视频需要时序分割与关键帧嵌入。统一检索层通过多模态联合查询，让 Agent 能够跨模态召回历史上下文，真正实现"看过、听过、看过视频的事情都能记住"的长期记忆能力。