OpenClaw的记忆系统,用在企业场景还是太粗糙了

用OpenClaw这类个人AI助手久了，很多人都会遇到一个棘手问题：当你询问上周刚确认的API限流方案时，AI翻出来的却是三个月前的过期内容。

其实记忆记录本身是存在的，核心问题在于，龙虾无法判断哪条记忆更重要、更具时效性。

而OpenClaw、NanoClaw这些产品默认的全量写入上下文方案，天生就存在时效性缺失的短板。

为了解决这个问题，我们以最近很火的NanoClaw（龙虾系轻量替代版）为实验载体，叠加使用Milvus 2.6版本的 Decay Function功能，看看它能否有效解决AI的信息选择困难症。

01 各种Claw的记忆模式到底有什么问题？

NanoClaw 在内，大部分龙虾类的记忆方式可以概括如下：每个对话 group 对应一个CLAUDE.md文件，Agent 启动时把整个文件塞进 context。

这种模式看似便捷，却存在两个致命问题。

第一，Token消耗过高。以Claude Sonnet 200k上下文窗口为基准，仅框架基础指令就占用了约17%的Token，还未计入用户的实际记忆内容，长期使用会大幅增加成本，甚至触发上下文窗口上限。

第二，记忆无优先级区分。这是最核心的问题：即便上下文没有占满，三个月前的历史讨论与昨日刚敲定的决策，会被模型同等看待。面对两条语义相近的内容（比如不同时期的API限流讨论），模型完全不会主动优先采信时间更近、更具时效性的信息，进而导致回复偏离实际需求。

要解这个问题，关键不在于盲目扩大上下文窗口，而是要在检索层把时间维度纳入排序权重。

02 Decay Function如何解决RAG相似内容的重要性排序？

提到AI助手的知识检索优化，很多人的第一反应是做RAG。把记忆转换成向量（embedding）存入向量库，检索时获取语义最相似的前k条内容。但这种纯向量检索，只能解决找相关内容的问题，无法解决一堆召回的相关内容中，谁更重要的问题。

为了解决这个问题，Milvus 2.6 推出了 Decay Function ：在向量相似度排序的基础上叠加时间衰减，用 Exponential 或 Gaussian 函数给旧记忆降权，距当前时间越远，得分折扣越大，从而让最新的有效记忆优先被召回。

03 实验验证：三种检索方式对比，看Decay的实际效果

为了直观验证时间衰减检索的效果，我们用独立脚本模拟NanoClaw积累三个月记忆后的状态，并设计了三组对比实验，分别测试不同检索策略的效果。

第一步：实验准备：环境搭建与数据模拟

首先需要安装Milvus 2.6版本，搭建基础环境，同时模拟3个月内的50条项目记忆（贴合OpenClaw实际使用场景）：

安装 milvus2.6

wget https://github.com/milvus-io/milvus/releases/download/v2.6.11/milvus-standalone-docker-compose.yml -O docker-compose.yml

docker compose up -d

环境依赖：

pip install pymilvus openai

建 collection，写入 50 条模拟记忆

这里我们模拟50条记忆，时间跨度3个月，分为两条主题线：25条API限流方案演进（从初步讨论到最终上线调参），25条项目基础设施决策（缓存、数据库、部署等），越接近当前时间的记忆，实用性越高：

import jsonimport timefrom datetime import datetime, timedeltafrom openai import OpenAIfrom pymilvus import MilvusClient, DataTypeclient = OpenAI()milvus = MilvusClient(    uri="http://your-milvus-host:19530",    user="root",    password="your-password")COLLECTION_NAME = "nanoclaw_memory"if milvus.has_collection(COLLECTION_NAME):    milvus.drop_collection(COLLECTION_NAME)schema = milvus.create_schema(auto_id=True, enable_dynamic_field=False)schema.add_field("id", DataType.INT64, is_primary=True)schema.add_field("text", DataType.VARCHAR, max_length=1000)schema.add_field("vector", DataType.FLOAT_VECTOR, dim=1536)schema.add_field("timestamp", DataType.INT64)schema.add_field("metadata", DataType.VARCHAR, max_length=500)milvus.create_collection(collection_name=COLLECTION_NAME, schema=schema)index_params = milvus.prepare_index_params()index_params.add_index(    field_name="vector",    index_type="HNSW",    metric_type="COSINE",    params={"M": 16, "efConstruction": 200})index_params.add_index(    field_name="metadata",    index_type="INVERTED",    index_name="metadata_group_index",    params={        "json_path": "metadata[\"group\"]",        "json_cast_type": "varchar"    })milvus.create_index(COLLECTION_NAME, index_params)milvus.load_collection(COLLECTION_NAME)# 50 条模拟记忆，时间跨度三个月now = datetime.now()memories = [    {"text": "项目决定用微服务架构，前后端分离，部署在 AWS。", "days_ago": 90},    {"text": "团队讨论过用 GraphQL，但最终因为学习成本放弃了，还是 REST。", "days_ago": 85},    {"text": "数据库选型时考虑过 MongoDB，最终选了 PostgreSQL。", "days_ago": 80},    {"text": "API 限流方案初步讨论：考虑 token bucket 或 sliding window。", "days_ago": 75},    {"text": "用户认证方案讨论：JWT vs Session，倾向 JWT。", "days_ago": 70},    {"text": "API 限流方案第二次讨论，sliding window 实现复杂度太高，倾向 token bucket。", "days_ago": 60},    {"text": "前端框架选 React，状态管理用 Zustand 而不是 Redux。", "days_ago": 55},    {"text": "CI/CD 用 GitHub Actions，部署脚本已经写好。", "days_ago": 50},    {"text": "日志方案选 Datadog，已经接入。", "days_ago": 45},    {"text": "API 网关用 Kong，限流插件测试中。", "days_ago": 40},    {"text": "Kong 限流插件测试完成，性能符合预期，准备上线。", "days_ago": 30},    {"text": "JWT 有效期定为 7 天，refresh token 有效期 30 天。", "days_ago": 28},    {"text": "PostgreSQL 连接池大小定为 20，经过压测验证。", "days_ago": 25},    {"text": "前端构建时间优化到 45 秒，用了 esbuild。", "days_ago": 22},    {"text": "API 文档用 Swagger 自动生成，已经部署到内网。", "days_ago": 20},    {"text": "API 限流方案最终确认：token bucket，每用户每分钟 60 次请求。", "days_ago": 14},    {"text": "压测结果：QPS 峰值 2000，P99 延迟 120ms，符合预期。", "days_ago": 12},    {"text": "上线前 checklist 完成，安全扫描通过。", "days_ago": 10},    {"text": "灰度发布策略：先放 5% 流量，观察 24 小时。", "days_ago": 8},    {"text": "监控告警阈值设置完成：错误率超 1% 触发 PagerDuty。", "days_ago": 7},    {"text": "灰度发布顺利，错误率 0.02%，准备全量上线。", "days_ago": 5},    {"text": "全量上线完成，API 限流生效，未见异常。", "days_ago": 4},    {"text": "用户反馈限流太严，部分场景需要提高到 120 次/分钟。", "days_ago": 3},    {"text": "和产品确认：付费用户限流提高到 120 次/分钟，免费用户保持 60 次。", "days_ago": 2},    {"text": "API 限流最终方案更新：免费 60 次/分钟，付费 120 次/分钟，已上线。", "days_ago": 1},    # ... 其余 25 条同理，覆盖其他话题]def get_embedding(text):    resp = client.embeddings.create(input=text, model="text-embedding-3-small")    return resp.data[0].embeddingprint("正在生成 embeddings...")data = []for i, m in enumerate(memories):    ts = int((now - timedelta(days=m["days_ago"])).timestamp())    data.append({        "text": m["text"],        "vector": get_embedding(m["text"]),        "timestamp": ts,        "metadata": json.dumps({"group": "project", "importance": "normal"}),    })    print(f"  [{i+1}/{len(memories)}] 完成")milvus.insert(collection_name=COLLECTION_NAME, data=data)print(f"写入完成，共 {len(data)} 条")

第二步：三组检索对比

A 组模拟CLAUDE.md全量写入，B 组纯语义检索，C 组加上 Exponential Decay 重排。

（参数说明：scale=7 天是项目决策场景的经验值——一周内的决策还算当前，一个月前约剩 6% 权重，三个月前接近 0；如果你的场景是年度规划跟踪，可以把 scale 拉到 30 天。）

import timefrom datetime import datetimefrom openai import OpenAIfrom pymilvus import MilvusClient, Function, FunctionTypeclient = OpenAI()milvus = MilvusClient(    uri="http://your-milvus-host:19530",    user="root",    password="your-password")COLLECTION_NAME = "nanoclaw_memory"TEST_QUERY = "API 限流方案最后怎么定的？"def get_embedding(text):    resp = client.embeddings.create(input=text, model="text-embedding-3-small")    return resp.data[0].embeddingquery_vector = get_embedding(TEST_QUERY)# A 组：全量塞入（模拟 CLAUDE.md）all_memories = milvus.query(    collection_name=COLLECTION_NAME,    filter="",    output_fields=["text", "timestamp"],    limit=50)# B 组：纯语义检索 top-5results_b = milvus.search(    collection_name=COLLECTION_NAME,    data=[query_vector],    limit=5,    output_fields=["text", "timestamp"])[0]# C 组：语义检索 + Exponential Decay 重排decay_ranker = Function(    name="memory_recency",    input_field_names=["timestamp"],    function_type=FunctionType.RERANK,    params={        "reranker": "decay",        "function": "exp",        "origin": int(time.time()),        "offset": 24 * 3600,    # 1 天内不打折        "decay": 0.5,           # 每过一个 scale，得分乘以 0.5        "scale": 7 * 24 * 3600  # 7 天半衰期    })results_c = milvus.search(    collection_name=COLLECTION_NAME,    data=[query_vector],    limit=5,    output_fields=["text", "timestamp"],    ranker=decay_ranker)[0]print("=== A 组（CLAUDE.md 全量，50 条全部塞进 prompt，此处展示前 5 条）===")for m in all_memories[:5]:    ts = datetime.fromtimestamp(m["timestamp"]).strftime("%Y-%m-%d")    print(f"  [{ts}] {m['text']}")print("\n=== B 组（纯语义检索 top-5）===")for r in results_b:    ts = datetime.fromtimestamp(r["entity"]["timestamp"]).strftime("%Y-%m-%d")    print(f"  [{ts}] {r['entity']['text']}")print("\n=== C 组（语义 + Decay 重排 top-5）===")for r in results_c:    ts = datetime.fromtimestamp(r["entity"]["timestamp"]).strftime("%Y-%m-%d")    print(f"  [{ts}] {r['entity']['text']}")

第三步：喂给 Agent 对比回答

from openai import OpenAIfrom pymilvus import MilvusClientfrom pymilvus import Function, FunctionTypeimport timefrom datetime import datetimeclient = OpenAI()milvus = MilvusClient(    uri="http://192.168.7.122:19530",    user="root",    password="Milvus"  )COLLECTION_NAME = "nanoclaw_memory"TEST_QUERY = "API 限流方案最后怎么定的？"def get_embedding(text):    resp = client.embeddings.create(input=text, model="text-embedding-3-small")    return resp.data[0].embeddingdef ask_agent(context, query):    resp = client.chat.completions.create(        model="gpt-5",        messages=[            {"role": "system", "content": f"你是一个项目助手，以下是你的记忆：\n\n{context}"},            {"role": "user", "content": query}        ]    )    return resp.choices[0].message.contentquery_vector = get_embedding(TEST_QUERY)# A 组上下文all_memories = milvus.query(    collection_name=COLLECTION_NAME,    filter="",    output_fields=["text"],    limit=50)context_a = "\n".join([m["text"] for m in all_memories])# B 组上下文results_b = milvus.search(    collection_name=COLLECTION_NAME,    data=[query_vector],    limit=5,    output_fields=["text"])[0]context_b = "\n".join([r["entity"]["text"] for r in results_b])# C 组上下文decay_ranker = Function(    name="memory_recency",    input_field_names=["timestamp"],    function_type=FunctionType.RERANK,    params={        "reranker": "decay",        "function": "exp",        "origin": int(time.time()),        "offset": 24 * 3600,          "decay": 0.5,          "scale": 7 * 24 * 3600          })results_c = milvus.search(    collection_name=COLLECTION_NAME,    data=[query_vector],    limit=5,    output_fields=["text"],    ranker=decay_ranker)[0]context_c = "\n".join([r["entity"]["text"] for r in results_c])print("=== Agent 回答对比 ===\n")print("【A 组 - CLAUDE.md 全量】")print(ask_agent(context_a, TEST_QUERY))print("\n【B 组 - 纯语义检索】")print(ask_agent(context_b, TEST_QUERY))print("\n【C 组 - 语义 + Decay】")print(ask_agent(context_c, TEST_QUERY))

第四步：实验结果分析（核心结论）

三组策略的核心差异的可总结为以下3点：

全量方案：50 条全进 prompt，信息最完整，但对 Agent 来说噪音也最多，早期的背景讨论和最新决策混在一起，没有优先级。

纯语义检索方案：过滤了无关内容，但把三个月前关于 token bucket 的初步讨论也一并拉了上来——两条语义接近，排名差不多，Agent 回答时因此夹带了早期讨论的干扰。

Decay 方案：最近几天的记录顶到前排，Agent 直接给出：

“免费 60 次/分钟、付费 120 次/分钟，已全量上线，运行稳定。”

不只结论对，连当前运行状态也一并给出。

不过 Decay 有一个明显的失效场景，值得单独说。

如果你问的是”当初为什么放弃 GraphQL”，这条记录在三个月前，Decay 会把它的权重压到接近 0，C 组根本召回不到。

因此，Decay 只适合找最新决策，不适合追溯历史原因。这不是设计缺陷，需要我们对它的能力边界做好设计。

04 进阶优化：分类优先，Decay为辅，构建完整记忆生命周期

Decay函数解决了旧记忆降权的问题，但没有解决哪些记忆该降权、哪些不该降权的问题——三个月前的架构决策（如“放弃GraphQL选REST”）和昨天的参数调整（如“限流次数修改”），本就不该被同等对待。

因此，比调优Decay参数更根本的解决方案是：在记忆写入阶段就进行分类，再结合Decay函数和NanoClaw/OpenClaw的调度器，构建完整的记忆生命周期闭环。

具体操作方式如下：

第一步：记忆分类（写入阶段）

我们可以将OpenClaw/NanoClaw的记忆分为以下3类

1. permanent（长期记忆）：长期稳定的架构决策、技术选型、团队约定（如“放弃GraphQL选REST”“数据库选PostgreSQL”）——不会随时间失效，无需衰减；

2. temporal（临时记忆）：进度更新、临时状态、会议纪要（如“灰度发布观察24小时”）——时效性极强，过期后无实际价值，需重点衰减；

3. review（待核查记忆）：参数、阈值、对外策略（如“API限流次数”“JWT有效期”）——当前有效，但未来可能被修改，需定期核查，无需衰减但需跟踪更新。

补充说明：temporal和review的核心区别在于，temporal是过期就没用，review是不一定过期，但可能被修改。比如限流次数从60次改为120次，旧记录不会自动失效，若不分类，会和新记录一起混入prompt，导致Agent给出错误答案。

关于如何做分类，我们可以用一次LLM调用将分类步骤插入前文的建库流程，具体代码如下：

def classify_memory(text: str) -> str:    """    对记忆文本做类型分类，返回 'permanent' | 'temporal' | 'review'    """    resp = client.chat.completions.create(        model="gpt-4o-mini",        messages=[            {                "role": "system",                "content": (                    "你是一个项目记忆分类器，将输入文本归入以下三类之一，只输出分类名，不要解释：\n"                    "- permanent：长期稳定的架构决策、技术选型、团队约定\n"                    "- temporal：进度更新、临时状态、会议纪要（时效性强，过期无用）\n"                    "- review：参数、阈值、对外策略（当前有效，但未来可能被修改，需定期核查）"                )            },            {"role": "user", "content": text}        ]    )    result = resp.choices[0].message.content.strip().lower()    return result if result in ("permanent", "temporal", "review") else "temporal"

Schema 里加一个字段：

schema.add_field("memory_type", DataType.VARCHAR, max_length=20)# 写入时：data.append({    "text": m["text"],    "vector": get_embedding(m["text"]),    "timestamp": ts,    "memory_type": classify_memory(m["text"]),   # ← 新增    "metadata": json.dumps({"group": "project"}),})

分类结果会持久化到Milvus中，后续检索直接按memory_type过滤，不用每次查询时再判断。

第二步：检索策略优化（按类型路由）

写入时分好类后，检索时可针对不同类型的记忆，采用不同的策略，精准匹配查询场景，避免Decay的失效问题：

#---- 找最新决策（问"现在是什么状态"）----# temporal类：应用 Decay，旧记录自动降权results_temporal = milvus.search(    collection_name=COLLECTION_NAME,    data=[query_vector],    filter='memory_type == "temporal"',    limit=3,    output_fields=["text", "timestamp", "memory_type"],ranker=decay_ranker)[0]# review 类：不做 Decay，但优先返回最新的（它们是当前有效值）results_review = milvus.search(    collection_name=COLLECTION_NAME,    data=[query_vector],    filter='memory_type == "review"',    limit=2,    output_fields=["text", "timestamp", "memory_type"])[0]# permanent 类：不做 Decay，语义相关就召回results_permanent = milvus.search(    collection_name=COLLECTION_NAME,    data=[query_vector],    filter='memory_type == "permanent"',    limit=2,    output_fields=["text", "timestamp", "memory_type"])[0]#合并三组，按优先级组装promptcontext = "\n".join([    *[r["entity"]["text"] for r in results_review],      # 优先：当前参数    *[r["entity"]["text"] for r in results_temporal],    # 其次：近期状态    *[r["entity"]["text"] for r in results_permanent],   # 背景：长期决策])

举例说明：查询“当初为什么放弃GraphQL”时，该记忆会被分类为permanent，检索时单独走permanent分支，Decay不会对其降权，能精准召回，解决了Decay的失效问题。

第三步：review类记忆定时核查

这是分类带来的第三个收益：review类记忆可以被 NanoClaw 的定时任务主动浮出来核查。

NanoClaw 的调度器每 60 秒轮询一次，支持 cron/interval/one-time 三种触发模式，可以把“检查 review 记忆”注册成一个定期任务：

# 每周一次，查出所有 review 类、超过 14 天未更新的记忆def fetch_stale_review_memories(days_threshold: int = 14):    cutoff_ts = int((datetime.now() - timedelta(days=days_threshold)).timestamp())    stale = milvus.query(        collection_name=COLLECTION_NAME,        filter=f'memory_type == "review" && timestamp < {cutoff_ts}',        output_fields=["id", "text", "timestamp"],    )    return stalestale_memories = fetch_stale_review_memories()if stale_memories:    summary = "\n".join([f"- [{datetime.fromtimestamp(m['timestamp']).strftime('%Y-%m-%d')}] {m['text']}"                         for m in stale_memories])    # 推送给 Agent 或通知渠道，请求确认这些参数是否仍然有效    ask_agent(        context=summary,        query="以上记录超过 14 天未更新，请逐条确认：是否仍然有效？如已变更请告知新值。"    )

到这里，整个方案的轮廓就完整了：通过写入分类+检索路由+定时核查，结合Milvus Decay函数和NanoClaw/OpenClaw调度器，这样就能形成完整的记忆管理闭环：

permanent（长期记忆）：写入后长期保留，检索时语义匹配即召回，不干扰最新决策；
temporal（临时记忆）：写入后随时间自动降权，减少噪音，过期后无需手动清理；
review（待核查记忆）：写入后定期浮出核查，确认有效则更新时间戳，无效则标记过期，避免错误。

05 什么时候值得给OpenClaw做这样的优化？

NanoClaw选择CLAUDE.md而非向量库，本质也是权衡后的结果：因为向量检索+Decay会增加一层不透明性，与OpenClaw/NanoClaw代码精简、可整体理解的设计思路存在一定的矛盾。

至于是否值得改造，核心看你的使用场景，以下是一些经验分享：

不值得改造的场景：对话量小、记忆条数少，CLAUDE.md完全够用，改造的收益覆盖不了引入的复杂度。

值得考虑的场景：长期运行、记忆以月为单位积累——项目跟踪、知识库问答、跨周的定时汇报。到了这个体量，时效性就从理论上的隐患变成了实际干扰。

如果你的场景里既有找最新决策也有找历史原因，Section 04 里的三分类框架可以直接迁移：permanent走语义召回，temporal走 Decay 重排，review由调度器主动驱动，各类型按自己的路径走，不需要在参数上折中。

作者介绍

Zilliz黄金写手：尹珉

阅读推荐
官宣：Zilliz Cloud&Milvus发布CLI工具与官方Skill，让AI Agent成为专业VDB运维与开发助手
Vector Graph RAG 开源！一套向量数据库同时搞定语义检索+RAG多跳


黄仁勋GTC演讲上，Milvus为什么能够站稳非结构化数据处理C位
用RAG的思路做agent知识管理，为什么跑不通