推荐语
OpenClaw记忆系统存在时效性问题？试试用Milvus 2.6的Decay Function解决AI助手的信息选择困难症。核心内容：1. 分析OpenClaw等AI助手记忆系统的两大核心问题：Token消耗过高和记忆无优先级区分2. 介绍Milvus 2.6的Decay Function如何通过时间衰减算法优化RAG检索3. 通过实验验证三种检索方式对比，展示Decay Function的实际效果
杨芳贤
53AI创始人/腾讯云(TVP)最具价值专家
用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实际使用场景）：
wget https://github.com/milvus-io/milvus/releases/download/v2.6.11/milvus-standalone-docker-compose.yml -O docker-compose.yml
pip install pymilvus openai
这里我们模拟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)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}, ]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)all_memories = milvus.query( collection_name=COLLECTION_NAME, filter="", output_fields=["text", "timestamp"], limit=50)results_b = milvus.search( collection_name=COLLECTION_NAME, data=[query_vector], limit=5, output_fields=["text", "timestamp"])[0]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", "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']}")
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)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])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])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))
全量方案：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的失效问题：
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]results_review = milvus.search( collection_name=COLLECTION_NAME, data=[query_vector], filter='memory_type == "review"', limit=2, output_fields=["text", "timestamp", "memory_type"])[0]results_permanent = milvus.search( collection_name=COLLECTION_NAME, data=[query_vector], filter='memory_type == "permanent"', limit=2, output_fields=["text", "timestamp", "memory_type"])[0]context = "\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类记忆可以被 NanoClaw 的定时任务主动浮出来核查。NanoClaw 的调度器每 60 秒轮询一次，支持 cron/interval/one-time 三种触发模式，可以把“检查 review 记忆”注册成一个定期任务：
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]) 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由调度器主动驱动，各类型按自己的路径走，不需要在参数上折中。
粤ICP备14082021号
免费POC，
零成本试错
