大模型长期记忆的工程化之痛：我是怎么被上下文窗口搞疯的

前言

做智能客服系统，用户聊到第 10 轮，大模型就开始"失忆"了。翻聊天记录？那是真痛苦。

长期记忆模块的问题，比想象中复杂得多。今天聊聊我们遇到的工程化挑战和应对方案。

一、底层原理

1.1 长期记忆管理的核心难点

大模型的记忆管理有三个层次：

graph TD
    A["对话上下文"] --> B["短期记忆"]
    B --> C["工作记忆"]
    C --> D["长期记忆"]
    D --> E["向量数据库"]
    D --> F["总结压缩"]
    D --> G["分层检索"]
    B --> H["Token 限制"]
    H --> I["滑动窗口"]
    I --> J["信息丢失"]

核心挑战：

• Token 窗口有限，不能无限塞
• 记忆压缩时丢失细节
• 检索时找不到相关信息
• 记忆的时效性管理

1.2 记忆方案对比

方案	容量	检索速度	信息损失
全部历史	很小	快	无
滑动窗口	中	快	大
总结压缩	大	中	中
向量检索	很大	中	小

二、快速上手

先看最简单的记忆方案，滑动窗口：

from typing import List, Dict, Any
from collections import deque

class SlidingWindowMemory:
    def __init__(self, window_size=10):
        self.window = deque(maxlen=window_size)
    
    def add(self, message: Dict[str, str]):
        self.window.append(message)
    
    def get_context(self) -> List[Dict[str, str]]:
        return list(self.window)
    
    def clear(self):
        self.window.clear()

memory = SlidingWindowMemory(5)
memory.add({"role": "user", "content": "你好"})
memory.add({"role": "assistant", "content": "你好，有什么可以帮助的？"})
print(memory.get_context())

这够简单，但窗口大小就是记忆瓶颈。

再看更好的实现：

class SummaryMemory:
    def __init__(self, llm, max_tokens=2000):
        self.llm = llm
        self.max_tokens = max_tokens
        self.history = []
        self.summary = ""
    
    def add(self, message: Dict):
        self.history.append(message)
        if self._estimate_tokens() > self.max_tokens:
            self._compress()
    
    def _estimate_tokens(self):
        total = len(self.summary.split())
        for msg in self.history:
            total += len(msg.get("content", "").split())
        return total
    
    def _compress(self):
        text = "\n".join(f"{m['role']}: {m['content']}" for m in self.history)
        prompt = f"压缩以下对话为摘要（保留关键信息）：\n{text}"
        self.summary = self.llm(prompt)
        self.history = []
    
    def get_context(self):
        context = [{"role": "system", "content": f"历史摘要：{self.summary}"}]
        context.extend(self.history)
        return context

三、核心 API / 深水区

3.1 记忆管理策略速查

策略	适用场景	优点	缺点
滑动窗口	短期对话	实现简单	丢失历史
总结压缩	长对话	压缩效果好	总结开销
向量检索	知识问答	容量大	检索可能不准
分层记忆	复杂场景	兼顾长短	实现复杂

3.2 向量检索记忆

import json
from typing import List, Dict, Any

class VectorStoreMemory:
    def __init__(self, embedding_func, store):
        self.embedding = embedding_func
        self.store = store
        self.session_memory = []
    
    def add(self, message: Dict):
        self.session_memory.append(message)
    
    def save_to_long_term(self):
        if len(self.session_memory) < 5:
            return
        
        text = "\n".join(m["content"] for m in self.session_memory)
        embedding = self.embedding(text)
        
        self.store.add(
            id=f"mem_{len(self.store)}",
            vector=embedding,
            metadata={
                "text": text,
                "timestamp": time.time()
            }
        )
        self.session_memory = []
    
    def retrieve(self, query: str, k=3) -> List[str]:
        query_embedding = self.embedding(query)
        results = self.store.similarity_search(query_embedding, k=k)
        return [r.metadata["text"] for r in results]

3.3 记忆优先级管理

class PriorityMemory:
    def __init__(self, max_items=100):
        self.max_items = max_items
        self.items = []
    
    def add(self, item: Dict, priority: int = 1):
        self.items.append({"item": item, "priority": priority})
        self.items.sort(key=lambda x: x["priority"], reverse=True)
        self.items = self.items[:self.max_items]
    
    def get_top(self, n=10):
        return [i["item"] for i in self.items[:n]]
    
    def increase_priority(self, key: str, amount: int = 1):
        for item in self.items:
            if item["item"].get("key") == key:
                item["priority"] += amount
                break
        self.items.sort(key=lambda x: x["priority"], reverse=True)

四、实战演练

完整的多层记忆系统：

import time
from typing import List, Dict, Any, Optional
from dataclasses import dataclass, field

@dataclass
class MemoryEntry:
    content: str
    timestamp: float
    type: str
    importance: int = 1

class LayeredMemory:
    def __init__(self, llm, vector_store):
        self.llm = llm
        self.working_memory = []
        self.short_term = []
        self.long_term = vector_store
        self.max_working = 20
        self.max_short = 100
    
    def add(self, entry: MemoryEntry):
        # 1. 加到工作记忆
        self.working_memory.append(entry)
        
        if len(self.working_memory) >= self.max_working:
            self._consolidate()
    
    def _consolidate(self):
        # 压缩到短期记忆
        text = "\n".join(e.content for e in self.working_memory)
        if len(text) > 500:
            prompt = f"压缩以下内容：{text[:1000]}"
            summary = self.llm(prompt)
            self.short_term.append(MemoryEntry(
                content=summary,
                timestamp=time.time(),
                type="summary",
                importance=5
            ))
        
        self.working_memory = []
        
        if len(self.short_term) >= self.max_short:
            self._archive()
    
    def _archive(self):
        # 归档到长期记忆
        for entry in self.short_term:
            if entry.importance >= 3:
                self.long_term.add(entry.content)
        self.short_term = []
    
    def retrieve(self, query: str, k=5) -> List[str]:
        results = []
        
        # 1. 工作记忆
        for entry in self.working_memory[-5:]:
            results.append(entry.content)
        
        # 2. 短期记忆
        for entry in self.short_term[-3:]:
            results.append(entry.content)
        
        # 3. 长期记忆
        long_results = self.long_term.similarity_search(query, k=2)
        results.extend(long_results)
        
        return results

memory = LayeredMemory(llm, vector_store)
memory.add(MemoryEntry("用户说想退款", time.time(), "user", 5))
memory.add(MemoryEntry("系统查询到订单", time.time(), "system", 3))
context = memory.retrieve("退款政策")
print(context)

五、避坑指南与最佳实践

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

⚠️ **警告：压缩太多会丢关键信息
压缩比不要超过 5:1，不然细节全没了。

✅ **推荐：分层记忆架构
工作记忆 + 短期 + 长期，兼顾速度和容量。

六、综合实战演示

生产级长期记忆系统：

import json
import time
from typing import Dict, List, Any, Optional
from collections import OrderedDict

class LongTermMemory:
    def __init__(self, llm, persist_path="memory.json"):
        self.llm = llm
        self.persist_path = persist_path
        self.episodes = []
        self.index = OrderedDict()
    
    def store_episode(self, episode: Dict):
        self.episodes.append({
            "content": episode,
            "timestamp": time.time(),
            "id": len(self.episodes)
        })
        
        # 生成摘要和关键词
        text = json.dumps(episode, ensure_ascii=False)
        keywords = self._extract_keywords(text)
        
        for kw in keywords:
            if kw not in self.index:
                self.index[kw] = []
            self.index[kw].append(len(self.episodes) - 1)
    
    def _extract_keywords(self, text):
        prompt = f"提取关键词（3-5个）：{text[:200]}"
        result = self.llm(prompt)
        return [kw.strip() for kw in result.split(",") if kw.strip()]
    
    def recall(self, query: str, k=5) -> List[Dict]:
        # 关键词匹配
        keywords = self._extract_keywords(query)
        episode_ids = set()
        for kw in keywords:
            for eid in self.index.get(kw, []):
                episode_ids.add(eid)
        
        # 关联意图
        matched = []
        for eid in list(episode_ids)[:k]:
            if eid < len(self.episodes):
                matched.append(self.episodes[eid])
        
        return matched
    
    def forget_old(self, max_age_hours=48):
        now = time.time()
        self.episodes = [
            e for e in self.episodes 
            if (now - e["timestamp"]) < max_age_hours * 3600
        ]
    
    def persist(self):
        data = {
            "episodes": self.episodes,
            "index": dict(self.index)
        }
        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.episodes = data.get("episodes", [])
            self.index = OrderedDict(data.get("index", {}))
        except:
            pass

memory = LongTermMemory(llm)
memory.store_episode({"action": "查询订单", "result": "已发货"})
memory.store_episode({"action": "退款申请", "result": "已退款"})

# 回忆
recalled = memory.recall("我的订单怎么了？")
print(recalled)

# 持久化
memory.persist()

七、总结

大模型长期记忆管理的工程化挑战：

• 窗口限制用滑动窗口解决
• 信息丢失用总结压缩
• 检索用向量+关键词
• 效率用分层架构

没有银弹，但组合起来能有效缓解"失忆"问题。