1. 为什么你的 Agent 总是“失忆”:从场景说起
如果你正在做智能体应用,大概率遇到过这种尴尬:用户上周明确说过“我司报销走钉钉审批”,这周再问,Agent 又推荐了一套飞书流程。你翻代码发现,历史对话确实存了,向量库也建了,可召回结果就是不对。问题不在模型笨,而在于你把“记忆”当成了“存储”。
Agent 记忆系统本质上是一套数据基础设施,不是把聊天记录往向量库里一塞就完事。它要回答四个问题:从对话里抽什么、新旧信息怎么合并、用什么结构存、查询时怎么找回正确证据。这四件事分别对应信息提取、记忆管理、记忆存储、信息检索四个模块。任何一个模块偷懒,长期运行后都会表现为“记不住”或“记错”。
这篇内容面向正在落地 Agent 的开发者,尤其是用 Python/Node 写后端、需要给智能体加长期记忆层的同学。我会用 TaoToken 作为统一的 LLM 调用通道,把四个模块串成一条可运行的链路,并给出config.toml和settings.json两份配置骨架,以及记忆写入、召回、衰减的验证动作。你不需要自己维护多套 API Key,也不用为每个模块单独接一家模型服务。
先说清楚 TaoToken 在这里的角色:它是一个统一的模型 API 通道,兼容 OpenAI 风格的接口,你可以在一个 Key 下调用不同模型来完成提取、摘要、检索重排等任务。官网入口是 https://taotoken.net/?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= ,API 基址是 https://taotoken.net/api 。下面所有配置都围绕这个基址展开。
2. 前置准备:用 TaoToken 统一 Key 打通四个模块
在动手写记忆层之前,先把调用通道固定下来。很多同学的做法是:提取用一个模型、摘要用一个模型、重排再用一个模型,结果三套 Key、三套限流、三套计费,排障时根本不知道是哪一环挂了。用 TaoToken 的好处是,你只需要一个 Key,就能在四个模块里按需切换模型。
2.1 获取 Key 与确认基址
登录后进入控制台,在 API Keys 页面创建一个新 Key。建议按项目命名,比如agent-memory-dev,方便后续轮换。创建后你会拿到一串以sk-开头的密钥,只显示一次,先复制到安全的地方。
接入文档在 https://taotoken.net/doc ,里面列出了当前支持的模型名和请求格式。基址统一用 https://taotoken.net/api ,不要带末尾斜杠,否则部分 SDK 会拼出双斜杠导致 404。
2.2 四个模块分别用什么模型
我的建议是分层用模型,而不是全用一个最贵的:
| 模块 | 任务 | 推荐模型档位 | 理由 |
|---|---|---|---|
| 信息提取 | 从对话抽实体、偏好、事实 | 中等 | 量大,要求稳定 JSON 输出 |
| 记忆管理 | 合并、更新、冲突判断 | 中高 | 需要语义判断,但不必最强 |
| 记忆存储 | 摘要压缩、分层归档 | 中等 | 可批量处理,控制成本 |
| 信息检索 | 查询改写、重排 | 中高 | 直接影响召回质量 |
你可以在config.toml里为每个模块指定不同的model字段,共用同一个api_key和base_url。这样既控制了成本,又避免了多 Key 管理的混乱。
2.3 环境变量与依赖
Python 侧建议用openaiSDK 即可,因为 TaoToken 兼容该协议。安装:
pip install openai tomli tomli-w numpyNode 侧用openai包同样可以。下面以 Python 为主演示,Node 的配置结构一致。
3. 可复制配置:config.toml 与 settings.json 骨架
这一节给出两份可直接落地的配置。config.toml负责模型通道和模块参数,settings.json负责记忆层的运行时策略。两者分离的好处是:换模型不动策略,调策略不动密钥。
3.1 config.toml:统一通道与模块模型
# config.toml [llm] api_key = "sk-你的TaoToken密钥" base_url = "https://taotoken.net/api" timeout = 60 max_retries = 3 [modules.extraction] model = "gpt-4o-mini" temperature = 0.1 max_tokens = 800 [modules.management] model = "gpt-4o" temperature = 0.2 max_tokens = 1200 [modules.storage] model = "gpt-4o-mini" temperature = 0.0 max_tokens = 600 [modules.retrieval] model = "gpt-4o" temperature = 0.0 max_tokens = 1000 [memory] short_term_ttl = 3600 mid_term_max_segments = 200 long_term_heat_threshold = 0.75 decay_half_life_days = 14这里的关键参数解释一下。short_term_ttl是短期记忆的存活秒数,超过就下沉到中期。mid_term_max_segments控制中期树的最大叶子数,超了触发聚合。long_term_heat_threshold是晋升到长期记忆的热度阈值,低于它的片段只降权不删除。decay_half_life_days是衰减半衰期,用来计算旧记忆的权重。
3.2 settings.json:记忆层运行时策略
{ "memory": { "layers": { "short_term": { "max_turns": 20, "store_raw": true }, "mid_term": { "structure": "tree", "segment_size": 5, "beam_width": 3 }, "long_term": { "structure": "vector", "top_k": 8, "min_score": 0.35 } }, "extraction": { "mode": "dual_track", "keep_raw_evidence": true, "schema": ["entity", "relation", "preference", "fact"] }, "management": { "conflict_policy": "versioned", "merge_strategy": "semantic_cluster", "filter_expired": true }, "retrieval": { "strategy": "hybrid", "use_query_rewrite": true, "use_rerank": true, "time_aware": true } } }dual_track表示结构化信息和原始片段双轨保存,这是避免“三元组压掉语气和细节”的关键。conflict_policy设为versioned,意味着新旧事实冲突时不直接覆盖,而是保留版本并标记当前有效项。time_aware打开后,检索会考虑时间戳,优先返回与查询时间相关的证据。
3.3 加载配置的代码
import tomli import json from openai import OpenAI def load_config(path="config.toml"): with open(path, "rb") as f: return tomli.load(f) def load_settings(path="settings.json"): with open(path, "r", encoding="utf-8") as f: return json.load(f) cfg = load_config() settings = load_settings() client = OpenAI( api_key=cfg["llm"]["api_key"], base_url=cfg["llm"]["base_url"], timeout=cfg["llm"]["timeout"], max_retries=cfg["llm"]["max_retries"], )到这里,通道和策略都固定了。接下来把四个模块接上。
4. 四模块落地:写入、召回、衰减的完整链路
4.1 信息提取:双轨保存,先保证据再谈结构
提取模块的任务是从一轮对话里抽出可长期保存的信息。我的做法是让模型输出 JSON,同时把原始对话片段一并存下。
EXTRACT_PROMPT = """从以下对话中提取长期记忆条目。 要求: 1. 输出 JSON 数组,每条包含 type、content、confidence。 2. type 取值:entity / relation / preference / fact。 3. 不要编造,只提取明确出现的信息。 4. 保留原始表述中的关键限定词。 对话: {conversation} """ def extract_memory(conversation: str): resp = client.chat.completions.create( model=cfg["modules"]["extraction"]["model"], temperature=cfg["modules"]["extraction"]["temperature"], max_tokens=cfg["modules"]["extraction"]["max_tokens"], messages=[{"role": "user", "content": EXTRACT_PROMPT.format(conversation=conversation)}], ) return resp.choices[0].message.content拿到结构化结果后,和原始片段一起写入短期记忆。注意keep_raw_evidence为 true 时,原始片段不参与摘要压缩,只做归档,供后续校验。
4.2 记忆管理:版本化更新,别直接覆盖
管理模块最容易踩的坑是“新事实直接覆盖旧事实”。正确做法是保留版本链。下面是一个冲突判断的示例:
MANAGE_PROMPT = """判断新记忆与已有记忆的关系。 已有记忆:{old} 新记忆:{new} 输出 JSON:{{"action": "add|update|merge|skip", "reason": "..."}} 如果新旧冲突,action 用 update,并保留旧版本。 """ def manage_memory(old: str, new: str): resp = client.chat.completions.create( model=cfg["modules"]["management"]["model"], temperature=cfg["modules"]["management"]["temperature"], max_tokens=cfg["modules"]["management"]["max_tokens"], messages=[{"role": "user", "content": MANAGE_PROMPT.format(old=old, new=new)}], ) return resp.choices[0].message.contentupdate时不要删除旧记录,而是给它打上superseded_by标记,并把新记录设为active。这样时间推理类查询才能回溯“之前是什么”。
4.3 记忆存储:三层结构,短期全量、中期树、长期向量
存储层按settings.json的三层来。短期记忆就是一个 FIFO 队列,超过max_turns就把最旧的按segment_size切段,送进中期树。中期树的叶子存片段摘要,父节点存聚合摘要。长期记忆只存热度超过阈值的片段向量。
import time import numpy as np class MemoryStore: def __init__(self, settings): self.short = [] self.mid_tree = {} self.long_vecs = [] self.long_meta = [] self.settings = settings def add_short(self, turn): self.short.append({"turn": turn, "ts": time.time()}) if len(self.short) > self.settings["memory"]["layers"]["short_term"]["max_turns"]: self._flush_short() def _flush_short(self): seg_size = self.settings["memory"]["layers"]["mid_term"]["segment_size"] while len(self.short) >= seg_size: seg = self.short[:seg_size] self.short = self.short[seg_size:] self._insert_mid(seg) def _insert_mid(self, seg): key = f"seg_{int(time.time()*1000)}" self.mid_tree[key] = {"leaves": seg, "heat": 0.0, "ts": time.time()} def decay(self): half_life = self.settings["memory"]["decay_half_life_days"] now = time.time() for key, node in self.mid_tree.items(): age_days = (now - node["ts"]) / 86400 node["heat"] *= 0.5 ** (age_days / half_life)衰减函数每轮对话后调用一次。热度低于阈值的节点不删除,只是不再参与晋升,检索时权重也降低。
4.4 信息检索:混合策略,先改写再重排
检索层用hybrid策略:先用 LLM 改写查询,再分别走向量和结构检索,最后重排。
REWRITE_PROMPT = """把用户问题改写成适合记忆检索的查询。 保留时间线索和实体名,去掉寒暄。 问题:{q} """ def rewrite_query(q: str): resp = client.chat.completions.create( model=cfg["modules"]["retrieval"]["model"], temperature=0.0, max_tokens=200, messages=[{"role": "user", "content": REWRITE_PROMPT.format(q=q)}], ) return resp.choices[0].message.content.strip()重排时把短期全量、中期 beam search 结果、长期向量 top-k 合并,按热度、时间、相似度加权排序。time_aware打开后,与查询时间接近的证据加分。
5. 验证请求:确认记忆写入、召回、衰减都生效
配置写完不算完,要跑三个验证动作。
5.1 验证写入
发一轮对话,检查短期队列和中期树是否更新:
store = MemoryStore(settings) store.add_short("用户说:我们团队用 Python 做后端,数据库是 PostgreSQL。") print(len(store.short)) # 应为 1连续灌入 25 轮后,len(store.short)应稳定在 20,中期树出现至少一个seg_节点。
5.2 验证召回
构造一个跨会话查询,看能否召回早期证据:
q = rewrite_query("我们后端用什么数据库?") print(q) # 期望输出类似:团队后端数据库 PostgreSQL然后用改写后的查询去长期向量里检索,min_score设为 0.35,应能命中包含 PostgreSQL 的片段。如果命中不了,先检查提取阶段是否把该事实标成了fact且confidence足够高。
5.3 验证衰减
手动把某个中期节点的ts改成 30 天前,调用decay(),观察heat是否按半衰期下降:
key = list(store.mid_tree.keys())[0] store.mid_tree[key]["ts"] -= 30 * 86400 store.decay() print(store.mid_tree[key]["heat"]) # 应明显低于初始值如果热度没降,检查decay_half_life_days是否被正确读取,以及decay()是否真的被调用。
5.4 用模型对话做快速回归
如果你想快速验证不同模型在提取和重排上的表现,可以直接在模型对话页面切换模型试跑同一段对话,对比 JSON 输出质量。入口在 https://taotoken.net/model-chat?utm_source=taotoken_aicg_blog_end&utm_content=model_chat&utm_campaign=rewrite 。这样不用改代码就能判断某个模型是否适合你的提取任务。
6. 本篇常见错排查
6.1 报错 401:Key 无效或基址写错
最常见的是base_url带了末尾斜杠,或者把官网地址当成了 API 地址。正确写法是https://taotoken.net/api,不带路径后缀。Key 如果复制时带了空格,也会 401,建议用strip()处理。
6.2 提取结果不是合法 JSON
模型偶尔会输出带 markdown 代码块的 JSON。处理方式是在解析前去掉json 和包裹,或者用json.loads前先做正则清洗。如果频繁出现,把temperature降到 0,并在 prompt 里强调“只输出 JSON,不要解释”。
6.3 召回总是返回最新记忆
这是近因偏差的典型表现。检查time_aware是否打开,以及长期检索是否只用了向量相似度。加入时间戳权重后,早期证据的得分会回升。另外确认conflict_policy是versioned,否则旧版本被覆盖后就真的找不回来了。
6.4 中期树无限增长
mid_term_max_segments没生效,通常是_insert_mid里没做上限判断。补一个检查:超过上限时,把热度最低的节点聚合到父节点,释放叶子。
6.5 衰减后记忆全没了
decay只降权不删除。如果发现节点消失,检查是否有别的地方调用了删除逻辑。另外heat初始值不要设成 0,否则乘任何系数都是 0,建议初始 1.0。
6.6 多模块并发时 Key 限流
四个模块共用一个 Key,高并发时可能触发限流。TaoToken 控制台可以看到用量,必要时在config.toml里给不同模块配不同 Key,或者加一个简单的令牌桶。接入文档里有关于重试和退避的说明,地址是 https://taotoken.net/doc 。
7. 长期编码与 Agent 场景的下一步
如果你打算把记忆层接到长期运行的编码 Agent 上,比如让它记住仓库约定、失败测试和用户偏好,那么调用量会明显上升,单次对话的 token 成本也需要控制。这种情况下可以关注 Coding Plan,它更适合持续性的编码和 Agent 任务,入口在 https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_content=coding_plan&utm_campaign=rewrite 。
回到记忆系统本身,四条工程原则值得贴在显示器上:层次结构比单层向量稳,原始证据不能丢,版本管理要显式,评测要覆盖上下文扩展和位置敏感性。把这四条落到config.toml和settings.json里,你的 Agent 才算真正有了长期可靠的记忆层。