企业官网建设流程全解析

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 的编排器里用变量存储；长期记忆才是真正需要设计存储模型的部分。

长期记忆又有两个方向：一种是结构化记忆，比如用户性别、会员等级、收货地址，适合用关系表或键值存储；另一种是语义记忆，比如“用户最近在关注户外跑步装备”，适合用向量检索。真实系统通常两者共存：结构化记忆负责精确查询，语义记忆负责模糊匹配。

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 numpy>=1.24 requests>=2.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.txt

memory_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_factory=lambda: datetime.utcnow().isoformat()) updated_at: str = field(default_factory=lambda: datetime.utcnow().isoformat()) access_count: int = 0 last_access_at: str = field(default_factory=lambda: datetime.utcnow().isoformat()) embedding: List[float] = field(default_factory=list)

字段设计时的几个考虑：

• 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_id=row[0], namespace=row[1], user_id=row[2], memory_type=row[3], content=row[4], importance=row[5], embedding=json.loads(row[6]), created_at=row[7], updated_at=row[8], access_count=row[9], last_access_at=row[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, dtype=np.float32) b = np.array(vec_b, dtype=np.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_func=None): 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(size=dim) 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(key=lambda x: x[0], reverse=True) 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_id=user_id, content=content, memory_type=memory_type, importance=importance, created_at=now, updated_at=now, embedding=new_embedding, last_access_at=now, ) 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 为例，核心需求是：

1. 用户第一次说“我最近想买跑步鞋”，Agent 要记住这个意图。
2. 用户提到“预算五百左右”，Agent 要把预算价格和跑步鞋偏好关联起来。
3. 下一轮用户问“有适合夏天的吗”，Agent 要能通过关键词“夏天”检索到之前的跑步鞋偏好和预算。
4. 用户说“还是看看休闲鞋吧”，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_type=mtype, importance=importance)

这段代码展示了最小闭环：输入多轮对话，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_k=5, threshold=0.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_k=1, threshold=0.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_k=3): 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

处理建议：

1. 新建干净虚拟环境，按requirements.txt顺序安装，排除版本冲突。
2. 如果使用sentence-transformers，先单独跑一次模型加载脚本，确认推理正常。
3. 将向量维度统一写进配置，所有 embedding 函数返回同一维度，不要依赖“刚好能跑”的形状。
4. 如果崩溃发生在faiss索引构建，考虑换成纯 NumPy 或提升到内存充足的环境再复现。

6.2 现象二：Java OutOfMemoryError / Native Memory Allocation Failed

如果 Agent 应用由 Java 提供 HTTP 服务，Python 侧做向量检索，常见报错有两类：

java.lang.OutOfMemoryError: Insufficient memory

Native 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 scale=MB

同时检查服务端是否在每次请求时都构建新模型对象或新向量索引，导致内存无法回收。推荐做法是：模型只加载一次，向量索引长期驻留，单次请求只做查询。

6.3 现象三：Agent 回复了，但记忆没有写入

比内存异常更隐蔽的是“逻辑看起来正常，但记忆没有生效”。常见原因：

• 记忆写入是异步任务，主线程未等待完成就返回了响应。
• 写入函数被放在异常分支里，工具调用失败后没有执行。
• 消息进入 Agent 时不是用户最终输入，而是子 Agent 内部输出。
• 多个请求并发写同一条记忆，后写覆盖先写。

检查时先确认写入调用是否真的执行：

# 在 save_memory 前打印输入 print("debug: save_memory called for", user_id, content)

再查存储层事务是否提交。SQLiteMemoryStore.insert中已经调用commit，但如果你改成异步或包事务，就要检查提交点和异常处理。

6.4 排查清单

7. 生产环境落地 Agent Memory 的最佳实践

7.1 学习环境、开发环境和生产环境的差异

同一套代码在不同环境里，关注点完全不同。学习环境要的是逻辑闭环，开发环境要的是可调试，生产环境要的是可维护。

学习代码与生产代码的接口应该保持一致，比如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: 768

7.3 关键参数速查

以上数值不是官方标准，要根据业务数据实测调整。向量库能帮你处理距离计算，但“该记住什么”永远是需要业务介入的策略问题。

7.4 生产发布前检查清单

上线前至少确认以下项目：

• 所有查询是否带user_id或租户维度，权限校验是否在 API 层完成。
• embedding 模型是否固定版本，历史和新增样本的向量维度是否一致。
• 记忆写入是否幂等，重复消息不会大量复制。
• 是否有记忆清理任务，内存和磁盘不会无限增长。
• 是否保留记忆变更日志，能否回滚到旧记忆状态。
• 检索超时时间是否设置，向量库挂掉时是否能降级到 SQL 精确查询。
• 模型提示词是否有注入风险，用户输入不能直接改写记忆规则。
• 是否对记忆读取次数做监控，用于发现检索热点和记忆膨胀。

7.5 扩展方向

如果这篇文章的代码你已经跑通，下一步可以按顺序做三件事：

1. 把_dummy_embedding替换为真实 embedding 模型，观察检索质量变化。
2. 把SQLiteMemoryStore改造成 PostgreSQL + pgvector 实现，保持MemoryService接口不变。
3. 在电商案例中加入“记忆抽取由大模型判断”的逻辑，让 Agent 自己决定把哪些信息写入长期记忆。

更完善的记忆系统还会引入记忆图、技能库和观察反思机制，但对新手来说，先把写入、检索、更新、遗忘这条主链路做扎实，比追概念更有价值。还可以对照pi agent、hermes agent、codex agent等开源项目的代码，看看它们把记忆放在哪个模块、用什么协议读写，随时调整自己的设计。

写完这套代码后，建议你把每一条记忆的“生命周期”画成一张状态流：从用户输入触发写入，到命中检索被注入提示词，再到访问次数变化和过期清理。这张图能帮你快速定位绝大多数 Agent 记忆问题。