AI Agent 记忆系统爆发：Claude-Mem 单周 14K 星 + MemPalace 22K 星——记忆正在成为 Agent 刚需基础设施

# Claude-Mem 风格的核心实现：捕获 → 压缩 → 检索
import json
import hashlib
from typing import List, Dict, Optional
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum

class MemoryType(Enum):
    """记忆类型枚举"""
    CODE_CHANGE = "code_change"      # 代码变更
    DECISION = "decision"             # 架构决策
    BUG_FIX = "bug_fix"              # Bug 修复
    PATTERN = "pattern"              # 设计模式
    CONTEXT = "context"              # 项目上下文

@dataclass
class MemoryEntry:
    """单条记忆条目"""
    id: str
    content: str                      # 压缩后的记忆内容
    memory_type: MemoryType           # 记忆类型
    source_session: str              # 来源会话 ID
    file_paths: List[str]            # 相关文件
    timestamp: datetime
    relevance_score: float = 1.0     # 相关性分数（会随时间衰减）
    tags: List[str] = field(default_factory=list)
    
    def decay(self, days: float, decay_rate: float = 0.95):
        """时间衰减：越旧的记忆相关性越低"""
        self.relevance_score *= (decay_rate ** days)
        return self

class AgentMemorySystem:
    """Agent 记忆系统核心"""
    
    def __init__(self, max_entries: int = 1000):
        self.memories: Dict[str, MemoryEntry] = {}
        self.max_entries = max_entries
        self.session_counter = 0
    
    def capture_session(self, operations: List[Dict]) -> MemoryEntry:
        """捕获并压缩一次会话的操作历史"""
        self.session_counter += 1
        session_id = f"session-{self.session_counter:04d}"
        
        # 模拟 AI 压缩：从原始操作中提取关键信息
        compressed = self._compress_operations(operations)
        
        entry = MemoryEntry(
            id=f"mem-{hashlib.md5(session_id.encode()).hexdigest()[:8]}",
            content=compressed["summary"],
            memory_type=self._classify(compressed),
            source_session=session_id,
            file_paths=compressed["files"],
            timestamp=datetime.now(),
            tags=compressed["tags"]
        )
        self.memories[entry.id] = entry
        return entry
    
    def _compress_operations(self, operations: List[Dict]) -> Dict:
        """压缩操作历史（模拟 AI 压缩逻辑）"""
        files_modified = set()
        patterns = []
        
        for op in operations:
            if op.get("type") == "file_write":
                files_modified.add(op["path"])
            if op.get("type") == "pattern_detected":
                patterns.append(op["pattern"])
        
        # 生成结构化摘要
        return {
            "summary": f"Modified {len(files_modified)} files: {', '.join(files_modified)}",
            "files": list(files_modified),
            "patterns": patterns,
            "tags": list(set(p.split("_")[0] for p in patterns))
        }
    
    def _classify(self, compressed: Dict) -> MemoryType:
        """分类记忆类型"""
        if "fix" in str(compressed.get("patterns", [])).lower():
            return MemoryType.BUG_FIX
        if "pattern" in str(compressed.get("patterns", [])).lower():
            return MemoryType.PATTERN
        return MemoryType.CODE_CHANGE
    
    def retrieve(self, query_files: List[str], 
                 query_type: Optional[MemoryType] = None,
                 top_k: int = 5) -> List[MemoryEntry]:
        """检索相关记忆"""
        scored = []
        for entry in self.memories.values():
            score = entry.relevance_score
            
            # 文件匹配
            file_overlap = set(query_files) & set(entry.file_paths)
            score += len(file_overlap) * 0.3
            
            # 类型匹配
            if query_type and entry.memory_type == query_type:
                score += 0.5
            
            scored.append((score, entry))
        
        scored.sort(reverse=True)
        return [entry for _, entry in scored[:top_k]]

# ===== 使用示例 =====
if __name__ == "__main__":
    memory = AgentMemorySystem()
    
    # 模拟一次会话的操作捕获
    ops = [
        {"type": "file_write", "path": "src/auth/login.py"},
        {"type": "file_write", "path": "src/auth/session.py"},
        {"type": "pattern_detected", "pattern": "auth_retry_pattern"},
    ]
    entry = memory.capture_session(ops)
    print(f"记忆已保存: {entry.id} - {entry.content}")
    
    # 新会话中检索相关记忆
    relevant = memory.retrieve(
        query_files=["src/auth/login.py"],
        query_type=MemoryType.CODE_CHANGE
    )
    print(f"检索到 {len(relevant)} 条相关记忆")

# MemPalace 风格的记忆宫殿实现
import json
from typing import Dict, List, Optional, Tuple
from dataclasses import dataclass, field
from enum import Enum
import math

class RoomType(Enum):
    """记忆宫殿中的房间类型"""
    CONFIG = "config"           # 项目配置
    STANDARDS = "standards"     # 代码规范
    BUGS = "bugs"              # 错误知识
    LESSONS = "lessons"        # 经验教训
    PATTERNS = "patterns"      # 设计模式
    CONTEXT = "context"        # 项目上下文

@dataclass
class AnchorKey:
    """AAAK 锚点键"""
    semantic_hash: str          # 语义哈希（抽象后的标识）
    weight: float               # 重要性权重
    room: RoomType              # 所属房间
    connections: List[str] = field(default_factory=list)  # 走廊连接
    
    def token_size(self) -> int:
        """估算 token 大小"""
        return max(1, int(math.log2(len(self.semantic_hash)) + 4))

@dataclass
class MemoryPalaceRoom:
    """记忆宫殿中的一个房间"""
    name: str
    room_type: RoomType
    memories: Dict[str, str] = field(default_factory=dict)
    anchors: Dict[str, AnchorKey] = field(default_factory=dict)
    corridors: Dict[str, List[str]] = field(default_factory=dict)  # 走廊: 目标房间列表
    
    def add_memory(self, key: str, content: str, 
                   memory_type: str = "general"):
        """添加记忆并创建锚点"""
        self.memories[key] = content
        self.anchors[key] = AnchorKey(
            semantic_hash=self._hash_semantic(content),
            weight=self._compute_weight(content),
            room=self.room_type
        )
    
    def _hash_semantic(self, content: str) -> str:
        """语义哈希：提取内容的抽象表示"""
        # 模拟语义哈希（实际使用 embedding 模型）
        words = content.lower().split()
        keywords = [w for w in words if len(w) > 4]
        return ":".join(sorted(set(keywords))[:5])
    
    def _compute_weight(self, content: str) -> float:
        """计算记忆重要性权重"""
        # 越长越重要，但边际递减
        return min(1.0, math.log(len(content) + 1) / 10)

class MemoryPalace:
    """记忆宫殿主系统"""
    
    def __init__(self):
        self.rooms: Dict[RoomType, MemoryPalaceRoom] = {}
        self._init_rooms()
        self._build_corridors()
    
    def _init_rooms(self):
        """初始化所有房间"""
        for room_type in RoomType:
            self.rooms[room_type] = MemoryPalaceRoom(
                name=room_type.value.capitalize(),
                room_type=room_type
            )
    
    def _build_corridors(self):
        """建立房间之间的走廊连接"""
        corridors = [
            (RoomType.STANDARDS, RoomType.BUGS),
            (RoomType.BUGS, RoomType.LESSONS),
            (RoomType.CONFIG, RoomType.STANDARDS),
            (RoomType.PATTERNS, RoomType.STANDARDS),
            (RoomType.CONTEXT, RoomType.CONFIG),
        ]
        for src, dst in corridors:
            self.rooms[src].corridors[dst.value] = [dst.value]
    
    def store(self, room_type: RoomType, key: str, content: str):
        """存储记忆到指定房间"""
        self.rooms[room_type].add_memory(key, content)
        print(f"📦 记忆已存入 [{room_type.value}] 房间: {key}")
    
    def retrieve(self, query: str, 
                 room_types: Optional[List[RoomType]] = None,
                 top_k: int = 5) -> List[Tuple[str, str, float]]:
        """跨房间检索记忆"""
        target_rooms = room_types or list(RoomType)
        results = []
        
        query_hash = self._query_hash(query)
        
        for room_type in target_rooms:
            room = self.rooms[room_type]
            for key, anchor in room.anchors.items():
                # 计算语义相似度（简化版）
                similarity = self._semantic_similarity(query_hash, anchor.semantic_hash)
                score = similarity * anchor.weight
                results.append((key, room.memories[key], score))
        
        # 按分数排序
        results.sort(key=lambda x: x[2], reverse=True)
        return results[:top_k]
    
    def _query_hash(self, query: str) -> str:
        """查询哈希"""
        words = query.lower().split()
        keywords = [w for w in words if len(w) > 4]
        return ":".join(sorted(set(keywords))[:5])
    
    def _semantic_similarity(self, q_hash: str, a_hash: str) -> float:
        """语义相似度计算（简化版）"""
        q_words = set(q_hash.split(":"))
        a_words = set(a_hash.split(":"))
        if not q_words or not a_words:
            return 0.0
        return len(q_words & a_words) / len(q_words | a_words)
    
    def get_token_budget(self) -> int:
        """计算当前记忆系统的总 token 开销"""
        total = 0
        for room in self.rooms.values():
            for anchor in room.anchors.values():
                total += anchor.token_size()
        return total

# ===== 使用示例 =====
if __name__ == "__main__":
    palace = MemoryPalace()
    
    # 存入不同类型的记忆
    palace.store(RoomType.CONFIG, "db_config", 
                "PostgreSQL 15, 连接池大小 20, 超时 30s")
    palace.store(RoomType.STANDARDS, "naming",
                "使用 snake_case 命名函数，PascalCase 命名类")
    palace.store(RoomType.BUGS, "auth_timeout",
                "OAuth token 过期时间设置为 3600s，需要实现自动刷新")
    palace.store(RoomType.LESSONS, "migration",
                "数据库迁移必须在低峰期执行，先备份再迁移")
    palace.store(RoomType.PATTERNS, "repo_pattern",
                "使用 Repository Pattern 隔离数据访问层")
    
    # 跨房间检索
    results = palace.retrieve("database configuration timeout")
    print(f"\n检索结果（{len(results)} 条）：")
    for key, content, score in results:
        print(f"  [{score:.2f}] {key}: {content}")
    
    print(f"\n总 Token 开销: {palace.get_token_budget()} tokens")

# 混合型 Agent 记忆系统：短期 + 长期 + 智能桥接
from typing import List, Dict, Optional
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from enum import Enum
import json

class MemoryTier(Enum):
    SHORT_TERM = "short_term"    # 短期记忆（自动压缩）
    LONG_TERM = "long_term"      # 长期记忆（记忆宫殿）
    EPISODIC = "episodic"        # 情景记忆（关键事件）

@dataclass
class HybridMemoryEntry:
    """混合记忆条目"""
    id: str
    tier: MemoryTier
    content: str
    metadata: Dict = field(default_factory=dict)
    created_at: datetime = field(default_factory=datetime.now)
    access_count: int = 0
    last_accessed: Optional[datetime] = None
    
    def record_access(self):
        self.access_count += 1
        self.last_accessed = datetime.now()

class HybridMemorySystem:
    """混合型记忆系统"""
    
    def __init__(self, 
                 short_term_window: int = 5,   # 保留最近 N 次会话
                 promotion_threshold: int = 3,  # 访问 N 次后升级
                 decay_days: float = 30):       # 30 天衰减
        self.short_term: List[HybridMemoryEntry] = []
        self.long_term: Dict[str, HybridMemoryEntry] = {}
        self.episodic: List[HybridMemoryEntry] = []
        
        self.short_term_window = short_term_window
        self.promotion_threshold = promotion_threshold
        self.decay_days = decay_days
    
    def store(self, content: str, 
              tier: MemoryTier = MemoryTier.SHORT_TERM,
              is_episodic: bool = False,
              **metadata) -> HybridMemoryEntry:
        """存储记忆"""
        entry = HybridMemoryEntry(
            id=f"mem-{len(self.short_term) + len(self.long_term):04d}",
            tier=tier,
            content=content,
            metadata=metadata
        )
        
        if is_episodic:
            self.episodic.append(entry)
        elif tier == MemoryTier.SHORT_TERM:
            self.short_term.append(entry)
            # 维护短期记忆窗口
            if len(self.short_term) > self.short_term_window:
                oldest = self.short_term.pop(0)
                # 频繁访问的短期记忆升级为长期记忆
                if oldest.access_count >= self.promotion_threshold:
                    self._promote(oldest)
        else:
            self.long_term[entry.id] = entry
        
        return entry
    
    def _promote(self, entry: HybridMemoryEntry):
        """将短期记忆升级为长期记忆"""
        entry.tier = MemoryTier.LONG_TERM
        self.long_term[entry.id] = entry
        print(f"⬆️ 记忆 {entry.id} 已从短期升级为长期")
    
    def retrieve(self, query: str, 
                 include_short: bool = True,
                 include_long: bool = True,
                 include_episodic: bool = True) -> List[HybridMemoryEntry]:
        """混合检索"""
        results = []
        
        # 短期记忆（按时间倒序）
        if include_short:
            results.extend(reversed(self.short_term))
        
        # 长期记忆（按访问频率排序）
        if include_long:
            long_memories = sorted(
                self.long_term.values(),
                key=lambda x: x.access_count,
                reverse=True
            )
            results.extend(long_memories)
        
        # 情景记忆（最近的关键事件）
        if include_episodic:
            results.extend(reversed(self.episodic[-5:]))
        
        # 记录访问
        for entry in results:
            entry.record_access()
        
        return results
    
    def get_context_summary(self) -> str:
        """生成当前上下文摘要（用于注入 Agent）"""
        parts = []
        
        # 短期记忆摘要
        if self.short_term:
            recent = self.short_term[-3:]
            parts.append("📌 最近会话：")
            for entry in recent:
                parts.append(f"  • {entry.content[:80]}...")
        
        # 长期记忆摘要（高频访问）
        frequent = sorted(
            self.long_term.values(),
            key=lambda x: x.access_count,
            reverse=True
        )[:3]
        if frequent:
            parts.append("\n📚 常用知识：")
            for entry in frequent:
                parts.append(f"  • [{entry.access_count}x] {entry.content[:60]}...")
        
        return "\n".join(parts)
    
    def stats(self) -> Dict:
        """记忆系统统计"""
        return {
            "short_term_count": len(self.short_term),
            "long_term_count": len(self.long_term),
            "episodic_count": len(self.episodic),
            "total_entries": len(self.short_term) + len(self.long_term) + len(self.episodic),
            "total_token_estimate": self._estimate_tokens()
        }
    
    def _estimate_tokens(self) -> int:
        """估算总 token 数"""
        total = 0
        for entries in [self.short_term, list(self.long_term.values()), self.episodic]:
            for entry in entries:
                # 粗略估算：每 4 个字符约 1 个 token
                total += len(entry.content) // 4
        return total

# ===== 使用示例：模拟一个 Agent 的工作日 =====
if __name__ == "__main__":
    memory = HybridMemorySystem(short_term_window=3)
    
    # 上午：开始新项目
    memory.store("项目初始化：使用 FastAPI + PostgreSQL", 
                 is_episodic=True)
    memory.store("数据库配置：PostgreSQL 15, 连接池 20",
                 tier=MemoryTier.LONG_TERM)
    
    # 中午：遇到问题
    memory.store("Bug：OAuth token 过期未处理，需要实现自动刷新",
                 is_episodic=True)
    memory.store("修复方案：添加 refresh_token 中间件",
                 tier=MemoryTier.SHORT_TERM)
    
    # 下午：继续开发
    memory.store("实现用户认证模块", tier=MemoryTier.SHORT_TERM)
    memory.store("使用 Repository Pattern 隔离数据层",
                 tier=MemoryTier.LONG_TERM)
    
    # 第二天：新会话开始
    print("=== 新会话开始，加载记忆 ===")
    context = memory.get_context_summary()
    print(context)
    
    # 检索相关信息
    relevant = memory.retrieve("authentication database")
    print(f"\n=== 检索到 {len(relevant)} 条相关记忆 ===")
    for entry in relevant:
        print(f"  [{entry.tier.value}] {entry.content[:60]}...")
    
    print(f"\n=== 记忆系统统计 ===")
    for k, v in memory.stats().items():
        print(f"  {k}: {v}")