1. 从一次线上事故说起：为什么我要去翻 TencentDB Agent Memory 的源码
前段时间团队在做一个基于 Codex 的智能编码助手，需求很明确：让 Agent 记住用户的历史偏好、项目上下文、常用代码片段，下次对话直接调用，不用每次从零开始。听起来是个标准的 Agent Memory 场景，我们一开始的方案也很朴素——把记忆存进 TencentDB，用向量检索做召回，Codex 侧通过工具调用读写。
上线第一周就出问题了。用户反馈"昨天让它记住的命名规范，今天又忘了"，日志里能看到记忆写入成功，但读取时返回的是另一条不相关的记录。更诡异的是，同一个 session 内连续两次调用，第一次能拿到正确记忆，第二次就串了。这种"时好时坏"的 bug 最折磨人，因为本地复现概率极低。
我当时的判断是：问题不在业务代码，而在 Agent Memory 的架构层。Codex 这类 Agent 框架对记忆的读写有自己的一套假设——比如它默认记忆是"无状态、幂等、可并发"的，而 TencentDB 作为持久化存储，天然带有事务、连接池、会话隔离这些概念。两边的假设一旦对不上，就会出现这种"看起来写进去了，读出来不对"的硬冲突。
所以我决定直接读 TencentDB Agent Memory 相关模块的源码，把数据从 Codex 发出到落库、再从库里召回返回 Codex 的完整链路捋一遍。这篇文章就是这次排查的完整记录，包括架构分层、关键数据结构、冲突点定位、以及最后我们怎么绕过去的。如果你也在做 Agent + 持久化记忆的方案，尤其是用 Codex 这类框架接外部数据库，这篇应该能帮你省掉至少两天的排查时间。
2. TencentDB Agent Memory 的整体架构拆解
2.1 三层结构：接入层、记忆管理层、存储层
先把架构图在脑子里画出来。TencentDB Agent Memory 这套东西，从源码目录结构看，大致分三层：
- 接入层（Adapter Layer）：负责和 Agent 框架对接，Codex 这边通过 tool call 或者 SDK 调用进来，参数格式转换、鉴权、限流都在这一层。
- 记忆管理层（Memory Manager）：核心逻辑层，负责记忆的写入策略、召回策略、生命周期管理（TTL、淘汰、合并）。
- 存储层（Storage Layer）：真正落到 TencentDB 的部分，包括表结构、索引、向量检索、事务封装。
这个分层本身不新鲜，很多 Agent Memory 方案都是这么切的。但 TencentDB 这套的特殊之处在于：记忆管理层和存储层之间的边界不是干净的。源码里能看到，Memory Manager 在某些路径下会直接拼 SQL 下发到存储层，绕过了一层抽象。这个设计在性能上有好处（少一次序列化），但也是后面冲突的根源之一。
我画一个简化的调用链，方便你对照源码看：
Codex Agent -> tool_call(memory_write / memory_read) -> Adapter.parse(params) -> MemoryManager.write() / recall() -> StorageClient.exec() // 这里可能直接拼 SQL -> TencentDB (MySQL / 向量引擎)关键点在于StorageClient.exec()这一步。它接收的不是结构化对象，而是半结构化的查询描述，里面混了 Agent 侧的 session 信息和存储侧的事务信息。两边语义不一致的时候，就是 bug 出现的时候。
2.2 记忆的数据模型：为什么不是简单的 KV
很多人第一反应是"记忆不就是 KV 吗，key 是用户 ID，value 是内容"。真读源码会发现完全不是。TencentDB Agent Memory 里一条记忆记录至少包含这些字段：
| 字段 | 含义 | 来源 |
|---|---|---|
| memory_id | 记忆唯一标识 | 存储层生成 |
| agent_id | 哪个 Agent 实例 | Codex 侧传入 |
| session_id | 会话标识 | Codex 侧传入 |
| scope | 作用域（session / user / project） | 业务侧指定 |
| content | 记忆正文 | 业务侧传入 |
| embedding | 向量表示 | 记忆管理层生成 |
| created_at / updated_at | 时间戳 | 存储层生成 |
| ttl | 过期时间 | 业务侧或默认策略 |
| version | 版本号 | 存储层维护 |
这里第一个坑就来了：session_id 的语义在 Codex 侧和存储侧不一致。Codex 的 session 是"一次对话上下文"，可能几分钟就结束；而存储侧的 session_id 被当成"隔离维度"，用来做分区和索引。当 Codex 在一个长对话里多次调用记忆接口，session_id 不变，但存储侧可能因为连接池切换、事务提交等原因，把同一个 session_id 的记录分到了不同的物理分区。召回时如果只按 session_id 查，就会漏。
2.3 写入路径：从 Codex tool call 到落库
写入路径的源码我逐行跟过，简化后是这样：
# 伪代码，基于源码逻辑还原 def write_memory(agent_id, session_id, content, scope="session"): # 1. Adapter 层参数校验 validate(agent_id, session_id, content) # 2. MemoryManager 生成 embedding embedding = embed_model.encode(content) # 3. 构造存储记录 record = { "agent_id": agent_id, "session_id": session_id, "scope": scope, "content": content, "embedding": embedding, "version": next_version(agent_id, session_id), } # 4. 下发存储层 storage_client.insert("agent_memory", record) # 5. 更新缓存（如果有） cache.set(f"{agent_id}:{session_id}", record)看起来没问题，但第 3 步的next_version是个隐患。源码里它是通过"查当前最大 version + 1"实现的，没有加锁。并发写入时两个请求可能拿到同一个 version，后写的覆盖先写的。这就是我们线上"记忆丢失"的直接原因之一。
2.4 召回路径：向量检索 + 结构化过滤
召回比写入复杂，因为它要同时做两件事：向量相似度检索 + 结构化条件过滤。源码里的顺序是先过滤后检索：
def recall(agent_id, session_id, query, top_k=5): # 1. 先按结构化条件缩小范围 candidates = storage_client.query( "agent_memory", where={"agent_id": agent_id, "session_id": session_id}, ) # 2. 在候选集上做向量检索 query_emb = embed_model.encode(query) scored = [(cosine(query_emb, c.embedding), c) for c in candidates] scored.sort(reverse=True) return [c for _, c in scored[:top_k]]这个顺序在数据量小的时候没问题，但数据量一大就崩——先过滤可能返回几万条，再在 Python 里算余弦相似度，延迟直接爆炸。更合理的做法是让存储层支持"向量索引 + 标量过滤"的联合查询，但 TencentDB 这套源码里，向量检索和结构化查询是两套独立的客户端，没有做下推优化。
3. Codex 接记忆的硬冲突：三个真实踩坑点
3.1 冲突一：session 生命周期不匹配
Codex 的 session 是短生命周期的，一次对话结束就销毁。但 Agent Memory 的设计假设是 session 相对稳定，用来做数据隔离。当 Codex 在同一个用户下连续开多个 session，每个 session 都往同一个 agent_id 下写记忆，召回时如果只按当前 session_id 过滤，就看不到其他 session 写的记忆。
我们当时的业务需求恰恰是"跨 session 记忆"——用户今天让 Agent 记住命名规范，明天新开对话也要能用。但源码默认的召回逻辑是 session 隔离的，导致跨 session 召回为空。
解决方案：把 scope 从 session 改成 user，召回时按 agent_id + user_id 过滤，session_id 只作为元数据记录，不参与过滤。改完之后跨 session 召回正常了，但引入了新问题——不同项目的记忆会串。所以又加了一层 project_id 过滤。最终过滤条件是agent_id + user_id + project_id，session_id 彻底退出过滤逻辑。
3.2 冲突二：并发写入的 version 竞争
前面提到的next_version无锁问题，在 Codex 场景下特别容易触发。因为 Codex 经常并行调用多个 tool，比如同时写"代码风格"和"依赖版本"两条记忆，两个请求几乎同时到达。
源码里的 version 机制本意是做乐观锁，但实现有缺陷：
# 源码逻辑（简化） def next_version(agent_id, session_id): row = db.query("SELECT MAX(version) FROM agent_memory WHERE ...") return (row.max or 0) + 1两个并发请求都读到 max=5，都返回 6，后写的覆盖先写的。我们线上看到的"记忆丢失"就是这么来的。
解决方案：不能改源码的情况下，我们在 Adapter 层加了一个分布式锁，key 是agent_id + session_id，写入前先抢锁。锁的粒度要控制好，太粗会影响吞吐，太细挡不住竞争。实测下来按agent_id + session_id加锁，QPS 从 200 降到 150 左右，可以接受。
如果允许改源码，更优雅的做法是用数据库的INSERT ... ON DUPLICATE KEY UPDATE version = version + 1，把 version 递增交给数据库保证原子性。
3.3 冲突三：向量检索与事务隔离的互斥
这个冲突最隐蔽。TencentDB 的事务隔离级别默认是 RR（可重复读），而向量检索走的是另一套连接，可能是 RC（读已提交）。当写入事务还没提交时，向量检索连接看不到新数据；等提交后，向量索引可能还没更新，又看不到。
表现出来就是：写入返回成功，立刻召回却查不到，过几秒再查又有了。这种"延迟可见"在人工测试时很难发现，但在自动化流程里会导致 Agent 行为不一致。
解决方案：写入后强制刷新向量索引，或者召回时加一个"最小可见延迟"——如果召回为空，等待 500ms 重试一次。我们选了后者，因为刷新索引的代价太高。实测下来 500ms 能覆盖 95% 的场景，剩下 5% 通过重试兜底。
4. 实操：从零搭一套可用的 Agent Memory 接入方案
4.1 环境准备与依赖确认
先把基础环境列清楚，避免你踩版本坑：
| 组件 | 版本要求 | 说明 |
|---|---|---|
| Python | 3.9+ | 源码用了dict有序特性 |
| TencentDB | MySQL 5.7+ 或兼容版 | 向量功能需要额外插件 |
| Codex SDK | 最新版 | 老版本 tool call 格式不同 |
| 向量模型 | 任意 768 维 | 源码默认 768，改维度要同步改表结构 |
安装依赖：
pip install tencentdb-agent-memory codex-sdk numpy注意tencentdb-agent-memory这个包在 PyPI 上有多个同名包，认准官方源。我一开始装了个第三方的，接口对不上，排查了半天。
4.2 表结构设计与索引优化
源码自带的建表语句比较粗糙，我按实际业务改了一版：
CREATE TABLE agent_memory ( memory_id BIGINT AUTO_INCREMENT PRIMARY KEY, agent_id VARCHAR(64) NOT NULL, user_id VARCHAR(64) NOT NULL, project_id VARCHAR(64) DEFAULT '', session_id VARCHAR(64) NOT NULL, scope VARCHAR(16) NOT NULL DEFAULT 'user', content TEXT NOT NULL, embedding BLOB NOT NULL, version INT NOT NULL DEFAULT 1, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, ttl INT DEFAULT 0, INDEX idx_agent_user_project (agent_id, user_id, project_id), INDEX idx_session (session_id), UNIQUE KEY uk_agent_session_version (agent_id, session_id, version) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;关键改动有三个：加了user_id和project_id支持跨 session 召回；加了uk_agent_session_version唯一索引，从数据库层面挡住 version 竞争；embedding用 BLOB 存，比 JSON 省空间。
4.3 写入流程的完整实现
写入这块我把源码的逻辑重写了一遍，核心是加锁 + 重试：
import hashlib from contextlib import contextmanager @contextmanager def memory_lock(agent_id, session_id, timeout=5): lock_key = f"mem_lock:{agent_id}:{session_id}" acquired = redis.set(lock_key, "1", nx=True, ex=timeout) if not acquired: raise LockTimeoutError(lock_key) try: yield finally: redis.delete(lock_key) def write_memory(agent_id, user_id, project_id, session_id, content, scope="user"): with memory_lock(agent_id, session_id): embedding = embed_model.encode(content) for attempt in range(3): try: version = get_next_version(agent_id, session_id) db.execute( "INSERT INTO agent_memory (...) VALUES (...)", (agent_id, user_id, project_id, session_id, scope, content, embedding.tobytes(), version) ) return True except DuplicateKeyError: continue raise WriteConflictError(agent_id, session_id)这里get_next_version还是查最大值，但因为外面有锁，并发问题解决了。DuplicateKeyError重试是兜底，防止锁失效的极端情况。
4.4 召回流程的优化实现
召回我做了两处优化：一是把过滤条件下推到 SQL，二是加缓存。
def recall(agent_id, user_id, project_id, query, top_k=5): cache_key = f"recall:{agent_id}:{user_id}:{project_id}:{hash(query)}" cached = redis.get(cache_key) if cached: return json.loads(cached) query_emb = embed_model.encode(query) # 过滤下推到 SQL，只取最近 1000 条候选 candidates = db.query( """SELECT memory_id, content, embedding FROM agent_memory WHERE agent_id=%s AND user_id=%s AND project_id=%s AND (ttl=0 OR created_at > NOW() - INTERVAL ttl SECOND) ORDER BY created_at DESC LIMIT 1000""", (agent_id, user_id, project_id) ) scored = [] for c in candidates: emb = np.frombuffer(c.embedding, dtype=np.float32) score = float(np.dot(query_emb, emb) / (np.linalg.norm(query_emb) * np.linalg.norm(emb))) scored.append((score, c)) scored.sort(key=lambda x: x[0], reverse=True) result = [{"memory_id": c.memory_id, "content": c.content, "score": s} for s, c in scored[:top_k]] redis.setex(cache_key, 60, json.dumps(result)) return result缓存 60 秒是个折中，太短没效果，太长会导致新写入的记忆看不到。如果你的业务对实时性要求高，可以改成写入时主动失效缓存。
5. 常见问题与排查技巧实录
5.1 记忆写入成功但召回为空
这是最高频的问题，排查顺序建议这样：
- 确认过滤条件：先看召回时的
agent_id / user_id / project_id和写入时是否完全一致。我遇到过因为 user_id 大小写不一致导致查不到的。 - 确认事务可见性：写入后立刻查，如果查不到，等 1 秒再查。能查到就是可见性问题，参考 3.3 的解决方案。
- 确认向量维度：写入和召回的 embedding 维度必须一致。源码默认 768，如果你换了模型没改表结构，向量会截断或补零，相似度全乱。
- 确认 TTL：如果记忆设了 TTL，检查是否已过期。源码里 TTL 是秒为单位，但有些文档写的是毫秒，容易搞错。
5.2 召回结果不相关
向量检索返回的结果和 query 语义不匹配，通常是这几个原因：
- embedding 模型不一致：写入用的模型 A，召回用的模型 B，向量空间不同，相似度无意义。
- content 被截断：源码里 content 字段有长度限制，超长会被截断，截断后的 embedding 和原文语义偏差大。
- 候选集太小：如果过滤后只剩几条，向量检索退化成随机排序。建议候选集至少 100 条。
5.3 并发写入导致记忆覆盖
前面详细讲过，核心是 version 竞争。除了加锁，还有一个技巧：写入时带上 content 的 hash，如果 hash 相同就跳过，避免重复写入触发 version 递增。
content_hash = hashlib.md5(content.encode()).hexdigest() existing = db.query("SELECT memory_id FROM agent_memory WHERE agent_id=%s AND session_id=%s AND content_hash=%s", ...) if existing: return existing.memory_id # 幂等返回这个改动让我们的重复写入率从 30% 降到 5% 以下。
5.4 问题速查表
| 现象 | 可能原因 | 快速验证 | 解决 |
|---|---|---|---|
| 写入成功召回空 | 过滤条件不一致 | 打印两边参数对比 | 统一参数 |
| 写入成功召回空 | 事务可见性延迟 | 等 1 秒重查 | 加重试 |
| 召回结果乱 | embedding 模型不一致 | 检查模型版本 | 统一模型 |
| 记忆被覆盖 | version 竞争 | 查 version 分布 | 加锁或唯一索引 |
| 召回延迟高 | 候选集太大 | 看 SQL 返回行数 | 下推过滤 + 限制候选数 |
| 跨 session 召回空 | scope 设置错误 | 检查 scope 字段 | 改成 user scope |
5.5 几个源码里没写但很重要的细节
- 连接池大小：源码默认连接池 10，高并发下不够用。我们调到 50 后，超时错误基本消失。
- 向量索引更新频率：TencentDB 的向量索引不是实时更新的，有延迟。如果业务要求强一致，得自己维护一份内存索引。
- TTL 清理：源码里 TTL 是查询时过滤，不是物理删除。数据量大了之后查询会变慢，建议加一个定时任务做物理清理。
- 日志级别：默认 INFO 级别会打印每条记忆的 content，涉及隐私。生产环境记得调到 WARN。
6. 一些个人体会和后续可扩展的方向
这套方案跑了一个多月，整体稳定。回头看，最大的收获不是解决了某个具体 bug，而是理解了 Agent Memory 这类系统的本质矛盾：Agent 框架假设记忆是轻量、无状态、可并发的，而持久化存储天然是重量、有状态、需要事务的。两边的假设不可能完全对齐，只能在中间层做适配。
我个人的经验是，接入层要做厚一点，把存储侧的复杂性挡在外面。Codex 侧只关心"写一条记忆"和"召回相关记忆"，至于 version 怎么管、锁怎么加、缓存怎么失效，都不应该让 Agent 感知。这样即使底层换存储，上层也不用改。
后续可以扩展的方向有几个：一是把向量检索换成专门的向量数据库，TencentDB 的向量能力毕竟不是主业；二是加一层记忆压缩，把相似记忆合并，减少存储和检索压力；三是做记忆的重要性评分，召回时优先返回高价值记忆，而不是单纯按相似度排序。这几个方向我还在试，有进展再单独写一篇。
