1. 长对话跑着跑着就失忆问题出在记忆清理策略上如果你正在做 AI Agent 的生产落地大概率遇到过这种场景用户跟 Agent 聊了三十多轮突然问「刚才说的那个退款金额是多少」Agent 一脸茫然地开始重新推荐商品。或者更糟——用户半年前提过「我对花生过敏」Agent 今天推荐了一份含花生酱的食谱。这类问题表面看是「模型记性差」根子上其实是记忆过期策略没设计好。AI Agent Harness Engineering 里的记忆管理模块核心要解决的就是「哪些记忆该留、哪些该清、什么时候清」。传统做法无非 FIFO、LRU、TTL 三件套但它们都有一个致命缺陷只看单一维度。FIFO 只看创建时间会把用户最早提的核心诉求清掉LRU 只看最近访问时间会把低频但极高价值的健康信息清掉TTL 更粗暴所有记忆统一过期时间完全不区分价值差异。这篇要聊的是基于访问频率加重要性的双因子动态清理算法。简单说就是给每条记忆算一个「综合留存得分」得分低的先清得分高的留着而且清理阈值会根据记忆池的占用率动态调整。适合正在做 Agent 生产落地、被 Token 成本和长对话准确率两头夹击的开发者。下面我会给出可复制的 config.toml 骨架、完整的验证动作以及本地跑通记忆清理流程的具体步骤。2. 前置准备TaoToken 接入与项目初始化在开始配置记忆清理算法之前你需要先有一个能跑通的大模型接口。我这边用的是 TaoToken 的 API 服务它兼容 OpenAI 的接口格式接入成本很低。如果你已经有其他可用的接口也可以直接替换 base_url 和 api_key。先拿到 API Key。访问 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi_keysutm_campaignrewrite 创建一个新的 Key复制出来备用。然后确认你的项目环境python --version # 需要 3.10 及以上 pip install langchain chromadb openai pydantic numpy python-dotenv如果你需要更详细的接入文档可以参考 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 里面有完整的接口说明和参数列表。环境变量文件.env这样写OPENAI_API_KEY你的TaoToken_API_Key OPENAI_BASE_URLhttps://taotoken.net/api注意base_url 不要加多余的路径后缀OpenAI SDK 会自动拼接/v1/chat/completions。如果你手动加了/v1反而会 404。项目目录结构建议这样组织agent-memory-cleaner/ ├── .env ├── config.toml ├── main.py ├── memory_item.py ├── importance_scorer.py ├── frequency_scorer.py ├── cleaner.py └── chroma_db/3. 可复制配置config.toml 骨架与参数说明把配置从代码里抽出来放到 config.toml是 Harness Engineering 的基本功。这样调参不用改代码也方便不同环境用不同配置。下面是我实测下来比较稳的一套骨架[memory] max_memory_size 10000 high_occupancy_threshold 0.8 low_occupancy_threshold 0.5 clean_batch_ratio 0.1 [scoring.weights] # 用户信息类重要性权重拉满频率权重压低 user_info_w_imp 0.9 user_info_w_freq 0.1 user_info_alpha 0.7 # 工具调用结果时效性强频率权重更高 tool_result_w_imp 0.4 tool_result_w_freq 0.6 tool_result_alpha 0.5 # 规划类记忆中等偏上 plan_w_imp 0.6 plan_w_freq 0.4 plan_alpha 0.6 # 反思类记忆 reflection_w_imp 0.7 reflection_w_freq 0.3 reflection_alpha 0.6 [scoring.importance] llm_weight 0.8 rule_weight 0.2 high_value_keywords [过敏, 病史, 密码, 合同, 退款, 赔偿, 约定, 身份证] low_value_keywords [天气, 几点, 笑话, 闲聊] [storage] hot_path ./chroma_db cold_enabled false cold_bucket agent-memory-cold [cleanup] trigger_on_occupancy true daily_cron 0 3 * * *几个关键参数解释一下。max_memory_size是热存储的最大容量超过这个数就会触发清理。high_occupancy_threshold设成 0.8意思是占用率到 80% 就开始清理不要等到 100% 才动手。clean_batch_ratio控制每次最多清理当前记忆的 10%避免一次清太多导致误删。权重配置里w_imp和w_freq加起来必须等于 1。alpha是访问频率得分里访问次数的权重越大越看重「被访问了多少次」越小越看重「最近一次访问是什么时候」。4. 核心算法实现双因子打分与动态阈值4.1 记忆条目数据结构先用 Pydantic 定义记忆条目把重要性得分、访问次数、最近访问时间这些字段都带上from pydantic import BaseModel, Field from datetime import datetime from typing import Optional, List class MemoryItem(BaseModel): id: str content: str embedding: Optional[List[float]] None importance_score: float Field(ge0, le10, default0) visit_count: int Field(default0, ge0) last_visit_time: datetime Field(default_factorydatetime.now) create_time: datetime Field(default_factorydatetime.now) memory_type: str other storage_layer: str hot4.2 重要性打分规则兜底加模型打分重要性打分不能全交给大模型因为模型偶尔会抽风。我的做法是规则先兜底命中高价值关键词直接给 10 分命中低价值关键词直接给 0 分剩下的才交给模型打分import os from openai import OpenAI from dotenv import load_dotenv load_dotenv() client OpenAI( api_keyos.getenv(OPENAI_API_KEY), base_urlos.getenv(OPENAI_BASE_URL) ) HIGH_VALUE {过敏, 病史, 密码, 合同, 退款, 赔偿, 约定, 身份证} LOW_VALUE {天气, 几点, 笑话, 闲聊} def score_importance(content: str, memory_type: str) - float: for kw in HIGH_VALUE: if kw in content: return 10.0 for kw in LOW_VALUE: if kw in content: return 0.0 base 6.0 if memory_type user_info else 3.0 prompt f给以下记忆打分0-10分只返回数字。 0分完全无价值的临时信息 5分中等价值短期可能有用 10分极高价值绝对不能丢失 记忆内容{content} 记忆类型{memory_type} try: resp client.chat.completions.create( modelgpt-3.5-turbo, messages[{role: user, content: prompt}], temperature0, max_tokens10 ) llm_score float(resp.choices[0].message.content.strip()) llm_score max(0, min(10, llm_score)) except Exception as e: print(f打分失败用基础分{e}) llm_score base return round(0.2 * base 0.8 * llm_score, 2)4.3 访问频率得分对数衰减避免误删访问频率得分要综合「访问次数」和「最近访问时间」两个因素。访问次数做归一化最近访问时间用对数衰减这样短期没访问的记忆不会得分暴跌import numpy as np from datetime import datetime def score_frequency(metadata: dict, max_visit: int, alpha: float) - float: visit_count int(metadata.get(visit_count, 0)) last_visit datetime.fromisoformat( metadata.get(last_visit_time, datetime.now().isoformat()) ) hours (datetime.now() - last_visit).total_seconds() / 3600 part1 alpha * (visit_count / max(max_visit, 1)) part2 (1 - alpha) * (1 / np.log(hours 2)) score (part1 part2) * 10 return round(max(0, min(10, score)), 2)为什么用ln而不是线性衰减举个例子线性衰减下10 小时没访问得分就归零了用ln的话10 小时没访问得分还有 0.4 左右100 小时也有 0.22。这样用户每周问一次的数据中间六天没访问也不会被清掉。4.4 综合得分与动态阈值综合得分就是两个维度加权def total_score(metadata: dict, max_visit: int, config: dict) - float: mtype metadata.get(memory_type, other) w_imp config.get(f{mtype}_w_imp, 0.5) w_freq config.get(f{mtype}_w_freq, 0.5) alpha config.get(f{mtype}_alpha, 0.6) s_imp float(metadata.get(importance_score, 0)) s_freq score_frequency(metadata, max_visit, alpha) return round(w_imp * s_imp w_freq * s_freq, 2)动态过期阈值根据占用率来算def expire_threshold(current: int, max_size: int) - float: if max_size 0: return 3.0 occ current / max_size if occ 0.5: return 3.0 elif occ 0.8: return 3.0 4 * (occ - 0.5) else: return 7.0占用率低于 50% 时阈值是 3基本不清理50% 到 80% 之间线性升到 7超过 80% 阈值锁在 7开始积极清理。5. 验证请求本地跑通清理流程并观察结果配置和代码都就位后写一个main.py来跑验证。先往记忆池里灌一批测试数据然后触发清理看返回结果import chromadb from memory_item import MemoryItem from importance_scorer import score_importance from cleaner import clean_expired_memory client chromadb.PersistentClient(path./chroma_db) collection client.get_or_create_collection(nameagent_memory) # 灌入测试记忆 test_memories [ (用户对花生过敏绝对不能推荐含花生的食品, user_info), (今天北京天气晴温度25度, tool_result), (用户要求全额退款订单号12345, user_info), (讲个笑话吧, other), (合同约定服务期限为2024年1月至2025年1月, user_info), ] for i, (content, mtype) in enumerate(test_memories): imp score_importance(content, mtype) collection.add( ids[fmem_{i}], documents[content], metadatas[{ importance_score: imp, visit_count: 0, last_visit_time: 2024-01-01T00:00:00, memory_type: mtype, storage_layer: hot }] ) # 触发清理 result clean_expired_memory(max_memory_count10000) print(result)跑完之后你会看到类似这样的返回{ cleaned_count: 2, moved_to_cold_count: 0, remaining_count: 3, expire_threshold: 3.0, occupancy: 0.0003 }因为占用率极低阈值只有 3.0所以只清掉了「天气」和「笑话」这两条低价值记忆。高价值的过敏信息、退款信息、合同条款都留下来了。如果你想验证高占用率下的清理效果可以把max_memory_count改成 5再跑一次result clean_expired_memory(max_memory_count5) print(result)这时候占用率变成 100%阈值升到 7.0清理会更激进。你可以观察哪些记忆被清掉、哪些被保留验证算法是否符合预期。6. 本篇常见错排查报错一openai.AuthenticationError: Incorrect API key provided检查.env里的OPENAI_API_KEY是否复制完整有没有多余空格。另外确认OPENAI_BASE_URL写的是https://taotoken.net/api不要加/v1后缀。报错二chromadb.errors.InvalidCollectionException集合名写错了或者还没创建。用get_or_create_collection而不是get_collection确保集合存在。报错三清理后重要记忆不见了检查importance_score字段是否正确写入。如果打分模块抛异常走了兜底逻辑可能所有记忆都拿了基础分。建议在score_importance里加日志确认每条记忆的实际得分。报错四ValueError: alpha must be between 0 and 1config.toml 里alpha值写错了。检查是不是把w_imp和w_freq搞混了或者某个类型的权重加起来不等于 1。报错五清理接口返回cleaned_count: 0但占用率很高检查expire_threshold的计算逻辑。如果max_memory_size设得太大占用率永远上不去阈值就一直是 3.0不会触发清理。把max_memory_size调小到实际记忆量的 1.2 倍左右比较合适。报错六minio.error.S3Error冷存储连接失败如果cold_enabled false清理时不会走冷存储逻辑。如果你开启了冷存储但 MinIO 没配好会报这个错。本地测试建议先关掉冷存储把cold_enabled设为false。7. 下一步把记忆清理接入你的 Agent 主流程本地跑通之后下一步就是把这个清理模块接入你的 Agent 主流程。两个接入点一是每次记忆被召回时调用update_memory_visit更新访问统计二是每次对话结束后检查占用率是否超过阈值超过就触发清理。如果你需要验证不同模型在记忆场景下的表现可以用 https://taotoken.net/models?utm_sourcetaotoken_aicg_blog_endutm_contentmodelsutm_campaignrewrite 上的模型对话功能快速对比。如果你在做长期的编码类 Agent需要更稳定的调用配额可以看看 https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding_planutm_campaignrewrite 的 Coding Plan。接入过程中遇到接口报错优先查 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi_keysutm_campaignrewrite 确认 Key 状态再对照 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 检查参数格式。我自己的经验是清理频率不要设太高每天凌晨跑一次定时清理就够了事件触发只在占用率超过 80% 时才启用。另外记得给清理操作加日志记录每次清了多少条、移了多少条到冷存储方便后续调参。