1. 项目概述：当AI学会“回头看”与“向前看”

最近在折腾一个挺有意思的开源项目，叫OpenClaw。这名字听着就有点“爪牙”的犀利感，但它核心的魅力，不在于多锋利的攻击性，而在于一种更接近人类思考方式的“记忆”机制。我们常说，一个强大的AI智能体，不能像金鱼一样只有七秒记忆，也不能像硬盘一样只会机械存储。它需要的是在复杂的任务流中，既能清晰地记住自己从哪里来（上下文），又能灵活地规划自己要到哪里去（长期目标）。OpenClaw提出的“双源记忆系统”，正是为了解决这个核心问题。

简单来说，你可以把它想象成一位经验丰富的侦探。在侦破一桩复杂案件时，他手边会有一个案件日志本，实时记录每一条线索、每一次询问、每一个现场细节，这是他的“工作记忆”，确保推理的连贯性。同时，他脑海里还有一个经验档案库，里面分类存放着以往破获的各类案件模式、罪犯心理侧写、物证鉴定知识，这是他的“长期记忆”，用于提供策略和灵感。OpenClaw的双源记忆，就是试图在AI智能体中构建这样一套协同工作的“日志本”和“档案库”。

这套系统不是为了炫技，而是为了解决大模型应用中的几个实实在在的痛点：面对长对话或复杂任务时，模型会不会忘了最初的指令？在多步骤规划中，AI能否参考过去的成功或失败经验来优化当前决策？当需要调用外部工具或知识时，如何快速精准地定位相关信息？双源记忆系统通过分离记忆的“时效性”与“功能性”，给出了一个工程上非常优雅的解法。无论你是想构建一个能进行深度、连贯对话的聊天助手，还是一个能自主完成复杂工作流的AI智能体，理解这套记忆架构，都能让你在系统设计上事半功倍。

2. 双源记忆系统的核心架构拆解

OpenClaw的双源记忆系统，其精妙之处在于它不是简单地将记忆分成“短期”和“长期”，而是从数据来源和功能角色两个维度进行了正交设计。理解这个设计，是掌握其所有代码实现的前提。

2.1 记忆的两种源头：外部观察与内部思考

记忆从何而来？OpenClaw将其清晰地划分为两类：

1. 观察记忆：这是智能体通过“感官”从外部环境直接获取的信息。在代码中，这通常对应着任务执行过程中产生的客观输出。例如：

   • 执行一个Shell命令后，终端返回的标准输出和错误输出。
   • 调用一个API后，返回的JSON数据或状态码。
   • 读取一个文件后，得到的文件内容。
   • 浏览一个网页后，提取到的关键文本。

   观察记忆的核心特征是客观、原始、高保真。它忠实地记录了“世界发生了什么”，是后续一切推理和决策的基石。在系统中，这部分记忆通常被结构化地存储，并附带丰富的元数据，如时间戳、来源工具、执行状态等，便于后续检索和溯源。

2. 思考记忆：这是智能体内部认知过程的产物，是它对观察记忆进行加工、推理、总结和规划后的结果。例如：

   • 看到命令执行出错后，分析得出的可能原因。
   • 阅读多篇文档后，归纳出的核心知识点。
   • 为了完成一个子目标，自行拆解出的下一步行动步骤。
   • 对当前任务整体进展的自我评估和反思。

   思考记忆的核心特征是主观、抽象、高信息密度。它代表了智能体的“内部独白”和“思维链”，是将原始数据转化为知识和策略的关键环节。这部分记忆往往以更自然语言化、更结构化的笔记形式存在。

关键设计洞察：将“观察”与“思考”分离，是避免记忆污染、实现清晰思维流的关键。想象一下，如果把错误信息和你的分析猜测混在一起，后续检索时就会引入噪音。OpenClaw强制区分二者，确保了记忆库的“干净”和推理链条的“可解释性”。

2.2 记忆的两种功能：情景缓冲与长期知识库

基于这两种记忆源头，OpenClaw又通过功能划分，构建了两个核心存储组件：

1. 情景缓冲：这是一个容量有限、但存取速度极快的“工作台”。它的主要职责是保持当前任务上下文的连贯性。

   • 内容：它滚动存储最近若干轮的“观察记忆”和“思考记忆”。例如，最近5次工具调用的结果和智能体对其的思考。
   • 作用：当智能体需要决定下一步行动时，情景缓冲中的信息会被自动地、完整地作为上下文提供给大语言模型。这确保了智能体不会患上“短期失忆”，能牢牢记住对话刚刚发生了什么、上一步做了什么、结果如何。
   • 类比：就像你电脑上正在编辑的文档窗口和打开的参考网页，是你手头正在处理工作的直接上下文。

2. 长期记忆库：这是一个容量巨大、但检索需要成本的“档案室”。它的主要职责是存储跨任务的、重要的经验与知识。

   • 内容：它选择性地存储那些被认为具有长期价值的“思考记忆”，以及与之强相关的关键“观察记忆”。例如，一个复杂问题排查成功的完整心路历程和关键命令；或者从多个类似任务中抽象出的通用解决模式。
   • 作用：当智能体遇到新问题或需要战略规划时，可以通过检索的方式，从长期记忆库中寻找相关的历史经验来参考。这赋予了智能体“学习”和“积累经验”的能力。
   • 类比：就像你个人知识库中的笔记、项目总结报告，是你需要时主动去查阅的宝贵资产。

两者的协同流程可以概括为：智能体在任务中不断产生“观察”和“思考”，它们首先流入“情景缓冲”维持连贯性。同时，一个独立的“记忆整理”线程会异步地评估缓冲中的内容，将有长期价值的部分，经过摘要、提炼、打标签后，存入“长期记忆库”。当新任务触发时，长期记忆库通过向量相似度检索等方式，将相关记忆“激活”并注入当前的情景缓冲，从而影响当下的决策。

3. 核心模块的代码级实现解析

理解了架构，我们深入到代码层面，看看OpenClaw是如何用具体的类和函数将这些概念落地的。这里我们聚焦几个最核心的模块。

3.1 记忆单元的数据结构定义

一切记忆的基石是记忆单元。OpenClaw通常会定义一个基础的数据类，例如MemoryUnit。

from dataclasses import dataclass from datetime import datetime from enum import Enum from typing import Any, Optional class MemoryType(Enum): OBSERVATION = "observation" THOUGHT = "thought" @dataclass class MemoryUnit: id: str # 唯一标识符，如UUID type: MemoryType # 观察 or 思考 content: str # 记忆的具体内容 timestamp: datetime # 创建时间 metadata: dict[str, Any] # 元数据，如来源工具、状态、嵌入向量等 importance_score: float = 0.0 # 重要性评分，用于长期记忆筛选 # 可能还有关联ID，用于链接相关的观察和思考

这个简单的结构体承载了所有关键信息。metadata字段是个百宝箱，对于观察记忆，可能包含{“tool”: “shell”, “command”: “ls -la”, “exit_code”: 0, “embedding”: [0.1, 0.2, ...]}；对于思考记忆，可能包含{“reflection_type”: “lesson_learned”, “related_observation_ids”: [“id1”, “id2”]}。

3.2 情景缓冲区的滚动管理

情景缓冲区通常实现为一个有容量限制的队列。它的核心方法是add和get_context。

from collections import deque from typing import List class ContextBuffer: def __init__(self, max_size: int = 10): self.buffer = deque(maxlen=max_size) # 固定长度的双端队列 self.max_size = max_size def add(self, memory_unit: MemoryUnit): """向缓冲区添加一个记忆单元""" self.buffer.append(memory_unit) # 可能触发一些轻量级的处理，如计算重要性初值 def get_context(self) -> str: """将缓冲区中的所有记忆，格式化为LLM可理解的提示词上下文""" context_parts = [] for mem in self.buffer: # 根据记忆类型，添加不同的前缀标识，帮助LLM区分 if mem.type == MemoryType.OBSERVATION: prefix = f"[Observation from {mem.metadata.get('tool', 'unknown')}]:" else: prefix = "[My thought]:" context_parts.append(f"{prefix} {mem.content}") return "\n".join(context_parts) # 返回一个连贯的文本块 def clear(self): """在任务边界清空缓冲区""" self.buffer.clear()

这个get_context方法返回的字符串，就是最终会被拼接到LLM系统提示词后面的“近期历史”。它的格式设计直接影响LLM的理解效果，清晰的标识前缀至关重要。

3.3 长期记忆库的存储与检索

长期记忆库的实现更为复杂，涉及向量化、存储和检索。OpenClaw可能会集成像ChromaDB、Weaviate或简单的FAISS这类向量数据库。

# 假设使用一个简单的向量数据库客户端 import numpy as np from some_vector_db import VectorStore class LongTermMemory: def __init__(self, vector_store_path: str): self.vector_store = VectorStore(persist_path=vector_store_path) self.encoder = SentenceTransformer('all-MiniLM-L6-v2') # 嵌入模型 def store(self, memory_unit: MemoryUnit): """存储一个记忆单元到长期记忆""" # 1. 只有当重要性分数高于阈值，或标记为“知识”时才存储 if memory_unit.importance_score < 0.7 and not memory_unit.metadata.get('is_knowledge'): return # 2. 为内容生成向量嵌入 embedding = self.encoder.encode(memory_unit.content) memory_unit.metadata['embedding'] = embedding.tolist() # 3. 准备存储的文档 doc = { 'id': memory_unit.id, 'content': memory_unit.content, 'type': memory_unit.type.value, 'metadata': memory_unit.metadata, 'embedding': embedding } self.vector_store.add_documents([doc]) def retrieve(self, query: str, top_k: int = 3) -> List[MemoryUnit]: """根据查询检索相关记忆""" # 1. 将查询语句也向量化 query_embedding = self.encoder.encode(query) # 2. 在向量库中进行相似度搜索 results = self.vector_store.similarity_search_by_vector(query_embedding, k=top_k) # 3. 将结果转换回MemoryUnit对象 retrieved_memories = [] for res in results: mu = MemoryUnit( id=res['id'], type=MemoryType(res['type']), content=res['content'], timestamp=datetime.fromisoformat(res['metadata'].get('timestamp')), metadata=res['metadata'] ) retrieved_memories.append(mu) return retrieved_memories def consolidate(self, new_memory: MemoryUnit, related_memories: List[MemoryUnit]): """记忆巩固：将新记忆与旧记忆融合，形成更抽象的知识""" # 这是一个高级功能，可能调用LLM进行摘要、去重或知识融合 # 例如：“我曾用`ps aux | grep python`找进程，也用`lsof -i :8080`找端口。总结：排查进程问题可从进程列表和网络端口两方面入手。” pass

retrieve方法是长期记忆发挥价值的入口。当智能体开始一个新任务或遇到瓶颈时，系统会用当前目标或问题作为query去检索，返回的相关记忆会被格式化后加入到当前的情景缓冲中，从而实现“经验借鉴”。

3.4 记忆的评估与重要性打分

什么样的记忆值得进入长期库？这离不开一个评估器。它可能是一个基于规则的启发式函数，也可能是一个微调的小型模型。

class MemoryEvaluator: @staticmethod def calculate_importance(memory: MemoryUnit) -> float: """计算记忆单元的重要性分数（0-1之间）""" base_score = 0.5 # 规则1：思考记忆通常比观察记忆更重要 if memory.type == MemoryType.THOUGHT: base_score += 0.2 # 规则2：包含错误或异常的记忆可能很重要（教训） if "error" in memory.content.lower() or "failed" in memory.content.lower(): base_score += 0.15 # 规则3：元数据中标记为关键步骤的记忆 if memory.metadata.get('is_critical_step'): base_score += 0.1 # 规则4：记忆的长度（过于简短的可能是噪音） if len(memory.content.split()) > 20: # 超过20个词 base_score += 0.05 # 规则5：来自特定重要工具的记忆（如代码执行、数据库查询） if memory.metadata.get('tool') in ['code_interpreter', 'sql_executor']: base_score += 0.05 return min(1.0, base_score) # 确保不超过1 @staticmethod def should_consolidate(memories: List[MemoryUnit]) -> bool: """判断一组相关记忆是否需要被巩固成知识""" if len(memories) < 2: return False # 如果这些记忆属于同一任务，且包含成功结论 if all(m.metadata.get('task_id') == memories[0].metadata.get('task_id') for m in memories): if any('succeeded' in m.content or 'solution' in m.content for m in memories): return True return False

这个评估器是策略的核心，你可以根据你的智能体专注的领域（如客服、编程、数据分析）定制更复杂的评分规则。

4. 系统工作流与线程协同实战

双源记忆系统不是静态的存储，而是一个动态的、多线程协同的工作流。理解数据如何在观察、思考、缓冲、长期库之间流动，是将其应用到实际项目的关键。

4.1 单任务循环内的记忆流

在一个典型的智能体决策循环中，记忆的流动如下：

1. 感知与行动：智能体根据当前情景缓冲提供的上下文，决定调用一个工具（如执行命令git status）。
2. 生成观察记忆：工具执行后，将输出结果（如“On branch main...”）封装成一个MemoryUnit(type=OBSERVATION, ...)，其元数据记录工具名、命令、成功状态。
3. 存入情景缓冲：该观察记忆被立刻添加到ContextBuffer。这保证了下一步的思考能基于最新事实。
4. 反思与规划：LLM基于包含了新观察的完整情景缓冲，进行分析、总结，并规划下一步。这个过程产生的文本（如“Git状态显示我在主分支，工作区干净，可以继续。”）被封装成MemoryUnit(type=THOUGHT, ...)。
5. 再次存入缓冲：该思考记忆也被加入情景缓冲。至此，缓冲中包含了“行动-结果-思考”的完整闭环。
6. 异步评估与归档：与此同时，一个后台线程或异步任务被触发，它调用MemoryEvaluator对刚产生的这对观察和思考记忆进行评分。如果分数超过阈值（例如，思考记忆提到了一个“重要教训”），则调用LongTermMemory.store()方法，将其向量化后存入长期记忆库。

这个循环使得情景缓冲始终是“热”的、最新的上下文，而长期记忆库则在后台默默地积累财富。

4.2 跨任务的知识检索与激活

当智能体开始一个全新的任务，或者在当前任务中发出一个全新的子查询时，长期记忆库就被激活了。

1. 查询生成：系统将当前的任务描述或用户问题（例如，“如何排查服务器上的Python进程内存泄漏？”）作为检索查询。
2. 向量检索：调用LongTermMemory.retrieve(query=“排查Python内存泄漏”, top_k=2)。
3. 记忆注入：检索返回的2条最相关的历史记忆（可能是过去成功使用过memory_profiler模块的记录，或者通过ps和grep定位进程的经验），被转换成文本格式。
4. 上下文重构：这些检索到的记忆，会被插入到当前情景缓冲的头部或作为一个独立的“相关经验”部分，与原有的对话历史一起，构成一个更丰富的提示词上下文，送给LLM。
5. 影响决策：LLM在生成下一步行动或回答时，就能自然地参考这些“经验之谈”，给出更专业、更准确的建议。

实操心得：检索时机与上下文窗口的权衡。不要在每个回合都进行全量检索，这会造成延迟和成本上升。合理的策略是：a) 任务开始时检索一次；b) 当用户问题发生显著转折时检索；c) 当智能体连续多次行动失败或陷入循环时检索。同时，要严格控制注入上下文的历史记忆条数，避免挤占宝贵的上下文窗口，导致最新的任务细节被“淹没”。

4.3 记忆的定期维护与优化

长期记忆库不是只进不出的。像我们的大脑需要睡眠来巩固和清理记忆一样，这个系统也需要维护。

1. 去重与融合：定期运行consolidate函数，对内容高度相似、但表述不同的记忆进行合并。例如，关于“安装Python包”的记忆，可能有10条，分别是用pip install、conda install、在虚拟环境中安装等。可以调用LLM生成一条更概括的记忆：“在Python环境中安装包，通常使用pip install package_name。若使用Anaconda，可用conda install。建议在虚拟环境中操作以隔离依赖。” 然后删除或归档那10条原始记忆。
2. 重要性衰减与清理：为每条长期记忆设置一个“访问热度”或“最后访问时间”。对于长期未被检索到、且重要性分数较低的记忆，可以将其迁移到更廉价的归档存储，或直接删除，避免向量数据库膨胀影响检索速度。
3. 知识图谱构建（进阶）：对于结构化强的记忆，可以尝试提取实体和关系，构建一个小型知识图谱。例如，从多次操作服务器的记忆中，提取出“Nginx”、“配置文件路径”、“重启命令”等实体及其关系，实现更精准的逻辑推理式检索，而非仅仅是语义相似度检索。

5. 常见问题、调试技巧与性能优化

在实际集成和开发基于OpenClaw双源记忆系统的智能体时，你会遇到一些典型问题。下面是我踩过坑后总结的一些排查思路和优化建议。

5.1 记忆检索不准或无关

这是最常见的问题。检索回来的记忆风马牛不相及，不仅无助于决策，还会干扰LLM。

• 可能原因与排查：

  1. 嵌入模型不匹配：你用的嵌入模型（如all-MiniLM-L6-v2）是通用模型，对你的专业领域（如医学、法律、编程）术语不敏感。
     • 解决：在领域文本上微调嵌入模型，或换用领域专用的嵌入模型（如针对代码的codebert）。
  2. 查询语句太泛：直接用用户问题“帮我写代码”检索，结果当然泛泛。
     • 解决：对查询进行重写或扩展。可以用一个轻量级LLM将用户问题改写成更利于检索的陈述句或关键词组合。例如，“帮我写个Python爬虫”重写为“Python网络爬虫实现步骤 requests BeautifulSoup 示例代码”。
  3. 记忆存储格式太脏：存储的content字段包含了大量无关的日志信息、错误堆栈，淹没了核心知识。
     • 解决：在存储前对记忆内容进行清洗和摘要。观察记忆可以只保留关键输出；思考记忆可以要求LLM提炼成“问题-解决方案-核心点”的格式再存储。
  4. 元数据未被利用：检索时只用了内容向量，忽略了宝贵的元数据过滤器。
     • 解决：采用混合检索。先根据metadata中的tool、task_type等字段进行过滤，再在缩小后的集合里做向量相似度搜索。

• 优化技巧：

  • 分库存储：不要所有记忆混在一个向量集合里。可以按任务类型、工具来源、项目ID建立不同的“集合”或“命名空间”，检索时先确定范围。
  • 递归检索（Query Decomposition）：对于复杂问题，先将其拆解成几个子问题，分别检索，再合并结果。例如，“如何部署一个高可用的Web服务？”拆解为“负载均衡配置”、“数据库主从复制”、“服务监控”分别检索。

5.2 上下文窗口爆炸与信息过载

情景缓冲滚动保存，长期记忆又不断注入，很容易导致提示词过长，超出模型上下文限制，且让LLM迷失重点。

• 可能原因与排查：

  1. 缓冲区长度的盲目设置：max_size设得太大，保留了过多陈旧细节。
  2. 长期记忆注入过多：top_k参数设置过大，把好几段长篇记忆都塞了进去。
  3. 记忆内容冗长：存储的思考记忆是一大段散文，而非精炼的要点。

• 优化技巧：

  • 动态缓冲区管理：不要固定长度。实现一个“重要性滑动窗口”，只保留重要性分数最高的N条记忆在缓冲区内，而不是最近N条。
  • 记忆摘要注入：从长期记忆库检索到原始记忆后，不要直接注入，而是用LLM对其生成一个一两句话的摘要，只注入摘要。如果需要细节，可以让LLM指明“我想查看关于XX记忆的详细步骤”，系统再根据ID取出完整内容进行“第二次注入”。
  • 结构化上下文模板：设计严格的提示词模板，将上下文分成几个明确的部分：

    [当前任务目标]: ... [近期关键步骤（最近3步）]: ... [相关历史经验（摘要，最多2条）]: ... [下一步行动指令]: ...

    强制每个部分有字数限制，让信息结构清晰。

5.3 记忆评估策略失效

重要性打分不准，导致该记的没记，不该记的记了一堆。

• 可能原因与排查：

  1. 规则过于简单：像前面示例的规则，可能无法捕捉复杂场景下的重要性。
  2. 缺乏反馈闭环：一条记忆被存储后，它后续是否被检索、检索后是否真正帮助任务成功，这个反馈没有用来调整评估策略。

• 优化技巧：

  • 引入学习机制：为每条被检索的记忆记录一个“有用性”反馈。当任务成功完成时，追溯本次任务中使用的记忆，为其“有用性”加分。这个分数可以反过来影响其重要性，或用于训练一个更精准的重要性预测模型。
  • 任务结果反向标注：在一个任务链结束后，用LLM对整个任务流进行回顾，让其自主标识出哪些观察和思考是“转折点”或“关键学习”，然后强制将这些记忆的重要性分数调高并存储。

5.4 系统性能瓶颈

随着记忆量增长，检索变慢，影响智能体响应速度。

• 优化技巧：
  • 分层存储：高频访问的热记忆放在内存或SSD支持的向量库（如FAISS）；低频的冷记忆放在磁盘型数据库或对象存储，检索时优先查热库。
  • 缓存检索结果：对常见的、通用的查询（如“如何开始一个新项目”、“错误处理通用原则”）的检索结果进行缓存，设定一个较短的有效期。
  • 批量异步操作：记忆的存储、评估、巩固等操作，尽量设计成异步任务，不要阻塞主决策循环。

调试时，最实用的方法是可视化记忆流。为系统添加详细的日志，记录每个记忆单元的ID、类型、分数、存储决策、检索命中情况。通过分析这些日志，你可以清晰地看到哪些记忆被频繁使用，哪些从未被触及，从而有针对性地调整你的评估策略和检索逻辑。记忆系统是智能体的“内功”，需要持续地观察、分析和调优，才能让它真正成为智能体进化的基石。