知识模块
🤖 Agent 知识模块
记忆压缩

记忆压缩与摘要

记忆压缩是解决上下文窗口限制的核心技术。通过智能摘要和关键信息提取,在保留核心信息的同时减少 Token 占用,实现记忆的高效管理。


一、为什么需要记忆压缩?

1.1 问题背景

┌─────────────────────────────────────────────────────────────┐
│                    记忆压缩的必要性                          │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  挑战 1:Context Window 限制                                 │
│  ┌─────────────────────────────────────────────────────┐   │
│  │ • GPT-3.5: 4K Token                                  │   │
│  │ • GPT-4: 8K-32K Token                                │   │
│  │ • GPT-4-Turbo: 128K Token                            │   │
│  │ • Claude: 200K Token                                 │   │
│  │                                                      │   │
│  │ 用户对话历史可能无限增长,但窗口有限!                 │   │
│  └─────────────────────────────────────────────────────┘   │
│                                                             │
│  挑战 2:成本控制                                            │
│  ┌─────────────────────────────────────────────────────┐   │
│  │ • 输入 Token 按量计费                                 │   │
│  │ • 100K Token 输入成本 >> 10K Token                   │   │
│  │ • 长上下文 = 高成本                                   │   │
│  └─────────────────────────────────────────────────────┘   │
│                                                             │
│  挑战 3:推理质量                                            │
│  ┌─────────────────────────────────────────────────────┐   │
│  │ • "Lost in the Middle" 现象                          │   │
│  │ • 过长上下文导致注意力分散                            │   │
│  │ • 关键信息被噪声淹没                                  │   │
│  └─────────────────────────────────────────────────────┘   │
│                                                             │
│  解决方案:记忆压缩                                          │
│  ┌─────────────────────────────────────────────────────┐   │
│  │ • 将长对话压缩为简洁摘要                              │   │
│  │ • 保留关键信息,丢弃冗余                              │   │
│  │ • 平衡信息完整性和 Token 效率                         │   │
│  └─────────────────────────────────────────────────────┘   │
│                                                             │
└─────────────────────────────────────────────────────────────┘

1.2 压缩效果示例

原始对话(约 500 Token):
────────────────────────────────────────────────────────────
User: 我想学习 Python,应该从哪里开始?
Assistant: 很好的选择!Python 是一门非常适合初学者的编程语言...
User: 有什么推荐的学习资源吗?
Assistant: 我推荐以下资源:1. 官方文档 2. 廖雪峰教程 3. Coursera 课程...
User: 我每天只有1小时学习时间,怎么安排?
Assistant: 每天1小时完全可以学好Python,建议这样安排...
User: 学完基础后应该学什么方向?
Assistant: Python有很多应用方向,比如Web开发、数据分析、机器学习...
User: 我对数据分析比较感兴趣
Assistant: 数据分析是个很好的方向!你可以学习 pandas、numpy...
User: 这些库难学吗?
Assistant: 这些库入门不难,建议从 pandas 开始...
────────────────────────────────────────────────────────────

压缩后摘要(约 50 Token):
────────────────────────────────────────────────────────────
用户想学习 Python,每天1小时学习时间。推荐了官方文档、
廖雪峰教程等资源。用户对数据分析方向感兴趣,计划学习
pandas 和 numpy 库。
────────────────────────────────────────────────────────────

压缩率:90%(500 Token → 50 Token)
信息保留:核心意图和偏好完整保留

二、记忆压缩策略

2.1 压缩策略分类

┌─────────────────────────────────────────────────────────────┐
│                    记忆压缩策略分类                          │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  1. 摘要压缩 (Summarization)                                │
│     ┌─────────────────────────────────────────────────┐    │
│     │ • 使用 LLM 生成摘要                              │    │
│     │ • 保留核心信息,丢弃细节                         │    │
│     │ • 适用于:长对话、历史记录                       │    │
│     └─────────────────────────────────────────────────┘    │
│                                                             │
│  2. 提取压缩 (Extraction)                                   │
│     ┌─────────────────────────────────────────────────┐    │
│     │ • 提取关键实体和事实                             │    │
│     │ • 结构化存储(如三元组)                         │    │
│     │ • 适用于:用户偏好、重要信息                     │    │
│     └─────────────────────────────────────────────────┘    │
│                                                             │
│  3. 截断压缩 (Truncation)                                   │
│     ┌─────────────────────────────────────────────────┐    │
│     │ • 直接丢弃旧内容                                 │    │
│     │ • 保留最近 N 轮                                  │    │
│     │ • 适用于:时效性强的内容                         │    │
│     └─────────────────────────────────────────────────┘    │
│                                                             │
│  4. 向量压缩 (Vector Compression)                           │
│     ┌─────────────────────────────────────────────────┐    │
│     │ • 将文本转换为向量表示                           │    │
│     │ • 使用向量检索代替原文                           │    │
│     │ • 适用于:需要语义检索的场景                     │    │
│     └─────────────────────────────────────────────────┘    │
│                                                             │
└─────────────────────────────────────────────────────────────┘

2.2 策略选择指南

场景推荐策略理由
长对话历史摘要压缩保留核心意图
用户偏好提取压缩结构化存储
临时信息截断压缩时效性强
知识库向量压缩支持语义检索

三、摘要压缩实现

3.1 对话摘要

from langchain_openai import ChatOpenAI
from langchain.prompts import PromptTemplate
from typing import List, Dict
 
class ConversationSummarizer:
    """对话摘要器"""
    
    def __init__(self, llm=None):
        self.llm = llm or ChatOpenAI(temperature=0)
        
        self.summary_prompt = PromptTemplate(
            template="""
            请将以下对话历史压缩为简洁的摘要。
            
            要求:
            1. 保留用户的意图和偏好
            2. 保留重要的决策和结论
            3. 使用简洁的语言
            4. 控制在 100 字以内
            
            对话历史:
            {conversation}
            
            摘要:
            """,
            input_variables=["conversation"]
        )
    
    def summarize(self, messages: List[Dict]) -> str:
        """生成对话摘要"""
        # 格式化对话
        conversation = self._format_conversation(messages)
        
        # 生成摘要
        chain = self.summary_prompt | self.llm
        result = chain.invoke({"conversation": conversation})
        
        return result.content.strip()
    
    def _format_conversation(self, messages: List[Dict]) -> str:
        """格式化对话"""
        lines = []
        for msg in messages:
            role = msg.get("role", "unknown")
            content = msg.get("content", "")
            lines.append(f"{role}: {content}")
        return "\n".join(lines)
 
# 使用示例
messages = [
    {"role": "user", "content": "我想学习 Python,应该从哪里开始?"},
    {"role": "assistant", "content": "很好的选择!Python 是一门非常适合初学者的编程语言..."},
    {"role": "user", "content": "有推荐的学习资源吗?"},
    {"role": "assistant", "content": "我推荐以下资源:1. 官方文档 2. 廖雪峰教程..."},
    {"role": "user", "content": "我对数据分析比较感兴趣"},
    {"role": "assistant", "content": "数据分析是个很好的方向!你可以学习 pandas..."}
]
 
summarizer = ConversationSummarizer()
summary = summarizer.summarize(messages)
print(summary)
# 输出:用户想学习Python,对数据分析方向感兴趣。推荐了官方文档、廖雪峰教程等学习资源,计划学习pandas库。

3.2 增量摘要

from typing import Optional
 
class IncrementalSummarizer:
    """增量摘要器"""
    
    def __init__(self, llm=None, max_summary_length: int = 500):
        self.llm = llm or ChatOpenAI(temperature=0)
        self.max_summary_length = max_summary_length
        self.current_summary: Optional[str] = None
        
        self.update_prompt = PromptTemplate(
            template="""
            现有摘要:
            {existing_summary}
            
            新对话内容:
            {new_content}
            
            请将现有摘要与新对话内容合并,生成更新后的摘要。
            要求:
            1. 保留所有重要信息
            2. 去除冗余和重复
            3. 保持简洁
            
            更新后的摘要:
            """,
            input_variables=["existing_summary", "new_content"]
        )
    
    def update(self, new_messages: List[Dict]) -> str:
        """增量更新摘要"""
        new_content = self._format_messages(new_messages)
        
        if self.current_summary is None:
            # 首次摘要
            self.current_summary = self._create_initial_summary(new_messages)
        else:
            # 增量更新
            chain = self.update_prompt | self.llm
            result = chain.invoke({
                "existing_summary": self.current_summary,
                "new_content": new_content
            })
            self.current_summary = result.content.strip()
        
        # 检查长度,必要时压缩
        if len(self.current_summary) > self.max_summary_length:
            self.current_summary = self._compress_summary()
        
        return self.current_summary
    
    def _create_initial_summary(self, messages: List[Dict]) -> str:
        """创建初始摘要"""
        summarizer = ConversationSummarizer(self.llm)
        return summarizer.summarize(messages)
    
    def _compress_summary(self) -> str:
        """压缩过长的摘要"""
        compress_prompt = PromptTemplate(
            template="""
            请将以下摘要进一步压缩,保留最核心的信息:
            
            原摘要:
            {summary}
            
            压缩后的摘要:
            """,
            input_variables=["summary"]
        )
        
        chain = compress_prompt | self.llm
        result = chain.invoke({"summary": self.current_summary})
        return result.content.strip()
    
    def _format_messages(self, messages: List[Dict]) -> str:
        """格式化消息"""
        return "\n".join([
            f"{m['role']}: {m['content']}" 
            for m in messages
        ])
 
# 使用示例
summarizer = IncrementalSummarizer()
 
# 第一轮对话
summary1 = summarizer.update([
    {"role": "user", "content": "我想学习 Python"},
    {"role": "assistant", "content": "很好的选择..."}
])
print(f"摘要1: {summary1}")
 
# 第二轮对话(增量更新)
summary2 = summarizer.update([
    {"role": "user", "content": "我对数据分析感兴趣"},
    {"role": "assistant", "content": "可以学习 pandas..."}
])
print(f"摘要2: {summary2}")

3.3 LangChain 集成

from langchain.memory import ConversationSummaryMemory
from langchain.chains import ConversationChain
from langchain_openai import OpenAI
 
# 使用 LangChain 内置的摘要记忆
llm = OpenAI(temperature=0)
memory = ConversationSummaryMemory(llm=llm)
 
conversation = ConversationChain(
    llm=llm,
    memory=memory,
    verbose=True
)
 
# 多轮对话
conversation.predict(input="我想学习 Python 编程")
conversation.predict(input="有推荐的教程吗?")
conversation.predict(input="我每天只有1小时学习时间")
 
# 查看摘要
print(memory.load_memory_variables({}))
# 输出类似:{'history': '用户想学习Python编程,询问了教程推荐,表示每天有1小时学习时间。'}

四、关键信息提取

4.1 实体和关系提取

from typing import List, Dict, Tuple
from dataclasses import dataclass
import json
 
@dataclass
class ExtractedFact:
    """提取的事实"""
    subject: str      # 主体
    relation: str     # 关系
    object: str       # 客体
    confidence: float # 置信度
 
class FactExtractor:
    """事实提取器"""
    
    def __init__(self, llm=None):
        self.llm = llm or ChatOpenAI(temperature=0)
        
        self.extraction_prompt = PromptTemplate(
            template="""
            请从以下文本中提取关键事实,以 JSON 格式输出。
            
            文本:
            {text}
            
            提取格式:
            {{
                "facts": [
                    {{"subject": "主体", "relation": "关系", "object": "客体"}}
                ]
            }}
            
            只输出 JSON,不要其他内容:
            """,
            input_variables=["text"]
        )
    
    def extract(self, text: str) -> List[ExtractedFact]:
        """提取事实"""
        chain = self.extraction_prompt | self.llm
        result = chain.invoke({"text": text})
        
        try:
            data = json.loads(result.content)
            facts = []
            for item in data.get("facts", []):
                facts.append(ExtractedFact(
                    subject=item.get("subject", ""),
                    relation=item.get("relation", ""),
                    object=item.get("object", ""),
                    confidence=1.0
                ))
            return facts
        except json.JSONDecodeError:
            return []
    
    def extract_from_conversation(self, messages: List[Dict]) -> List[ExtractedFact]:
        """从对话中提取事实"""
        all_facts = []
        for msg in messages:
            if msg["role"] == "user":
                facts = self.extract(msg["content"])
                all_facts.extend(facts)
        return all_facts
 
# 使用示例
extractor = FactExtractor()
 
text = "我叫小明,是一名 Python 开发者,我对机器学习很感兴趣"
facts = extractor.extract(text)
 
for fact in facts:
    print(f"{fact.subject} --[{fact.relation}]--> {fact.object}")
# 输出:
# 小明 --[是]--> Python 开发者
# 小明 --[感兴趣]--> 机器学习

4.2 用户画像提取

from typing import Dict, Any
from dataclasses import dataclass, field, asdict
 
@dataclass
class UserProfile:
    """用户画像"""
    user_id: str
    name: str = ""
    profession: str = ""
    interests: List[str] = field(default_factory=list)
    preferences: Dict[str, Any] = field(default_factory=dict)
    goals: List[str] = field(default_factory=list)
    
    def to_dict(self) -> Dict:
        return asdict(self)
 
class ProfileExtractor:
    """用户画像提取器"""
    
    def __init__(self, llm=None):
        self.llm = llm or ChatOpenAI(temperature=0)
        
        self.extraction_prompt = PromptTemplate(
            template="""
            请从以下对话中提取用户画像信息,以 JSON 格式输出。
            
            对话:
            {conversation}
            
            提取格式:
            {{
                "name": "用户姓名(如果提到)",
                "profession": "职业(如果提到)",
                "interests": ["兴趣1", "兴趣2"],
                "preferences": {{
                    "language": "语言偏好",
                    "style": "沟通风格偏好"
                }},
                "goals": ["目标1", "目标2"]
            }}
            
            未提及的字段留空。只输出 JSON:
            """,
            input_variables=["conversation"]
        )
    
    def extract(self, messages: List[Dict], user_id: str) -> UserProfile:
        """提取用户画像"""
        conversation = "\n".join([
            f"{m['role']}: {m['content']}" 
            for m in messages
        ])
        
        chain = self.extraction_prompt | self.llm
        result = chain.invoke({"conversation": conversation})
        
        try:
            data = json.loads(result.content)
            return UserProfile(
                user_id=user_id,
                name=data.get("name", ""),
                profession=data.get("profession", ""),
                interests=data.get("interests", []),
                preferences=data.get("preferences", {}),
                goals=data.get("goals", [])
            )
        except json.JSONDecodeError:
            return UserProfile(user_id=user_id)
    
    def update_profile(
        self, 
        existing: UserProfile, 
        new_messages: List[Dict]
    ) -> UserProfile:
        """增量更新用户画像"""
        new_profile = self.extract(new_messages, existing.user_id)
        
        # 合并画像
        if new_profile.name:
            existing.name = new_profile.name
        if new_profile.profession:
            existing.profession = new_profile.profession
        existing.interests = list(set(existing.interests + new_profile.interests))
        existing.preferences.update(new_profile.preferences)
        existing.goals = list(set(existing.goals + new_profile.goals))
        
        return existing
 
# 使用示例
extractor = ProfileExtractor()
 
messages = [
    {"role": "user", "content": "我叫小明,是一名后端开发工程师"},
    {"role": "assistant", "content": "你好小明!有什么可以帮助你的?"},
    {"role": "user", "content": "我对机器学习和数据分析很感兴趣"},
    {"role": "assistant", "content": "很好的方向!"},
    {"role": "user", "content": "我希望能在一年内转型成为 AI 工程师"}
]
 
profile = extractor.extract(messages, "user_123")
print(json.dumps(profile.to_dict(), ensure_ascii=False, indent=2))

五、记忆遗忘机制

5.1 遗忘策略设计

┌─────────────────────────────────────────────────────────────┐
│                    记忆遗忘策略                              │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  1. TTL 过期遗忘                                            │
│     ┌─────────────────────────────────────────────────┐    │
│     │ • 设置记忆的生存时间(Time-To-Live)             │    │
│     │ • 超时后自动删除                                 │    │
│     │ • 适用于:临时信息、时效性内容                   │    │
│     └─────────────────────────────────────────────────┘    │
│                                                             │
│  2. LRU 淘汰遗忘                                            │
│     ┌─────────────────────────────────────────────────┐    │
│     │ • 最近最少使用(Least Recently Used)            │    │
│     │ • 当容量达到上限时淘汰最久未访问的记忆           │    │
│     │ • 适用于:有限容量的记忆池                       │    │
│     └─────────────────────────────────────────────────┘    │
│                                                             │
│  3. 重要性遗忘                                              │
│     ┌─────────────────────────────────────────────────┐    │
│     │ • 根据重要性评分决定是否遗忘                     │    │
│     │ • 低重要性记忆优先遗忘                           │    │
│     │ • 适用于:区分核心信息和次要信息                 │    │
│     └─────────────────────────────────────────────────┘    │
│                                                             │
│  4. 冗余去重遗忘                                            │
│     ┌─────────────────────────────────────────────────┐    │
│     │ • 检测相似或重复的记忆                           │    │
│     │ • 合并或删除冗余内容                             │    │
│     │ • 适用于:重复表达同一信息的记忆                 │    │
│     └─────────────────────────────────────────────────┘    │
│                                                             │
└─────────────────────────────────────────────────────────────┘

5.2 遗忘机制实现

import time
from typing import Dict, List, Optional
from dataclasses import dataclass, field
from datetime import datetime, timedelta
 
@dataclass
class MemoryItem:
    """记忆项"""
    id: str
    content: str
    importance: float = 0.5
    created_at: float = field(default_factory=time.time)
    last_accessed: float = field(default_factory=time.time)
    access_count: int = 0
    ttl: Optional[int] = None  # 秒,None 表示永不过期
    metadata: Dict = field(default_factory=dict)
 
class MemoryForgettingManager:
    """记忆遗忘管理器"""
    
    def __init__(
        self,
        max_items: int = 1000,
        default_ttl: Optional[int] = None,
        min_importance: float = 0.1
    ):
        self.max_items = max_items
        self.default_ttl = default_ttl
        self.min_importance = min_importance
        self.memories: Dict[str, MemoryItem] = {}
    
    def add_memory(
        self, 
        memory_id: str, 
        content: str, 
        importance: float = 0.5,
        ttl: Optional[int] = None,
        metadata: Dict = None
    ):
        """添加记忆"""
        self.memories[memory_id] = MemoryItem(
            id=memory_id,
            content=content,
            importance=importance,
            ttl=ttl or self.default_ttl,
            metadata=metadata or {}
        )
        
        # 检查是否需要淘汰
        if len(self.memories) > self.max_items:
            self._evict_lru()
    
    def access_memory(self, memory_id: str) -> Optional[str]:
        """访问记忆"""
        if memory_id not in self.memories:
            return None
        
        memory = self.memories[memory_id]
        
        # 检查是否过期
        if self._is_expired(memory):
            del self.memories[memory_id]
            return None
        
        # 更新访问信息
        memory.last_accessed = time.time()
        memory.access_count += 1
        
        return memory.content
    
    def cleanup(self) -> int:
        """清理过期和低重要性记忆"""
        removed = 0
        
        # 1. 清理过期记忆
        expired = [
            mid for mid, m in self.memories.items() 
            if self._is_expired(m)
        ]
        for mid in expired:
            del self.memories[mid]
            removed += 1
        
        # 2. 清理低重要性记忆
        low_importance = [
            mid for mid, m in self.memories.items()
            if m.importance < self.min_importance
        ]
        for mid in low_importance:
            del self.memories[mid]
            removed += 1
        
        return removed
    
    def _is_expired(self, memory: MemoryItem) -> bool:
        """检查是否过期"""
        if memory.ttl is None:
            return False
        return time.time() - memory.created_at > memory.ttl
    
    def _evict_lru(self):
        """LRU 淘汰"""
        if not self.memories:
            return
        
        # 找到最久未访问的记忆
        lru_id = min(
            self.memories.keys(),
            key=lambda x: (
                self.memories[x].importance,  # 先按重要性
                self.memories[x].last_accessed  # 再按访问时间
            )
        )
        del self.memories[lru_id]
    
    def get_stats(self) -> Dict:
        """获取统计信息"""
        return {
            "total_memories": len(self.memories),
            "avg_importance": sum(m.importance for m in self.memories.values()) / len(self.memories) if self.memories else 0,
            "avg_access_count": sum(m.access_count for m in self.memories.values()) / len(self.memories) if self.memories else 0
        }
 
# 使用示例
manager = MemoryForgettingManager(max_items=100, default_ttl=3600)  # 1小时TTL
 
# 添加记忆
manager.add_memory("mem_1", "用户偏好中文", importance=0.9, ttl=86400)  # 1天
manager.add_memory("mem_2", "临时任务:提醒开会", importance=0.3, ttl=3600)  # 1小时
manager.add_memory("mem_3", "用户喜欢简洁回答", importance=0.8)
 
# 访问记忆
content = manager.access_memory("mem_1")
print(f"访问记忆: {content}")
 
# 清理
removed = manager.cleanup()
print(f"清理了 {removed} 条记忆")
 
# 统计
stats = manager.get_stats()
print(f"统计: {stats}")

六、完整的记忆管理系统

6.1 系统架构

from typing import List, Dict, Optional, Any
from dataclasses import dataclass
import chromadb
import json
 
@dataclass
class MemoryConfig:
    """记忆配置"""
    max_context_tokens: int = 4000
    max_history_messages: int = 20
    summary_threshold: int = 10
    compression_ratio: float = 0.2
    ttl_default: int = 86400  # 1天
    min_importance: float = 0.3
 
class IntelligentMemorySystem:
    """智能记忆管理系统"""
    
    def __init__(
        self,
        config: MemoryConfig = None,
        llm=None,
        vector_db=None
    ):
        self.config = config or MemoryConfig()
        self.llm = llm or ChatOpenAI(temperature=0)
        
        # 初始化组件
        self.summarizer = ConversationSummarizer(self.llm)
        self.fact_extractor = FactExtractor(self.llm)
        self.forgetting_manager = MemoryForgettingManager(
            max_items=10000,
            default_ttl=self.config.ttl_default,
            min_importance=self.config.min_importance
        )
        
        # 向量存储(长期记忆)
        self.vector_store = vector_db or chromadb.Client().create_collection("memory")
        
        # 短期记忆
        self.short_term_memory: List[Dict] = []
        self.summary: str = ""
    
    def add_message(self, role: str, content: str, metadata: Dict = None):
        """添加消息"""
        message = {
            "role": role,
            "content": content,
            "metadata": metadata or {},
            "timestamp": time.time()
        }
        self.short_term_memory.append(message)
        
        # 检查是否需要压缩
        if len(self.short_term_memory) >= self.config.summary_threshold:
            self._compress_short_term()
        
        # 提取关键事实到长期记忆
        if role == "user":
            self._extract_and_store_facts(content)
    
    def _compress_short_term(self):
        """压缩短期记忆"""
        # 生成摘要
        old_messages = self.short_term_memory[:-self.config.max_history_messages]
        
        if old_messages:
            new_summary = self.summarizer.summarize(old_messages)
            
            if self.summary:
                # 合并新旧摘要
                self.summary = self._merge_summaries(self.summary, new_summary)
            else:
                self.summary = new_summary
            
            # 保留最近的对话
            self.short_term_memory = self.short_term_memory[-self.config.max_history_messages:]
    
    def _merge_summaries(self, old_summary: str, new_summary: str) -> str:
        """合并摘要"""
        merge_prompt = PromptTemplate(
            template="""
            合并以下两个摘要为一个简洁的摘要:
            
            摘要1:{old}
            摘要2:{new}
            
            合并后的摘要:
            """,
            input_variables=["old", "new"]
        )
        
        chain = merge_prompt | self.llm
        result = chain.invoke({"old": old_summary, "new": new_summary})
        return result.content.strip()
    
    def _extract_and_store_facts(self, text: str):
        """提取并存储事实"""
        facts = self.fact_extractor.extract(text)
        
        for fact in facts:
            fact_text = f"{fact.subject} {fact.relation} {fact.object}"
            self.vector_store.add(
                documents=[fact_text],
                metadatas=[{
                    "type": "fact",
                    "subject": fact.subject,
                    "relation": fact.relation,
                    "importance": fact.confidence
                }],
                ids=[f"fact_{hash(fact_text)}"]
            )
    
    def get_context(self, query: str = None) -> str:
        """获取推理上下文"""
        context_parts = []
        
        # 1. 添加摘要
        if self.summary:
            context_parts.append(f"[历史摘要]\n{self.summary}\n")
        
        # 2. 添加近期对话
        if self.short_term_memory:
            context_parts.append("[近期对话]")
            for msg in self.short_term_memory[-10:]:
                context_parts.append(f"{msg['role']}: {msg['content']}")
        
        # 3. 检索相关长期记忆
        if query:
            results = self.vector_store.query(
                query_texts=[query],
                n_results=3
            )
            if results["documents"]:
                context_parts.append("\n[相关记忆]")
                for doc in results["documents"][0]:
                    context_parts.append(f"- {doc}")
        
        return "\n".join(context_parts)
    
    def save_session(self):
        """保存会话到长期记忆"""
        # 将摘要和关键对话存入向量数据库
        if self.summary:
            self.vector_store.add(
                documents=[self.summary],
                metadatas=[{"type": "session_summary", "timestamp": time.time()}],
                ids=[f"session_{int(time.time())}"]
            )
    
    def clear(self):
        """清空短期记忆"""
        self.short_term_memory.clear()
        self.summary = ""

七、面试高频问题

Q1: 记忆压缩会丢失重要信息吗?如何平衡?

答案要点

平衡策略:

1. 分层保留
   • 摘要保留核心意图
   • 原文保留关键对话
   • 向量存储支持检索

2. 重要性筛选
   • 评估每条信息的重要性
   • 高重要性信息特殊处理
   • 低重要性信息可压缩

3. 多副本存储
   • 原文存入向量数据库
   • 摘要用于日常上下文
   • 需要时可检索原文

4. 用户确认
   • 关键决策前确认用户意图
   • 避免因信息丢失导致错误

Q2: 什么时候应该触发记忆压缩?

答案要点

触发时机:

1. 定时触发
   • 每隔 N 轮对话
   • 每隔一段时间

2. 阈值触发
   • Token 数接近上限
   • 消息数超过阈值

3. 事件触发
   • 会话结束
   • 任务完成
   • 用户明确要求

最佳实践:
• 结合多种触发条件
• 异步执行压缩
• 保留压缩前备份

Q3: 如何评估记忆压缩的效果?

答案要点

评估指标:

1. 压缩率
   • 压缩后 Token / 原始 Token
   • 目标:50%-90%

2. 信息保留率
   • 关键信息是否保留
   • 需要人工评估或 LLM 评估

3. 任务影响
   • 压缩后任务完成率
   • 对话质量评分

评估方法:
• 人工抽样检查
• LLM 评估信息完整性
• A/B 测试对比效果

Q4: 记忆遗忘会不会删除用户想要保留的信息?

答案要点

防止误删策略:

1. 重要性标记
   • 让用户或 LLM 标记重要信息
   • 高重要性信息永不删除

2. 用户确认
   • 删除前询问用户
   • 提供"回收站"机制

3. 软删除
   • 标记为删除而非真正删除
   • 可从备份恢复

4. 分级存储
   • 核心信息:永久存储
   • 普通信息:定期清理
   • 临时信息:快速遗忘

Q5: 长上下文模型还需要记忆压缩吗?

答案要点

仍然需要:

1. 成本考量
   • 200K Token 输入成本很高
   • 压缩可显著降低成本

2. 注意力问题
   • "Lost in the Middle" 现象仍存在
   • 精简上下文提升推理质量

3. 无限历史
   • 用户交互历史可能无限
   • 任何窗口都有上限

4. 响应效率
   • 处理长上下文更慢
   • 压缩提升响应速度

最佳实践:
• 压缩 + 长上下文结合
• 核心信息精简输入
• 次要信息需要时检索

八、总结

核心概念回顾

概念定义关键要点
记忆压缩减少记忆占用的技术降低 Token、保留核心
摘要压缩LLM 生成摘要长对话压缩首选
提取压缩提取关键事实结构化存储
记忆遗忘清理过期/低价值记忆TTL、LRU、重要性
信息平衡压缩 vs 信息完整多层存储、重要性分级

一句话总结

记忆压缩通过摘要生成、关键信息提取和遗忘机制,在保留核心信息的同时大幅减少 Token 占用,是解决上下文窗口限制的核心技术。

设计口诀

记忆压缩设计口诀:
上下窗口有限制,压缩摘要来解决
核心信息要保留,冗余细节可丢弃
分层存储多备份,需要时能找回它
遗忘机制定期清,重要信息永远留

最后更新:2026年3月18日