Agent Memory 是最近 Agent 应用开发中讨论最多、也最容易踩坑的一块。很多人用 LLM 写完意图识别和工具调用后发现再做对话上下文、长期用户偏好、历史行为复盘时模型几乎每次都是“重新认识用户”原因就是没有设计好 Agent 的记忆体系。Agent Memory 解决的不只是“记住上次说了什么”还包括什么时候写入、什么时候修改、什么时候遗忘、以及如何把记忆变成检索结果让模型使用。这篇文章会从记忆的存储模型讲起用一套可运行的 Python 示例实现记忆的写入、检索和修改再以电商场景为例把用户偏好记忆接进推荐和客服流程最后给出从内存异常到记忆不生效的排查路径。整篇按“概念、环境、实现、案例、验证、排错、上线”展开。如果你正在做 Agent 应用、智能客服、AI 导购、个人助理这类项目或者只是想把“带记忆的 Agent”从 Demo 推进到可维护状态这篇文章的思路可以直接拿来做基础骨架。1. Agent Memory 的本质先搞清楚要记住什么再谈怎么存1.1 没有记忆的 Agent 为什么像失忆患者传统 LLM 对话接口本身是无状态的。每次调用模型时你传入多少上下文模型就在多少上下文中推理上一次请求中的信息不会自动保留。所谓“Agent 有记忆”本质上是在模型外部构建一套记忆服务把需要跨请求保留的信息存下来经过筛选后注入到下一轮提示词中。没有记忆时用户说“我喜欢白色运动鞋”下一轮说“推荐一款适合跑步的”模型可能忽略颜色偏好用户说“上次那件衣服不合适”模型不知道是哪件用户购买两次后又问“我是不是买过这个牌子”Agent 只能猜。所以记忆不是增强功能而是 Agent 在真实业务中的基础能力。这里需要区分三个容易混淆的概念会话上下文Conversation Context一轮请求内的消息列表通常存在内存或 Redis请求结束后可丢弃或滚动写入日志。短期工作记忆Working Memory当前任务执行过程中需要临时保存的信息例如工具调用中间结果。长期记忆Long-term Memory跨会话保留的用户偏好、事实信息、行为摘要通常由数据库或向量存储承载。1.2 记忆在技术上的分层把记忆拆成三层后实现方式完全不同。会话上下文可以直接用 LLM 的 messages 数组携带最多做长度裁剪短期工作记忆通常在 Agent 的编排器里用变量存储长期记忆才是真正需要设计存储模型的部分。记忆类型生命周期存储介质典型数据会话上下文单次请求内内存、Redis当前对话消息、工具返回工作记忆单次任务执行内存变量、状态对象检索结果、临时结论长期记忆跨会话、长期关系库、KV、向量库用户偏好、历史行为、知识摘要长期记忆又有两个方向一种是结构化记忆比如用户性别、会员等级、收货地址适合用关系表或键值存储另一种是语义记忆比如“用户最近在关注户外跑步装备”适合用向量检索。真实系统通常两者共存结构化记忆负责精确查询语义记忆负责模糊匹配。1.3 存储与修改为什么比写入更麻烦很多人做记忆的第一版只实现了“写入”和“读取”上线后才发现问题记忆重复写入。用户三次说“我爱喝美式”库里存了三段几乎一样的内容检索时互相干扰。记忆无法更新。用户从“我住北京”改成“我搬去上海了”旧记忆还在模型可能同时引用两个地点。记忆没有衰减。三个月前的兴趣依旧高权重推荐内容越来越不贴近现状。记忆没有权限。多用户共用一张记忆表A 用户的偏好泄漏到 B 用户。所以存储修改不是简单的UPDATE而是要考虑“这段记忆是否已存在”“重要程度如何变化”“是否应该覆盖旧值”“什么时候该被清理”。这也是本文代码实战部分的核心。注意Agent Memory 的第一版建议先做“可解释、可回滚”的存储而不是先上复杂检索。能用 SQL 查清楚的场景不要过早引入向量库。2. 准备 Agent Memory 实验环境选型不是越重越好2.1 技术选型与架构学习阶段不需要一开始就上分布式向量库。Neo4j、Milvus、Redis 集群这些组件在只有几千条记忆时只会增加排错成本。推荐这条路径语言使用 Python 3.10便于验证算法和对接大模型 SDK。元数据使用 SQLite一个文件就能保存学习时可视化方便。向量检索先用 NumPy 做暴力余弦相似度检索理解原理后再替换成 pgvector、Chroma 或 Milvus。大模型接口先用抽象类封装可以接 OpenAI 风格接口也可以接本地 Ollama避免绑定具体厂商。这个架构的优点是依赖少单机可跑代码逻辑透明。生产环境再按同样的接口替换底层存储即可。2.2 环境依赖创建项目目录后先准备虚拟环境mkdir agent-memory-demo cd agent-memory-demo python -m venv .venv source .venv/bin/activate # Windows 下执行 .venv\Scripts\activate pip install numpy按需安装向量库和大模型 SDK。下面的最小示例只需要numpy和requests# requirements.txt numpy1.24 requests2.31如果你需要本地计算句向量可以再安装sentence-transformers但首次运行会下载模型网络环境较慢时容易失败。为了先把流程跑通本文示例用确定性哈希向量代替真实 embedding生产环境替换为真实 embedding 即可。2.3 项目目录结构建议保持文件职责单一方便后续替换存储层agent-memory-demo/ ├── config.py ├── memory_schema.py ├── memory_store.py ├── memory_service.py ├── agent_demo.py └── requirements.txtmemory_schema.py定义记忆数据结构和建表语句。memory_store.py负责 SQLite 读写memory_service.py负责写入策略、检索策略和过期清理agent_demo.py是电商场景的入口。3. 实现记忆的写入、检索与修改一个可运行的 Python 框架3.1 先定义记忆的数据结构一份记忆记录至少要包含记忆内容、所属用户或命名空间、重要程度、创建时间、更新时间、访问次数和最后访问时间。向量值单独存放避免把大数组塞进关系表。# memory_schema.py from dataclasses import dataclass, field from datetime import datetime from typing import List, Optional dataclass class MemoryItem: user_id: str content: str memory_type: str fact # fact / preference / behavior / summary importance: float 0.5 # 0~1越接近 1 越重要 namespace: str default memory_id: Optional[int] None created_at: str field(default_factorylambda: datetime.utcnow().isoformat()) updated_at: str field(default_factorylambda: datetime.utcnow().isoformat()) access_count: int 0 last_access_at: str field(default_factorylambda: datetime.utcnow().isoformat()) embedding: List[float] field(default_factorylist)字段设计时的几个考虑user_id是隔离边界所有查询必须先过滤用户。memory_type区分偏好和事实后续可以按类型设置不同权重。importance决定这条记忆是否值得长期保留。access_count和last_access_at用于温冷数据淘汰。3.2 用 SQLite 做存储层SQLite 不需要额外服务适合学习阶段。建表时把user_id、memory_type、updated_at加上索引检索时避免全表扫描。# memory_store.py import sqlite3 import json from typing import List, Optional from memory_schema import MemoryItem CREATE_TABLE_SQL CREATE TABLE IF NOT EXISTS memory ( memory_id INTEGER PRIMARY KEY AUTOINCREMENT, namespace TEXT NOT NULL, user_id TEXT NOT NULL, memory_type TEXT NOT NULL, content TEXT NOT NULL, importance REAL NOT NULL, embedding TEXT NOT NULL, created_at TEXT NOT NULL, updated_at TEXT NOT NULL, access_count INTEGER NOT NULL DEFAULT 0, last_access_at TEXT NOT NULL ); CREATE INDEX IF NOT EXISTS idx_memory_user ON memory(user_id); CREATE INDEX IF NOT EXISTS idx_memory_type ON memory(memory_type); class SQLiteMemoryStore: def __init__(self, db_path: str agent_memory.db): self.conn sqlite3.connect(db_path) self.conn.executescript(CREATE_TABLE_SQL) def insert(self, item: MemoryItem) - int: cur self.conn.execute( INSERT INTO memory (namespace, user_id, memory_type, content, importance, embedding, created_at, updated_at, access_count, last_access_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?), (item.namespace, item.user_id, item.memory_type, item.content, item.importance, json.dumps(item.embedding), item.created_at, item.updated_at, item.access_count, item.last_access_at), ) self.conn.commit() return cur.lastrowid def update(self, item: MemoryItem) - None: self.conn.execute( UPDATE memory SET content?, importance?, updated_at?, access_count?, last_access_at?, memory_type? WHERE memory_id? AND user_id?, (item.content, item.importance, item.updated_at, item.access_count, item.last_access_at, item.memory_type, item.memory_id, item.user_id), ) self.conn.commit() def get(self, memory_id: int, user_id: str) - Optional[MemoryItem]: row self.conn.execute( SELECT memory_id, namespace, user_id, memory_type, content, importance, embedding, created_at, updated_at, access_count, last_access_at FROM memory WHERE memory_id? AND user_id?, (memory_id, user_id), ).fetchone() if not row: return None return self._row_to_item(row) def list_by_user(self, user_id: str) - List[MemoryItem]: rows self.conn.execute( SELECT memory_id, namespace, user_id, memory_type, content, importance, embedding, created_at, updated_at, access_count, last_access_at FROM memory WHERE user_id? ORDER BY importance DESC, updated_at DESC, (user_id,), ).fetchall() return [self._row_to_item(r) for r in rows] def delete(self, memory_id: int, user_id: str) - None: self.conn.execute(DELETE FROM memory WHERE memory_id? AND user_id?, (memory_id, user_id)) self.conn.commit() def _row_to_item(self, row) - MemoryItem: return MemoryItem( memory_idrow[0], namespacerow[1], user_idrow[2], memory_typerow[3], contentrow[4], importancerow[5], embeddingjson.loads(row[6]), created_atrow[7], updated_atrow[8], access_countrow[9], last_access_atrow[10], )这里的查询都带上user_id是为了避免多用户串记忆。实际生产环境中还应在应用层鉴权后再调用存储层不能让用户传入一个user_id就能读别人的数据。3.3 用余弦相似度实现语义检索真实向量库帮我们省掉了相似度计算和索引构建。学习阶段用 NumPy 写一个暴力检索能清楚看到“向量值、相似度、top_k”的全过程。# memory_service.py import numpy as np from typing import List from datetime import datetime from memory_schema import MemoryItem from memory_store import SQLiteMemoryStore def cosine_similarity(vec_a: List[float], vec_b: List[float]) - float: a np.array(vec_a, dtypenp.float32) b np.array(vec_b, dtypenp.float32) if np.linalg.norm(a) 0 or np.linalg.norm(b) 0: return 0.0 return float(np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))) class MemoryService: def __init__(self, store: SQLiteMemoryStore, embed_funcNone): self.store store self.embed_func embed_func or self._dummy_embedding staticmethod def _dummy_embedding(text: str, dim: int 16) - List[float]: # 仅用于学习流程生产环境请替换为真实 embedding 模型 seed sum(ord(c) * (i 1) for i, c in enumerate(text)) rng np.random.default_rng(seed) vec rng.normal(sizedim) norm np.linalg.norm(vec) return (vec / norm).tolist() def search(self, user_id: str, query: str, top_k: int 5, threshold: float 0.0) - List[MemoryItem]: query_vec self.embed_func(query) items self.store.list_by_user(user_id) scored [] for item in items: if not item.embedding: continue score cosine_similarity(query_vec, item.embedding) if score threshold: scored.append((score, item)) scored.sort(keylambda x: x[0], reverseTrue) return [item for _, item in scored[:top_k]]使用np.random.default_rng(seed)并不是真正可用的语义文本向量它只是让同一句子稳定得到同一向量从而能验证检索逻辑。真实项目里要替换成sentence-transformers、OpenAI Embedding 或本地模型。3.4 写入与修改去重、覆盖、衰减带记忆的 Agent 最关键的策略在写入阶段。直接 insert 会导致重复记忆越来越多因此save_memory要先做语义相似度检查如果和已有记忆高度相似就更新已有记录而不是新增。def save_memory(self, user_id: str, content: str, memory_type: str fact, importance: float 0.5, same_threshold: float 0.92) - MemoryItem: new_embedding self.embed_func(content) same_item self._find_same(user_id, content, new_embedding, same_threshold) now datetime.utcnow().isoformat() if same_item: same_item.content content same_item.importance max(same_item.importance, importance) same_item.updated_at now same_item.access_count 1 same_item.embedding new_embedding self.store.update(same_item) return same_item item MemoryItem( user_iduser_id, contentcontent, memory_typememory_type, importanceimportance, created_atnow, updated_atnow, embeddingnew_embedding, last_access_atnow, ) item.memory_id self.store.insert(item) return item def _find_same(self, user_id: str, content: str, embedding: List[float], threshold: float) - MemoryItem | None: items self.store.list_by_user(user_id) for item in items: if not item.embedding: continue score cosine_similarity(embedding, item.embedding) if score threshold: return item return None def forget_old_memories(self, user_id: str, max_days: int 30, max_importance: float 0.85) - int: items self.store.list_by_user(user_id) now datetime.utcnow() deleted 0 for item in items: last_access datetime.fromisoformat(item.last_access_at) age_days (now - last_access).days if age_days max_days and item.importance max_importance: self.store.delete(item.memory_id, user_id) deleted 1 return deleted这里的去重阈值默认设成 0.92含义是两个句子在向量空间中高度相似时视为同一条记忆。阈值过高会重复过低会合并不同信息实际使用时需要通过样本测试调整。修改的常见场景有两种用户主动纠正比如“我目前住在上海之前说的北京是老家”。此时需要把旧记忆标为过期或直接更新内容再由大模型判断是否保留衍生记忆。新行为推翻旧偏好比如用户最近一周只购买无糖饮料。此时要提升新记忆的权重降低旧记忆的importance。4. 电商实战把记忆管理接到用户偏好和智能客服场景4.1 需求拆解电商场景最能体现 Agent Memory 的价值。以一个“AI 导购 客服”Agent 为例核心需求是用户第一次说“我最近想买跑步鞋”Agent 要记住这个意图。用户提到“预算五百左右”Agent 要把预算价格和跑步鞋偏好关联起来。下一轮用户问“有适合夏天的吗”Agent 要能通过关键词“夏天”检索到之前的跑步鞋偏好和预算。用户说“还是看看休闲鞋吧”Agent 要更新偏好方向而不是把旧偏好直接删除而是降低权重或标记历史。这里的记忆不是简单存聊天记录而是抽取“用户画像事实”后做结构化保存。4.2 对话事件的记忆抽取逻辑在真实 Agent 中抽取记忆可以由大模型完成从用户话语中提取三元组(用户, 维度, 值)。为了演示存储与修改逻辑这里用手写规则加关键词抽取代替。# agent_demo.py from memory_schema import MemoryItem from memory_store import SQLiteMemoryStore from memory_service import MemoryService store SQLiteMemoryStore(agent_memory.db) service MemoryService(store) DEMO_USER_ID u_10001 def extract_memories(user_input: str): memories [] if 跑步鞋 in user_input or 跑鞋 in user_input: memories.append((preference, 用户最近对跑步鞋感兴趣, 0.8)) if 预算 in user_input and 百 in user_input: memories.append((fact, 用户购物预算在五百元左右, 0.7)) if 夏天 in user_input: memories.append((fact, 用户当前季节偏好夏季透气款, 0.6)) if 休闲鞋 in user_input: memories.append((preference, 用户近期转向休闲鞋, 0.9)) return memories for user_input in [ 我最近想买跑步鞋预算五百左右, 有适合夏天的吗, 还是看看休闲鞋吧, ]: for mtype, content, importance in extract_memories(user_input): service.save_memory(DEMO_USER_ID, content, memory_typemtype, importanceimportance)这段代码展示了最小闭环输入多轮对话Agent 抽取记忆并写入。save_memory内部会通过相似度判断如果发现“跑步鞋”和“转向休闲鞋”不是同一概念就不会误合并如果用户重复说“预算五百”第二次也不会新增重复记录而会更新访问次数和重要性。4.3 在对话生成前注入检索结果当用户提出新问题时Agent 从记忆中检索 top_k 条最相关内容拼接到系统提示词里再调用大模型。def build_prompt(user_id: str, user_query: str) - str: relevant service.search(user_id, user_query, top_k5, threshold0.35) memory_lines [] for item in relevant: memory_lines.append( f- [{item.memory_type}, 重要度 {item.importance:.2f}] {item.content} ) memory_text \n.join(memory_lines) if memory_lines else 当前没有相关记忆。 system_prompt f你是电商导购助手。回答前先阅读用户记忆 {memory_text} 如果记忆中有相关偏好结合偏好回答不要编造记忆中没有的信息。 return system_prompt, relevant在实际大模型调用阶段system_prompt和user_query会组成请求体{ model: qwen-plus, messages: [ {role: system, content: 你是电商导购助手。回答前先阅读用户记忆\n- [fact, 重要度 0.70] 用户购物预算在五百元左右\n- [preference, 重要度 0.80] 用户最近对跑步鞋感兴趣}, {role: user, content: 有适合夏天的吗} ] }这里可以看到记忆注入的关键点只注入与当前请求相关的记忆而不是把用户全部历史都塞进提示词。否则提示词会越来越长成本高且容易互相干扰。4.4 记忆修改的触发场景记忆修改不能只发生在写入时。在客服场景里用户情绪化表达或售后反馈往往包含需要覆盖的旧记忆。下面是一个简化策略def handle_user_correction(user_id: str, correction: str, old_content: str): candidates service.search(user_id, old_content, top_k1, threshold0.8) if candidates: item candidates[0] item.content correction item.importance min(1.0, item.importance 0.1) store.update(item)这段代码解决的问题是当用户明确纠正时Agent 应该修改旧记录而不是新建记录。比如用户之前说“我喜欢白色”现在说“其实我更喜欢黑色”更新后旧记录中的白色不再出现在后续检索中。注意直接覆盖可能导致历史解读丢失。生产系统通常会加一个superseded_by字段让旧记录保留但不参与默认检索这是本文简化版没有展开的部分。5. 运行验证与结果分析5.1 运行最小案例在项目目录执行python agent_demo.py正常流程下程序应依次处理三句用户输入写入记忆后打印保存状态。如果要看检索结果可以再加一段输出print(当前用户记忆:) for item in store.list_by_user(DEMO_USER_ID): print(f [{item.memory_id}] {item.content} 重要度{item.importance:.2f} 访问{item.access_count}) print(\n用户查询适合夏天的运动鞋:) for item in service.search(DEMO_USER_ID, 适合夏天的运动鞋, top_k3): print(f sim{cosine_similarity(service.embed_func(适合夏天的运动鞋), item.embedding):.2f} - {item.content})预期需要看到三类结果同一句话重复处理两次时记忆表里不会出现两条完全相同的内容。查询“适合夏天的运动鞋”时能检索到“跑步鞋”和“夏季透气”相关记忆。用户纠正“休闲鞋”后新偏好的importance应该高于旧偏好或者旧偏好被覆盖。这里要特别提醒示例用的_dummy_embedding只保证相同输入得到相同向量不能保证语义相近的句子相似度高。你在本地验证业务效果时一定要换成真实 embedding否则检索顺序没有参考价值。5.2 验证去重与覆盖去重逻辑是否生效可以直接查 SQLite 数据sqlite3 agent_memory.db SELECT memory_id, user_id, content, importance, access_count FROM memory;如果“预算五百”被用户提了三次应该只看到一条记录且access_count大于 1。如果出现多条几乎相同的记录说明same_threshold设置过严或者向量维度、embedding 质量有问题。5.3 验证记忆隔离多用户场景下必须确认user_id不会串数据store SQLiteMemoryStore(agent_memory.db) service_a MemoryService(store) print(service_a.search(u_10001, 夏天的运动鞋)) print(service_a.search(u_20002, 夏天的运动鞋))最终打印结果应该是u_10001有记忆路径u_20002没有记忆或只有自己的记忆。如果u_20002查到了别人的偏好说明查询语句漏掉了user_id过滤。6. 常见失败现象与排查路径6.1 现象一进程崩溃退出码 3221225477 / 0xc0000005很多 Agent 项目在本地运行时报出Process exited with code 3221225477十六进制是0xc0000005对应 Windows 的内存访问违规。这通常不来自 Python 解释器本体而是来自 C 扩展层例如numpy版本与 Python 版本不匹配。faiss-cpu、onnxruntime、transformers等 C 扩展访问了非法内存。向量维度不一致导致 C 层在内存拷贝时越界。排查步骤python -c import numpy; print(numpy.__version__) python -c import faiss; print(faiss.__version__)如果安装了多个深度学习库先检查依赖冲突pip check处理建议新建干净虚拟环境按requirements.txt顺序安装排除版本冲突。如果使用sentence-transformers先单独跑一次模型加载脚本确认推理正常。将向量维度统一写进配置所有 embedding 函数返回同一维度不要依赖“刚好能跑”的形状。如果崩溃发生在faiss索引构建考虑换成纯 NumPy 或提升到内存充足的环境再复现。6.2 现象二Java OutOfMemoryError / Native Memory Allocation Failed如果 Agent 应用由 Java 提供 HTTP 服务Python 侧做向量检索常见报错有两类java.lang.OutOfMemoryError: Insufficient memoryNative memory allocation (malloc) failed to allocate 2046256 bytes for chunk第一类大多是 JVM 堆内存不足第二类是堆外内存不足。堆外内存常被缓存、JNI、NIO、线程栈和 C 扩展占用。排查顺序# 查看 JVM 参数 jinfo -flags pid # 查看堆使用 jstat -gcutil pid 1000 # 查看 Native 内存 jcmd pid VM.native_memory summary scaleMB同时检查服务端是否在每次请求时都构建新模型对象或新向量索引导致内存无法回收。推荐做法是模型只加载一次向量索引长期驻留单次请求只做查询。6.3 现象三Agent 回复了但记忆没有写入比内存异常更隐蔽的是“逻辑看起来正常但记忆没有生效”。常见原因记忆写入是异步任务主线程未等待完成就返回了响应。写入函数被放在异常分支里工具调用失败后没有执行。消息进入 Agent 时不是用户最终输入而是子 Agent 内部输出。多个请求并发写同一条记忆后写覆盖先写。检查时先确认写入调用是否真的执行# 在 save_memory 前打印输入 print(debug: save_memory called for, user_id, content)再查存储层事务是否提交。SQLiteMemoryStore.insert中已经调用commit但如果你改成异步或包事务就要检查提交点和异常处理。6.4 排查清单问题现象优先检查可能原因处理方向记忆表里有大量重复记录save_memory的去重阈值same_threshold过高或 embedding 不稳定降低阈值统一 embedding 模型检索结果明显不相关threshold和 top_k向量相似度阈值太低或模型维度过小调整阈值换真实 embedding用户 A 看到用户 B 的记忆查询 SQLlist_by_user 漏掉 user_id检查存储层过滤条件进程崩溃退出码 0xc0000005依赖安装C 扩展版本不匹配重建虚拟环境pin 版本修改不生效update 方法未按 memory_id user_id 更新更新条件必须带用户隔离字段记忆表无限膨胀forget_old_memories未执行过期清理增加定时任务和访问衰减7. 生产环境落地 Agent Memory 的最佳实践7.1 学习环境、开发环境和生产环境的差异同一套代码在不同环境里关注点完全不同。学习环境要的是逻辑闭环开发环境要的是可调试生产环境要的是可维护。维度学习/本地开发/测试生产存储SQLiteMySQL/PostgreSQLPostgreSQL pgvector 或 Milvusembedding随机哈希sentence-transformers稳定 embedding API 或模型服务检索暴力余弦FAISS 本地索引分布式向量库数据隔离按 user_id 过滤按租户过滤租户角色权限记忆清理手动脚本定时任务异步 job 监控观察性print日志文件链路追踪、指标告警学习代码与生产代码的接口应该保持一致比如MemoryService的save_memory、search、forget_old_memories方法底层存储替换时调用方不需要变化。7.2 记忆一致性与过期策略生产系统必须定义记忆的一致性和生命周期规则用户主动更新了资料后相关记忆要级联修改不能保留两个互相矛盾的常识。长期不访问的记忆要阶段性降权而不是直接删除避免误伤低频但重要的偏好。删除记忆要留审计日志尤其是客服、医疗、法律等高合规业务。记忆写入失败不应阻塞主流程建议先返回用户应答再异步重试写入但服务端要标记“本轮记忆未落库”避免用户下次被要求重新说明。一个实用的过期参数组合如下memory: default_importance: 0.5 search_top_k: 5 search_threshold: 0.35 same_threshold: 0.92 forget_max_days: 30 forget_min_importance: 0.85 embedding_dim: 7687.3 关键参数速查参数含义默认参考值调大影响调小影响top_k每次注入提示词的记忆条数3~8信息更多干扰和成本上升信息不足容易忽略关键偏好threshold检索过滤的最低相似度0.3~0.5结果更精准但可能漏掉弱相关记忆召回变多噪音增加same_threshold判断两条记忆是否相同的阈值0.88~0.95容易判定为不同重复增多容易误合并不同含义importance记忆权重0.5越高越难被清理越低越容易丢失forget_max_days多少天未访问被清理30保留更多历史历史丢得快embedding_dim向量维度384~1536表达更强占用更大表达较弱检索偏差以上数值不是官方标准要根据业务数据实测调整。向量库能帮你处理距离计算但“该记住什么”永远是需要业务介入的策略问题。7.4 生产发布前检查清单上线前至少确认以下项目所有查询是否带user_id或租户维度权限校验是否在 API 层完成。embedding 模型是否固定版本历史和新增样本的向量维度是否一致。记忆写入是否幂等重复消息不会大量复制。是否有记忆清理任务内存和磁盘不会无限增长。是否保留记忆变更日志能否回滚到旧记忆状态。检索超时时间是否设置向量库挂掉时是否能降级到 SQL 精确查询。模型提示词是否有注入风险用户输入不能直接改写记忆规则。是否对记忆读取次数做监控用于发现检索热点和记忆膨胀。7.5 扩展方向如果这篇文章的代码你已经跑通下一步可以按顺序做三件事把_dummy_embedding替换为真实 embedding 模型观察检索质量变化。把SQLiteMemoryStore改造成 PostgreSQL pgvector 实现保持MemoryService接口不变。在电商案例中加入“记忆抽取由大模型判断”的逻辑让 Agent 自己决定把哪些信息写入长期记忆。更完善的记忆系统还会引入记忆图、技能库和观察反思机制但对新手来说先把写入、检索、更新、遗忘这条主链路做扎实比追概念更有价值。还可以对照pi agent、hermes agent、codex agent等开源项目的代码看看它们把记忆放在哪个模块、用什么协议读写随时调整自己的设计。写完这套代码后建议你把每一条记忆的“生命周期”画成一张状态流从用户输入触发写入到命中检索被注入提示词再到访问次数变化和过期清理。这张图能帮你快速定位绝大多数 Agent 记忆问题。