6.2 可写入式智能体记忆构建

为了赋予智能体真正的学习和适应能力，需要构建完整的可写入式记忆系统。本节涵盖记忆的读写对称性设计、Frontmatter 元数据结构、写入时机策略以及 OpenClaw 的混合检索机制，最后讨论并发环境下的冲突解决方案。

6.2.1 记忆系统的读写对称性

传统的智能体记忆系统往往是 单向的：Agent 读取预设的记忆文件，但无法主动更新。这限制了智能体的自适应能力。真正强大的智能体系统应该支持 可写入式记忆——Agent 作为主动的参与者，既读取也创建和更新记忆。

可写入式记忆的关键特性：

1. 主动性：Agent 在对话过程中识别值得记忆的信息，自主决定何时写入

2. 原子性：记忆写入操作应该是原子的，避免部分更新导致的不一致

3. 可审计性：每次写入都应该留下痕迹，支持版本回滚

4. 并发安全：在多会话并发场景下，防止记忆冲突

6.2.2 记忆文件的 Frontmatter 结构

为了支持可写入式记忆，我们采用 Markdown Frontmatter 结构，将元数据与内容分离：

---
type: episodic            # 记忆类型:user/feedback/project/reference/episodic
version: 3                # 版本号,每次更新递增
last_modified: 2024-01-15T14:30:00Z
modified_by: session_id   # 修改者的会话标识
confidence: 0.85          # 信息可信度(0-1),自动学习可降低
expiry: 2024-06-15        # 过期时间,可选
tags: [user_preference, performance]
---

# 记忆内容

实际的内容从这里开始...

各字段的含义：

• type：记忆的语义类别

  • user：用户档案信息（不改变的属性）

  • feedback：用户反馈和评估

  • project：项目或任务上下文

  • reference：可复用的代码、命令、模式

  • episodic：具体事件的记录

• version：支持版本追踪，允许回滚

• last_modified：ISO 8601 时间戳

• modified_by：记录是哪个会话或智能体实例修改的

• confidence：由智能体自评的信息可信度，用于后续搜索排序

• expiry：自动清理过期记忆

• tags：用于分类和检索的标签集

6.2.3 记忆写入时机策略

Agent 应该在以下时刻考虑写入记忆：

显式写入信号 （高优先级）：

• 用户明确要求“记住这个”或“保存这个信息”

• 系统检测到错误或失败，需要记录教训

• 完成一个重要的里程碑或任务

隐式写入信号 （中优先级）：

• 首次遇到新的用户偏好（如代码风格、工作习惯）

• 发现系统性的性能问题或优化机会

• 完成一个新的项目或学习领域

实时写入信号 （低优先级）：

• 每个会话结束后的自动总结

• 长对话中的中间检查点

写入决策的伪代码：

async def decide_write_memory(session_context, content):
    """判断是否应该写入记忆"""

    signals = []

    # 检查显式信号
    if "remember" in content.lower() or "记住" in content:
        signals.append(("explicit_user", 0.95))

    # 检查隐式信号
    if is_new_user_preference(content, session_context):
        signals.append(("new_preference", 0.7))

    if is_error_or_lesson_learned(content):
        signals.append(("learning", 0.8))

    if is_milestone_completion(content):
        signals.append(("milestone", 0.85))

    # 综合评分
    max_score = max([score for _, score in signals]) if signals else 0

    if max_score > 0.6:  # 阈值
        return True, dict(signals)
    return False, {}

6.2.4 Claude Code 的记忆类型分类

Claude Code 将记忆划分为四个类型，各有不同的更新策略：

User Profiles：

• 存储用户的基本属性、技能、工作风格

• 更新频率：低（通常用户主动告知）

• 示例：

  User: Alex
  - Programming level: Intermediate Python, Advanced JS
  - Preferred style: Concise, comments only for complex logic
  - Work context: Startup, rapid iteration preferred

Feedback Records：

• 用户对智能体建议的反馈

• 更新频率：每次用户评价时更新

• 示例：

  Feedback [2024-01-14]:
  - Rejected: Suggested async/await pattern
    Reason: Prefers synchronous code in this codebase
  - Approved: Refactoring suggestion for API layer

Project Context：

• 当前项目的架构、进度、关键文件

• 更新频率：每个功能完成后更新

• 示例：

  Project: WebApp v2.0
  Last activity: 2024-01-15
  - Modules: auth (done), api (in progress), ui (planned)
  - Key files: src/main.py, tests/
  - Architecture decision: Monolithic with plugins

Reference Materials：

• 可复用的代码片段、命令、文档链接

• 更新频率：中等（发现新的有用模式时添加）

• 示例：

  ## Useful Commands
  - Test: pytest -v --cov
  - Deploy: make deploy-prod
  
  ## Code Patterns
  - Factory pattern for data access
  - Decorator for logging

6.2.5 OpenClaw 的 memory_search 混合检索

OpenClaw 的可写入记忆通过 memory_search 实现高效的读取，支持 混合检索：

关键词检索阶段：

def keyword_search(query: str, memory_base: str) -> List[str]:
    """在 MEMORY.md 和日志中执行关键词匹配"""
    results = []
    keywords = query.lower().split()

    for memory_file in get_memory_files():
        content = read_file(memory_file)
        matches = [line for line in content.split('\n')
                   if any(kw in line.lower() for kw in keywords)]
        results.extend(matches)

    return results

向量检索阶段：

def semantic_search(query: str, embedding_index) -> List[str]:
    """使用向量相似度查找相关内容"""
    query_embedding = embed(query)

    scores = []
    for memory_id, stored_embedding in embedding_index.items():
        similarity = cosine_similarity(query_embedding, stored_embedding)
        scores.append((memory_id, similarity))

    # 返回相似度最高的 Top-K（按相似度排序，而非默认的按 memory_id 排序）
    return [mid for mid, _ in sorted(scores, key=lambda x: x[1], reverse=True)[:5]]

混合排序：

def hybrid_search(query: str, memory_base: str, embedding_index) -> List[str]:
    """组合关键词和向量搜索结果"""
    kw_results = keyword_search(query, memory_base)
    sem_results = semantic_search(query, embedding_index)

    # 融合:先返回同时匹配两者的结果,再返回单一匹配
    combined = set(kw_results) & set(sem_results)
    rest = (set(kw_results) | set(sem_results)) - combined

    return list(combined) + list(rest)

这种混合检索的优势在于：

• 精确性：关键词搜索捕获精确的事实

• 泛化性：向量搜索发现语义相似但词汇不同的信息

• 效率：分阶段执行避免全向量扫描

6.2.6 写入冲突解决

在多会话并发环境下，多个智能体实例可能同时修改同一个记忆文件。冲突解决策略：

版本号机制：

def write_memory_atomic(memory_id: str, new_content: str,
                        expected_version: int) -> bool:
    """原子式写入,基于版本号"""
    current = read_memory(memory_id)

    if current['version'] != expected_version:
        # 版本冲突,进行 3-way merge（基础版本从版本历史中取写入方所基于的那一版）
        base = read_memory_version(memory_id, expected_version)
        return merge_versions(base['content'], current['content'], new_content)

    new_version = {
        'version': expected_version + 1,
        'last_modified': now(),
        'modified_by': current_session_id(),
        'content': new_content
    }

    return atomic_write(memory_id, new_version)

3-way Merge 策略： 当版本冲突时，比较三个版本的差异（基础版本、当前版本、新版本），智能合并：

def merge_versions(base, current, new):
    """3-way merge 用于记忆文件"""
    # 简化示例:按行级别的 diff
    base_lines = base.split('\n')
    curr_lines = current.split('\n')
    new_lines = new.split('\n')

    merged = []
    for i, (b, c, n) in enumerate(zip(base_lines, curr_lines, new_lines)):
        if c == n:  # 无冲突
            merged.append(c)
        elif b == c:  # 只有新版本改动
            merged.append(n)
        elif b == n:  # 只有当前版本改动
            merged.append(c)
        else:  # 双向改动,标记为冲突
            merged.append(f"<<< CONFLICT >>>\n{c}\n---\n{n}\n>>>")

    return '\n'.join(merged)

6.2.7 记忆写入工具的接口

对智能体暴露的记忆写入接口应该简洁明确：

class MemoryWriter:
    """Agent 可调用的记忆写入工具"""

    async def save_user_profile(self, key: str, value: any) -> bool:
        """保存用户属性"""
        pass

    async def record_feedback(self, action: str, approved: bool,
                              reason: str) -> bool:
        """记录用户反馈"""
        pass

    async def update_project_context(self, updates: dict) -> bool:
        """更新项目信息"""
        pass

    async def add_reference(self, category: str, content: str,
                            tags: List[str]) -> bool:
        """添加参考资料"""
        pass

    async def record_lesson(self, lesson: str, confidence: float) -> bool:
        """记录学到的教训"""
        pass

每次写入都应该触发异步验证和索引更新，确保后续的检索不会出现不一致的状态。

下一节将探讨如何从多个记忆源组装出完整的上下文，为智能体的决策提供支持。