第16章 MemGPT / Letta —— Agent 记忆的「操作系统」

论文: MemGPT: Towards LLMs as Operating Systems (2023)项目: github.com/letta-ai/letta (原 MemGPT)16.1 核心思想把 LLM 当作操作系统类比计算机操作系统管理内存的方式操作系统内存层次:┌──────────┐ 最快/最小 │ 寄存器 │ ├──────────┤ │ L1/L2 │ │ Cache │ ├──────────┤ │ RAM │ ← 主内存 ├──────────┤ │ 磁盘 │ ← 持久化 └──────────┘ 最慢/最大MemGPT 的内存层次:┌──────────────┐ 最快/在上下文中 │ Core Memory │ ← 当前对话 记忆块 │ (上下文内) │ (类似 RAM) ├──────────────┤ │ Recall Memory │ ← 完整对话历史 │ (上下文外) │ (磁盘可检索) ├──────────────┤ │ Archival Memory│ ← 外部知识库 │ (向量数据库) │ (归档存储) └──────────────┘ 最慢/最大核心洞察LLM 的上下文窗口 操作系统的 RAM上下文有限但通过「换页」技术可以管理无限容量的信息OS 用虚拟内存换页 → MemGPT 用记忆管理工具换页16.2 MemGPT 的内存管理机制 架构示意┌────────────────────────────────────────────────────┐ │ MemGPT Agent │ │ │ │ ┌──────────────────────────────────────────────┐ │ │ │ Core Memory (上下文内) │ │ │ │ ┌──────────┐ ┌──────────┐ ┌──────────────┐ │ │ │ │ │ System │ │ Memory │ │ Conversation │ │ │ │ │ │ Prompt │ │ Blocks │ │ History │ │ │ │ │ └──────────┘ └──────────┘ └──────────────┘ │ │ │ └──────────────────────────────────────────────┘ │ │ │ │ │ ┌──────────┴──────────┐ │ │ │ 对话历史管理工具 │ │ │ │ ───────────────── │ │ │ │ conversation_search │ ← 搜索完整历史 │ │ │ core_memory_append │ ← 编辑记忆块 │ │ │ core_memory_replace │ ← 替换记忆块 │ │ └─────────────────────┘ │ │ │ │ │ ┌────────────────────┴──────────────────────────┐ │ │ │ 外部记忆上下文外 │ │ │ │ ┌────────────────┐ ┌──────────────────────┐ │ │ │ │ │ Recall Memory │ │ Archival Memory │ │ │ │ │ │ (对话历史全文) │ │ (向量库 知识检索) │ │ │ │ │ │ archival_memory │ │ archival_memory │ │ │ │ │ │ _search │ │ _insert │ │ │ │ │ └────────────────┘ └──────────────────────┘ │ │ │ └──────────────────────────────────────────────┘ │ └────────────────────────────────────────────────────┘核心工具Agent 通过 Function Calling 调用send_message: 向用户发送消息MemGPT 唯一的消息出口core_memory_append: 向 Core Memory 块追加信息core_memory_replace: 替换 Core Memory 块的内容conversation_search: 搜索完整对话历史archival_memory_insert: 将信息存入归档记忆向量化archival_memory_search: 从归档记忆中检索16.3 Heartbeat 机制 —— Agent 的「自我意识」传统 Agent 的问题工具调用 → 返回结果 → 给用户 → 结束Agent 没有机会检查「我做得对不对」Heartbeat 的解决方案每个工具的返回结果中包含一个 request_heartbeat 参数如果设为 true → 工具执行完毕后控制权交还给 AgentAgent 可以检查结果 → 发现错误 → 自动重试代码示意# 工具返回结果{ tool_call_id: call_123, content: 搜索结果为空, request_heartbeat: true // ← 请求心跳 }# Agent 收到心跳后→ 搜索结果为空我换一个关键词试试→ 再次调用 search(换一个搜索词)→ 这次找到了这类似于你在命令行执行 ls 之后shell 不会直接退出而是等你输入下一个命令Heartbeat 就是给 Agent 的「命令提示符」16.4 Sleep-Time Compute —— 后台记忆整理MemGPT v2 的创新2025年。核心思想Agent 在「休眠」时用户不互动时启动一个后台 Agent 来整理记忆。做什么浏览最近的对话历史提取重要信息 → 写入 Core Memory删除过时的信息更新用户画像合并重复的记忆类比就像你睡觉时大脑在整理白天的记忆不重要的事情遗忘重要的事情巩固效果Letta 在 LoCoMo 基准测试上达到 74.0% 准确率仅用文件系统 GPT-4o-mini16.5 Letta Filesystem ——「文件系统 最好的记忆工具」2025年8月Letta 发表了一篇震撼论文Is a Filesystem All You Need?→ LoCoMo 基准测试 74.0%超过专用记忆工具 Mem0 的 68.5%为什么文件系统做记忆效果惊人文件操作是 LLM 最熟悉的范式grep / open / close / search_files这些工具在 LLM 训练数据中大量出现Agent 对它们的使用非常自然简单即正确专用记忆工具引入额外复杂性多了出错的环节Agent 需要理解专用工具的特殊语义无需学习新范式每个记忆工具Mem0/Zep/LangMem都有自己的 APILLM 需要「学会」每个新工具而文件操作是 LLM 的「母语」核心启示面试金句Agent 的记忆质量取决于 Agent 管理上下文的能力而不是记忆工具的复杂度。文件系统 好用工具 最佳记忆方案。16.5.1 Heartbeat 与 Sleep-Time 的工程实现 —— 不只是「定时器」▍ Heartbeat 的成本计算 —— 「自我纠错」要花多少钱面试官问「Heartbeat 每次都要多调一次 LLM成本怎么控制」每次 Heartbeat 触发时Agent 要把「当前状态 工具结果」发给 LLM让 LLM 判断是否需要进一步行动。这确实增加了成本。控制策略a) 选择性 Heartbeat —— 不是每个工具都请求心跳只在「关键工具」数据库写入、API 调用执行后触发b) 使用廉价模型做 Heartbeat —— Heartbeat 判断是「是否需要继续」不要求深度推理 → 用 gpt-4o-mini 就够c) Heartbeat 超时限制 —— 最多触发 3 次心跳避免死循环d) Token 预算驱动 —— 累计消耗超过阈值 → 停止心跳直接总结经验数据合理的 Heartbeat 机制增加 15-25% 的额外成本但减少 30-50% 的错误率 → ROI 是正的。▍ Sleep-Time Compute 的调度设计Letta 的 Sleep-Time Compute 不是简单的 cron 任务。它需要判断什么时候触发用户 5 分钟无交互1 小时处理哪些记忆最近 10 轮对话所有对话用什么模型整理记忆不需要 GPT-4o小模型够用工程实现建议用户交互结束后启动一个 30 秒的倒计时倒计时内用户再次交互 → 取消倒计时倒计时结束 → 启动 Sleep-Time Agent用廉价模型Sleep-Time Agent 扫描本轮对话 → 提取要点 → 更新 Core Memory完成后 Agent 回到待机状态注意Sleep-Time Compute 不是实时需求 → 可以异步执行不阻塞用户的下一次交互。这也是为什么 Letta 使用后台 Agent 的原因。▍ 记忆压缩的工程权衡 —— 保留什么丢弃什么MemGPT 记忆管理最核心的决策「什么信息值得进入 Core Memory」面试可以主动分析值得保留 → 用户偏好「我喜欢简洁回答」、事实信息「我是北京人」、任务上下文「正在做税务报表已完成 Q1-Q3」可以丢弃 → 临时搜索词、失败的尝试、已过时的数据、与当前任务无关的闲聊内容工业界的经验法则Core Memory 控制在 2000 tokens 以内留给对话历史足够的空间。超过上限 → 触发记忆整理合并相似、删除过时、归档低频16.6 MemGPT 架构的简化实现下面实现一套 MemGPT 风格的 Agent 记忆管理模拟器核心三组件CoreMemoryBlock —— 存储 Agent 的核心身份信息人类/角色Heartbeat —— 定时触发反思模拟 Sleep-Time ComputeMemoryManager —— 统一的记忆读写接口这套架构的精髓在于Agent 不是被动地被调用而是像操作系统一样持续运行定期自我审视和整理记忆。 对应的代码实现core_memory_appendcore_memory_replacearchival_memory_insertarchival_memory_searchcompact_conversationheartbeat_should_continuedemo_memgpt_workflowMemGPTAgentfrom typing import Optional import json import time import hashlib class MemGPTAgent: MemGPT 架构的简化实现 —— 展示核心概念。 实现 1. Core Memory 块System Human Persona 2. 记忆编辑工具core_memory_append/replace 3. Conversation Search对话历史搜索 4. 上下文压缩对话历史过长时自动压缩 def __init__(self, system_prompt: str ): 初始化 MemGPT Agent。 Args: system_prompt: 系统提示词。 # Core Memory 块 self.core_memory { human: 用户偏好和背景信息, persona: AI助手专业且友好, system: system_prompt or 你是一个有用的 AI 助手。, } # 完整对话历史Recall Memory self.conversation [] # 归档记忆简化版字典存储 self.archival {} # 已压缩的摘要 self.compacted_summary def _build_context(self, max_recent: int 10) - list[dict]: 构建发给 LLM 的上下文。 包含 1. Core Memory 块 2. 压缩摘要如果有 3. 最近 N 条对话 Args: max_recent: 最大保留的最近消息数。 Returns: 上下文消息列表。 context [] # Core Memory 组装 core_text \n.join( f### {key}\n{value} for key, value in self.core_memory.items() ) context.append({ role: system, content: fmemory\n{core_text}\n/memory, }) if self.compacted_summary: context.append({ role: system, content: fsummary\n{self.compacted_summary}\n/summary, }) # 最近 N 条对话 recent self.conversation[-max_recent:] if self.conversation else [] context.extend(recent) return context def core_memory_append(self, block: str, content: str): 向 Core Memory 块追加内容。 Args: block: 块名human/persona。 content: 追加的内容。 if block in self.core_memory: self.core_memory[block] \n content def core_memory_replace(self, block: str, content: str): 替换 Core Memory 块的内容。 Args: block: 块名。 content: 新内容。 if block in self.core_memory: self.core_memory[block] content def archival_memory_insert(self, key: str, content: str): 向归档记忆插入一条信息。 Args: key: 检索键。 content: 信息内容。 self.archival[key] { content: content, timestamp: time.time(), } def archival_memory_search(self, query: str) - str: 从归档记忆搜索简化版关键词匹配。 Args: query: 搜索词。 Returns: 匹配的结果。 results [] for key, entry in self.archival.items(): if query.lower() in key.lower() or query.lower() in entry[content].lower(): results.append(f[{key}] {entry[content]}) return \n.join(results[-5:]) if results else 未找到相关信息 def compact_conversation(self): 对话压缩早期对话 → 摘要放入 compacted_summary。 这是 MemGPT 最核心的操作之一 当对话历史过长时对早期部分做摘要压缩。 if len(self.conversation) 20: return # 取最早的 10 条对话做压缩 old_part self.conversation[:10] self.conversation self.conversation[10:] # 简化的摘要生成 topics set() for msg in old_part: for word in msg.get(content, ).split(): if len(word) 3: topics.add(word.lower()) new_summary ( f此前讨论了以下话题{, .join(list(topics)[:10])}... ) if self.compacted_summary: self.compacted_summary \n new_summary else: self.compacted_summary new_summary def heartbeat_should_continue(self, last_tool_result: str) - bool: 判断是否需要 Heartbeat继续给 Agent 控制权。 如果工具调用失败或者返回不充分 返回 True 让 Agent 继续思考和行动。 Args: last_tool_result: 上一步工具调用的结果。 Returns: 是否需要继续。 # 模拟判断如果结果包含错误信号 error_signals [未找到, 错误, 失败, 抱歉, 无法] return any(sig in last_tool_result for sig in error_signals) def demo_memgpt_workflow(): 演示 MemGPT 的完整工作流。 print( * 60) print( MemGPT 记忆架构演示) print( * 60) agent MemGPTAgent() # Step 1: 展示初始 Core Memory print(\n 初始 Core Memory:) for key, value in agent.core_memory.items(): print(f [{key}] {value}) # Step 2: Agent 学习用户信息 → 写入 Core Memory print(\n 用户告诉 Agent 自己的偏好) agent.core_memory_append(human, 喜欢简洁的回答风格) agent.core_memory_append(human, 是Python开发者经常需要代码帮助) agent.core_memory_replace(persona, Python 专家助手回答简洁专业) print( 更新后的 Core Memory:) for key, value in agent.core_memory.items(): print(f [{key}] {value}) # Step 3: 模拟对话 print(\n 模拟对话) dialogs [ (user, 帮我写一个快速排序算法), (assistant, 好的这是一个 Python 快速排序实现\ndef quicksort(arr):\n if len(arr) 1:\n return arr\n pivot arr[0]\n left [x for x in arr[1:] if x pivot]\n right [x for x in arr[1:] if x pivot]\n return quicksort(left) [pivot] quicksort(right)), (user, 谢谢再帮我分析一下时间复杂度), (assistant, 快速排序的平均时间复杂度是 O(n log n)最坏情况是 O(n²)...), ] for role, content in dialogs: agent.conversation.append({role: role, content: content}) print(f 当前对话数: {len(agent.conversation)}) # Step 4: 写入归档记忆 print(\n 归档记忆) agent.archival_memory_insert( quicksort, 快速排序O(n log n) 平均O(n²) 最坏非稳定排序 ) agent.archival_memory_insert( python_best_practice, Python编码规范PEP 8, 类型注解, 单元测试 ) agent.archival_memory_insert( user_preference, 用户偏好简洁代码不喜欢过度注释 ) print(f 已存入 {len(agent.archival)} 条归档记忆) # Step 5: 搜索归档记忆 print(\n 搜索归档记忆) result agent.archival_memory_search(排序) print(f 搜索「排序」→ {result}) # Step 6: 模拟对话压缩 print(\n ️ 对话压缩演示) # 模拟大量对话 for i in range(50): agent.conversation.append({ role: user, content: f问题{i}这是一条测试消息, }) agent.conversation.append({ role: assistant, content: f回答{i}这是一条回复, }) print(f 压缩前: {len(agent.conversation)} 条消息) agent.compact_conversation() print(f 压缩后: {len(agent.conversation)} 条消息) print(f 摘要: {agent.compacted_summary[:100]}...) # Step 7: Heartbeat 机制演示 print(\n Heartbeat 机制演示) test_cases [ (查询成功找到 3 条结果, False), (未找到相关数据请更换搜索词, True), (工具调用失败API 超时, True), (计算完成结果为 42, False), ] for result, expected_heartbeat in test_cases: needs_heartbeat agent.heartbeat_should_continue(result) status ✅ if needs_heartbeat expected_heartbeat else ❌ print(f {status} 「{result[:25]}...」→ 需要心跳: {needs_heartbeat}) # Step 8: 展示 LLM 上下文 print(\n 发给 LLM 的上下文结构) context agent._build_context(max_recent3) for i, msg in enumerate(context): role msg[role] content msg[content][:80].replace(\n, ) print(f [{i}] {role}: {content}...)16.7 本章总结核心要点回顾MemGPT 把 LLM 当操作系统管理Core Memory RAM上下文中快速访问Recall Memory 磁盘对话历史全文可检索Archival Memory 归档向量库 外部知识Heartbeat 机制工具调用后把控制权交还给 AgentAgent 可以检查结果、发现错误、自动重试类似命令行的「命令提示符」Sleep-Time Compute (MemGPT v2)Agent 休眠时后台整理记忆提取重要信息、删除冗余、更新画像Letta Filesystem 的启示文件系统做记忆 简单 效果好LoCoMo 74.0%超过专用工具的 68.5%Agent 不需要复杂的记忆工具需要好用的工具面试速记MemGPT 的核心创新是什么→ OS 级内存管理Core Memory Recall Archival→ 通过工具让 Agent 自己管理自己的记忆→ Heartbeat 机制实现自我纠错→ 「文件系统就是最好的记忆工具」 对应的代码实现from typing import Optional import json import time import hashlib if __name__ __main__: print(╔══════════════════════════════════════════════════════╗) print(║ 第16章MemGPT / Letta 记忆架构 ║) print(║ Core Memory · Heartbeat · Sleep-Time · Filesystem ║) print(╚══════════════════════════════════════════════════════╝) demo_memgpt_workflow() print(\n▶ MemGPT 内存层次对应关系) print(- * 50) layers [ (Core Memory, RAM, 当前上下文 可编辑记忆块), (Recall Memory, 磁盘, 完整对话历史可搜索), (Archival Memory, 归档, 向量数据库外部知识), ] for mem, os, desc in layers: print(f {mem:18s} ↔ {os:6s} → {desc}) print(\n✅ 第16章完成)