OpenClaw Memory 模块完整分析

一、项目背景

OpenClaw 是一个本地优先的个人 AI 助手，支持多种消息通道（WhatsApp、Telegram、Slack 等）。Memory 模块为 AI Agent 提供语义记忆搜索能力——Agent 可以在 Markdown 记忆文件和历史会话中进行向量 + 关键词的混合检索。

二、整体架构

┌─────────────────────────────────────────────────┐
│                  入口层 (index.ts)               │
│  getMemorySearchManager() → 选择后端策略          │
├────────────┬────────────────────────┬───────────┤
│  QMD 后端   │  FallbackManager      │ Builtin 后端│
│ (外部CLI)   │  (主备自动切换)        │ (核心实现)   │
├─────────────┴───────────────────────┴───────────┤
│               MemoryIndexManager                │
│   ┌──────────┬──────────┬──────────┐            │
│   │ SyncOps  │Embedding │ Search   │            │
│   │(文件监听) │ Ops(索引) │(混合搜索) │             │
│   └──────────┴──────────┴──────────┘            │
├─────────────────────────────────────────────────┤
│            存储层: SQLite + FTS5 + sqlite-vec    │
├─────────────────────────────────────────────────┤
│  Embedding Providers: OpenAI|Gemini|Voyage|     │
│  Mistral|Local(node-llama-cpp)                  │
└──────────────────────────────────────────────────┘

三、核心设计详解

1. 统一接口 (MemorySearchManager)

export type MemorySource = "memory" | "sessions";

export type MemorySearchResult = {
  path: string;
  startLine: number;
  endLine: number;
  score: number;
  snippet: string;
  source: MemorySource;
  citation?: string;
};
// ...
export interface MemorySearchManager {
  search(query, opts?): Promise<MemorySearchResult[]>;
  readFile(params): Promise<{ text: string; path: string }>;
  status(): MemoryProviderStatus;
  sync?(params?): Promise<void>;
  probeEmbeddingAvailability(): Promise<MemoryEmbeddingProbeResult>;
  probeVectorAvailability(): Promise<boolean>;
  close?(): Promise<void>;
}

这是整个模块的核心抽象——无论底层用什么后端（builtin SQLite 还是外部 QMD CLI），对上层暴露统一接口。

2. 后端策略选择 (search-manager.ts)

export async function getMemorySearchManager(params: {
  cfg: OpenClawConfig;
  agentId: string;
  purpose?: "default" | "status";
}): Promise<MemorySearchManagerResult> {
  const resolved = resolveMemoryBackendConfig(params);
  if (resolved.backend === "qmd" && resolved.qmd) {
    // ... 尝试 QMD 后端，失败则 fallback 到 builtin
    const wrapper = new FallbackMemoryManager({
      primary,
      fallbackFactory: async () => {
        const { MemoryIndexManager } = await import("./manager.js");
        return await MemoryIndexManager.get(params);
      },
    }, () => QMD_MANAGER_CACHE.delete(cacheKey));
    // ...
  }
  // 默认使用 builtin
  const manager = await MemoryIndexManager.get(params);
  return { manager };
}

设计亮点：

• 策略模式 + 懒加载：通过 dynamic import 延迟加载后端实现
• FallbackMemoryManager：代理模式，主后端失败自动切换到备用后端，对上层透明
• 缓存驱逐：失败时自动从缓存中移除，下次请求可以重试

3. 混合搜索引擎 (hybrid.ts)

export async function mergeHybridResults(params: {
  vector: HybridVectorResult[];
  keyword: HybridKeywordResult[];
  vectorWeight: number;
  textWeight: number;
  mmr?: Partial<MMRConfig>;
  temporalDecay?: Partial<TemporalDecayConfig>;
}): Promise<Array<{ path; startLine; endLine; score; snippet; source }>> {
  // 1. 按 ID 合并向量和关键词结果
  // 2. 加权融合分数: score = vectorWeight * vectorScore + textWeight * textScore
  // 3. 时间衰减: 旧记忆分数降低
  // 4. MMR 重排序: 增加结果多样性
}

这是搜索的核心——三层融合管线：

阶段	算法	作用
加权融合	score = w_v × vectorScore + w_t × textScore	平衡语义相似度和关键词匹配
时间衰减	指数衰减 e^(-λ × age)	让近期记忆权重更高
MMR 重排序	λ × relevance - (1-λ) × max_similarity	增加结果多样性，避免重复

4. 时间衰减机制 (temporal-decay.ts)

export function toDecayLambda(halfLifeDays: number): number {
  return Math.LN2 / halfLifeDays;
}

export function calculateTemporalDecayMultiplier(params: {
  ageInDays: number;
  halfLifeDays: number;
}): number {
  const lambda = toDecayLambda(params.halfLifeDays);
  return Math.exp(-lambda * clampedAge);
}

核心设计：

• 日期命名文件 memory/2024-01-15.md 自动从文件名提取时间
• 常青文件 MEMORY.md 和非日期命名的 memory/*.md 不衰减（核心知识）
• fallback 到 mtime：无法从文件名解析日期时，使用文件修改时间

5. MMR 多样性重排序 (mmr.ts)

export function computeMMRScore(relevance: number, maxSimilarity: number, lambda: number): number {
  return lambda * relevance - (1 - lambda) * maxSimilarity;
}

使用 Jaccard 相似度（基于 token 集合的交集/并集）来衡量结果间的相似程度，避免返回大量重复内容。比起用向量余弦相似度做 MMR，Jaccard 更轻量且无需额外嵌入计算。

6. Markdown 分块策略 (internal.ts)

export function chunkMarkdown(
  content: string,
  chunking: { tokens: number; overlap: number },
): MemoryChunk[] {
  const maxChars = Math.max(32, chunking.tokens * 4);
  const overlapChars = Math.max(0, chunking.overlap * 4);
  // 按行扫描，达到 maxChars 时 flush
  // flush 后保留尾部 overlapChars 作为重叠区
}

设计要点：

• token 估算：tokens × 4 转为字符数（粗略但高效）
• 滑动窗口重叠：chunk 之间有 overlap，避免语义在边界处被截断
• 超长行切割：单行超过 maxChars 时自动分段
• 每个 chunk 记录 startLine/endLine，支持精确引用

7. Embedding Provider 工厂 (embeddings.ts)

export async function createEmbeddingProvider(
  options: EmbeddingProviderOptions,
): Promise<EmbeddingProviderResult> {
  // auto 模式: local → openai → gemini → voyage → mistral
  // 指定模式: primary → fallback
  // 所有 API key 缺失: 返回 null provider (FTS-only mode)
}

三层降级策略：

1. auto 模式：依次尝试 local → openai → gemini → voyage → mistral
2. 指定 + fallback：用户指定的 provider 失败时切换到 fallback
3. 全部失败 → FTS-only：纯关键词搜索，仍可用但质量降低

8. 数据库 Schema (memory-schema.ts)

// meta: 索引元信息 (model/provider/版本)
// files: 文件记录 (path, hash, mtime, source)
// chunks: 文本块 (id, path, text, embedding, model)
// embedding_cache: 嵌入缓存 (provider+model+hash → embedding)
// chunks_fts: FTS5 全文搜索虚拟表
// chunks_vec: sqlite-vec 向量搜索虚拟表

9. 同步机制 (manager-sync-ops.ts)

同步有多种触发方式：

触发方式	场景
watch	chokidar 文件监听，debounce 后触发
session-start	新会话开始时预热
session-delta	会话文件增长超过阈值（字节/消息数）
search	搜索时如果 dirty 则先同步
interval	定时同步（可配置分钟数）

重建索引采用安全替换策略：先写入临时 DB，完成后原子交换，失败则回滚。

10. 实例缓存 + 单例

const INDEX_CACHE = new Map<string, MemoryIndexManager>();

static async get(params): Promise<MemoryIndexManager | null> {
  const key = `${agentId}:${workspaceDir}:${JSON.stringify(settings)}`;
  const existing = INDEX_CACHE.get(key);
  if (existing) return existing;
  // ... 创建新实例
  INDEX_CACHE.set(key, manager);
  return manager;
}

用 agentId + workspaceDir + settings 作为缓存 key，保证同一配置只有一个 Manager 实例，避免重复打开数据库和文件监听器。

四、参考价值总结

如果你想在自己的项目中实现类似的记忆/知识检索系统，这个模块有以下核心参考价值：

维度	设计模式	参考价值
接口抽象	MemorySearchManager 接口	将搜索、同步、状态查询统一抽象，后端可替换
混合搜索	Vector + BM25 加权融合	兼顾语义理解和精确匹配，比单纯向量搜索更鲁棒
结果优化	MMR + 时间衰减	解决结果重复和旧信息权重过高的问题
降级策略	Provider 三层 fallback + FTS-only	无 API key 也能用，极大提升了可用性
主备切换	FallbackMemoryManager 代理模式	QMD 后端失败自动切到 builtin，对上层透明
增量同步	hash 对比 + 文件监听 + delta 阈值	只重新索引变化的文件，会话增量基于字节/消息数阈值
安全重建	临时 DB → 原子交换 → 失败回滚	全量重建索引时不影响在线查询
嵌入缓存	SQLite embedding_cache 表	避免重复调用 API，重建索引时可复用历史嵌入
分块策略	行级滑动窗口 + overlap	保留行号信息，支持精确引用，overlap 防止语义断裂
实例管理	缓存 Map + 复合 key	避免重复实例，正确处理 close 和缓存驱逐

这套架构特别适合以下场景复用：

1. RAG 系统——需要在本地文档上做语义搜索
2. 知识库检索——混合搜索 + 时间衰减适合持续更新的知识
3. Agent 工具——作为 AI Agent 的长期记忆组件
4. 离线优先应用——SQLite 本地存储 + 可选远程 Embedding 的架构

从零开始LangChain：构建你的第一个AI应用工作流

引言

2022年ChatGPT横空出世，让全世界见识了大型语言模型（LLM）的魔力。但你知道吗？有一个叫LangChain的框架其实比ChatGPT还早，最近它发布了1.0+版本，成为了AI应用开发的“明星框架”。

LangChain是什么？拆开名字：Lang（语言） + Chain（链）。它把大语言模型和一系列任务节点像链条一样连接起来，形成一个工作流。就像n8n、Coze这些可视化工具把节点串起来一样，LangChain用代码的方式帮你搭建AI应用。

这篇文章我会带你从零开始，一步步用LangChain写几个小例子，从最简单的模型调用，到用“链”组合复杂的任务流程。所有代码都基于ES Module（type: "module"），你可以直接复制运行。

环境准备：先跑通一个最简单的例子

在开始之前，确保你安装了Node.js（18+版本），然后创建一个新项目，安装必要的依赖：

npm init -y
npm install dotenv langchain @langchain/deepseek

在package.json中加入 "type": "module"，这样我们就可以使用import语法。

创建一个.env文件，放你的DeepSeek API密钥（如果没有可以去platform.deepseek.com申请）：

DEEPSEEK_API_KEY=你的密钥

现在，写第一个脚本main.js：

import 'dotenv/config'
import { ChatDeepSeek } from '@langchain/deepseek'

// 初始化模型
const model = new ChatDeepSeek({
  model: 'deepseek-reasoner',
  temperature: 0,  // 控制随机性，0表示最确定
})

// 调用模型
const res = await model.invoke('用一句话解释什么是RAG')
console.log(res.content)

运行node main.js，你应该能看到模型输出的回答。

这段代码做了什么？

• ChatDeepSeek是一个“适配器”，LangChain用它来统一不同大模型的接口。以后换GPT、Claude，只需要改一下import和配置，其余代码几乎不用动。
• model.invoke是最核心的方法，把问题传给模型，然后得到回答。
• API密钥从环境变量自动读取，不用写在代码里，安全又方便。

这就是LangChain最基础的用法：把大模型当成一个可调用的函数。

第一章：更灵活的提问——提示词模板

直接写死的提问太死板了。如果我想让模型扮演不同角色、限制回答长度，每次都拼接字符串会很麻烦。LangChain提供了PromptTemplate，像填空一样生成提示词。

新建1.js：

import 'dotenv/config'
import { ChatDeepSeek } from '@langchain/deepseek'
import { PromptTemplate } from '@langchain/core/prompts'

// 定义一个模板
const prompt = PromptTemplate.fromTemplate(`
你是一个{role}。
请用不超过{limit}字回答以下问题：
{question}
`)

// 填入具体内容
const promptStr = await prompt.format({
  role: '前端面试官',
  limit: '50',
  question: '什么是闭包'
})

console.log('生成的提示词：', promptStr)

// 调用模型
const model = new ChatDeepSeek({
  model: 'deepseek-reasoner',
  temperature: 0.7,
})

const res = await model.invoke(promptStr)
console.log('回答：', res.content)

运行后，你会看到模型根据“前端面试官”的身份，用不超过50字解释了闭包。

• 如图

PromptTemplate让我们把提示词的结构和内容分离，方便复用。比如你可以换一个角色问同样的问题，只需改format的参数即可。

第二章：什么是“链”？用pipe连接节点

上面的例子还是两步走：先生成提示词，再调用模型。LangChain的核心理念是“链”（Chain），它可以把多个步骤像管道一样连接起来，成为一个可执行的单元。

新建2.js：

import 'dotenv/config'
import { ChatDeepSeek } from '@langchain/deepseek'
import { PromptTemplate } from '@langchain/core/prompts'

const model = new ChatDeepSeek({
  model: 'deepseek-reasoner',
  temperature: 0.7,
})

const prompt = PromptTemplate.fromTemplate(`
你是一个前端专家，用一句话解释：{topic}
`)

// 用 pipe 把 prompt 和 model 连接成一个链
const chain = prompt.pipe(model)
//打印chain可以看到它的类型为RunnableSequence，和它的节点
console.log(chain)
// 直接调用链，传入变量
const res = await chain.invoke({
  topic: '闭包'
})

console.log(res.content)

• 执行效果图

1. 执行打印chain

2. 执行打印输出结果

注意看，prompt.pipe(model)返回了一个新的对象，它也是一个“可运行”的链。我们调用chain.invoke({ topic: '闭包' })，内部会自动执行：

1. 用传入的{topic: '闭包'}填充prompt模板，生成提示词。
2. 把提示词传给model，得到回答。
3. 返回最终结果。

整个过程就像工厂流水线：原材料（topic）进入第一道工序（prompt模板），产物（提示词）直接传给下一道工序（模型），最后产出成品（回答）。

这就是LangChain最基础的链：RunnableSequence（可运行序列）。你不需要手动调用两次，代码更简洁，逻辑更清晰。

第三章：组合多个链——复杂任务的工作流

现实中的AI任务往往不止一步。比如我想让模型先详细解释一个概念，然后把这个解释总结成三个要点。这需要两个步骤，而且第二步要用到第一步的输出。

LangChain提供了RunnableSequence来组合多个链。我们新建3.js：

import { ChatDeepSeek } from '@langchain/deepseek'
import { PromptTemplate } from '@langchain/core/prompts'
import { RunnableSequence } from '@langchain/core/runnables'
import 'dotenv/config'

const model = new ChatDeepSeek({
  model: 'deepseek-reasoner',
  temperature: 0.7,
})

// 第一步：详细解释概念
const explainPrompt = PromptTemplate.fromTemplate(`
你是一个前端专家，请详细介绍以下概念：{topic}
要求：覆盖定义、原理、使用方式，不超过300字。
`)

// 第二步：总结要点
const summaryPrompt = PromptTemplate.fromTemplate(`
请将以下前端概念解释总结为3个核心要点（每个要点不超过20字）：
{explaination}
`)

// 分别构建两个链
const explainChain = explainPrompt.pipe(model)
//我们打印explainChain也可以看到它的类型和节点
console.log(explainChain)
const summaryChain = summaryPrompt.pipe(model)

// 用 RunnableSequence 组合它们
const fullChain = RunnableSequence.from([
  // 第一步：输入 topic -> 得到详细解释
  async (input) => {
    const res = await explainChain.invoke({ topic: input.topic })
    return res.content  // 将解释传给下一步
  },
  // 第二步：拿到上一步的解释 -> 生成总结
  async (explaination) => {
    const res = await summaryChain.invoke({ explaination })
    return `知识点：${explaination}\n总结：${res.content}`
  },
])

// 执行完整链
const res = await fullChain.invoke({
  topic: '闭包'
})
console.log(res)

• 效果图

1. 打印explainChain

2.打印输出结果

这段代码稍微复杂一点，但逻辑很清晰：

• explainChain：输入topic，输出详细解释。
• summaryChain：输入explaination（详细解释），输出总结。
• 我们用RunnableSequence.from([...])把两个步骤串起来。数组里的每个元素是一个函数，接收上一步的输出，返回下一步的输入。
• 最后调用fullChain.invoke({ topic: '闭包' })，内部自动执行两步，并把最终结果返回。

运行后你会看到模型先给出了关于闭包的详细解释（不超过300字），然后给出了三个要点总结。整个流程自动化完成，无需人工介入。

深入理解：LangChain的适配器模式与可拔插设计

你可能注意到，所有代码中我们只引用了@langchain/deepseek这一个具体模型包。如果我想换成OpenAI的GPT，该怎么做？只需要：

npm install @langchain/openai

然后把import { ChatDeepSeek }改成import { ChatOpenAI }，model参数稍作调整即可，其余代码几乎不用动。

这就是LangChain的“适配器模式”。它定义了一套统一的接口（如invoke、stream等），各个模型厂商通过适配器实现这些接口。这样一来，你的业务逻辑和具体模型解耦，大模型更新换代再快，你只需换一个适配器，不用重写应用。

总结

通过这几个小例子，我们走过了LangChain的入门之路：

1. 基础调用：用适配器连接大模型，执行最简单的问答。
2. 提示词模板：用PromptTemplate动态构造输入，让提示词更灵活。
3. 简单链：用pipe把模板和模型连接起来，形成可复用单元。
4. 复杂链：用RunnableSequence组合多个链，实现多步骤工作流。

LangChain不仅仅是“链”，它还是一个完整的AI应用开发框架，提供了记忆、工具调用、代理（Agent）等高级功能。但无论多复杂的功能，底层都离不开我们今天学到的核心思想：把任务拆分成节点，用链条连接，让流程自动化。

现在你已经掌握了LangChain的基本功，可以尝试用它搭建更酷的应用了，比如文档问答机器人、自动化报告生成器等等。如果在实践中遇到问题，欢迎在评论区留言交流。