AI Agent 记忆系统工程：四层架构、Mem0/Zep/Letta/LangMem 生产选型与 2026 最新进展

// Agent 四层记忆系统数据模型
interface WorkingMemory {
  sessionId: string;
  messages: ChatMessage[];
  toolResults: ToolResult[];
  reasoningSteps: string[];
  maxEntries: number;  // 通常 10-50
}

interface EpisodicMemory {
  eventId: string;
  sessionId: string;
  timestamp: Date;
  actor: string;        // "user" | "agent" | "tool"
  action: string;
  outcome: "success" | "failure" | "partial";
  metadata: Record<string, unknown>;
  embedding: number[];  // 用于语义检索
}

interface SemanticMemory {
  memoryId: string;
  type: "fact" | "preference" | "relationship";
  namespace: string;    // e.g. "/users/user-123/preferences"
  content: string;
  source: {
    kind: "extracted" | "explicit" | "inferred";
    sessionId: string;
    evidenceEventIds: string[];
  };
  confidence: number;   // 0-1
  writtenAt: Date;
  lastConfirmedAt: Date;
  expiresAt: Date | null;
  supersedes: string | null;  // 记忆版本链
  embedding: number[];
}

interface ProceduralMemory {
  procedureId: string;
  name: string;         // e.g. "deploy-to-production"
  steps: ProcedureStep[];
  successRate: number;
  avgDuration: number;
  lastUsed: Date;
  validityBasis: {
    kind: "file_content" | "config" | "api_contract";
    ref: string;
    checksum: string;
  };
  embedding: number[];
}

# Mem0 记忆系统集成示例
# pip install mem0ai

from mem0 import Memory

# 初始化记忆客户端
config = {
    "llm": {
        "provider": "openai",
        "config": {"model": "gpt-4o-mini", "temperature": 0}
    },
    "vector_store": {
        "provider": "qdrant",
        "config": {
            "host": "localhost",
            "port": 6333,
            "collection_name": "agent_memory"
        }
    },
    "graph_store": {
        "provider": "neo4j",
        "config": {
            "url": "bolt://localhost:7687",
            "username": "neo4j",
            "password": "your-password"
        }
    },
    # 记忆提取配置
    "memory_config": {
        "extraction_prompt": "从对话中提取值得记住的事实、偏好和关系",
        "deduplication": True,       # 自动去重
        "confidence_threshold": 0.7, # 只存储高置信度记忆
        "auto_consolidate": True     # 自动合并相似记忆
    }
}

memory = Memory.from_config(config)

# === 记忆写入：从对话中自动提取 ===
conversation = [
    {"role": "user", "content": "我下周要去上海出差，帮我订浦东的酒店"},
    {"role": "assistant", "content": "好的，我帮你搜索浦东的酒店。你偏好哪个价位？"},
    {"role": "user", "content": "500-800 一晚就行，最好离地铁站近"}
]

# Mem0 自动提取记忆
result = memory.add(
    messages=conversation,
    user_id="user_123",
    metadata={"session": "travel-planning"}
)
# 提取的记忆：
# - "用户计划下周去上海出差"
# - "用户偏好浦东区域的酒店"
# - "用户预算 500-800 元/晚"
# - "用户偏好靠近地铁站的酒店"

# === 记忆检索：为下次对话提供上下文 ===
memories = memory.search(
    query="用户的酒店偏好",
    user_id="user_123",
    limit=5
)

for mem in memories:
    print(f"[{mem['metadata']['category']}] {mem['memory']}")
    # 输出：
    # [preference] 用户偏好靠近地铁站的酒店
    # [preference] 用户预算 500-800 元/晚
    # [plan] 用户计划去上海出差

# === 记忆更新：处理事实变更 ===
memory.add(
    messages=[{"role": "user", "content": "我搬家了，现在住在朝阳区"}],
    user_id="user_123"
)
# Mem0 自动检测到与旧记忆"用户住在海淀区"冲突
# 创建新版本，保留旧版本作为历史记录

# Zep 时间感知记忆集成
# pip install zep-python

from zep_cloud.client import Zep
from zep_cloud.types import Message

client = Zep(api_key="your-api-key")

# 创建会话并添加消息
session = client.memory.create(session_id="session-001")

messages = [
    Message(role_type="user", role="用户", content="我在用 AWS 的 EC2"),
    Message(role_type="assistant", role="助手", content="了解，你使用 EC2 的哪个区域？"),
    Message(role_type="user", role="用户", content="us-east-1，主要是 t3.medium"),
]

client.memory.add(session_id="session-001", messages=messages)

# === 时间范围查询：Zep 的独特能力 ===
# 查询"过去 7 天用户提到的所有基础设施变更"
from datetime import datetime, timedelta

history = client.memory.search(
    session_id="session-001",
    text="基础设施变更",
    search_type="similarity",
    # Zep 特有：时间过滤
    start_date=(datetime.now() - timedelta(days=7)).isoformat(),
)

# === 事实版本追踪 ===
# 当用户说"我们迁移到 GCP 了"
client.memory.add(
    session_id="session-001",
    messages=[Message(role_type="user", role="用户",
                      content="我们已经从 AWS 迁移到 GCP 了")]
)

# Zep 保留完整的变更历史：
# v1: "用户使用 AWS EC2 us-east-1" (2026-06-01)
# v2: "用户迁移到 GCP" (2026-06-15) supersedes v1

import math
import time
from dataclasses import dataclass, field
from typing import Optional
from enum import Enum

class MemoryStatus(Enum):
    ACTIVE = "active"
    STALE = "stale"
    PENDING_CLEANUP = "pending_cleanup"
    ARCHIVED = "archived"

@dataclass
class MemoryEntry:
    id: str
    content: str
    category: str  # "fact" | "preference" | "procedure"
    timestamp: float
    importance: float = 0.5  # 0-1, 由 LLM 评估
    access_count: int = 0
    last_accessed: float = 0
    confidence: float = 0.8
    status: MemoryStatus = MemoryStatus.ACTIVE
    supersedes: Optional[str] = None
    tags: list = field(default_factory=list)

    @property
    def age_days(self) -> float:
        return (time.time() - self.timestamp) / 86400

    def calculate_weight(self) -> float:
        """综合权重 = 基础权重 × 时间衰减 × 使用增强 × 重要性"""
        now = time.time()

        # 时间衰减：重要性越高，衰减越慢
        # 高重要性记忆 λ=0.005（半衰期 ~138 天）
        # 普通记忆 λ=0.02（半衰期 ~35 天）
        # 低重要性 λ=0.05（半衰期 ~14 天）
        lam = 0.05 - (self.importance * 0.045)
        time_decay = math.exp(-lam * self.age_days)

        # 使用增强：频繁访问的记忆获得权重提升
        usage_boost = math.log(1 + self.access_count) * 0.3

        # 置信度因子
        confidence_factor = self.confidence

        return self.importance * time_decay * (1 + usage_boost) * confidence_factor

class MemoryStore:
    def __init__(self, cleanup_threshold: float = 0.1):
        self.memories: dict[str, MemoryEntry] = {}
        self.cleanup_threshold = cleanup_threshold

    def add(self, entry: MemoryEntry) -> str:
        self.memories[entry.id] = entry
        return entry.id

    def search(self, query_embedding: list, top_k: int = 5) -> list:
        """检索时按综合权重排序，而非纯相似度"""
        results = []
        for mem in self.memories.values():
            if mem.status == MemoryStatus.ARCHIVED:
                continue
            # 实际使用中这里会有向量相似度计算
            weight = mem.calculate_weight()
            results.append((mem, weight))

        # 按权重降序排列
        results.sort(key=lambda x: x[1], reverse=True)
        return [r[0] for r in results[:top_k]]

    def run_maintenance(self):
        """定期维护：衰减检测 + 过时标记 + 清理"""
        now = time.time()
        for mem in self.memories.values():
            weight = mem.calculate_weight()

            # 权重低于阈值 → 标记为待清理
            if weight < self.cleanup_threshold:
                mem.status = MemoryStatus.PENDING_CLEANUP

            # 30 天未访问 → 标记为过时
            days_since_access = (now - mem.last_accessed) / 86400
            if days_since_access > 30 and mem.status == MemoryStatus.ACTIVE:
                mem.status = MemoryStatus.STALE
                mem.confidence *= 0.95  # 置信度衰减

            # 过时记忆 60 天后归档
            if mem.status == MemoryStatus.STALE and days_since_access > 90:
                mem.status = MemoryStatus.ARCHIVED

    def consolidate(self, similar_pairs: list[tuple[str, str]]):
        """语义合并：将高度相似的记忆合并为一条"""
        for id_a, id_b in similar_pairs:
            mem_a = self.memories.get(id_a)
            mem_b = self.memories.get(id_b)
            if not mem_a or not mem_b:
                continue

            # 保留更新的、重要性更高的记忆
            if mem_b.timestamp > mem_a.timestamp and mem_b.importance >= mem_a.importance:
                mem_b.supersedes = id_a
                mem_b.access_count += mem_a.access_count
                mem_a.status = MemoryStatus.ARCHIVED
            else:
                mem_a.supersedes = id_b
                mem_a.access_count += mem_b.access_count
                mem_b.status = MemoryStatus.ARCHIVED

from dataclasses import dataclass
from enum import Enum
from typing import Optional
import hashlib

class SecurityLevel(Enum):
    PUBLIC = 1       # 通用知识
    INTERNAL = 2     # 用户偏好
    CONFIDENTIAL = 3 # 支付/身份
    RESTRICTED = 4   # 医疗/法律

class MemorySource(Enum):
    EXPLICIT = "explicit"    # 用户明确声明
    INFERRED = "inferred"    # 从对话推断
    EXTRACTED = "extracted"  # 从文档提取

@dataclass
class SecurityGate:
    """CAMS 记忆安全门控"""

    # 敏感字段模式（匹配这些模式的记忆需要额外验证）
    SENSITIVE_PATTERNS = [
        "email", "password", "phone", "payment",
        "credit_card", "ssn", "bank_account"
    ]

    # 安全级别映射
    LEVEL_MAP = {
        "preference": SecurityLevel.INTERNAL,
        "fact": SecurityLevel.INTERNAL,
        "identity": SecurityLevel.CONFIDENTIAL,
        "payment": SecurityLevel.CONFIDENTIAL,
        "medical": SecurityLevel.RESTRICTED,
        "legal": SecurityLevel.RESTRICTED,
    }

    def validate_write(self, memory_content: str, source: MemorySource,
                       category: str) -> tuple[bool, str]:
        """验证记忆写入是否安全"""

        # 检查 1：敏感字段检测
        is_sensitive = any(p in memory_content.lower()
                          for p in self.SENSITIVE_PATTERNS)

        if is_sensitive and source != MemorySource.EXPLICIT:
            return False, "敏感信息需要用户显式确认"

        # 检查 2：安全级别检查
        level = self.LEVEL_MAP.get(category, SecurityLevel.INTERNAL)
        if level >= SecurityLevel.CONFIDENTIAL:
            if source != MemorySource.EXPLICIT:
                return False, f"{category} 类记忆需要显式声明"

        # 检查 3：推断记忆的置信度上限
        if source == MemorySource.INFERRED:
            max_confidence = 0.8  # 推断记忆置信度不超过 0.8
        else:
            max_confidence = 1.0

        return True, f"安全级别: {level.name}, 最大置信度: {max_confidence}"

    def detect_conflict(self, new_memory: str, existing_memories: list,
                        threshold: float = 0.85) -> Optional[dict]:
        """检测新记忆与现有记忆的冲突"""
        import difflib

        for existing in existing_memories:
            similarity = difflib.SequenceMatcher(
                None, new_memory.lower(), existing.lower()
            ).ratio()

            if similarity > threshold:
                # 高度相似但不同 → 可能是更新
                return {
                    "type": "potential_update",
                    "existing": existing,
                    "new": new_memory,
                    "similarity": similarity,
                    "action": "require_confirmation"
                }

        return None