Day 13：持久化记忆 - 让 Agent 拥有长期记忆

深街酒徒

432人浏览 · 2026-06-12 17:38:03

深街酒徒 · 2026-06-12 17:38:03 发布

🤖 系列：Java工程师转AI Agent 3个月学习计划
👤 作者：宸丶一 | 28岁Java程序员，正在学习 AI Agent 开发中ing…
🎯 今日目标： 持久化记忆、四种记忆类型、SQLite FTS5
💬 个人格言： 代码改不改变世界我不知道，但先让我准时下班。

📖 前言

在 Day 12 的学习中，我们掌握了人机协作和 Goal/停止条件，学会了如何让 Agent 和人类一起工作，以及如何防止"乐观停止"。

今天，我们进入 Day 13 的学习：持久化记忆。

如果说 Day 12 是"人机协作"，那么 Day 13 就是"持久化记忆"。持久化记忆让 Agent 在多次会话间保持对项目的理解，避免重复学习，就像人类的长期记忆一样。

特别的是，今天我们还结合了小米开源的 MiMo-Code 框架，学习了其中的持久化记忆机制，并采用了深度思想的学习模式，通过开放性思考题来加深理解。

🎯 学习目标

理解持久化记忆的核心概念
掌握四种记忆类型的设计
学习基于 SQLite FTS5 的实现
结合 MiMo-Code 框架学习最佳实践

📚 核心概念

1. 什么是持久化记忆？

定义： Agent 在多次会话间保持对项目的理解

Java 对标： 数据库 + 缓存

理解： 持久化记忆就像人类的长期记忆，让 Agent 能够记住之前学过的东西，避免重复学习。

本质： 持久化记忆的本质是"数据持久化"，让 Agent 在多次会话间保持对项目的理解。

为什么需要持久化记忆：

效率提升 — 避免重复学习
体验优化 — 用户说"继续"时，Agent 能接上
质量保证 — 保持一致性

优势：

更懂用户
更人性化
人机交互更好

劣势：

效率变慢（需要加载记忆）
存储成本增加
隐私风险

2. 四种记忆类型

参考 MiMo-Code 的设计：

记忆类型	说明	存储周期	Java 对标
项目记忆	跨会话持久的项目知识	长期	配置文件
会话检查点	结构化状态快照	中期	快照
笔记暂存	Agent 临时记录区	短期	临时文件
任务进展	逐任务日志	逐任务	任务日志

四种记忆类型关系图：

项目记忆：

作用：保存项目知识，避免重复学习
场景：跨会话保持项目理解
示例：项目规则、架构决策、学习模式

会话检查点：

作用：保存会话状态，支持会话恢复
场景：用户说"继续"时，能接上
示例：当前任务、对话历史、Agent 状态

笔记暂存：

作用：保存临时笔记，支持上下文理解
场景：Agent 记录临时信息
示例：处理过程中的中间结果

任务进展：

作用：保存任务进度，支持任务追踪
场景：长任务的进度管理
示例：任务状态、完成进度

3. SQLite FTS5

定义： SQLite 的全文搜索引擎

Java 对标： Elasticsearch

特点：

轻量级
高效
易用

为什么选择 SQLite FTS5：

轻量级 — 不需要额外的服务
高效 — 支持中文分词
易用 — API 简单

使用场景：

中小规模应用
全文搜索
笔记检索

SQLite FTS5 架构图：

🔄 持久化记忆工作流

流程图：

执行流程：

加载记忆 — 从存储中加载项目记忆、会话检查点等
处理任务 — 根据记忆上下文处理任务
保存记忆 — 将新的记忆保存到存储中
返回结果 — 返回处理结果

🆚 四种记忆类型对比

记忆类型	存储周期	存储方式	使用场景
项目记忆	长期	JSON/SQLite	项目知识、规则、决策
会话检查点	中期	JSON/SQLite	会话状态、对话历史
笔记暂存	短期	JSON/SQLite	临时笔记、中间结果
任务进展	逐任务	JSON/SQLite	任务状态、完成进度

简单说：

项目记忆 — 长期保存，不会丢失
会话检查点 — 中期保存，会话结束可能丢失
笔记暂存 — 短期保存，随时可能清理
任务进展 — 逐任务保存，任务完成可能清理

💻 代码示例

示例 1：持久化记忆基础

执行流程：

用户输入: 查询北京天气
    ↓
加载记忆: 项目记忆、会话检查点
    ↓
处理任务: 北京今天晴，25度，微风
    ↓
保存记忆: 会话检查点、笔记、任务进展
    ↓
返回结果: 北京今天晴，25度，微风

代码：

from typing import TypedDict
from langgraph.graph import StateGraph, END
import json
import time

# 定义记忆管理器
class MemoryManager:
    def __init__(self):
        self.project_memory = {}  # 项目记忆
        self.session_checkpoint = {}  # 会话检查点
        self.notes = []  # 笔记暂存
        self.task_progress = {}  # 任务进展
    
    def save_project_memory(self, key: str, value: str):
        """保存项目记忆"""
        self.project_memory[key] = value
    
    def load_project_memory(self, key: str) -> str:
        """加载项目记忆"""
        return self.project_memory.get(key, "")
    
    def save_session_checkpoint(self, session_id: str, current_task: str, task_result: str):
        """保存会话检查点"""
        self.session_checkpoint = {
            "session_id": session_id,
            "current_task": current_task,
            "task_result": task_result
        }
    
    def save_notes(self, content: str):
        """保存笔记"""
        self.notes.append(content)
    
    def save_task_progress(self, task_id: str, status: str, progress: str):
        """保存任务进展"""
        self.task_progress[task_id] = {
            "status": status,
            "progress": progress
        }

# 定义记忆 Agent
def memory_agent(state: dict, memory_manager: MemoryManager) -> dict:
    """记忆 Agent：管理持久化记忆"""
    current_task = state.get("current_task", "")
    
    # 加载项目记忆
    project_memory = memory_manager.load_project_memory("project_name")
    
    # 处理任务
    if "天气" in current_task:
        result = "北京今天晴，25度，微风"
    else:
        result = f"已处理：{current_task}"
    
    # 保存记忆
    memory_manager.save_session_checkpoint("session_001", current_task, result)
    memory_manager.save_notes(f"处理了任务: {current_task}")
    memory_manager.save_task_progress("task_001", "完成", f"完成了任务: {current_task}")
    
    return {**state, "task_result": result}

# 创建工作流图
def create_memory_graph(memory_manager: MemoryManager) -> StateGraph:
    graph = StateGraph(dict)
    graph.add_node("memory", lambda state: memory_agent(state, memory_manager))
    graph.set_entry_point("memory")
    graph.add_edge("memory", END)
    return graph

# 编译并运行
memory_manager = MemoryManager()
memory_manager.save_project_memory("project_name", "Python 学习项目")

graph = create_memory_graph(memory_manager)
app = graph.compile()

result = app.invoke({
    "messages": [{"role": "user", "content": "查询北京天气"}],
    "current_task": "查询北京天气",
    "task_result": ""
})
print(result["task_result"])

示例 2：基于 SQLite FTS5 的实现

执行流程：

用户输入: 查询北京天气
    ↓
加载记忆: 从 SQLite 加载项目记忆
    ↓
处理任务: 北京今天晴，25度，微风
    ↓
保存记忆: 保存到 SQLite 数据库
    ↓
全文搜索: 搜索相关笔记
    ↓
返回结果: 北京今天晴，25度，微风

代码：

import sqlite3
import time

# 定义 SQLite FTS5 记忆存储
class SQLiteFTS5Memory:
    def __init__(self, db_path: str = "memory.db"):
        self.db_path = db_path
        self.conn = sqlite3.connect(db_path)
        self.create_tables()
    
    def create_tables(self):
        """创建表"""
        cursor = self.conn.cursor()
        
        # 创建项目记忆表
        cursor.execute("""
            CREATE TABLE IF NOT EXISTS project_memory (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                key TEXT NOT NULL,
                value TEXT NOT NULL,
                timestamp TEXT NOT NULL
            )
        """)
        
        # 创建 FTS5 全文索引
        cursor.execute("""
            CREATE VIRTUAL TABLE IF NOT EXISTS memory_fts USING fts5(
                content,
                content_rowid='rowid'
            )
        """)
        
        self.conn.commit()
    
    def save_project_memory(self, key: str, value: str):
        """保存项目记忆"""
        cursor = self.conn.cursor()
        timestamp = time.strftime("%Y-%m-%d %H:%M:%S")
        
        cursor.execute("""
            INSERT OR REPLACE INTO project_memory (key, value, timestamp)
            VALUES (?, ?, ?)
        """, (key, value, timestamp))
        
        self.conn.commit()
    
    def load_project_memory(self, key: str) -> str:
        """加载项目记忆"""
        cursor = self.conn.cursor()
        cursor.execute("SELECT value FROM project_memory WHERE key = ?", (key,))
        result = cursor.fetchone()
        return result[0] if result else ""
    
    def save_note(self, content: str):
        """保存笔记"""
        cursor = self.conn.cursor()
        timestamp = time.strftime("%Y-%m-%d %H:%M:%S")
        
        cursor.execute("""
            INSERT INTO memory_fts (content)
            VALUES (?)
        """, (content,))
        
        self.conn.commit()
    
    def search_notes(self, query: str) -> list:
        """搜索笔记（全文搜索）"""
        cursor = self.conn.cursor()
        
        cursor.execute("""
            SELECT content
            FROM memory_fts
            WHERE content MATCH ?
            LIMIT 10
        """, (query,))
        
        results = cursor.fetchall()
        return [{"content": r[0]} for r in results]
    
    def close(self):
        """关闭数据库连接"""
        self.conn.close()

# 使用示例
memory = SQLiteFTS5Memory()
memory.save_project_memory("project_name", "Python 学习项目")
memory.save_note("处理了任务: 查询北京天气")

# 全文搜索
results = memory.search_notes("任务")
print(f"搜索结果: {len(results)} 条")

memory.close()

🎓 学习收获

1. 核心概念理解

持久化记忆 = 数据库 + 缓存
四种记忆类型 = 项目记忆、会话检查点、笔记暂存、任务进展
SQLite FTS5 = 轻量级全文搜索引擎

2. 记忆类型对比

记忆类型	存储周期	使用场景
项目记忆	长期	项目知识、规则、决策
会话检查点	中期	会话状态、对话历史
笔记暂存	短期	临时笔记、中间结果
任务进展	逐任务	任务状态、完成进度

3. 技术实现

存储方式 — JSON 文件、SQLite 数据库
检索方式 — 关键词搜索、全文搜索（FTS5）
更新机制 — 手动更新、自动更新、版本控制

🧠 深度思想学习模式

为什么采用深度思想学习模式？

传统学习模式：

代码填空 — 重复性高，容易枯燥
固定答案 — 限制思维，缺乏创新
被动学习 — 缺乏主动思考

深度思想学习模式：

开放性设计 — 激发创造力
深度分析 — 培养分析能力
创新设计 — 培养创新思维
主动思考 — 提升学习效果

深度思想学习模式的特点

特点	说明	效果
开放性	没有标准答案	激发创造力
深度性	需要深入分析	培养分析能力
创新性	需要提出创新想法	培养创新思维
主动性	需要主动思考	提升学习效果