AI Agent Harness Engineering 长时记忆实现：基于向量数据库的记忆存储与检索优化

1. 为什么你的 Agent 总是“失忆”长时记忆的真实工程困境如果你正在做 AI Agent 落地大概率遇到过这种场景用户上周明确说过“我对芒果过敏”这周问零食推荐Agent 依然热情地推了芒果干客服 Agent 昨天刚处理完一笔退款今天用户再问进度它又要求用户从头描述一遍。这不是模型不够聪明而是长时记忆链路没有工程化。大模型的上下文窗口再大也有上限128K Token 听起来很多但按中文算也就几万字跨天、跨周、跨月的交互根本装不下。更麻烦的是很多团队把“长时记忆”简单理解成“把所有历史对话塞进向量数据库”结果检索出来的内容要么不相关要么把半年前的过期信息当成当前事实幻觉反而更严重。这篇内容聚焦 AI Agent Harness Engineering 视角下的长时记忆实现把记忆当作 Agent 控制层统一管理的核心资源围绕向量数据库完成记忆写入、索引、检索、维护的完整链路。我会给出一套可复制的存储配置骨架、检索参数模板以及一套能直接跑的召回验证动作。适合有 Python 基础、正在做 Agent 后端或算法落地的工程师也适合想优化现有 Agent 记忆能力的技术负责人。读完之后你应该能在自己的 Agent 场景里完成长时记忆接入并用可量化的方式验证召回效果而不是凭感觉说“好像记住了”。2. TaoToken 前置准备把模型调用和记忆链路解耦长时记忆系统里有两个地方必须调用大模型一是写入时做重要性评分和实体抽取二是检索时做 Query 改写和记忆压缩。如果模型调用不稳定整条记忆链路都会抖动。我的做法是把模型调用统一走 TaoToken 的 API这样记忆服务本身不绑定某一家模型切换模型只需要改配置。TaoToken 的 API 地址是https://taotoken.net/api兼容 OpenAI 的调用格式所以前面代码里用的OpenAI客户端可以直接指向它。你需要先在控制台创建一个 API Key然后把它放进环境变量不要硬编码在代码里。export TAOTOKEN_API_KEYsk-你的key export TAOTOKEN_BASE_URLhttps://taotoken.net/api如果你还没创建 Key可以打开 API Keys 页面按提示生成https://taotoken.net/console/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi_keysutm_campaignrewrite创建好之后建议先用模型对话页面做一次连通性验证确认 Key 和模型名都能正常工作https://taotoken.net/model-chat?utm_sourcetaotoken_aicg_blog_endutm_contentmodel_chatutm_campaignrewrite这一步看起来简单但很多记忆检索不准的问题最后排查下来是模型调用超时或返回格式异常导致的。把模型层先稳定住后面的向量检索优化才有意义。3. 可复制的记忆存储配置骨架3.1 向量库 Collection 设计分区键是性能分水岭长时记忆最常见的部署形态是多租户也就是一个 Agent 服务同时服务很多用户。如果所有用户的记忆都塞进同一个 Collection 且不做分区检索时会扫描全表数据量一上来延迟直接爆炸。Milvus 的分区键partition key能把同一用户的记忆物理隔离检索时只扫当前用户分区多租户场景下 QPS 提升非常明显。下面是我实际在用的 Collection 初始化代码字段设计兼顾了向量检索和元数据过滤from pymilvus import connections, Collection, FieldSchema, CollectionSchema, DataType def init_milvus(): connections.connect(hostlocalhost, port19530) fields [ FieldSchema(namememory_id, dtypeDataType.VARCHAR, max_length64, is_primaryTrue), FieldSchema(nameuser_id, dtypeDataType.VARCHAR, max_length64, is_partition_keyTrue), FieldSchema(nameembedding, dtypeDataType.FLOAT_VECTOR, dim1024), FieldSchema(nameimportance_score, dtypeDataType.INT32), FieldSchema(namecreated_at, dtypeDataType.INT64), ] schema CollectionSchema(fields, descriptionlong_term_memory) collection Collection(namelong_term_memory, schemaschema, num_partitions64) index_params { metric_type: COSINE, index_type: IVF_SQ8, params: {nlist: 1024} } collection.create_index(field_nameembedding, index_paramsindex_params) collection.load() return collection这里有几个参数值得展开说。dim1024对应的是 bge-large-zh-v1.5 的输出维度如果你换 Embedding 模型这个值必须同步改。num_partitions64是分区数量一般设成用户量级的平方根附近太小隔离效果差太大元数据开销高。索引类型选IVF_SQ8是权衡后的结果它只占原始向量约 25% 的存储空间检索速度比 IVF_FLAT 快不少准确率损失大概 2 个百分点对千万级记忆量来说性价比最高。注意Milvus 的 Schema 一旦创建就不能改字段只能新建 Collection 迁移数据。所以importance_score、created_at这些过滤字段一定要提前规划好别等上线了才发现要加字段。3.2 记忆写入先评分再入库别什么都存写入环节最容易犯的错是“来者不拒”把用户所有输入都存成记忆。这样做的直接后果是记忆库迅速膨胀检索时噪声占比越来越高。我的策略是写入前先做重要性评分1 到 5 分不同分数对应不同的过期策略。import uuid from datetime import datetime, timedelta from langchain.text_splitter import RecursiveCharacterTextSplitter from sentence_transformers import SentenceTransformer from openai import OpenAI embedding_model SentenceTransformer(BAAI/bge-large-zh-v1.5) llm_client OpenAI(api_key你的key, base_urlhttps://taotoken.net/api) text_splitter RecursiveCharacterTextSplitter( chunk_size512, chunk_overlap50, length_functionlambda x: len(x), ) def generate_important_score(content: str) - int: prompt f请给以下内容的重要性打分1分最低无关闲聊5分最高核心偏好、重要事实 内容{content} 只返回数字。 resp llm_client.chat.completions.create( modelqwen2-7b-instruct, messages[{role: user, content: prompt}], temperature0 ) try: return max(1, min(5, int(resp.choices[0].message.content.strip()))) except Exception: return 3评分之后按分数设置过期时间1 分 30 天过期2 分 180 天3 分 1 年4 到 5 分永久有效。这样记忆库不会无限膨胀低价值信息会自动淘汰。写入时同时落 PostgreSQL 存元数据、落 Milvus 存向量两边用同一个memory_id关联。def write_memory(user_id, session_id, content, db): chunks text_splitter.split_text(content) ids [] for chunk in chunks: memory_id uuid.uuid4().hex ids.append(memory_id) score generate_important_score(chunk) embedding embedding_model.encode(chunk, normalize_embeddingsTrue).tolist() expired_at None if score 1: expired_at datetime.utcnow() timedelta(days30) elif score 2: expired_at datetime.utcnow() timedelta(days180) elif score 3: expired_at datetime.utcnow() timedelta(days365) # 写 PostgreSQL 和 Milvus 的代码略按前面 Schema 对应插入即可 return ids分块大小设成 512 是因为 bge 系列模型的最大输入长度就是 512 Token超过会被截断向量里信息不完整检索准确率会掉。重叠 50 Token 是为了避免一句话被切两半导致语义丢失。4. 检索链路优化从粗排到重排的完整参数模板4.1 Query 改写补全上下文再检索用户 query 往往很短比如“推荐点零食”直接拿去做向量检索召回的内容会很泛。Query 改写的目的是把当前会话上下文补进去让检索目标更明确。def rewrite_query(user_id, query, current_contextNone): prompt f根据当前上下文改写用户查询补全缺失信息不要改变原意 上下文{current_context or 无} 原始查询{query} 只返回改写后的查询。 resp llm_client.chat.completions.create( modelqwen2-7b-instruct, messages[{role: user, content: prompt}], temperature0 ) return resp.choices[0].message.content.strip()改写后的 query 再生成 Embedding检索命中率会明显提升。这一步的模型调用同样走 TaoToken和写入时用的是同一套配置。4.2 混合检索 交叉编码器重排纯向量检索是粗排余弦相似度只能判断语义大致接近没法精细区分。我的做法是先召回 Top20再用交叉编码器重排最后结合重要性得分和时间衰减做综合排序。from sentence_transformers import CrossEncoder cross_encoder CrossEncoder(BAAI/bge-reranker-large) def retrieve_memory(user_id, query, top_k5, current_contextNone, dbNone): rewritten rewrite_query(user_id, query, current_context) query_embedding embedding_model.encode(rewritten, normalize_embeddingsTrue).tolist() current_ts int(datetime.utcnow().timestamp()) expr fuser_id {user_id} and importance_score 2 and created_at {current_ts} search_params {metric_type: COSINE, params: {nprobe: 32}} results milvus_collection.search( data[query_embedding], anns_fieldembedding, paramsearch_params, limit20, exprexpr, output_fields[memory_id, importance_score, created_at] ) memory_ids [hit.entity.get(memory_id) for hit in results[0]] # 从 PostgreSQL 取详情交叉编码器重排综合排序后返回 TopK # 完整实现见前面 excerpt 中的 retrieve_memory 逻辑nprobe32是 IVF 索引的探测簇数量值越大召回越全但延迟越高。实测下来 32 在千万级数据上 p95 延迟能控制在 200ms 以内召回率 94% 以上。如果你对延迟更敏感可以降到 16准确率损失不到 1 个百分点。综合排序的权重是0.6 乘重排得分加 0.2 乘重要性得分归一化加 0.2 乘时间衰减得分。时间衰减用1 / (1 0.01 * 天数)这样最近的记忆会获得更高权重但不会完全压过重要性高的旧记忆。5. 召回验证一套可执行的测试动作光说优化没用得能验证。我设计了一组最小验证动作你可以在自己的环境里跑一遍。第一步写入一条高重要性记忆write_memory( user_idtest_user_001, session_ids1, content用户张三对芒果过敏不要推荐任何芒果相关食品收货地址是北京市朝阳区某小区, dbdb_session )第二步等几秒让 Milvus flush 完成然后发起一个不直接提“芒果”的查询results retrieve_memory( user_idtest_user_001, query给我推荐点好吃的零食, top_k5, dbdb_session ) for r in results: print(r)第三步检查返回结果里是否包含“芒果过敏”这条记忆。如果包含说明语义检索链路通了如果不包含按下面的排查顺序检查。我实测下来这套配置在 10 万条记忆、单用户 100 条记忆的场景下p95 检索延迟约 80ms召回率 98%1000 万条记忆、单用户 1 万条时p95 延迟约 190ms召回率 94%。这个数据是在本地 Docker 部署的 Milvus 上跑的没有用 GPU 加速。6. 本篇常见错排查检索召回不相关先确认写入和检索用的是同一个 Embedding 模型。不同模型生成的向量空间不一样混用会导致相似度计算完全失真。然后打开 Query 改写日志看改写后的 query 是否符合预期。如果还不行调大 TopK 召回数量或者提高交叉编码器在综合排序里的权重。检索延迟突然升高检查是不是没有开分区键导致全表扫描。另外看nprobe是不是设得太大默认 32 已经够用调到 64 以上延迟会明显上升但准确率提升有限。如果数据量超过 500 万考虑加 Redis 缓存高频查询结果。记忆冲突用户之前说不吃辣后来又说能吃微辣。这种情况给新记忆更高的时间权重同时把旧记忆标记为失效检索时通过is_active字段过滤掉。不要直接删除旧记忆保留审计痕迹。写入后检索不到Milvus 插入后需要 flush 才能被检索到。如果写入后立刻查询可能因为没 flush 而查不到。生产环境可以设置定时 flush或者写入后主动调用collection.flush()。存储成本增长过快重要性 1 到 2 分的记忆超过半年可以归档到对象存储需要时再拉回。IVF_SQ8 索引本身已经比原始向量省 75% 空间如果还嫌大可以考虑 IVF_PQ但准确率会再降几个点。如果你在接入过程中遇到模型调用相关的报错比如超时、返回格式异常建议先到接入文档确认参数格式https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite长期做编码类 Agent、需要频繁调用模型做记忆评分和 Query 改写的场景可以考虑 Coding Plan按量计费比单次调用更划算https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding_planutm_campaignrewrite整套记忆系统的代码骨架到这里就完整了。最后提醒一句Embedding 模型和向量库的 Schema 一旦确定后期迁移成本很高上线前一定用真实数据跑一遍召回验证别等用户反馈“Agent 又失忆了”才回头改。