大模型长上下文处理与记忆压缩：从“金鱼记忆“到“长期记忆“的工程实践

大模型长上下文处理与记忆压缩从金鱼记忆到长期记忆的工程实践一、大模型的记忆瓶颈上下文窗口就是全部记忆大模型的推理能力受限于上下文窗口长度。GPT-4 的 128K token 窗口看似很大但在实际应用中系统提示词、对话历史、工具调用结果、检索文档共同占据上下文空间留给核心推理的空间往往不足 20K token。当对话轮次增多或文档过长时早期信息被截断或稀释模型表现出遗忘——忘记之前讨论的结论、重复提问、前后矛盾。长上下文处理的工程方案分为两类扩展窗口让模型看得更多和压缩记忆让模型记住关键。扩展窗口依赖模型架构改进如 RoPE 外推、Ring Attention压缩记忆则通过摘要、检索和结构化存储减少需要传入上下文的信息量。后者是当前更实用的工程方案。二、记忆压缩的架构与策略记忆压缩的核心思想是只保留对当前推理有用的信息。对话历史中大部分内容是过程性信息中间推理步骤、工具调用细节只有少部分是结论性信息最终决策、关键事实。压缩策略需要区分这两类信息保留后者、丢弃前者。flowchart TD A[完整对话历史br/50K tokens] -- B[记忆分类器br/区分关键 vs 过程信息] B -- C[关键信息br/结论、决策、事实br/~5K tokens] B -- D[过程信息br/中间步骤、工具调用br/~45K tokens] C -- E[结构化记忆存储br/向量数据库 / 知识图谱] D -- F[摘要压缩br/50K → 2K tokens] E -- G[检索增强br/按相关性召回] F -- H[压缩摘要br/保留关键脉络] G -- I[组装上下文br/系统提示 压缩摘要 召回记忆 当前输入] H -- I subgraph 记忆层次 J[工作记忆br/当前对话窗口br/~8K tokens] K[短期记忆br/最近 N 轮摘要br/~4K tokens] L[长期记忆br/向量检索库br/按需召回] end I -- J I -- K I -- L三层记忆架构工作记忆当前对话窗口内的原始信息无需压缩短期记忆最近 N 轮对话的摘要保留关键脉络长期记忆所有历史信息的结构化存储按相关性检索召回三、记忆压缩系统的实现# memory_compression.py — 大模型记忆压缩系统 # 设计意图通过摘要压缩和检索增强将无限对话历史 # 压缩为有限的上下文输入解决大模型的遗忘问题 import hashlib import json from dataclasses import dataclass, field from typing import List, Dict, Optional, Tuple from datetime import datetime dataclass class Message: 对话消息 role: str # user / assistant / system / tool content: str timestamp: datetime field(default_factorydatetime.now) metadata: Dict field(default_factorydict) dataclass class MemoryBlock: 记忆块 id: str content: str # 原始内容或摘要 summary: Optional[str] None # 压缩后的摘要 tokens: int 0 importance: float 0.5 # 重要性评分 0-1 category: str general # fact / decision / process / tool_call embedding: Optional[List[float]] None created_at: datetime field(default_factorydatetime.now) class MemoryCompressor: 记忆压缩器将长对话历史压缩为有限的上下文 def __init__(self, llm_client, max_context_tokens: int 8000): self.llm llm_client self.max_context_tokens max_context_tokens self.memory_blocks: List[MemoryBlock] [] def add_message(self, message: Message): 添加消息到记忆库 block MemoryBlock( idself._generate_id(message), contentmessage.content, tokensself._estimate_tokens(message.content), importanceself._assess_importance(message), categoryself._classify_message(message), ) self.memory_blocks.append(block) def compress(self) - str: 压缩记忆库生成适合上下文窗口的摘要 # Step 1: 按重要性排序 important_blocks [ b for b in self.memory_blocks if b.importance 0.7 ] process_blocks [ b for b in self.memory_blocks if b.importance 0.7 ] # Step 2: 保留高重要性信息的原文 important_content [] token_budget self.max_context_tokens // 2 # 一半预算给关键信息 used_tokens 0 for block in reversed(important_blocks): # 最近的优先 if used_tokens block.tokens token_budget: important_content.append(block.content) used_tokens block.tokens # Step 3: 对过程性信息进行摘要压缩 process_content [b.content for b in process_blocks] if process_content: summary self._generate_summary(process_content) else: summary # Step 4: 组装压缩后的上下文 compressed_parts [] if important_content: compressed_parts.append( Key Information ) compressed_parts.extend(reversed(important_content)) if summary: compressed_parts.append( Conversation Summary ) compressed_parts.append(summary) return \n\n.join(compressed_parts) def retrieve(self, query: str, top_k: int 5) - List[str]: 检索与当前查询相关的记忆 # 基于关键词的简单检索生产环境应使用向量检索 query_keywords set(query.lower().split()) scored_blocks [] for block in self.memory_blocks: content_keywords set(block.content.lower().split()) overlap len(query_keywords content_keywords) if overlap 0: score overlap / max(len(query_keywords), 1) scored_blocks.append((block, score)) scored_blocks.sort(keylambda x: x[1], reverseTrue) return [block.content for block, _ in scored_blocks[:top_k]] def _assess_importance(self, message: Message) - float: 评估消息的重要性 importance 0.5 # 默认中等重要性 # 用户明确表达的偏好或决策 → 高重要性 decision_keywords [ 决定, 选择, 要求, 必须, 不要, prefer, decide, must, dont, ] if any(kw in message.content.lower() for kw in decision_keywords): importance 0.9 # 事实性信息 → 高重要性 fact_keywords [ 名字是, 地址是, 日期是, 编号是, my name is, the address is, ] if any(kw in message.content.lower() for kw in fact_keywords): importance 0.95 # 工具调用结果 → 中等重要性 if message.role tool: importance 0.4 # 系统消息 → 高重要性 if message.role system: importance 1.0 return importance def _classify_message(self, message: Message) - str: 分类消息类型 if message.role system: return system elif message.role tool: return tool_call elif any(kw in message.content for kw in [决定, 选择, 要求]): return decision else: return general def _generate_summary(self, contents: List[str]) - str: 生成对话摘要 combined \n.join(contents[-20:]) # 最多取最近 20 条 prompt fSummarize the following conversation, preserving: 1. Key decisions and conclusions 2. Important facts and numbers 3. Unresolved questions or action items Omit: - Greetings and pleasantries - Repetitive confirmations - Detailed tool call logs Conversation: {combined} Summary: # 实际部署中调用 LLM 生成摘要 # 此处返回简化版本 return f[Summary of {len(contents)} messages: key points preserved] def _generate_id(self, message: Message) - str: 生成记忆块 ID content_hash hashlib.md5(message.content.encode()).hexdigest()[:8] return fmem_{content_hash}_{int(message.timestamp.timestamp())} def _estimate_tokens(self, text: str) - int: 估算 token 数量 # 简化估算中文约 1.5 字/token英文约 4 字符/token return len(text) // 2 class LongTermMemoryStore: 长期记忆存储向量数据库接口 def __init__(self, embedding_clientNone): self.embedding_client embedding_client self.store: Dict[str, MemoryBlock] {} def store_memory(self, block: MemoryBlock): 存储记忆块 if self.embedding_client: block.embedding self.embedding_client.embed(block.content) self.store[block.id] block def search(self, query: str, top_k: int 5) - List[MemoryBlock]: 搜索相关记忆 if not self.embedding_client: # 降级关键词搜索 return self._keyword_search(query, top_k) # 向量搜索 query_embedding self.embedding_client.embed(query) scored [] for block in self.store.values(): if block.embedding: similarity self._cosine_similarity( query_embedding, block.embedding ) scored.append((block, similarity)) scored.sort(keylambda x: x[1], reverseTrue) return [block for block, _ in scored[:top_k]] def _keyword_search(self, query: str, top_k: int) - List[MemoryBlock]: 关键词搜索降级方案 query_words set(query.lower().split()) scored [] for block in self.store.values(): content_words set(block.content.lower().split()) overlap len(query_words content_words) if overlap 0: scored.append((block, overlap)) scored.sort(keylambda x: x[1], reverseTrue) return [block for block, _ in scored[:top_k]] def _cosine_similarity(self, a: List[float], b: List[float]) - float: 计算余弦相似度 dot sum(x * y for x, y in zip(a, b)) norm_a sum(x ** 2 for x in a) ** 0.5 norm_b sum(x ** 2 for x in b) ** 0.5 return dot / (norm_a * norm_b 1e-8) class MemoryManager: 记忆管理器集成压缩和检索 def __init__(self, llm_client, max_context_tokens: int 8000): self.compressor MemoryCompressor(llm_client, max_context_tokens) self.long_term LongTermMemoryStore() def process_turn( self, user_message: Message ) - Tuple[str, List[str]]: 处理一轮对话返回压缩上下文和检索结果 # 添加到记忆 self.compressor.add_message(user_message) # 压缩短期记忆 compressed self.compressor.compress() # 检索长期记忆 retrieved self.compressor.retrieve(user_message.content) # 将高重要性记忆存入长期存储 for block in self.compressor.memory_blocks: if block.importance 0.8 and block.id not in self.long_term.store: self.long_term.store_memory(block) return compressed, retrieved四、记忆压缩的 Trade-offs压缩的信息损失摘要压缩不可避免地丢失细节。一个 50K token 的对话压缩为 2K token 的摘要信息保留率约 4%。关键信息的保留依赖重要性评估的准确性——如果评估错误关键决策可能被当作过程信息丢弃。建议对重要性评估设置保守阈值宁可多保留也不要误删。检索的相关性向量检索基于语义相似度但语义相似不等于推理需要。用户问上次讨论的部署方案是什么检索可能返回包含部署关键词的无关内容。需要结合时间衰减最近的记忆权重更高和结构化标签决策类记忆优先召回提升检索精度。压缩延迟摘要生成需要调用 LLM每次压缩约需 1-3 秒。在实时对话场景中这个延迟可能影响用户体验。建议在后台异步执行压缩对话进行中使用上一轮的压缩结果。记忆一致性长期记忆中可能存在矛盾信息——用户在不同时间表达了不同的偏好。检索时可能同时召回两条矛盾的记忆导致模型困惑。需要引入记忆版本管理新记忆覆盖旧记忆或标注记忆的时间戳供模型判断。五、总结大模型记忆压缩通过三层架构工作记忆 → 短期记忆 → 长期记忆将无限对话历史压缩为有限的上下文输入。重要性评估区分关键信息和过程信息摘要压缩保留关键脉络向量检索按需召回历史记忆。但压缩的信息损失、检索的相关性、压缩延迟和记忆一致性是需要权衡的因素。在实际落地中建议对重要性评估设置保守阈值结合时间衰减和结构化标签优化检索后台异步执行压缩引入记忆版本管理处理矛盾信息。记忆压缩的目标不是记住一切而是在有限的上下文窗口内保留对当前推理最有价值的信息。