当前位置：首页 > news >正文

大模型长期记忆机制：长上下文记忆管理的工程化挑战与应对方案

news 2026/6/5 13:37:08

大模型长期记忆机制：长上下文记忆管理的工程化挑战与应对方案

大模型长期记忆工程化的三个"致命错误"，本文全都犯了

前言

做大模型应用一年，长期记忆这个问题折腾了本文最久。踩了无数坑，犯了很多错误。

今天把这些经验总结出来，希望大家少走弯路。本文犯了三个致命错误，每一个都差点让项目夭折。

一、底层原理

1.1 长期记忆的三个层次

长期记忆不是"存起来"那么简单，它有三个层次：

graph TD A["用户对话"] --> B["短期记忆"] B --> C["工作记忆"] C --> D["长期记忆"] D --> E["向量数据库"] D --> F["摘要存储"] D --> G["知识图谱"] B --> H["Token 超限"] H --> I["截断"] I --> J["信息丢失"] C --> K["重要性评分"] K --> L["高优先级保存"]

三个层次：

短期记忆：当前对话上下文
工作记忆：正在处理的信息
长期记忆：持久化存储

1.2 三个致命错误

错误	表现	后果
无限存储	存了 10 万条历史	OOM
过度压缩	压缩比 10:1	丢失关键信息
忽略检索	全量加载	响应慢

二、快速上手

2.1 错误 1：无限存储

# 错误：无限存储 class BadMemory: def __init__(self): self.history = [] # 无限增长 def add(self, msg): self.history.append(msg) # 没有裁剪，OOM 预定

2.2 正确做法

class GoodMemory: def __init__(self, max_size=1000): self.history = [] self.max_size = max_size def add(self, msg): self.history.append(msg) if len(self.history) > self.max_size: self.history = self.history[-self.max_size:]

三、核心 API / 深水区

3.1 记忆管理速查

| 策略 | 做法 | 效果 |
| 限长 | 最大 1000 条 | 防止 OOM |
| 摘要 | LLM 压缩 | 保留要点 |
| 分层 | 短/中/长 | 兼顾效率 |
| 过期 | TTL | 自动清理 |

3.2 错误 2：过度压缩

# 错误：过度压缩 def bad_summarize(text): return text[:50] # 太短了 # 正确：摘要保留关键信息 def good_summarize(text, llm): prompt = f"""请压缩以下内容，保留关键信息（压缩比不超过 3:1）： {text}""" return llm(prompt)

3.3 错误 3：忽略检索

# 错误：全量加载 def bad_get_context(self): return "\n".join(self.history) # 10000 条全加载 # 正确：检索后加载 def good_get_context(self, query): relevant = self.retrieve(query, k=5) return "\n".join(relevant)

四、实战演练

4.1 正确的长期记忆实现

from typing import List, Dict, Any, Optional from dataclasses import dataclass import time import json @dataclass class MemoryItem: content: str timestamp: float importance: int class CorrectLongTermMemory: def __init__(self, llm, max_items=1000): self.llm = llm self.max_items = max_items self.short_term: List[MemoryItem] = [] self.long_term: List[MemoryItem] = [] self.summaries: List[str] = [] def add(self, content: str, importance: int = 1): item = MemoryItem( content=content, timestamp=time.time(), importance=importance ) # 短期记忆 self.short_term.append(item) # 自动整理 if len(self.short_term) > 50: self._consolidate() if len(self.long_term) > self.max_items: self._cleanup() def _consolidate(self): high_importance = [i for i in self.short_term if i.importance >= 5] low_importance = [i for i in self.short_term if i.importance < 5] # 高优先级保留全部 self.long_term.extend(high_importance) # 低优先级生成摘要 if low_importance: text = "\n".join(i.content for i in low_importance) prompt = f"压缩以下内容（保留关键信息）：\n{text}" summary = self.llm(prompt[:1000]) self.summaries.append(summary) self.short_term = self.short_term[-10:] def _cleanup(self): # 按重要性排序，保留高 self.long_term.sort(key=lambda x: x.importance, reverse=True) self.long_term = self.long_term[:self.max_items] def get_context(self, query: str) -> str: parts = [] # 1. 最近的短期记忆 for item in self.short_term[-5:]: parts.append(item.content) # 2. 长期记忆中相似的内容 for item in self.long_term[-10:]: if self._is_relevant(query, item.content): parts.append(item.content) # 3. 相关摘要 for summary in self.summaries[-3:]: if self._is_relevant(query, summary): parts.append(f"[摘要] {summary}") return "\n".join(parts) def _is_relevant(self, query: str, content: str) -> bool: # 简化版相关性判断 query_words = set(query.lower().split()) content_words = set(content.lower().split()) common = query_words & content_words return len(common) > 0 def save(self, path="memory.json"): data = { "short_term": [{"content": i.content, "importance": i.importance} for i in self.short_term], "long_term": [{"content": i.content, "importance": i.importance} for i in self.long_term], "summaries": self.summaries } with open(path, "w") as f: json.dump(data, f, ensure_ascii=False) def load(self, path="memory.json"): try: with open(path) as f: data = json.load(f) self.short_term = [MemoryItem(**d, timestamp=time.time()) for d in data.get("short_term", [])] self.long_term = [MemoryItem(**d, timestamp=time.time()) for d in data.get("long_term", [])] self.summaries = data.get("summaries", []) except: pass memory = CorrectLongTermMemory(llm) memory.add("用户说想买手机", importance=8) memory.add("用户说预算 5000", importance=6) context = memory.get_context("手机") print(context) memory.save()

五、避坑指南与最佳实践

💡技巧：重要性评分
不是所有信息都值钱，给记忆打分，保留高分的。

⚠️警告：压缩比不超过 5:1
压缩太多会丢细节，5:1 是一个安全界限。

✅推荐：持久化到文件
进程重启后记忆不丢失，这是最低要求。

六、综合实战演示

6.1 生产级长期记忆系统

from typing import Dict, List, Any, Optional from datetime import datetime import json class ProductionLongTermMemory: def __init__(self, llm, user_id: str, persist_path: str): self.llm = llm self.user_id = user_id self.persist_path = persist_path self.messages: List[Dict] = [] self.key_points: List[str] = [] self._load() def add_message(self, role: str, content: str): self.messages.append({ "role": role, "content": content, "time": datetime.now().isoformat() }) self._extract_key_points(content) self._trim() self._save() def _extract_key_points(self, content: str): prompt = f"提取关键信息（不超过 50 字）：{content}" key_point = self.llm(prompt) if key_point: self.key_points.append(key_point) if len(self.key_points) > 50: self.key_points = self.key_points[-50:] def _trim(self): if len(self.messages) > 100: # 保留最新的 50 条 old = self.messages[:-50] self.messages = self.messages[-50:] # 旧消息生成摘要 text = "\n".join(m["content"] for m in old) summary = self.llm(f"总结：{text[:1000]}") self.key_points.append(f"[摘要] {summary}") def get_context(self, token_limit=2000) -> str: parts = [] # 关键点 for kp in self.key_points[-10:]: parts.append(f"关键: {kp}") # 最近消息 for msg in self.messages[-10:]: parts.append(f"{msg['role']}: {msg['content']}") return "\n".join(parts) def _save(self): data = { "messages": self.messages, "key_points": self.key_points } with open(self.persist_path, "w") as f: json.dump(data, f, ensure_ascii=False) def _load(self): try: with open(self.persist_path) as f: data = json.load(f) self.messages = data.get("messages", []) self.key_points = data.get("key_points", []) except: pass memory = ProductionLongTermMemory(llm, "user_001", "memory.json") memory.add_message("user", "本文想买一台电脑") memory.add_message("assistant", "请问您的预算是多少？") memory.add_message("user", "预算 8000 左右") print(memory.get_context())