OpenClaw记忆架构：从短期上下文到长期记忆的工程实现

┌─────────────────────────────────────────────┐
│ Layer 4: 损失无关回忆 (Lossless Recall)      │  ← LCM压缩与精确回溯
├─────────────────────────────────────────────┤
│ Layer 3: 长期记忆 (Long-term Memory)         │  ← MEMORY.md 策略性记忆
├─────────────────────────────────────────────┤
│ Layer 2: 每日记忆 (Daily Memory)             │  ← 按天组织的原始记录
├─────────────────────────────────────────────┤
│ Layer 1: 工作记忆 (Working Memory)           │  ← 当前会话上下文窗口
└─────────────────────────────────────────────┘

class WorkingMemory:
    """管理当前会话的上下文窗口"""
    
    def __init__(self, max_tokens: int = 128000):
        self.max_tokens = max_tokens
        self.messages: list[Message] = []
        self.token_counter = TokenCounter()
    
    def add(self, message: Message):
        """添加消息，超出限制时触发压缩"""
        self.messages.append(message)
        
        if self.token_counter.count(self.messages) > self.max_tokens * 0.85:
            self._compact()
    
    def _compact(self):
        """上下文压缩策略"""
        # 保留最近N轮完整对话
        recent = self.messages[-10:]
        
        # 将较早的消息压缩为摘要
        older = self.messages[:-10]
        summary = self._summarize(older)
        
        self.messages = [summary] + recent

# memory/2026-07-29.md

## 今日事件

### 10:30 - 项目讨论
- 用户要求设计一个RAG系统架构
- 讨论了向量数据库选型：Qdrant vs Milvus
- 决定使用Qdrant，原因是部署简单且性能足够
- 用户偏好：重视部署简洁性高于极致性能

### 14:00 - 代码审查
- 审查了同事提交的Python代码
- 发现3个潜在问题：SQL注入风险、N+1查询、缺失错误处理
- 已在PR中标注

### 16:30 - 学习记录
- 了解了vLLM的FP8-KV插件
- 关键发现：KV Cache量化可减少50%显存，质量几乎无损
- 后续需要测试在我们的场景下是否同样有效

class DailyMemory:
    """每日记忆管理"""
    
    def __init__(self, memory_dir: str):
        self.memory_dir = Path(memory_dir)
    
    def append(self, event: MemoryEvent):
        """追加事件到当天记忆"""
        today = datetime.now().strftime("%Y-%m-%d")
        path = self.memory_dir / f"{today}.md"
        
        timestamp = datetime.now().strftime("%H:%M")
        entry = f"\n### {timestamp} - {event.title}\n{event.content}\n"
        
        with open(path, "a", encoding="utf-8") as f:
            f.write(entry)
    
    def read_day(self, date: str) -> str:
        """读取指定日期的记忆"""
        path = self.memory_dir / f"{date}.md"
        if path.exists():
            return path.read_text(encoding="utf-8")
        return ""
    
    def search(self, query: str, days: int = 7) -> list[MemoryEntry]:
        """搜索最近N天的记忆"""
        results = []
        for i in range(days):
            date = (datetime.now() - timedelta(days=i)).strftime("%Y-%m-%d")
            content = self.read_day(date)
            if query.lower() in content.lower():
                results.append(MemoryEntry(date=date, content=content))
        return results

# MEMORY.md

## 用户画像
- 技术栈：Python, TypeScript, Go
- 偏好：优先选择开源方案，重视代码可维护性
- 工作风格：先调研再决策，喜欢看到对比数据
- 时区：Asia/Shanghai (UTC+8)

## 项目记忆
### RAG系统项目
- 架构决策：Qdrant + BGE-M3 + GPT-5
- 部署环境：8×A100-80G, Kubernetes
- 关键指标：P99延迟 < 1s, 吞吐量 > 50 QPS
- 已知问题：长查询时检索精度下降，待优化

## 经验教训
- vLLM的gpu_memory_utilization设置不要超过0.92，否则容易OOM
- LangChain的RetrievalQA在生产环境性能不佳，建议直接用原生SDK
- Qdrant的payload过滤比预过滤快3-5倍

class LongTermMemory:
    """长期记忆：策略性提炼与维护"""
    
    def __init__(self, memory_file: str, daily_memory: DailyMemory):
        self.memory_file = Path(memory_file)
        self.daily_memory = daily_memory
        self.content = self._load()
    
    def _load(self) -> str:
        if self.memory_file.exists():
            return self.memory_file.read_text(encoding="utf-8")
        return "# MEMORY.md\n\n"
    
    def consolidate(self, days: int = 7):
        """从每日记忆中提炼长期记忆"""
        recent_daily = []
        for i in range(days):
            date = (datetime.now() - timedelta(days=i)).strftime("%Y-%m-%d")
            content = self.daily_memory.read_day(date)
            if content:
                recent_daily.append(content)
        
        if not recent_daily:
            return
        
        # 用LLM进行记忆提炼
        prompt = f"""以下是过去{days}天的每日记忆记录。请提取值得长期记住的信息：
1. 用户偏好和习惯
2. 重要的架构/技术决策
3. 经验教训
4. 待办事项

每日记忆：
{chr(10).join(recent_daily)}

当前长期记忆：
{self.content}

请输出更新后的完整长期记忆。保留已有内容中仍然相关的部分，添加新发现，删除过时信息。"""
        
        updated = self.llm.complete(prompt)
        self.content = updated
        self.memory_file.write_text(updated, encoding="utf-8")

对话消息流:
msg1 → msg2 → msg3 → [压缩] → sum_1 → msg4 → msg5 → [压缩] → sum_2 → msg6
                    ↑                                    ↑
                    └── 可展开恢复原始消息 ──────────────┘

class LosslessRecall:
    """损失无关回忆：基于摘要DAG的精确回溯"""
    
    def __init__(self, storage):
        self.storage = storage  # 持久化存储
    
    def compress(self, messages: list[Message]) -> str:
        """将消息序列压缩为摘要"""
        summary_id = f"sum_{uuid.uuid4().hex[:12]}"
        
        # 生成摘要
        summary_text = self._generate_summary(messages)
        
        # 存储摘要及其子节点
        self.storage.save_summary(
            id=summary_id,
            content=summary_text,
            children=[m.id for m in messages],
            metadata={
                "timestamp": datetime.now().isoformat(),
                "message_count": len(messages),
                "token_count": sum(m.token_count for m in messages),
            }
        )
        
        return summary_id
    
    def expand(self, summary_id: str, query: str = None) -> str:
        """展开摘要，恢复底层细节"""
        summary = self.storage.get_summary(summary_id)
        
        # 递归展开子节点
        children = []
        for child_id in summary.children:
            if child_id.startswith("sum_"):
                # 子节点也是摘要，递归展开
                children.append(self.expand(child_id, query))
            else:
                # 叶子节点是原始消息
                msg = self.storage.get_message(child_id)
                children.append(msg.content)
        
        return "\n".join(children)
    
    def grep(self, pattern: str, scope: str = "both") -> list[Match]:
        """在记忆中搜索模式"""
        matches = []
        
        # 搜索原始消息
        if scope in ("messages", "both"):
            matches.extend(self.storage.search_messages(pattern))
        
        # 搜索摘要
        if scope in ("summaries", "both"):
            matches.extend(self.storage.search_summaries(pattern))
        
        return matches

class MemoryRetriever:
    """多路记忆检索"""
    
    def __init__(self, working: WorkingMemory, daily: DailyMemory, 
                 long_term: LongTermMemory, lcm: LosslessRecall):
        self.working = working
        self.daily = daily
        self.long_term = long_term
        self.lcm = lcm
    
    def recall(self, query: str, context: dict) -> RecallResult:
        """根据查询回忆相关信息"""
        results = RecallResult(query=query)
        
        # 1. 工作记忆：当前上下文中搜索
        results.working = self._search_working(query)
        
        # 2. 每日记忆：最近7天关键词搜索
        results.daily = self.daily.search(query, days=7)
        
        # 3. 长期记忆：语义搜索
        results.long_term = self._semantic_search(
            self.long_term.content, query
        )
        
        # 4. LCM：历史对话精确回溯
        results.lcm = self.lcm.grep(query, scope="both")
        
        # 合并去重并排序
        return self._merge_and_rank(results, context)
    
    def _merge_and_rank(self, results: RecallResult, ctx: dict) -> RecallResult:
        """合并多路结果并按相关性排序"""
        all_items = []
        
        # 工作记忆权重最高（当前上下文最相关）
        for item in results.working:
            all_items.append((item, 1.0, "working"))
        
        # 长期记忆次之（策略性重要信息）
        for item in results.long_term:
            all_items.append((item, 0.8, "long_term"))
        
        # LCM精确匹配
        for item in results.lcm:
            all_items.append((item, 0.7, "lcm"))
        
        # 每日记忆（原始记录，可能有噪音）
        for item in results.daily:
            all_items.append((item, 0.5, "daily"))
        
        # 按权重排序
        all_items.sort(key=lambda x: x[1], reverse=True)
        results.ranked = all_items
        return results

class MemorySecurity:
    """记忆安全策略"""
    
    # 敏感信息检测模式
    SENSITIVE_PATTERNS = [
        (r'\b\d{16,19}\b', '信用卡号'),
        (r'\b\d{11}\b', '手机号'),
        (r'[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}', '邮箱'),
        (r'password\s*[:=]\s*\S+', '密码'),
    ]
    
    def sanitize(self, content: str) -> str:
        """脱敏处理"""
        for pattern, label in self.SENSITIVE_PATTERNS:
            content = re.sub(pattern, f'[{label}已脱敏]', content)
        return content
    
    def access_control(self, memory_type: str, context: dict) -> bool:
        """访问控制：不同场景下的记忆访问权限"""
        # 在群聊中不加载长期记忆（防止泄露个人信息）
        if context.get('channel') == 'group':
            return memory_type in ('working', 'daily')
        
        # 在共享环境中限制访问
        if context.get('shared', False):
            return memory_type == 'working'
        
        # 私有会话中允许所有记忆
        return True