> For the complete documentation index, see [llms.txt](https://yeasy.gitbook.io/harness_engineering_guide/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://yeasy.gitbook.io/harness_engineering_guide/di-er-bu-fen-harness-he-xin-zi-xi-tong/06_memory/6.2_writable_memory.md). # 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**： * 用户对智能体建议的反馈 * 更新频率：每次用户评价时更新 * 示例： ``` 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（按相似度排序，而非默认的按 memory_id 排序） return [mid for mid, _ in sorted(scores, key=lambda x: x[1], 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（基础版本从版本历史中取写入方所基于的那一版） 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 策略**： 当版本冲突时，比较三个版本的差异（基础版本、当前版本、新版本），智能合并： ```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 ``` 每次写入都应该触发异步验证和索引更新，确保后续的检索不会出现不一致的状态。 下一节将探讨如何从多个记忆源组装出完整的上下文，为智能体的决策提供支持。