微信扫码
添加专属顾问
OpenClaw记忆系统存在时效性问题?试试用Milvus 2.6的Decay Function解决AI助手的信息选择困难症。 核心内容: 1. 分析OpenClaw等AI助手记忆系统的两大核心问题:Token消耗过高和记忆无优先级区分 2. 介绍Milvus 2.6的Decay Function如何通过时间衰减算法优化RAG检索 3. 通过实验验证三种检索方式对比,展示Decay Function的实际效果
第一步:实验准备:环境搭建与数据模拟
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
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)} 条")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']}")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))“免费 60 次/分钟、付费 120 次/分钟,已全量上线,运行稳定。”
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"}),})#---- 找最新决策(问"现在是什么状态")----# 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], # 背景:长期决策])
# 每周一次,查出所有 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 天未更新,请逐条确认:是否仍然有效?如已变更请告知新值。" )53AI,企业落地大模型首选服务商
产品:场景落地咨询+大模型应用平台+行业解决方案
承诺:免费POC验证,效果达标后再合作。零风险落地应用大模型,已交付160+中大型企业
2026-08-31
史上“最大龙虾”发布:OpenClaw 2.0要解决Agent的三大难题
2026-08-31
OpenClaw 迎来史上最大更新,龙虾之父在线答网友问
2026-08-31
刚刚,OpenClaw 史上最大更新来了,快凉的龙虾拿到了大结果
2026-07-23
别把聊天记录全塞进知识库:AI 记忆到底怎样才省钱、找得准?
2026-07-13
网易智企AI Native实践:人人都会用AI之后,真正的难题才开始
2026-06-30
运维界的 OpenClaw 来了!
2026-06-30
刚刚,OpenClaw和Cursor杀入手机!Agent从此塞进口袋
2026-06-21
openclaw深度实践(四种场景:企业提效参考)
2026-07-23
2026-04-09
2026-04-07
2026-04-02
2026-03-30
2026-03-30
2026-03-26
2026-03-24
欢迎您使用【53AI 官方网站】(以下简称“本网站”或“我们”)。本《会员服务协议》(以下简称“本协议”)是您(以下简称“会员”或“用户”)与【深圳市博思协创网络科技有限公司】之间关于注册、登录及使用本网站会员服务所订立的法律协议。
在您注册或登录前,请务必审慎阅读、充分理解各条款内容,特别是免除或限制责任的条款、知识产权条款、争议解决条款等。此类条款将以加粗形式提示您注意。 当您通过微信公众号授权、手机验证码验证或其他方式成功登录本网站时,即视为您已完全理解并同意接受本协议的全部内容。
一、 定义
本网站:指由【深圳市博思协创网络科技有限公司】运营的,域名为【53ai.com】的网站及相关移动端页面。
会员服务:指本网站向注册会员提供的知识库文章查阅、内容检索及其他相关增值服务。
知识库内容:指本网站发布的包括但不限于文字、图表、数据、研究报告、行业分析等数字化内容资源。
二、 账号注册与登录
登录方式:本网站支持以下登录方式,您可根据实际情况选择:
微信公众号授权登录:您同意将您的微信OpenID信息授权给本网站,用于创建或关联会员账号。
手机验证码登录:您需提供真实有效的手机号码,并通过短信验证码完成身份验证与登录/注册。
账号安全:您的账号仅限您本人使用,禁止赠与、借用、租用、转让或售卖。因您保管不善导致的账号被盗、密码泄露等损失,由您自行承担。
实名认证:根据相关法律法规要求,我们可能要求您在特定功能下完成实名认证。如您拒绝提供,可能无法使用部分或全部服务。
未成年人保护:若您未满18周岁,请在法定监护人的陪同下阅读本协议,并在征得监护人同意后使用本服务。
三、 服务内容与规范
知识库查阅权限:会员登录后,有权按照其会员等级对应的权限范围,在线浏览、检索本网站知识库中的相关文章及内容。
服务变更:我们有权根据业务发展需要,调整、变更或终止部分服务内容,并将以网站公告、公众号消息等方式提前通知。
禁止行为:您在使用服务时不得实施以下行为:
利用技术手段批量爬取、下载、转存知识库内容;
将知识库内容用于商业目的或未经授权地向第三方传播;
干扰本网站正常运行或侵犯其他用户合法权益;
发布违法违规信息或从事违反公序良俗的活动。
四、 知识产权声明
权利归属:本网站知识库中的排版设计、软件代码等内容的知识产权均归【公司全称】或原权利人所有,受《中华人民共和国著作权法》等法律保护。
有限许可:本网站授予会员一项非独占、不可转让、不可转授权的普通许可,仅限于个人学习、研究之目的在线查阅知识库内容。
侵权追责:未经书面许可,任何单位或个人不得以任何形式复制、转载、摘编、镜像、汇编或以其他方式使用上述内容。一经发现,我们保留追究其法律责任的权利。
五、 个人信息保护
我们重视对您个人信息的保护。关于我们如何收集、使用、存储和保护您的个人信息,请单独阅读 《隐私政策》。
您通过微信公众号授权或手机号验证所提供的信息,我们将严格按照《个人信息保护法》的规定处理,仅用于身份识别、服务提供及安全验证等必要用途。
您可以随时通过网站设置或联系客服行使查阅、更正、删除个人信息及撤回授权同意的权利。
六、 免责声明
内容准确性:知识库内容仅供参考,不构成专业建议。我们不对其完整性、准确性、时效性作任何明示或暗示的保证,您应自行判断并承担使用风险。
不可抗力:因自然灾害、政策法规变化、网络故障、第三方平台接口异常(如微信接口维护、运营商短信通道故障)等不可抗力导致的服务中断或延迟,我们不承担违约责任。
第三方链接:本网站可能包含指向第三方网站的链接,该等网站的内容和服务不受我们控制,请您自行甄别风险。
七、 违约责任
如您违反本协议约定,我们有权视情节采取警告、限制功能、暂停服务、注销账号等措施,并保留要求赔偿损失的权利。
如因您的违约行为导致我们遭受行政处罚、第三方索赔或商誉损失,您应承担全部赔偿责任(包括但不限于罚款、赔偿金、律师费、公证费等)。
八、 法律适用与争议解决
本协议的订立、执行和解释均适用中华人民共和国大陆地区法律。
因本协议产生的或与本协议有关的任何争议,双方应友好协商解决;协商不成的,任何一方均可向【公司所在地】有管辖权的人民法院提起诉讼。
九、 其他
本协议构成双方就本服务达成的完整协议,取代此前任何口头或书面约定。
本协议任一条款被认定为无效或不可执行的,不影响其他条款的效力。
我们对本协议享有最终解释权,并在法律允许的范围内保留随时修改的权利。修改后的协议一经公布即生效,继续使用服务即视为同意修订内容。