IT架构师

奔赴山海，保持热爱

0%

AI Agent 记忆系统设计：OpenClaw 三层记忆架构实战

发表于 更新于 分类于 ， ，
本文字数： 20k 阅读时长 ≈ 36 分钟

AI Agent 记忆系统设计：OpenClaw 三层记忆架构实战

摘要：记忆是 AI Agent 实现持续学习和个性化服务的核心能力。本文深入解析 OpenClaw 的三层记忆架构设计：全局记忆（MEMORY.md）、每日记忆（YYYY-MM-DD.md）、项目记忆（projects/*/memory.md）。从架构设计、实现细节、性能优化到实际应用案例，全面揭秘如何让 AI Agent 拥有”长期记忆”和”短期记忆”，实现真正的个性化服务。

关键词：AI Agent、记忆系统、OpenClaw、上下文管理、RAG、架构设计

一、背景与挑战

1.1 为什么 AI Agent 需要记忆？

想象一下，如果你每天都要重新认识你的朋友，忘记昨天说过的话、做过的事，那会是怎样的体验？

没有记忆的 AI Agent 面临的困境：

1
2
3
4
5
6
7
8
用户：帮我继续昨天的 CrystalForge 测试优化
Agent：抱歉，我不记得昨天做过什么。能重新描述一下需求吗？

用户：还是用老配置部署
Agent：请问"老配置"是指哪个版本？上次部署是什么时候？

用户：金刚最近怎么样？
Agent：请问"金刚"是谁？是您的宠物、项目还是其他？

痛点分析：

问题 影响 严重度
上下文丢失 用户需要重复描述 🔴 高
个性化缺失 无法提供定制服务 🔴 高
学习效率低 每次从零开始 🟡 中
信任感差 像和陌生人对话 🔴 高

1.2 记忆系统的核心需求

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
graph TB
    subgraph "功能性需求"
        F1[长期记忆存储]
        F2[短期上下文缓存]
        F3[快速检索查询]
        F4[记忆更新机制]
        F5[记忆遗忘策略]
    end
    
    subgraph "非功能性需求"
        NF1[低延迟 < 100ms]
        NF2[高可用 99.9%]
        NF3[可扩展 GB→TB]
        NF4[安全性 加密存储]
        NF5[可维护 易调试]
    end
    
    F1 --> NF1
    F2 --> NF1
    F3 --> NF1
    F4 --> NF5
    F5 --> NF5

1.3 设计目标

指标 目标值 实际达成
记忆检索延迟 < 100ms 45ms
记忆存储容量 10GB+ 50GB
上下文窗口利用 > 80% 92%
记忆准确率 > 95% 97%
系统可用性 99.9% 99.95%

二、架构设计

2.1 三层记忆架构总览

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
graph TB
    subgraph "L1 - 全局记忆 (长期)"
        MEMORY[MEMORY.md<br/>核心事实/偏好/规则]
        SOUL[SOUL.md<br/>身份定义]
        USER[USER.md<br/>用户信息]
        AGENTS[AGENTS.md<br/>工作规范]
    end
    
    subgraph "L2 - 每日记忆 (短期)"
        Today[memory/YYYY-MM-DD.md<br/>今日工作日志]
        Yesterday[memory/YYYY-MM-DD-1.md<br/>昨日记录]
    end
    
    subgraph "L3 - 项目记忆 (上下文)"
        P1[projects/P1/memory.md<br/>CrystalForge]
        P2[projects/P2/memory.md<br/>TrailSync]
        P3[projects/P3/memory.md<br/>OpenClaw Extension]
        P4[projects/P4/memory.md<br/>Blog System]
    end
    
    subgraph "Agent Core"
        Query[记忆查询引擎]
        Retrieve[检索模块]
        Update[更新模块]
        Forget[遗忘策略]
    end
    
    User[用户请求] --> Query
    Query -->|读取 | MEMORY
    Query -->|读取 | Today
    Query -->|读取 | P1
    Query -->|写入 | Update
    Update --> MEMORY
    Update --> Today
    Update --> P1
    Today -.->|7 天后 | Forget

2.2 各层职责划分

L1 - 全局记忆（长期记忆）

特点：

  • 📌 持久化：除非手动删除，否则永久保存
  • 🎯 高价值：核心事实、用户偏好、重要规则
  • 🔒 高安全：仅主会话可访问，群聊隔离
  • 📊 小体积：通常 < 100KB

内容示例：

1
2
3
4
5
6
7
8
9
10
11
12
## John
- John 的爱犬叫"金刚"，体力很好（曾一起跑过 15 公里）。
- John 偏好在 Feishu 直接收到结果，不希望每次去服务器查看。

## Assistant Preferences / Working Style
- 重要自动化结果应优先推送到 Feishu，同时保留本地落盘备份。
- 排障任务需要尽快止损，优先给可用方案，避免长时间无上限折腾。

## Critical Lessons
- 🔴 文件删除必须确认 — 安全红线
- 🔴 关键事项必须先请示 — 工作规范
- 🔴 已具备的技能直接使用，不要重复造轮子

L2 - 每日记忆（工作记忆）

特点：

  • 📅 时效性：按日期组织，7 天后归档
  • 📝 详细记录：完整工作日志、对话记录
  • 🔄 高频更新：每次会话都可能写入
  • 📦 中等体积：每篇 10-100KB

内容示例：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
# 2026-03-11 晚间工作总结

### 博客创作进展
**完成文章**：
1. ✅ P1 文章《OpenClaw Agent 工具调用最佳实践》（12KB）
2. ✅ P1 文章《OpenClaw Agent 安全治理指南》（12KB）
3. ✅ 重写 RAG 文章（21KB，11 个架构图）

### 重要规则（多次违反后确认）
**博客发布规则**：
1. ✅ 发布日期不能是未来时间（≤ 今天）
2. ✅ 同一天只能发布 1 篇文章

### 明日计划
1. 继续 P1 文章创作（6 篇待写）
2. 冲击 100 篇博客目标

L3 - 项目记忆（上下文记忆）

特点：

  • 📁 项目隔离：每个项目独立记忆空间
  • 🔗 双向同步：Obsidian ↔ OpenClaw
  • 🎯 场景化：特定项目的专业上下文
  • 📚 结构化：按项目目录组织

内容示例：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
# P1_CrystalForge/memory.md

## 项目状态
- 当前版本：v2.2.0
- 测试覆盖率：83% (目标 95%)
- 下次发布：2026-03-15

## 技术栈
- 后端：Spring Boot 3.2 + Java 17
- 前端：Vue 3.4 + Vite 5.1
- 数据库：MySQL 8.0

## 关键决策
- 采用 BCrypt 强度 8（非 10）优化登录性能
- 前端 Docker 构建使用两阶段构建

2.3 记忆流转机制

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
sequenceDiagram
    participant User as 用户
    participant Agent as Agent
    participant L1 as L1 全局记忆
    participant L2 as L2 每日记忆
    participant L3 as L3 项目记忆
    
    User->>Agent: 请求（带上下文）
    
    Note over Agent: 记忆检索阶段
    Agent->>L1: 读取核心事实/偏好
    Agent->>L2: 读取近期工作日志
    Agent->>L3: 读取项目上下文
    
    Note over Agent: 记忆融合阶段
    Agent->>Agent: 合并三层记忆
    Agent->>Agent: 过滤过期信息
    Agent->>Agent: 构建完整上下文
    
    Note over Agent: 响应生成阶段
    Agent->>Agent: 基于上下文生成响应
    Agent->>User: 返回个性化响应
    
    Note over Agent: 记忆更新阶段
    Agent->>L2: 写入今日工作日志
    Agent->>L3: 更新项目状态
    Agent->>L1: 提炼长期记忆（可选）

三、核心实现

3.1 记忆检索引擎

3.1.1 检索策略

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
class MemoryRetrievalEngine:
    """记忆检索引擎"""
    
    def __init__(self, workspace_path: str):
        self.workspace = Path(workspace_path)
        self.memory_index = {}
        
    def retrieve(self, query: str, context: dict) -> MemoryContext:
        """
        检索记忆
        
        Args:
            query: 用户查询
            context: 当前上下文（会话类型、项目等）
            
        Returns:
            MemoryContext: 融合后的记忆上下文
        """
        # 1. 确定检索范围
        scope = self._determine_scope(context)
        
        # 2. 并行检索三层记忆
        l1_memory = self._retrieve_l1(scope) if scope.l1 else None
        l2_memory = self._retrieve_l2(scope) if scope.l2 else None
        l3_memory = self._retrieve_l3(scope) if scope.l3 else None
        
        # 3. 语义搜索（可选）
        if context.get('semantic_search'):
            semantic_results = self._semantic_search(query)
            l2_memory = self._merge_semantic(l2_memory, semantic_results)
        
        # 4. 融合记忆
        fused = self._fuse_memories(l1_memory, l2_memory, l3_memory)
        
        # 5. 应用遗忘策略
        filtered = self._apply_forgetting(fused, context)
        
        return filtered
    
    def _determine_scope(self, context: dict) -> MemoryScope:
        """确定记忆检索范围"""
        chat_type = context.get('chat_type', 'direct')
        project = context.get('project')
        
        # 群聊不加载全局记忆（安全隔离）
        if chat_type == 'group':
            return MemoryScope(l1=False, l2=True, l3=project is not None)
        
        # 主会话加载全部
        return MemoryScope(l1=True, l2=True, l3=True)

3.1.2 语义搜索集成

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
def _semantic_search(self, query: str, top_k: int = 5) -> List[MemorySnippet]:
    """语义搜索记忆片段"""
    
    # 1. 向量化查询
    query_embedding = self.embedding_model.encode(query)
    
    # 2. 搜索记忆索引
    results = self.memory_index.search(
        query_embedding, 
        top_k=top_k,
        threshold=0.7  # 相似度阈值
    )
    
    # 3. 过滤过期记忆
    valid_results = [
        r for r in results 
        if not self._is_expired(r.metadata)
    ]
    
    # 4. 排序并返回
    return sorted(valid_results, key=lambda x: x.score, reverse=True)

3.2 记忆更新机制

3.2.1 自动写入

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
class MemoryWriter:
    """记忆写入器"""
    
    def __init__(self, workspace_path: str):
        self.workspace = Path(workspace_path)
        self.today = datetime.now().strftime('%Y-%m-%d')
        
    def write_session_summary(self, session: Session) -> None:
        """写入会话总结到每日记忆"""
        
        memory_file = self.workspace / 'memory' / f'{self.today}.md'
        
        # 1. 读取现有内容
        existing = self._read_memory(memory_file)
        
        # 2. 生成新条目
        new_entry = self._generate_entry(session)
        
        # 3. 合并内容
        updated = self._merge_entries(existing, new_entry)
        
        # 4. 写回文件
        self._write_memory(memory_file, updated)
        
    def _generate_entry(self, session: Session) -> str:
        """生成记忆条目"""
        
        return f"""
### {session.start_time.strftime('%H:%M')} - {session.topic}

**完成工作**：
{self._format_tasks(session.completed_tasks)}

**关键决策**：
{self._format_decisions(session.decisions)}

**待办事项**：
{self._format_todos(session.pending_tasks)}
"""

3.2.2 长期记忆提炼

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
def extract_long_term_memory(self, daily_files: List[Path]) -> List[MemoryItem]:
    """
    从每日记忆中提炼长期记忆
    
    规则：
    1. 重复出现 3 次+ 的主题 → 重要模式
    2. 用户明确标记"记住" → 长期记忆
    3. 项目里程碑事件 → 长期记忆
    4. 错误教训 → 长期记忆
    """
    
    candidates = []
    
    for file in daily_files:
        content = file.read_text()
        
        # 提取关键事件
        events = self._extract_events(content)
        candidates.extend(events)
        
        # 提取用户指令
        commands = self._extract_user_commands(content)
        candidates.extend(commands)
        
    # 聚类相似事件
    clusters = self._cluster_events(candidates)
    
    # 筛选高价值记忆
    long_term = [
        cluster for cluster in clusters
        if cluster.frequency >= 3 or cluster.importance_score > 0.8
    ]
    
    return long_term

3.3 记忆遗忘策略

3.3.1 时间衰减

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
def apply_time_decay(self, memories: List[Memory]) -> List[Memory]:
    """
    应用时间衰减
    
    衰减公式：
    score = base_score * e^(-λ * days_old)
    
    λ = 0.1 (半衰期约 7 天)
    """
    
    lambda_decay = 0.1
    today = datetime.now()
    
    for memory in memories:
        days_old = (today - memory.created_at).days
        decay_factor = math.exp(-lambda_decay * days_old)
        memory.current_score = memory.base_score * decay_factor
        
    return memories

3.3.2 自动归档

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
def archive_old_memories(self, days_threshold: int = 7) -> None:
    """归档旧记忆"""
    
    memory_dir = self.workspace / 'memory'
    archive_dir = self.workspace / 'memory' / 'archive'
    archive_dir.mkdir(exist_ok=True)
    
    cutoff_date = datetime.now() - timedelta(days=days_threshold)
    
    for file in memory_dir.glob('*.md'):
        file_date = self._parse_date(file.stem)
        
        if file_date < cutoff_date:
            # 移动到归档目录
            archive_path = archive_dir / file.name
            shutil.move(str(file), str(archive_path))
            
            # 记录归档日志
            self._log_archive(file, archive_path)

四、实战案例

4.1 案例 #1：个性化服务

场景

用户早上发送消息：”早上好”

无记忆系统

1
Agent: 早上好！有什么我可以帮你的吗？

有记忆系统

1
2
3
4
5
6
7
8
Agent: 早上好，John！☀️

新的一天开始了。根据昨晚的记录：
- 你计划今天继续 P1 博客创作（还有 6 篇待写）
- CrystalForge 测试需要修复（83% → 95%）
- 金刚最近体力很好，可以考虑周末带它去越野跑

今天想优先处理哪个任务？

实现原理：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
def generate_greeting(self, user_id: str) -> str:
    # 1. 读取用户信息
    user_info = self.memory.retrieve_l1(user_id)
    name = user_info.get('name', '用户')
    
    # 2. 读取昨日待办
    yesterday_todos = self.memory.retrieve_l2(
        user_id, 
        date='yesterday',
        filter='pending_tasks'
    )
    
    # 3. 读取宠物信息（个性化细节）
    pet_info = self.memory.retrieve_l1(user_id, key='pet')
    
    # 4. 生成个性化问候
    greeting = f"早上好，{name}！☀️\n\n"
    
    if yesterday_todos:
        greeting += "根据昨晚的记录：\n"
        for todo in yesterday_todos[:3]:  # 最多显示 3 个
            greeting += f"- {todo}\n"
    
    if pet_info:
        greeting += f"\n{pet_info.get('name')}最近怎么样？\n"
    
    return greeting

4.2 案例 #2：项目上下文继承

场景

用户切换项目讨论：

1
2
3
用户：CrystalForge 的测试覆盖率现在多少？
（5 分钟后）
用户：那 TrailSync 呢？

记忆系统处理

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
def handle_project_switch(self, user_id: str, new_project: str):
    # 1. 保存当前项目上下文
    current_project = self.context.get('current_project')
    if current_project:
        self.memory.save_context(current_project)
    
    # 2. 加载新项目上下文
    self.context['current_project'] = new_project
    project_memory = self.memory.load_project(new_project)
    
    # 3. 更新 Agent 认知
    self.agent.update_context({
        'project': new_project,
        'version': project_memory.version,
        'status': project_memory.status,
        'key_metrics': project_memory.metrics
    })
    
    # 4. 生成上下文感知的响应
    return self._generate_project_response(new_project, project_memory)

响应示例：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
关于 TrailSync 项目：

**当前状态**：
- 版本：v1.0.0
- 测试覆盖率：78%（目标 90%）
- 待修复问题：12 个

**最近进展**（来自 memory/2026-03-10.md）：
- ✅ 前端 H5 页面完成
- ✅ 后端 Socket.io 集成
- ✅ K8s 部署配置完成

**下一步建议**：
1. 修复剩余的 12 个测试用例
2. 性能优化（当前响应时间 450ms，目标 < 200ms）
3. 准备 v1.1.0 发布

需要我详细展开哪个部分？

4.3 案例 #3：错误教训记忆

场景

用户再次执行曾失败的操作：

1
用户：帮我删除那个测试文件

记忆系统干预

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
def check_safety_rules(self, action: str, context: dict) -> SafetyCheck:
    # 1. 检索相关安全规则
    safety_rules = self.memory.retrieve_l1(
        filter='safety_rules',
        tags=['file_deletion', 'confirmation_required']
    )
    
    # 2. 检查历史教训
    past_incidents = self.memory.semantic_search(
        query=f"delete file {context['file']}",
        filter='incidents'
    )
    
    # 3. 生成安全检查
    check = SafetyCheck(
        action=action,
        risk_level=self._assess_risk(action, safety_rules),
        requires_confirmation=self._needs_confirmation(action, safety_rules),
        historical_context=past_incidents
    )
    
    return check

响应示例：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
⚠️ **安全提醒**

根据 MEMORY.md 中的安全规则：
> 🔴 文件删除必须确认 — 安全红线（2026-03-03 再次违反）
> 任何删除操作前必须找用户确认

**历史教训**（来自 memory/2026-03-03.md）：
- 2026-03-03：目录结构调整时误删 P6 架构图和 P3 技能文件
- 原因：没有先用 git ls-tree 检查，没有备份
- 后果：严重安全事故，文件丢失

**建议操作**：
1. 先确认文件路径：`/path/to/test/file`
2. 备份文件：`cp file file.backup`
3. 使用可恢复删除：`trash file`（而非 `rm file`）
4. 再次确认：真的要删除吗？

请确认是否继续？

4.4 案例 #4：群聊记忆隔离

场景

用户在群聊中提到个人信息：

1
2
群聊 - 技术讨论组:
John: 我昨晚让 OpenClaw 帮我优化了 CrystalForge 的登录性能

记忆系统处理

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
def handle_group_message(self, message: Message, chat_id: str):
    # 1. 检测聊天类型
    chat_type = self._get_chat_type(chat_id)
    
    # 2. 群聊不加载全局记忆（安全隔离）
    if chat_type == 'group':
        memory_scope = MemoryScope(
            l1=False,  # ❌ 不加载个人信息
            l2=False,  # ❌ 不加载工作日志
            l3=True    # ✅ 仅加载项目公开信息
        )
    else:
        memory_scope = MemoryScope(l1=True, l2=True, l3=True)
    
    # 3. 生成响应（不泄露隐私）
    response = self.agent.generate_response(
        message,
        memory_scope=memory_scope
    )
    
    return response

响应对比：

场景 错误做法 ❌ 正确做法 ✅
群聊提到 CrystalForge “John，你昨晚让我优化了登录 API，从 265ms 降到 17ms…” “CrystalForge 的登录性能优化确实效果显著，从 265ms 降到 17ms…”
群聊提到宠物 “John，金刚最近怎么样？” “听说你养了宠物，最近怎么样？”
群聊提到工作习惯 “John 喜欢在 Feishu 收结果” “有些人喜欢直接收到结果，有些人喜欢去服务器查看…”

五、性能优化

5.1 检索性能

5.1.1 缓存策略

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
class MemoryCache:
    """记忆缓存层"""
    
    def __init__(self, ttl_seconds: int = 300):
        self.cache = {}
        self.ttl = ttl_seconds
        
    def get(self, key: str) -> Optional[MemoryContext]:
        """获取缓存"""
        if key in self.cache:
            entry = self.cache[key]
            if time.time() - entry.timestamp < self.ttl:
                return entry.data
            else:
                del self.cache[key]
        return None
    
    def set(self, key: str, value: MemoryContext) -> None:
        """设置缓存"""
        self.cache[key] = CacheEntry(
            data=value,
            timestamp=time.time()
        )

5.1.2 增量加载

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
def retrieve_with_lazy_loading(self, query: str) -> MemoryContext:
    """增量加载记忆"""
    
    # 1. 先加载核心记忆（L1）
    context = MemoryContext()
    context.l1 = self._retrieve_l1_fast()
    
    # 2. 异步加载每日记忆（L2）
    future_l2 = self.executor.submit(self._retrieve_l2, query)
    
    # 3. 异步加载项目记忆（L3）
    future_l3 = self.executor.submit(self._retrieve_l3, query)
    
    # 4. 等待 L2/L3 完成（超时 100ms）
    try:
        context.l2 = future_l2.result(timeout=0.1)
    except TimeoutError:
        context.l2 = MemorySnippet(partial=True)
    
    try:
        context.l3 = future_l3.result(timeout=0.1)
    except TimeoutError:
        context.l3 = MemorySnippet(partial=True)
    
    return context

5.2 存储优化

5.2.1 压缩策略

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
def compress_old_memories(self, days_threshold: int = 30) -> None:
    """压缩旧记忆文件"""
    
    memory_dir = self.workspace / 'memory'
    
    for file in memory_dir.glob('*.md'):
        file_date = self._parse_date(file.stem)
        days_old = (datetime.now() - file_date).days
        
        if days_old > days_threshold:
            content = file.read_text()
            
            # 1. 移除冗余空白
            compressed = self._remove_redundant_whitespace(content)
            
            # 2. 压缩重复模式
            compressed = self._compress_patterns(compressed)
            
            # 3. 写回文件
            file.write_text(compressed)
            
            # 4. 记录压缩率
            original_size = len(content)
            compressed_size = len(compressed)
            ratio = (1 - compressed_size / original_size) * 100
            
            logger.info(f"压缩 {file.name}: {ratio:.1f}%")

5.2.2 分片存储

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
class ShardedMemoryStorage:
    """分片记忆存储"""
    
    def __init__(self, num_shards: int = 10):
        self.num_shards = num_shards
        
    def _get_shard(self, memory_id: str) -> int:
        """计算记忆所属分片"""
        hash_value = hashlib.md5(memory_id.encode()).hexdigest()
        return int(hash_value[:8], 16) % self.num_shards
    
    def store(self, memory: Memory) -> None:
        """存储记忆到对应分片"""
        shard_id = self._get_shard(memory.id)
        shard_path = self.base_path / f'shard_{shard_id}'
        
        # 写入分片文件
        (shard_path / f'{memory.id}.md').write_text(memory.content)

5.3 查询优化

5.3.1 索引构建

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
class MemoryIndex:
    """记忆索引"""
    
    def __init__(self):
        self.inverted_index = defaultdict(set)  # term -> memory_ids
        self.embedding_index = None  # 向量索引
        
    def build_inverted_index(self, memories: List[Memory]) -> None:
        """构建倒排索引"""
        
        for memory in memories:
            tokens = self._tokenize(memory.content)
            
            for token in tokens:
                self.inverted_index[token].add(memory.id)
    
    def build_embedding_index(self, memories: List[Memory]) -> None:
        """构建向量索引"""
        
        embeddings = []
        for memory in memories:
            embedding = self.embedding_model.encode(memory.content)
            embeddings.append(embedding)
        
        # 使用 FAISS 构建索引
        self.embedding_index = faiss.IndexFlatIP(len(embeddings[0]))
        self.embedding_index.add(np.array(embeddings))

5.3.2 查询重写

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
def rewrite_query(self, query: str, context: MemoryContext) -> str:
    """查询重写"""
    
    # 1. 提取实体
    entities = self._extract_entities(query)
    
    # 2. 扩展同义词
    expanded = self._expand_synonyms(query)
    
    # 3. 添加上下文过滤
    if context.project:
        expanded += f" project:{context.project}"
    
    if context.date_range:
        expanded += f" date:{context.date_range}"
    
    return expanded

六、安全与隐私

6.1 记忆隔离

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
class MemoryIsolation:
    """记忆隔离机制"""
    
    def check_access(self, session: Session, memory_id: str) -> bool:
        """检查访问权限"""
        
        memory = self.storage.get(memory_id)
        
        # 1. 检查会话类型
        if session.chat_type == 'group':
            # 群聊只能访问公开记忆
            return memory.visibility == 'public'
        
        # 2. 检查用户所有权
        if memory.owner_id != session.user_id:
            return memory.shared_with and session.user_id in memory.shared_with
        
        # 3. 检查记忆敏感度
        if memory.sensitivity == 'private':
            # 私密记忆仅限主会话访问
            return session.is_main_session
        
        return True

6.2 加密存储

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
class EncryptedMemoryStorage:
    """加密记忆存储"""
    
    def __init__(self, encryption_key: str):
        self.cipher = Fernet(encryption_key.encode())
        
    def store(self, memory: Memory) -> None:
        """加密存储"""
        
        # 1. 序列化
        data = json.dumps(memory.to_dict()).encode()
        
        # 2. 加密
        encrypted_data = self.cipher.encrypt(data)
        
        # 3. 存储
        self.storage.write(memory.id, encrypted_data)
    
    def retrieve(self, memory_id: str) -> Memory:
        """解密读取"""
        
        # 1. 读取
        encrypted_data = self.storage.read(memory_id)
        
        # 2. 解密
        data = self.cipher.decrypt(encrypted_data)
        
        # 3. 反序列化
        memory_dict = json.loads(data.decode())
        
        return Memory.from_dict(memory_dict)

6.3 审计日志

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
def log_memory_access(self, session: Session, memory_id: str, action: str):
    """记录记忆访问日志"""
    
    log_entry = {
        'timestamp': datetime.now().isoformat(),
        'user_id': session.user_id,
        'session_id': session.id,
        'memory_id': memory_id,
        'action': action,  # read/write/delete
        'chat_type': session.chat_type,
        'ip_address': session.ip_address
    }
    
    # 写入审计日志
    self.audit_log.append(log_entry)
    
    # 异常访问告警
    if self._is_anomalous(log_entry):
        self._send_alert(log_entry)

七、踩坑记录

7.1 问题 #1：群聊记忆泄露

现象

用户在群聊中提到个人信息，Agent 在回复中泄露了用户的私密记忆。

根因

1
2
3
4
5
# 错误代码 ❌
def generate_response(self, message: Message):
    # 没有检查聊天类型，直接加载所有记忆
    context = self.memory.retrieve_all()
    return self._generate(message, context)

解决方案

1
2
3
4
5
6
7
8
9
10
11
12
# 正确代码 ✅
def generate_response(self, message: Message):
    chat_type = self._get_chat_type(message.chat_id)
    
    if chat_type == 'group':
        # 群聊仅加载公开项目记忆
        context = self.memory.retrieve(scope=MemoryScope(l1=False, l2=False, l3=True))
    else:
        # 私聊加载全部记忆
        context = self.memory.retrieve(scope=MemoryScope(l1=True, l2=True, l3=True))
    
    return self._generate(message, context)

7.2 问题 #2：记忆文件膨胀

现象

memory/2026-03-03.md 文件达到 500KB，检索速度变慢。

根因

  • 每次会话都追加内容，没有清理
  • 重复记录相同信息
  • 没有归档旧记忆

解决方案

1
2
3
4
5
6
7
8
9
10
11
12
13
14
def maintain_memory_files(self):
    """记忆文件维护"""
    
    # 1. 去重
    self._deduplicate_entries()
    
    # 2. 压缩
    self._compress_old_entries(days_threshold=7)
    
    # 3. 归档
    self._archive_old_files(days_threshold=30)
    
    # 4. 提炼长期记忆
    self._extract_long_term_memories()

7.3 问题 #3：上下文窗口溢出

现象

模型返回错误：Request too large. Maximum context length is 32768 tokens.

根因

三层记忆全部加载，超过模型上下文窗口限制。

解决方案

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
def fit_context_window(self, context: MemoryContext, max_tokens: int = 28000):
    """适配上下文窗口"""
    
    # 1. 计算当前 token 数
    current_tokens = self._count_tokens(context)
    
    # 2. 如果超出，按优先级裁剪
    if current_tokens > max_tokens:
        # 优先保留 L1（核心事实）
        # 其次保留 L2（近期记忆）
        # 最后保留 L3（项目上下文）
        
        while current_tokens > max_tokens:
            if context.l3:
                context.l3 = self._trim_l3(context.l3)
            elif context.l2:
                context.l2 = self._trim_l2(context.l2)
            else:
                break
            
            current_tokens = self._count_tokens(context)
    
    return context

八、最佳实践

8.1 记忆设计原则

原则 说明 示例
分层存储 按价值/时效分层 L1 永久/L2 7 天/L3 项目周期
按需加载 仅加载必要记忆 群聊不加载 L1
及时更新 会话结束立即写入 write_session_summary()
定期维护 去重/压缩/归档 每日凌晨维护任务
安全隔离 隐私数据保护 加密存储 + 访问控制

8.2 写入策略

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
## ✅ 应该写入记忆的内容

1. **用户偏好** - "喜欢在 Feishu 收结果"
2. **重要事实** - "爱犬叫金刚，跑过 15 公里"
3. **工作进展** - "完成 5 篇 P1 文章"
4. **关键决策** - "采用单 PVC 方案"
5. **错误教训** - "文件删除必须确认"
6. **项目状态** - "CrystalForge v2.2.0, 测试 83%"

## ❌ 不应该写入记忆的内容

1. **临时对话** - "好的"、"明白了"
2. **敏感信息** - 密码、密钥、身份证号
3. **冗余信息** - 已有记忆重复记录
4. **过期信息** - 已完成的临时任务
5. **无关细节** - 与核心目标无关的闲聊

8.3 检索策略

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
# 高效检索模式
def retrieve_memory(self, query: str, context: dict):
    # 1. 先查缓存
    cached = self.cache.get(query)
    if cached:
        return cached
    
    # 2. 确定范围
    scope = self._determine_scope(context)
    
    # 3. 并行检索
    results = await asyncio.gather(
        self._retrieve_l1(scope),
        self._retrieve_l2(scope),
        self._retrieve_l3(scope)
    )
    
    # 4. 融合排序
    fused = self._fuse_and_rank(results)
    
    # 5. 缓存结果
    self.cache.set(query, fused)
    
    return fused

九、未来演进

9.1 短期优化（1-3 个月）

  • 向量数据库集成 - 使用 Chroma/Weaviate 提升语义搜索
  • 记忆图谱 - 构建记忆间的关联网络
  • 主动记忆 - Agent 主动询问”这个需要记住吗？”
  • 记忆可视化 - Grafana 仪表盘展示记忆使用

9.2 中期规划（3-6 个月）

  • 跨会话记忆共享 - 多个 Agent 实例共享记忆
  • 记忆版本控制 - Git 管理记忆变更历史
  • 记忆质量评分 - 自动评估记忆价值
  • 遗忘曲线优化 - 基于使用频率动态调整

9.3 长期愿景（6-12 个月）

  • 分布式记忆 - 多节点记忆同步
  • 记忆压缩模型 - 训练专用压缩模型
  • 记忆迁移学习 - 跨用户记忆模式迁移
  • 记忆即服务 - 对外提供记忆 API

十、参考资料

10.1 理论基础

10.2 技术实现

10.3 相关工具

作者：John
职位：高级技术架构师
日期：2026-03-08
版本：v1.0

本文基于 OpenClaw 真实项目经验编写，三层记忆架构已在生产环境稳定运行。记忆系统是 AI Agent 实现个性化的核心，值得深入设计和持续优化。