Agent 记忆系统设计实战：六层持久化策略、遗忘机制与框架选型指南

interface MemoryBlock {
  id: string;                    // 唯一标识
  type: "fact" | "preference" | "task" | "observation";
  content: string;               // 记忆内容
  importance: number;            // 重要性评分 (0-1)
  createdAt: Date;               // 创建时间
  lastAccessedAt: Date;          // 最后访问时间
  accessCount: number;           // 访问次数
  expiresAt?: Date;              // 可选过期时间
  tags: string[];                // 标签（用于检索）
  source: string;                // 来源（哪轮对话产生的）
}

# 全文对话日志持久化
import sqlite3
import json
from datetime import datetime

class ConversationLogger:
    def __init__(self, db_path: str = "conversations.db"):
        self.conn = sqlite3.connect(db_path)
        self._init_tables()

    def _init_tables(self):
        self.conn.execute("""
            CREATE TABLE IF NOT EXISTS messages (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                session_id TEXT NOT NULL,
                role TEXT NOT NULL,
                content TEXT NOT NULL,
                timestamp DATETIME DEFAULT CURRENT_TIMESTAMP,
                metadata JSON
            )
        """)
        self.conn.execute("""
            CREATE INDEX IF NOT EXISTS idx_session
            ON messages(session_id, timestamp)
        """)
        self.conn.commit()

    def log(self, session_id: str, role: str, content: str, metadata: dict = None):
        self.conn.execute(
            "INSERT INTO messages (session_id, role, content, metadata) VALUES (?, ?, ?, ?)",
            (session_id, role, content, json.dumps(metadata or {}))
        )
        self.conn.commit()

    def get_recent(self, session_id: str, limit: int = 50) -> list:
        cursor = self.conn.execute(
            "SELECT role, content, timestamp FROM messages WHERE session_id = ? ORDER BY timestamp DESC LIMIT ?",
            (session_id, limit)
        )
        return [{"role": r, "content": c, "timestamp": t} for r, c, t in cursor.fetchall()]

async def compress_conversation(messages: list, max_tokens: int = 500) -> str:
    """将对话历史压缩为摘要"""
    prompt = f"""请将以下对话压缩为不超过 {max_tokens} token 的摘要。
保留关键事实、决策和用户偏好，丢弃寒暄和重复内容。

对话内容：
{format_messages(messages)}

压缩摘要："""

    summary = await llm.generate(prompt)
    return summary

async def extract_entities(messages: list) -> dict:
    """从对话中提取结构化实体"""
    prompt = """从以下对话中提取所有实体信息，按类别组织：
- 人物：姓名、关系、角色
- 偏好：用户表达的喜好和习惯
- 事实：提到的具体事实和数据
- 任务：正在进行的任务和状态
- 技术：提到的技术栈和工具

对话内容：
{format_messages(messages)}"""

    entities = await llm.generate(prompt, response_format="json")
    return entities

class TTLMemory:
    def __init__(self, default_ttl_hours: int = 168):  # 默认7天
        self.default_ttl = default_ttl_hours
        self.store = {}

    def add(self, key: str, value: str, ttl_hours: int = None, importance: float = 0.5):
        ttl = ttl_hours or self.default_ttl
        # 高重要性记忆 TTL 更长
        adjusted_ttl = ttl * (1 + importance * 2)
        self.store[key] = {
            "value": value,
            "expires_at": time.time() + adjusted_ttl * 3600,
            "importance": importance,
            "created_at": time.time()
        }

    def get(self, key: str) -> Optional[str]:
        if key not in self.store:
            return None
        entry = self.store[key]
        if time.time() > entry["expires_at"]:
            del self.store[key]  # 过期删除
            return None
        return entry["value"]

from collections import OrderedDict

class LRUMemory(OrderedDict):
    def __init__(self, max_size: int = 1000):
        super().__init__()
        self.max_size = max_size

    def __setitem__(self, key, value):
        if key in self:
            self.move_to_end(key)
        super().__setitem__(key, value)
        if len(self) > self.max_size:
            oldest = next(iter(self))
            del self[oldest]

import math

def decay_importance(initial_importance: float, hours_since_access: float, half_life_hours: float = 168) -> float:
    """指数衰减的重要性计算"""
    decay_rate = math.log(2) / half_life_hours
    return initial_importance * math.exp(-decay_rate * hours_since_access)

# 示例：初始重要性 0.8 的记忆
# 7天后（半衰期）：0.4
# 14天后：0.2
# 21天后：0.1 → 可能触发删除

from letta import create_client

client = create_client()

# 创建带记忆的 Agent
agent = client.create_agent(
    name="memory_agent",
    model="openai/gpt-4o",
    memory_blocks=[
        {
            "label": "human",
            "value": "用户信息：薛帅，AI 工程师，偏好简洁代码风格"
        },
        {
            "label": "system",
            "value": "你是 AI 技术助手，负责帮助用户解决技术问题"
        }
    ],
    tools=["core_memory_append", "core_memory_replace", "archival_memory_insert", "archival_memory_search"]
)

# Agent 可以自主决定更新记忆
response = agent.chat("记住：我下周要做一个关于 RAG 优化的技术分享")
# Agent 内部会调用 core_memory_append 保存这个信息

# 下次对话时，Agent 会主动回忆
response = agent.chat("帮我准备 RAG 分享的提纲")
# Agent 从记忆中检索到上周的请求

from langmem import create_memory_manager

manager = create_memory_manager(
    model="openai/gpt-4o",
    vector_store=QdrantVectorStore(collection="agent_memories"),
    extraction_prompt="从对话中提取用户偏好、事实和技术知识"
)

# 自动从对话提取记忆
async def process_turn(user_msg: str, assistant_msg: str, user_id: str):
    # 提取并存储记忆
    memories = await manager.extract_and_store(
        conversation=[
            {"role": "user", "content": user_msg},
            {"role": "assistant", "content": assistant_msg}
        ],
        user_id=user_id
    )

    # 检索相关记忆用于下一轮
    relevant = await manager.search(
        query=user_msg,
        user_id=user_id,
        limit=5
    )
    return relevant

from mem0 import Memory

m = Memory.from_config({
    "vector_store": {
        "provider": "qdrant",
        "config": {"host": "localhost", "port": 6333}
    },
    "llm": {
        "provider": "openai",
        "config": {"model": "gpt-4o-mini"}
    }
})

# 添加记忆（自动提取关键信息）
m.add("我喜欢用 Python，最近在学 Rust", user_id="xueshuai")
# 自动提取：偏好 Python、正在学习 Rust

# 搜索记忆
results = m.search("薛帅用什么编程语言？", user_id="xueshuai")
# 返回：偏好 Python，正在学 Rust

# 获取所有记忆
all_memories = m.get_all(user_id="xueshuai")

class MemoryGuard:
    """记忆安全防护层"""

    # 不允许通过对话修改的关键记忆
    PROTECTED_KEYS = {"security_policy", "approval_rules", "access_control"}

    # 敏感操作需要确认
    SENSITIVE_ACTIONS = {"delete_data", "send_email", "modify_config"}

    def validate_memory_write(self, key: str, value: str, source: str) -> bool:
        """验证记忆写入是否安全"""
        # 检查是否尝试修改受保护的 key
        if key in self.PROTECTED_KEYS:
            logger.warning(f"Blocked protected memory write: {key}")
            return False

        # 检查是否包含危险指令
        dangerous_patterns = [
            r"跳过.*检查",
            r"忽略.*安全",
            r"不需要.*确认",
            r"always.*bypass",
        ]
        for pattern in dangerous_patterns:
            if re.search(pattern, value):
                logger.warning(f"Blocked dangerous memory content: {value[:100]}")
                return False

        return True

    def sanitize_retrieved_memories(self, memories: list, query: str) -> list:
        """过滤检索到的记忆，防止间接注入"""
        safe_memories = []
        for mem in memories:
            # 检查记忆内容是否包含可执行指令
            if self.contains_executable_instruction(mem["content"]):
                logger.warning(f"Filtered memory with executable instruction")
                continue
            safe_memories.append(mem)
        return safe_memories

import asyncio
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from enum import Enum
from typing import Optional
import json

class MemoryLevel(Enum):
    L0_WORKING = "working"      # 上下文窗口
    L1_HOT = "hot"              # Redis, TTL 24h
    L2_WARM = "warm"            # 向量DB, 30天
    L3_COLD = "cold"            # 向量DB + 知识图谱, 永久

@dataclass
class Memory:
    id: str
    content: str
    level: MemoryLevel
    user_id: str
    importance: float = 0.5
    tags: list = field(default_factory=list)
    created_at: datetime = field(default_factory=datetime.now)
    last_accessed: datetime = field(default_factory=datetime.now)
    access_count: int = 0
    expires_at: Optional[datetime] = None
    source_session: Optional[str] = None

class ProductionMemorySystem:
    """生产级 Agent 记忆系统"""

    MAX_WORKING_MEMORIES = 20  # L0 最多 20 个记忆块

    def __init__(self, redis_client, vector_db, llm):
        self.redis = redis_client
        self.vector_db = vector_db
        self.llm = llm
        self.working_memory: dict[str, list[Memory]] = {}  # session_id -> memories

    async def remember(self, session_id: str, user_id: str, content: str,
                       importance: float = 0.5, tags: list = None) -> str:
        """存储新记忆"""
        memory = Memory(
            id=generate_id(),
            content=content,
            level=MemoryLevel.L0_WORKING,
            user_id=user_id,
            importance=importance,
            tags=tags or [],
            source_session=session_id
        )

        # 存入工作记忆
        if session_id not in self.working_memory:
            self.working_memory[session_id] = []
        self.working_memory[session_id].append(memory)

        # 如果工作记忆满了，触发压缩
        if len(self.working_memory[session_id]) > self.MAX_WORKING_MEMORIES:
            await self._compress_working_memory(session_id, user_id)

        return memory.id

    async def recall(self, session_id: str, user_id: str, query: str,
                     max_results: int = 5) -> list[Memory]:
        """检索相关记忆"""
        results = []

        # 1. 先查工作记忆（最快）
        working = self.working_memory.get(session_id, [])
        working_results = await self._search_in_memories(working, query, max_results)
        results.extend(working_results)

        # 2. 查热记忆（Redis）
        hot_results = await self._search_hot_memory(user_id, query, max_results - len(results))
        results.extend(hot_results)

        # 3. 查温/冷记忆（向量DB）
        if len(results) < max_results:
            cold_results = await self._search_vector_db(
                user_id, query, max_results - len(results)
            )
            results.extend(cold_results)

        # 更新访问统计
        for mem in results:
            mem.last_accessed = datetime.now()
            mem.access_count += 1

        return results[:max_results]

    async def _compress_working_memory(self, session_id: str, user_id: str):
        """压缩工作记忆：将不重要的记忆降级到 L1/L2"""
        memories = self.working_memory[session_id]

        # 按重要性排序
        memories.sort(key=lambda m: m.importance, reverse=True)

        # 保留最重要的 N 个在工作记忆中
        to_keep = memories[:self.MAX_WORKING_MEMORIES]
        to_demote = memories[self.MAX_WORKING_MEMORIES:]

        # 将降级的记忆压缩并存入下层
        for mem in to_demote:
            if mem.importance > 0.7:
                # 高重要性 → L1 热记忆
                await self._promote_to_hot(user_id, mem)
            elif mem.importance > 0.3:
                # 中等重要性 → L2 温记忆
                await self._store_to_vector_db(user_id, mem, ttl_days=30)
            # 低重要性 → 直接丢弃

        self.working_memory[session_id] = to_keep

    async def consolidate(self, user_id: str):
        """记忆整合：定期将情景记忆抽象为语义记忆"""
        # 获取最近 7 天的情景记忆
        recent = await self.vector_db.search(
            collection="episodic",
            filter={"user_id": user_id, "created_after": datetime.now() - timedelta(days=7)},
            limit=100
        )

        # 用 LLM 从情景记忆中提取语义知识
        prompt = f"""从以下 {len(recent)} 条记忆片段中，提取出稳定的知识和规律：
- 用户的偏好和习惯
- 反复出现的事实
- 可以总结出的规律

记忆片段：
{json.dumps([m.content for m in recent], ensure_ascii=False)}

提取的知识（JSON 格式）："""

        semantic_knowledge = await self.llm.generate(prompt, response_format="json")

        # 存入语义记忆
        for item in semantic_knowledge:
            await self.vector_db.upsert(
                collection="semantic",
                id=generate_id(),
                vector=await self.vector_db.embed(item["content"]),
                metadata={
                    "user_id": user_id,
                    "content": item["content"],
                    "type": item["type"],
                    "confidence": item.get("confidence", 0.8),
                    "created_at": datetime.now().isoformat()
                }
            )