1. 为什么你的 Agent 总是“失忆”从短期缓存到长期知识图谱的真实痛点AI Agent 的记忆机制说白了就是让智能体在多次对话、多个任务之间记住该记的东西忘掉该忘的东西。它适合所有正在用大模型 API 做智能客服、个人助理、自动化工作流的开发者。你大概率遇到过这种场景用户第一轮说“我对芒果过敏”第五轮问“推荐个甜品”Agent 兴高采烈地推了芒果班戟。或者处理一份 80 页的合同模型读到后面忘了前面的付款条款输出自相矛盾。再或者多轮对话超过十几轮token 窗口告急只能粗暴截断对话连贯性直接崩掉。这些问题的根子不在模型不够强而在 Harness Engineering 层缺少一套分层的记忆骨架。短期缓存负责当前会话的连贯工作记忆负责当前任务的推理长期知识图谱负责跨会话的知识复用。三者各司其职又通过统一的调度器串起来。我试过把这套骨架落到一个真实项目里配合 TaoToken 的统一 Key/API 通道整个接入过程比想象中顺滑很多。下面我会把 config.toml 和 settings.json 的骨架直接给你再演示一次完整的记忆读写链路验证。2. TaoToken 前置统一 Key 与 API 通道的接入准备在动手写记忆系统之前先把模型调用通道理顺。TaoToken 提供统一的 API 入口你不需要在代码里散落多个厂商的 base_url 和 key。官网地址是 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content API 入口是 https://taotoken.net/api 。注意 API 地址后面不加 UTM 参数保持干净。你需要先拿到一个 API Key。进入控制台创建即可https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_contentconsoleutm_campaignrewrite 。创建完成后在 API Keys 页面复制密钥https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi-keysutm_campaignrewrite 。这个 Key 会同时用于对话模型和 embedding 模型记忆系统里的实体抽取、向量生成都走同一条通道。如果你后续要做长期编码或 Agent 类任务可以关注 Coding Planhttps://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding-planutm_campaignrewrite 。模型对话调试入口在https://taotoken.net/models?utm_sourcetaotoken_aicg_blog_endutm_contentmodelsutm_campaignrewrite 。接入文档在https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 。ClaudeCodeAnthropic 相关配置参考https://taotoken.net/claudecode-anthropic?utm_sourcetaotoken_aicg_blog_endutm_contentclaudecode-anthropicutm_campaignrewrite 。注意所有请求都通过 https://taotoken.net/api 发起不要在代码里硬编码其他第三方地址。Key 只存在服务端环境变量或配置文件里不要提交到公开仓库。3. 可复制配置config.toml 与 settings.json 骨架3.1 config.toml记忆系统的主配置这个文件放在项目根目录负责定义三层记忆的存储参数、模型通道和调度策略。你可以直接复制把api_key替换成自己的。# config.toml - Agent 记忆系统主配置 [llm] base_url https://taotoken.net/api api_key sk-your-taotoken-key chat_model gpt-4o-mini embedding_model text-embedding-3-small timeout 30 max_retries 3 [short_term_memory] backend redis redis_url redis://localhost:6379/0 max_tokens 4000 ttl_seconds 86400 recent_rounds 5 [working_memory] backend faiss embedding_dim 1536 max_fragments 100 top_k 5 [long_term_fact_memory] backend faiss index_path ./data/ltm_fact.index metadata_path ./data/ltm_fact_metadata.npy top_k 3 score_alpha 0.4 score_beta 0.3 score_gamma 0.3 decay_lambda 0.01 [long_term_graph_memory] backend neo4j uri bolt://localhost:7687 user neo4j password Agent123456 max_entity_types 10 [memory_harness] enable_graph_retrieval true enable_fact_retrieval true context_max_tokens 60003.2 settings.json运行时环境变量映射有些部署环境不方便读 toml可以用 settings.json 做一层映射。它和 config.toml 二选一即可我一般用 toml 做主配置json 做本地覆盖。{ TAOTOKEN_BASE_URL: https://taotoken.net/api, TAOTOKEN_API_KEY: sk-your-taotoken-key, REDIS_URL: redis://localhost:6379/0, NEO4J_URI: bolt://localhost:7687, NEO4J_USER: neo4j, NEO4J_PASSWORD: Agent123456, STM_MAX_TOKENS: 4000, WM_MAX_FRAGMENTS: 100, LTM_TOP_K: 3, GRAPH_ENABLED: true }3.3 配置加载代码在 Python 里读取这两个文件初始化记忆调度器。import tomllib import json import os from pathlib import Path def load_config(config_path: str config.toml) - dict: with open(config_path, rb) as f: cfg tomllib.load(f) # 环境变量覆盖 if os.getenv(TAOTOKEN_API_KEY): cfg[llm][api_key] os.getenv(TAOTOKEN_API_KEY) if os.getenv(TAOTOKEN_BASE_URL): cfg[llm][base_url] os.getenv(TAOTOKEN_BASE_URL) return cfg def load_settings(settings_path: str settings.json) - dict: p Path(settings_path) if p.exists(): return json.loads(p.read_text(encodingutf-8)) return {} if __name__ __main__: cfg load_config() settings load_settings() print(LLM base_url:, cfg[llm][base_url]) print(STM max_tokens:, cfg[short_term_memory][max_tokens]) print(Graph enabled:, cfg[memory_harness][enable_graph_retrieval])运行后你应该看到 base_url 指向 https://taotoken.net/api STM 的 token 上限是 4000图谱检索开关为 true。这一步确认配置链路通了再往下写记忆模块。4. 三层记忆模块的实现与验证4.1 短期缓存按 token 动态截断短期记忆存在 Redis 里核心逻辑不是按轮数截断而是按实际 token 数动态裁剪。这样不同长度的消息不会把窗口撑爆。import redis import tiktoken import time from typing import List, Dict from pydantic import BaseModel class Message(BaseModel): role: str content: str timestamp: float time.time() class ShortTermMemory: def __init__(self, redis_url: str, max_tokens: int 4000, ttl: int 86400): self.redis redis.from_url(redis_url) self.max_tokens max_tokens self.ttl ttl self.tokenizer tiktoken.get_encoding(cl100k_base) def _count_tokens(self, messages: List[Message]) - int: total 0 for msg in messages: total len(self.tokenizer.encode(msg.content)) 4 return total 3 def add_message(self, session_id: str, role: str, content: str) - None: key fstm:{session_id} raw self.redis.lrange(key, 0, -1) messages [Message.model_validate_json(m) for m in raw] if raw else [] messages.append(Message(rolerole, contentcontent)) while self._count_tokens(messages) self.max_tokens: messages.pop(0) self.redis.delete(key) self.redis.rpush(key, *[m.model_dump_json() for m in messages]) self.redis.expire(key, self.ttl) def get_messages(self, session_id: str) - List[Dict]: key fstm:{session_id} raw self.redis.lrange(key, 0, -1) return [Message.model_validate_json(m).model_dump() for m in raw] if raw else []验证动作连续写入 10 条长消息观察 Redis 里保留的条数是否随 token 数动态变化。stm ShortTermMemory(redis_urlredis://localhost:6379/0, max_tokens500) for i in range(10): stm.add_message(sess_001, user, f这是第{i}条测试消息内容长度适中用于验证截断逻辑。) msgs stm.get_messages(sess_001) print(保留条数:, len(msgs)) print(首条内容:, msgs[0][content] if msgs else 空)如果 max_tokens 设为 500最终保留的条数会明显少于 10说明动态截断生效。4.2 工作记忆任务级语义检索工作记忆用 FAISS 做任务内的片段检索每个 task_id 一个独立索引。import faiss import numpy as np import uuid from sentence_transformers import SentenceTransformer class WorkingMemory: def __init__(self, embed_model: str BAAI/bge-small-zh-v1.5, max_fragments: int 100): self.model SentenceTransformer(embed_model) self.max_fragments max_fragments self.dimension 512 self.task_indices: Dict[str, faiss.IndexFlatL2] {} self.task_metadata: Dict[str, List[Dict]] {} def add_fragment(self, task_id: str, content: str, fragment_type: str general) - str: if task_id not in self.task_indices: self.task_indices[task_id] faiss.IndexFlatL2(self.dimension) self.task_metadata[task_id] [] emb self.model.encode([content])[0].astype(float32) self.task_indices[task_id].add(np.array([emb])) fid str(uuid.uuid4()) self.task_metadata[task_id].append({ fragment_id: fid, content: content, type: fragment_type, timestamp: time.time() }) if len(self.task_metadata[task_id]) self.max_fragments: self.task_metadata[task_id].pop(0) all_emb self.model.encode([f[content] for f in self.task_metadata[task_id]]).astype(float32) self.task_indices[task_id] faiss.IndexFlatL2(self.dimension) self.task_indices[task_id].add(all_emb) return fid def retrieve(self, task_id: str, query: str, top_k: int 5) - List[Dict]: if task_id not in self.task_indices: return [] q_emb self.model.encode([query]).astype(float32) distances, indices self.task_indices[task_id].search(q_emb, top_k) results [] for idx in indices[0]: if idx len(self.task_metadata[task_id]): results.append(self.task_metadata[task_id][idx]) return results验证动作往同一个 task 里写入三条片段然后查询其中一条的语义近似问题。wm WorkingMemory() wm.add_fragment(task_001, 用户需要分析2024年Q3的销售数据重点关注华东区) wm.add_fragment(task_001, 华东区Q3销售额为1200万同比增长15%) wm.add_fragment(task_001, 华南区Q3销售额为800万同比下降3%) res wm.retrieve(task_001, 华东区卖了多少, top_k2) for r in res: print(r[content])输出应该优先命中华东区那条说明语义检索在工作记忆层生效。4.3 长期事实记忆重要度评分排序长期事实记忆存用户偏好、业务规则这类跨会话知识检索时用综合评分排序。class LongTermFactMemory: def __init__(self, embed_model: str BAAI/bge-small-zh-v1.5, index_path: str ./data/ltm_fact.index): self.model SentenceTransformer(embed_model) self.dimension 512 self.index_path index_path self.metadata_path index_path.replace(.index, _metadata.npy) try: self.index faiss.read_index(index_path) self.metadata np.load(self.metadata_path, allow_pickleTrue).tolist() except Exception: self.index faiss.IndexFlatL2(self.dimension) self.metadata [] def add_fact(self, user_id: str, content: str, importance: float 0.5) - str: emb self.model.encode([content])[0].astype(float32) self.index.add(np.array([emb])) fid str(uuid.uuid4()) self.metadata.append({ fact_id: fid, user_id: user_id, content: content, importance: importance, use_count: 0, create_time: time.time(), last_use_time: time.time() }) faiss.write_index(self.index, self.index_path) np.save(self.metadata_path, self.metadata) return fid def retrieve(self, user_id: str, query: str, top_k: int 3, alpha: float 0.4, beta: float 0.3, gamma: float 0.3) - List[Dict]: if not self.metadata: return [] q_emb self.model.encode([query]).astype(float32) distances, indices self.index.search(q_emb, 20) now time.time() candidates [] for rank, idx in enumerate(indices[0]): if idx len(self.metadata): continue meta self.metadata[idx] if meta[user_id] ! user_id: continue R 1 / (1 distances[0][rank]) delta_t (now - meta[create_time]) / 86400 T np.exp(-0.01 * delta_t) I meta[importance] * (meta[use_count] / (meta[use_count] 1)) S alpha * R beta * T gamma * I candidates.append({**meta, score: S}) candidates.sort(keylambda x: x[score], reverseTrue) top candidates[:top_k] for cand in top: for meta in self.metadata: if meta[fact_id] cand[fact_id]: meta[use_count] 1 meta[last_use_time] now break np.save(self.metadata_path, self.metadata) return top验证动作写入一条高重要度的过敏信息然后用新会话查询甜品推荐。ltm LongTermFactMemory(index_path./data/ltm_fact.index) ltm.add_fact(u001, 用户张三对芒果过敏不能吃含芒果的食物, importance0.9) res ltm.retrieve(u001, 推荐几个好吃的甜品, top_k2) for r in res: print(r[content], | score:, round(r[score], 4))输出应该包含过敏那条且评分靠前。这说明长期事实记忆在跨会话场景下能正确召回。4.4 长期知识图谱实体关系抽取与查询知识图谱用 Neo4j 存实体和关系实体抽取走 TaoToken 的对话模型。from neo4j import GraphDatabase from openai import OpenAI import json class LongTermGraphMemory: def __init__(self, uri: str, user: str, password: str, llm_client: OpenAI): self.driver GraphDatabase.driver(uri, auth(user, password)) self.llm llm_client def close(self): self.driver.close() def _extract(self, content: str, user_id: str) - dict: prompt f从以下文本抽取实体和关系只输出JSON 文本{content} 用户ID{user_id} 格式{{entities:[{{id:e1,type:人物,value:张三}}],relations:[{{from:e1,to:e2,relation:领导}}]}} resp self.llm.chat.completions.create( modelgpt-4o-mini, messages[{role: user, content: prompt}], temperature0 ) return json.loads(resp.choices[0].message.content) def add_to_graph(self, content: str, user_id: str) - None: data self._extract(content, user_id) with self.driver.session() as session: for ent in data.get(entities, []): session.run( MERGE (e:Entity {id: $id, type: $type, value: $value, user_id: $uid}), ident[id], typeent[type], valueent[value], uiduser_id ) for rel in data.get(relations, []): session.run( MATCH (a:Entity {id: $from_id}), (b:Entity {id: $to_id}) MERGE (a)-[r:RELATION {type: $rel_type}]-(b), from_idrel[from], to_idrel[to], rel_typerel[relation] ) def query_graph(self, cypher: str) - List[Dict]: with self.driver.session() as session: result session.run(cypher) return [dict(record) for record in result]验证动作写入一条“张三是李四的领导李四在腾讯上班”然后查询李四的上级。llm_client OpenAI(base_urlhttps://taotoken.net/api, api_keysk-your-taotoken-key) graph LongTermGraphMemory(bolt://localhost:7687, neo4j, Agent123456, llm_client) graph.add_to_graph(张三是李四的领导李四在腾讯上班, u001) res graph.query_graph(MATCH (a:Entity {value:李四})-[:RELATION]-(b:Entity) RETURN b.value AS leader) print(res)输出应该返回张三说明图谱写入和查询链路通了。5. 记忆读写链路完整验证与常见错排查5.1 调度器串联三层记忆把短期、工作、长期三层串起来组装上下文并更新记忆。class MemoryHarness: def __init__(self, cfg: dict, llm_client: OpenAI): self.stm ShortTermMemory(cfg[short_term_memory][redis_url], cfg[short_term_memory][max_tokens]) self.wm WorkingMemory(max_fragmentscfg[working_memory][max_fragments]) self.ltm_fact LongTermFactMemory(index_pathcfg[long_term_fact_memory][index_path]) self.ltm_graph LongTermGraphMemory(cfg[long_term_graph_memory][uri], cfg[long_term_graph_memory][user], cfg[long_term_graph_memory][password], llm_client) def build_context(self, session_id: str, task_id: str, user_id: str, query: str) - str: parts [] stm_msgs self.stm.get_messages(session_id) if stm_msgs: parts.append(### 最近对话) for m in stm_msgs[-5:]: parts.append(f{m[role]}: {m[content]}) wm_res self.wm.retrieve(task_id, query, top_k3) if wm_res: parts.append(### 当前任务相关) for r in wm_res: parts.append(f- {r[content]}) ltm_res self.ltm_fact.retrieve(user_id, query, top_k2) if ltm_res: parts.append(### 历史偏好) for r in ltm_res: parts.append(f- {r[content]}) return \n.join(parts) def update(self, session_id: str, task_id: str, user_id: str, query: str, response: str, task_finished: bool False) - None: self.stm.add_message(session_id, user, query) self.stm.add_message(session_id, assistant, response) self.wm.add_fragment(task_id, f问{query} 答{response}) if task_finished: for frag in self.wm.task_metadata.get(task_id, []): self.ltm_fact.add_fact(user_id, frag[content], importance0.6) self.wm.task_metadata.pop(task_id, None) self.wm.task_indices.pop(task_id, None)5.2 一次完整读写验证cfg load_config() llm_client OpenAI(base_urlcfg[llm][base_url], api_keycfg[llm][api_key]) harness MemoryHarness(cfg, llm_client) harness.update(s001, t001, u001, 我叫张三对芒果过敏, 好的记住了, task_finishedFalse) harness.update(s001, t001, u001, 我在做甜品推荐任务, 明白, task_finishedFalse) ctx harness.build_context(s001, t001, u001, 推荐几个甜品) print(ctx)输出应该同时包含最近对话、当前任务片段和历史偏好。如果历史偏好里没有过敏信息说明长期事实记忆的写入或检索有问题。5.3 常见错排查报错一openai.AuthenticationError或 401检查 config.toml 里的api_key是否以sk-开头以及 base_url 是否为https://taotoken.net/api。如果用了环境变量覆盖确认TAOTOKEN_API_KEY已导出。可以在模型对话页面先手动发一条消息验证 Key 有效https://taotoken.net/models?utm_sourcetaotoken_aicg_blog_endutm_contentmodelsutm_campaignrewrite 。报错二redis.exceptions.ConnectionErrorRedis 没启动或端口不对。用docker ps确认容器在跑或者把redis_url改成实际地址。短期记忆依赖 Redis 的 list 结构连不上会直接抛异常。报错三faiss.IndexFlatL2维度不匹配embedding 模型换了但dimension没改。bge-small-zh-v1.5 是 512 维text-embedding-3-small 是 1536 维。检查 config.toml 里embedding_dim和代码里self.dimension是否一致。报错四Neo4j 认证失败默认密码在首次启动后需要修改。如果用的是neo4j:5.20-community镜像NEO4J_AUTH环境变量设置的密码就是登录密码。确认uri是bolt://localhost:7687而不是http://。报错五知识图谱抽取返回非 JSON模型输出带了 markdown 代码块标记。在_extract里加一层清洗去掉json 和再解析。或者把 temperature 设为 0并在 prompt 里强调“只输出 JSON不要任何其他字符”。报错六长期记忆检索召回为空检查user_id是否一致。写入时用u001检索时也必须用u001。另外确认index_path和metadata_path指向同一目录且文件有写入权限。6. 语义一致的 CTA把记忆骨架接到你的 Agent 里这套三层记忆骨架的核心价值在于短期缓存保连贯工作记忆保任务长期知识图谱保复用。你不需要一次性全上可以先从短期缓存加长期事实记忆开始跑通读写链路后再引入图谱。配置里的config.toml和settings.json可以直接复制到项目里把api_key换成你自己的即可。如果你在接入过程中遇到 Key 或通道问题先去 API Keys 页面确认密钥状态https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi-keysutm_campaignrewrite 。接入文档里有完整的请求示例和参数说明https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 。需要长期跑编码或 Agent 任务的话Coding Plan 的额度模型更适合持续调用https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding-planutm_campaignrewrite 。最后留一个实用技巧每次任务结束后把工作记忆里 importance 高于 0.7 的片段归档到长期事实记忆低于 0.3 的直接丢弃。这样你的 Agent 会越用越懂用户而不是越用越臃肿。