1. 记忆系统 II 到底要解决什么问题:从“能写”到“能找”的工程落地
OpenClaw 的记忆系统如果只做到“写入”,那它本质上就是一个冷数据仓库。你在第 6 篇里已经让 Agent 能把用户偏好、任务结论、关键事实落盘到MEMORY.md和 daily log,但真正跑起来你会发现一个尴尬的现实:Agent 在下一轮对话里根本想不起来上一轮存了什么。原因很简单——写入和召回是两条独立的链路,只做写入不做召回,长期记忆就是一堆躺在磁盘上的死数据。
这一篇要补的正是“能找能用”这一环。具体来说,我们要用 Python + FastAPI 搭一个 Agent 记忆读写服务,把记忆的召回接口、会话启动自动注入、压缩摘要归档这三件事串起来,同时把 settings 里的 endpoint 统一改到 TaoToken 的 API 通道,让记忆服务里的 LLM 调用(比如摘要压缩、后续的语义重排)走同一个 Key 和 Base URL,不用在多个 provider 之间来回切换配置。
适合谁看?如果你正在手搓 OpenClaw 的最小闭环,已经跑通了/v1/chat和会话隔离,但发现 Agent 的“记性”时好时坏,或者你想给自己的 Agent 加一层可解释的长期记忆,这篇的代码和配置可以直接抄。核心检索词就三个:OpenClaw 记忆系统、Python FastAPI 记忆服务、Agent 长期记忆召回。下面所有步骤都是可跟做的,我会给出完整的路由、配置片段和一次真实的写入/召回验证动作。
先说清楚这一篇的边界:召回先用轻量关键词检索,优点是零外部依赖、稳定、好调试;语义召回留给第 8 篇的 RAG 升级。你要做的是先把“能找”这条链路跑通,再谈“找得准”。
2. TaoToken 前置:把 settings 的 endpoint 统一到一条 API 通道
在动手写记忆服务之前,先把配置层理顺。手搓 OpenClaw 的过程中最容易踩的坑就是:Agent 主循环调一个 provider,记忆摘要调另一个,RAG 重排又换一个,最后 Key 散落在三四个文件里,出问题根本不知道是哪条链路挂了。我的做法是把所有 LLM 调用收敛到 TaoToken 的统一通道,settings 里只保留一份 Base URL 和一份 Key。
TaoToken 在这里扮演的角色是统一的 API 网关:你拿一个 Key,就能通过兼容 OpenAI 协议的接口访问多家模型。对记忆系统来说,这意味着摘要压缩、关键词扩展、后续的 rerank 都可以复用同一个 client,不用为每个能力单独配一套鉴权。官网入口在 https://taotoken.net/?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= ,API 根地址是 https://taotoken.net/api ,注意 API 地址后面不加 UTM 参数,直接用于代码里的base_url。
你需要准备的东西只有两样:一个 API Key,以及确认你要用的 Model ID。Key 在控制台的 API Keys 页面创建,地址是 https://taotoken.net/console/api-keys?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= 。创建时建议按用途命名,比如openclaw-memory,方便后面排查是哪个服务在调用。Model ID 则取决于你当前用的模型,记忆摘要这种任务用轻量模型就够,主 Agent 循环可以用能力更强的。
这里要强调一个工程习惯:不要把 Key 硬编码进manager.py或agent.py。正确做法是走环境变量或.env,settings 只读环境变量。这样你在本地、容器、CI 里切换配置时不用改代码。下面这段是我实际在用的.env结构,你可以直接复制:
# .env TAOTOKEN_API_KEY=sk-你的key TAOTOKEN_BASE_URL=https://taotoken.net/api TAOTOKEN_MODEL=gpt-4o-mini LONG_TERM_MEMORY_BOOTSTRAP_CHARS=1200 MEMORY_STORE_PATH=./data/memory对应的settings.py用 pydantic-settings 读取,保持和 OpenClaw 原项目的字段命名一致,这样第 6 篇里已经写好的settings.long_term_memory_bootstrap_chars不用改:
# openclaw_py/app/core/settings.py from pydantic_settings import BaseSettings class Settings(BaseSettings): taotoken_api_key: str taotoken_base_url: str = "https://taotoken.net/api" taotoken_model: str = "gpt-4o-mini" long_term_memory_bootstrap_chars: int = 1200 memory_store_path: str = "./data/memory" class Config: env_file = ".env" settings = Settings()如果你用的是 Claude Code 这类工具做辅助开发,接入时同样是三件套:Base URL 填https://taotoken.net/api,Key 填上面创建的,Model ID 填你选的模型。这三者在任何兼容 OpenAI 协议的客户端里都是一致的,记忆服务里的 client 也照这个来。配置文档在 https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= ,遇到字段对不上时先查文档再改代码。
3. 可复制配置:FastAPI 记忆读写服务的路由与 settings 片段
这一节是整篇的核心,给出可以直接跑起来的 FastAPI 路由和配置。目标是把记忆的写入、召回、启动注入三个动作暴露成 HTTP 接口,同时让记忆服务内部的 LLM 调用走 TaoToken。
先看目录结构,保持和 OpenClaw 原项目的分层对齐:
openclaw_py/ app/ core/ settings.py agent.py memory/ manager.py store.py api/ memory_routes.py main.py data/ memory/ MEMORY.md daily/store.py负责落盘,manager.py负责召回逻辑,memory_routes.py暴露接口。先写 store,它管的是MEMORY.md的读写和 daily log 追加:
# openclaw_py/app/memory/store.py import os from pathlib import Path from datetime import datetime class MemoryStore: def __init__(self, base_path: str): self.base = Path(base_path) self.base.mkdir(parents=True, exist_ok=True) self.memory_file = self.base / "MEMORY.md" self.daily_dir = self.base / "daily" self.daily_dir.mkdir(exist_ok=True) if not self.memory_file.exists(): self.memory_file.write_text("# MEMORY\n", encoding="utf-8") def list_all(self) -> dict: entries = {} for line in self.memory_file.read_text(encoding="utf-8").splitlines(): if line.startswith("- ") and ": " in line: key, value = line[2:].split(": ", 1) entries[key.strip()] = value.strip() return entries def upsert(self, key: str, value: str) -> None: entries = self.list_all() entries[key] = value lines = ["# MEMORY"] + [f"- {k}: {v}" for k, v in entries.items()] self.memory_file.write_text("\n".join(lines) + "\n", encoding="utf-8") def append_compaction_summary(self, summary: str) -> None: day = datetime.now().strftime("%Y-%m-%d") log = self.daily_dir / f"{day}.md" with log.open("a", encoding="utf-8") as f: f.write(f"\n## {datetime.now().isoformat()}\n{summary}\n")然后是manager.py里的召回接口,这段和第 6 篇的写入逻辑衔接,关键词检索保持轻量:
# openclaw_py/app/memory/manager.py import re from .store import MemoryStore class MemoryManager: def __init__(self, store: MemoryStore, bootstrap_chars: int = 1200): self.store = store self.bootstrap_chars = bootstrap_chars def recall(self, query: str, top_k: int | None = None) -> str: normalized_query = (query or "").strip().lower() all_entries = self.store.list_all() if normalized_query in {"", "*", "all", "everything"}: return "\n".join(f"- {k}: {v}" for k, v in all_entries.items()) query_terms = [t for t in re.split(r"\s+", normalized_query) if t] scored = [] for key, value in all_entries.items(): haystack = f"{key} {value}".lower() score = sum(1 for term in query_terms if term in haystack) if score > 0: scored.append((score, key, value)) scored.sort(reverse=True) if top_k: scored = scored[:top_k] return "\n".join(f"- {k}: {v}" for _, k, v in scored) def boot_context(self, max_chars: int | None = None) -> str: budget = max_chars or self.bootstrap_chars snapshot = self.recall("*") if len(snapshot) > budget: snapshot = snapshot[:budget] return snapshot接着是 FastAPI 路由,把写入和召回暴露出来:
# openclaw_py/app/api/memory_routes.py from fastapi import APIRouter, HTTPException from pydantic import BaseModel from ..memory.manager import MemoryManager from ..memory.store import MemoryStore from ..core.settings import settings router = APIRouter(prefix="/v1/memory", tags=["memory"]) store = MemoryStore(settings.memory_store_path) manager = MemoryManager(store, settings.long_term_memory_bootstrap_chars) class WriteReq(BaseModel): key: str value: str class RecallReq(BaseModel): query: str top_k: int | None = None @router.post("/write") def write_memory(req: WriteReq): if not req.key.strip(): raise HTTPException(status_code=400, detail="key 不能为空") store.upsert(req.key, req.value) return {"ok": True, "key": req.key} @router.post("/recall") def recall_memory(req: RecallReq): result = manager.recall(req.query, req.top_k) return {"ok": True, "query": req.query, "result": result} @router.get("/boot") def boot_context(): return {"ok": True, "context": manager.boot_context()}最后在main.py里挂载路由:
# openclaw_py/app/main.py from fastapi import FastAPI from .api.memory_routes import router as memory_router app = FastAPI(title="OpenClaw Memory Service") app.include_router(memory_router)启动命令:
uvicorn openclaw_py.app.main:app --host 0.0.0.0 --port 8000 --reload到这里,记忆读写服务就跑起来了。注意settings.py里的taotoken_base_url和taotoken_api_key目前还没被直接用到,因为关键词召回是纯本地的。但下一步做摘要压缩和后续 RAG 重排时,client 会读这两个字段,所以现在就把配置放好,避免后面再改一遍。
4. 验证请求:一次记忆写入与召回的成功结果
配置写完了,必须做一次真实的写入/召回验证,否则你不知道链路是不是通的。我用 curl 演示,你也可以用 Postman 或 Python requests。
第一步,写入两条记忆:
curl -X POST http://127.0.0.1:8000/v1/memory/write \ -H "Content-Type: application/json" \ -d '{"key": "user_prefer_lang", "value": "用户偏好 Python 和 FastAPI"}' curl -X POST http://127.0.0.1:8000/v1/memory/write \ -H "Content-Type: application/json" \ -d '{"key": "project_stack", "value": "OpenClaw 记忆系统使用 MEMORY.md 加 daily log 分层"}'预期返回:
{"ok": true, "key": "user_prefer_lang"} {"ok": true, "key": "project_stack"}第二步,用 query 召回。这里故意用“FastAPI”这个只出现在第一条记忆里的词,验证关键词打分是否生效:
curl -X POST http://127.0.0.1:8000/v1/memory/recall \ -H "Content-Type: application/json" \ -d '{"query": "FastAPI", "top_k": 3}'预期返回:
{ "ok": true, "query": "FastAPI", "result": "- user_prefer_lang: 用户偏好 Python 和 FastAPI" }第三步,验证会话启动注入。这个接口返回的是全量快照,受long_term_memory_bootstrap_chars预算限制:
curl http://127.0.0.1:8000/v1/memory/boot预期返回里能看到两条记忆都被拼进了 context。如果字符数超过 1200,会被截断,这是故意的——防止长期记忆把主上下文撑爆。
第四步,把召回结果接进 Agent 循环。在agent.py的 session 初始化处加一段:
memory_snapshot = self.memory.boot_context( max_chars=settings.long_term_memory_bootstrap_chars, ) if memory_snapshot: self.history.append({"role": "system", "content": memory_snapshot})跑一次对话,观察 Agent 是否在回复里体现出“知道用户偏好 Python”。如果体现出来了,说明写入、召回、注入三段链路全部打通。这一步的实测结果是:Agent 在首轮回复里主动提到“你之前偏好 Python 和 FastAPI”,说明 boot_context 注入生效。
第五步,验证压缩摘要归档。当会话历史被压缩时,调用append_compaction_summary,然后去data/memory/daily/下看当天的 md 文件是否追加了内容。这一步的意义是:压缩不是删除历史,而是把历史迁移到更便宜的存储层,排查时还能追溯。
5. 本篇常见错排查:401、local proxy failed、reading choices 与 OAuth
记忆服务跑起来后,报错基本集中在鉴权和响应解析两类。下面按真实报错逐条对照。
401 Unauthorized:最常见。原因通常是.env里的TAOTOKEN_API_KEY没被读到,或者 Key 前后有空格。排查顺序:先确认settings.taotoken_api_key打印出来不是空字符串,再确认请求头里Authorization: Bearer <key>拼对了。如果你在容器里跑,检查环境变量有没有透传进去。Key 本身失效的话,去 https://taotoken.net/console/api-keys?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= 重新生成一个。
local proxy failed:这个报错通常出现在你本地网络层做了额外转发,或者base_url写成了带路径的地址。正确写法是https://taotoken.net/api,不要在后面加/v1或/chat/completions,SDK 会自己拼。如果你在 settings 里写成了https://taotoken.net/api/v1,就会出现路径重复导致连接失败。改回根地址即可。
reading choices 相关报错:典型信息是KeyError: 'choices'或list index out of range。这说明响应体里没有choices字段,通常是请求根本没成功,返回的是错误 JSON。排查方法:把原始响应print(response.text)打出来,看是不是 401 或 429。429 是限流,降低并发或换轻量模型即可。另一个原因是 Model ID 写错,服务端返回了错误结构,确认TAOTOKEN_MODEL和文档里的一致。
OAuth 相关报错:如果你用 Claude Code 或类似工具接入,报 OAuth 失败,说明你走的是账号授权流程而不是 API Key 流程。记忆服务里统一用 API Key,不要混用 OAuth。三件套再确认一遍:Base URL 是https://taotoken.net/api,Key 是控制台创建的,Model ID 是你要用的模型。这三者缺一不可,任何客户端接入都照这个来。
召回结果为空:不是报错但很常见。先确认MEMORY.md里的格式是- key: value,因为list_all是按这个格式解析的。如果你的写入用了别的分隔符,召回会解析不到。再确认 query 里的词确实出现在 key 或 value 里,关键词检索不做同义词扩展,这是第 8 篇 RAG 要解决的。
boot_context 被截断:如果发现注入的记忆不完整,检查long_term_memory_bootstrap_chars的值。默认 1200 字符,记忆条目多的时候会截断。可以调大,但别调太大,否则主上下文会被长期记忆挤占,影响当前对话质量。
6. 语义一致 CTA:把记忆服务接上统一通道继续往下走
到这里,OpenClaw 记忆系统 II 的工程落地就完成了:写入、召回、启动注入、压缩归档四条链路都跑通,settings 里的 endpoint 也统一到了 TaoToken 的 API 通道。你现在有一个可以独立部署的 FastAPI 记忆服务,Agent 在会话启动时能自动拿到长期偏好,召回接口可以按 query 取回相关记忆。
下一步的衔接很明确:第 8 篇会把当前的关键词召回升级成 RAG,引入 BM25、向量召回、RRF 融合和 rerank,让召回从“字面匹配”变成“语义匹配”。到那时,记忆服务里的 LLM 调用会变多,统一通道的价值就体现出来了——摘要、扩展、重排全走同一个 Key 和 Base URL,不用为每个能力单独配鉴权。
如果你还没创建 Key,去 https://taotoken.net/console/api-keys?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= 建一个,命名成openclaw-memory方便区分。接入过程中字段对不上,查 https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= 。想先验证模型通不通,用模型对话页面发一条测试消息:https://taotoken.net/models?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= 。如果你打算长期做 Agent 编码和记忆系统迭代,Coding Plan 更适合持续调用:https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= 。
最后留一个我踩过的坑:记忆服务的 store 和 manager 一定要做依赖注入,不要在每个路由里 new 一个实例。否则多进程部署时,每个 worker 各写各的MEMORY.md,会出现写入覆盖。用单例或 FastAPI 的 lifespan 管理,能省掉后面很多排查时间。