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

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

## 6.2.1 记忆系统的读写对称性

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

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

1. **主动性**：Agent 在对话过程中识别值得记忆的信息，自主决定何时写入
2. **原子性**：记忆写入操作应该是原子的，避免部分更新导致的不一致
3. **可审计性**：每次写入都应该留下痕迹，支持版本回滚
4. **并发安全**：在多会话并发场景下，防止记忆冲突

## 6.2.2 记忆文件的 Frontmatter 结构

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

```yaml
---
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 应该在以下时刻考虑写入记忆：

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

* 用户明确要求“记住这个”或“保存这个信息”
* 系统检测到错误或失败，需要记录教训
* 完成一个重要的里程碑或任务

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

* 首次遇到新的用户偏好（如代码风格、工作习惯）
* 发现系统性的性能问题或优化机会
* 完成一个新的项目或学习领域

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

* 每个会话结束后的自动总结
* 长对话中的中间检查点

写入决策的伪代码：

```python
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**：

* 用户对智能体建议的反馈
* 更新频率：每次用户评价时更新
* 示例：

  ```yaml
  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**：

* 可复用的代码片段、命令、文档链接
* 更新频率：中等（发现新的有用模式时添加）
* 示例：

  ```yaml
  ## 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 实现高效的读取，支持 **混合检索**：

**关键词检索阶段**：

```python
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
```

**向量检索阶段**：

```python
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
    return [mid for mid, _ in sorted(scores, reverse=True)[:5]]
```

**混合排序**：

```python
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 写入冲突解决

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

**版本号机制**：

```python
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
        return merge_versions(current, 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 策略**： 当版本冲突时，比较三个版本的差异（基础版本、当前版本、新版本），智能合并：

```python
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 记忆写入工具的接口

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

```python
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
```

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

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


---

# Agent Instructions: Querying This Documentation

If you need additional information that is not directly available in this page, you can query the documentation dynamically by asking a question.

Perform an HTTP GET request on the current page URL with the `ask` query parameter:

```
GET https://yeasy.gitbook.io/harness_engineering_guide/di-er-bu-fen-harness-he-xin-zi-xi-tong/06_memory/6.2_writable_memory.md?ask=<question>
```

The question should be specific, self-contained, and written in natural language.
The response will contain a direct answer to the question and relevant excerpts and sources from the documentation.

Use this mechanism when the answer is not explicitly present in the current page, you need clarification or additional context, or you want to retrieve related documentation sections.