跳转至

记忆架构设计

引言

设计一个完整的 Agent 记忆架构，需要综合考虑存储层级、检索效率、记忆更新和遗忘机制。本节介绍 MemGPT 等前沿方案，以及实际的记忆架构设计模式。

MemGPT：将 LLM 视为操作系统

核心思想

Packer et al.（2023）提出 MemGPT，将 LLM 的上下文窗口类比为操作系统的主内存（RAM），外部存储类比为磁盘。LLM 自身充当"处理器"，通过函数调用管理内存的读写。

graph TB
    subgraph "MemGPT 架构"
        subgraph "主上下文（Main Context / RAM）"
            SP[系统指令<br/>System Prompt]
            WM[工作上下文<br/>Working Context]
            FIFO[对话缓冲<br/>FIFO Queue]
        end

        subgraph "外部存储（Disk）"
            RM[(回忆存储<br/>Recall Storage)]
            AM[(归档存储<br/>Archival Storage)]
        end

        LLM[LLM 处理器] -->|core_memory_append| WM
        LLM -->|core_memory_replace| WM
        LLM -->|conversation_search| RM
        LLM -->|archival_memory_insert| AM
        LLM -->|archival_memory_search| AM

        FIFO -->|溢出| RM
    end

关键机制

  1. 核心记忆（Core Memory）：始终在上下文中的关键信息（用户画像、系统状态）
  2. 回忆存储（Recall Storage）：对话历史的向量化存储
  3. 归档存储（Archival Storage）：长期知识的向量化存储
  4. 自主管理：LLM 通过函数调用自主决定何时读写记忆
# MemGPT 风格的记忆管理函数
MEMGPT_TOOLS = [
    {
        "name": "core_memory_append",
        "description": "向核心记忆的指定部分追加信息",
        "parameters": {
            "section": {"type": "string", "enum": ["human", "persona"]},
            "content": {"type": "string"}
        }
    },
    {
        "name": "core_memory_replace",
        "description": "替换核心记忆中的特定内容",
        "parameters": {
            "section": {"type": "string"},
            "old_content": {"type": "string"},
            "new_content": {"type": "string"}
        }
    },
    {
        "name": "archival_memory_search",
        "description": "搜索归档记忆",
        "parameters": {
            "query": {"type": "string"},
            "page": {"type": "integer"}
        }
    },
    {
        "name": "archival_memory_insert",
        "description": "向归档记忆写入信息",
        "parameters": {
            "content": {"type": "string"}
        }
    },
    {
        "name": "conversation_search",
        "description": "搜索历史对话",
        "parameters": {
            "query": {"type": "string"},
            "page": {"type": "integer"}
        }
    }
]

分层记忆架构（Tiered Memory）

热/温/冷分层

graph TB
    subgraph "Hot Tier（热层）"
        H1[上下文窗口内的信息]
        H2[核心记忆 / 用户画像]
        H3[当前任务状态]
    end

    subgraph "Warm Tier（温层）"
        W1[最近 N 轮对话历史]
        W2[频繁访问的知识]
        W3[活跃任务的相关文档]
    end

    subgraph "Cold Tier（冷层）"
        C1[历史对话归档]
        C2[不常用的知识文档]
        C3[已完成任务的记录]
    end

    H1 -.->|溢出/老化| W1
    W1 -.->|归档| C1
    C1 -.->|按需检索| W1
    W1 -.->|注入上下文| H1

分层策略

层级 存储位置 访问延迟 容量 内容
Hot LLM 上下文 0ms 数 K tokens 系统提示、核心记忆、当前对话
Warm 内存缓存 + 向量 DB 10-100ms 数 MB 近期历史、高频知识
Cold 持久化存储 100ms-1s 数 GB 全量历史、完整知识库 
class TieredMemory:
    def __init__(self, hot_budget=4000, warm_size=1000):
        self.hot = HotMemory(max_tokens=hot_budget)
        self.warm = WarmMemory(max_items=warm_size)  # 内存中的向量缓存
        self.cold = ColdMemory()  # 持久化向量数据库

    def retrieve(self, query, max_results=5):
        """分层检索：先热后温再冷"""
        # 1. 热层：直接在上下文中
        hot_results = self.hot.search(query)
        if len(hot_results) >= max_results:
            return hot_results[:max_results]

        # 2. 温层：内存中的向量搜索
        warm_results = self.warm.search(query, k=max_results - len(hot_results))
        combined = hot_results + warm_results
        if len(combined) >= max_results:
            return combined[:max_results]

        # 3. 冷层：持久化存储搜索
        cold_results = self.cold.search(query, k=max_results - len(combined))

        # 将冷层命中的结果提升到温层
        for result in cold_results:
            self.warm.promote(result)

        return (combined + cold_results)[:max_results]

    def store(self, content, importance=5):
        """根据重要性决定存储层级"""
        if importance >= 8:
            self.hot.add(content)    # 高重要性：直接进入上下文
            self.warm.add(content)   # 同时在温层备份
        elif importance >= 4:
            self.warm.add(content)   # 中等重要性：温层
        else:
            self.cold.add(content)   # 低重要性：冷层

记忆巩固（Memory Consolidation）

灵感：人类的记忆巩固

人类在睡眠期间进行记忆巩固：将短期记忆转化为长期记忆，整合新旧知识，提取模式和规律。

Agent 的记忆巩固

class MemoryConsolidation:
    def __init__(self, episodic_memory, semantic_memory, llm):
        self.episodic = episodic_memory
        self.semantic = semantic_memory
        self.llm = llm

    def consolidate(self):
        """定期执行记忆巩固"""
        # 1. 提取近期经验
        recent_episodes = self.episodic.get_recent(hours=24)

        # 2. 从经验中提取知识
        knowledge = self.llm.generate(
            f"从以下经验中提取通用知识和模式：\n{recent_episodes}\n"
            f"输出格式：(主语, 关系, 宾语) 三元组列表"
        )

        # 3. 存入语义记忆
        for triple in knowledge:
            self.semantic.add_knowledge(
                triple["subject"], triple["predicate"], triple["object"]
            )

        # 4. 生成反思
        reflection = self.llm.generate(
            f"基于最近的经验，有什么重要的观察或教训？\n{recent_episodes}"
        )

        # 5. 将反思作为高重要性情景存储
        self.episodic.store_episode({
            "type": "reflection",
            "content": reflection,
            "importance": 9,
        })

        return {"knowledge_extracted": len(knowledge), "reflection": reflection}

重要性评估

重要性评分

不是所有信息都同等重要。重要性评分决定记忆的存储层级和保留时间。

def score_importance(content, llm):
    """使用 LLM 评估信息的重要性（1-10）"""
    score = llm.generate(
        f"在1到10的范围内，评估以下信息对一个 AI 助手而言的长期重要性。\n"
        f"1=完全无关紧要（闲聊），10=极其重要（关键偏好/知识）。\n"
        f"信息: {content}\n"
        f"只输出数字:"
    )
    return int(score.strip())

基于规则的重要性评估

def rule_based_importance(content, metadata):
    """基于规则的快速重要性评估"""
    score = 5  # 基础分

    # 用户明确请求记住的信息
    if "记住" in content or "remember" in content.lower():
        score += 3

    # 包含个人偏好
    if any(kw in content for kw in ["喜欢", "讨厌", "偏好", "prefer"]):
        score += 2

    # 任务成功/失败的经验
    if metadata.get("outcome") == "failure":
        score += 2  # 失败经验更值得记住

    # 包含具体数字或事实
    if any(char.isdigit() for char in content):
        score += 1

    return min(score, 10)

遗忘曲线与记忆衰减

Ebbinghaus 遗忘曲线

记忆的保持量随时间指数衰减：

\[R = e^{-t/S}\]

其中：

  • \(R\) 是记忆保持率（retention）
  • \(t\) 是经过的时间
  • \(S\) 是记忆强度（strength），与复习次数和重要性相关

实现

import math
from datetime import datetime

class MemoryWithDecay:
    def __init__(self):
        self.memories = []

    def compute_retention(self, memory):
        """计算记忆保持率"""
        hours_since_access = (
            datetime.now() - memory["last_accessed"]
        ).total_seconds() / 3600

        # 记忆强度 S 与访问次数和重要性相关
        strength = memory["access_count"] * 0.5 + memory["importance"] * 2

        retention = math.exp(-hours_since_access / max(strength, 1))
        return retention

    def decay_and_cleanup(self, threshold=0.1):
        """衰减并清理保持率过低的记忆"""
        to_keep = []
        to_archive = []

        for mem in self.memories:
            retention = self.compute_retention(mem)
            mem["retention"] = retention

            if retention >= threshold:
                to_keep.append(mem)
            else:
                to_archive.append(mem)

        self.memories = to_keep
        return to_archive  # 返回归档的记忆

    def access(self, memory_id):
        """访问记忆时更新元数据（间隔重复效应）"""
        for mem in self.memories:
            if mem["id"] == memory_id:
                mem["access_count"] += 1
                mem["last_accessed"] = datetime.now()
                break

实用设计模式

模式 1：用户画像渐进构建

class UserProfile:
    """随对话渐进构建用户画像"""

    def __init__(self, llm):
        self.profile = {
            "preferences": {},
            "background": {},
            "goals": {},
            "communication_style": {},
        }
        self.llm = llm

    def update_from_conversation(self, messages):
        """从对话中提取用户信息"""
        extracted = self.llm.generate(
            f"从以下对话中提取用户的偏好、背景、目标等信息。\n"
            f"对话: {messages}\n"
            f"当前画像: {self.profile}\n"
            f"输出更新后的画像（JSON）:"
        )
        self.profile = merge_profiles(self.profile, extracted)

模式 2：任务上下文持久化

class TaskContext:
    """跨会话的任务上下文"""

    def __init__(self, store):
        self.store = store

    def save_checkpoint(self, task_id, state):
        """保存任务检查点"""
        self.store.save({
            "task_id": task_id,
            "state": state,
            "progress": state.get("progress", 0),
            "next_steps": state.get("next_steps", []),
            "timestamp": datetime.now().isoformat(),
        })

    def resume_task(self, task_id):
        """恢复任务上下文"""
        checkpoint = self.store.get_latest(task_id)
        if checkpoint:
            return (
                f"任务进度: {checkpoint['progress']}%\n"
                f"下一步: {checkpoint['next_steps']}\n"
                f"上次更新: {checkpoint['timestamp']}"
            )
        return None

模式 3：记忆索引与标签

class TaggedMemory:
    """带标签的记忆系统，支持多维度检索"""

    def __init__(self, vector_store):
        self.vector_store = vector_store

    def store(self, content, tags, importance=5):
        self.vector_store.add(
            documents=[content],
            metadatas=[{
                "tags": ",".join(tags),
                "importance": importance,
                "timestamp": datetime.now().isoformat(),
            }],
            ids=[generate_id()]
        )

    def search_by_tag(self, tag, query=None, n=10):
        """按标签搜索，可选地结合语义搜索"""
        filters = {"tags": {"$contains": tag}}

        if query:
            return self.vector_store.query(
                query_texts=[query], n_results=n, where=filters
            )
        else:
            return self.vector_store.get(where=filters, limit=n)

综合架构示例

graph TB
    subgraph "Agent 综合记忆架构"
        INPUT[输入] --> ROUTER{记忆路由}

        ROUTER -->|存储| IMP[重要性评估]
        IMP -->|高| HOT[热层<br/>核心记忆]
        IMP -->|中| WARM[温层<br/>向量缓存]
        IMP -->|低| COLD[冷层<br/>归档存储]

        ROUTER -->|检索| QP[查询处理]
        QP --> SEARCH[分层搜索]
        HOT --> SEARCH
        WARM --> SEARCH
        COLD --> SEARCH
        SEARCH --> RERANK[重排序]
        RERANK --> CTX[上下文注入]

        subgraph "后台任务"
            CONS[记忆巩固<br/>定期执行]
            DECAY[衰减清理<br/>定期执行]
            CONS --> WARM
            DECAY --> COLD
        end
    end

    CTX --> LLM[LLM 推理]
    LLM --> OUTPUT[输出]
    OUTPUT -->|反馈| ROUTER

实践建议

设计清单

  • [ ] 明确记忆的分层策略（热/温/冷）
  • [ ] 实现重要性评估机制
  • [ ] 设计遗忘/衰减策略防止记忆膨胀
  • [ ] 实现记忆巩固管线（从情景到语义）
  • [ ] 添加记忆冲突检测和解决机制
  • [ ] 监控记忆系统的检索质量和延迟

注意事项

  1. 隐私：用户记忆数据需要妥善保护
  2. 一致性：避免存储相互矛盾的记忆
  3. 可解释性：用户应能查看和管理 Agent 对自己的"记忆"
  4. 成本：向量数据库的存储和查询成本需要控制

参考文献

  • Packer, C., et al. (2023). "MemGPT: Towards LLMs as Operating Systems"
  • Park, J. S., et al. (2023). "Generative Agents: Interactive Simulacra of Human Behavior"
  • Ebbinghaus, H. (1885). "Über das Gedächtnis"
  • Zhang, Z., et al. (2024). "A Survey on the Memory Mechanism of Large Language Model based Agents"
  • Modarressi, A., et al. (2024). "MemLLM: Finetuning LLMs to Use An Explicit Read-Write Memory"

评论 #