Codex接入Agent Memory：源码揭示的三大硬冲突与解法

第一次跑通 Codex 接进 TencentDB Agent Memory 的时候我差点以为这事已经结了。结果真的沉下心来去读那套记忆服务的源码才发现所谓的接记忆根本不是往对话历史里塞一段上下文那么简单——两边对记忆的底层假设几乎是反的。这篇博文把我在源码里碰到的三处硬冲突完整还原出来任务边界的对不上、向量召回和 token 预算打架、版本号被 checkpoint 分支冲垮。适合正在做 Agent 记忆接入、或者想搞明白这类存储/索引/适配层代码的人按图索骥去复现顺便避开我们踩过的那几个大坑。1. 先说结论这个项目到底在折腾什么1.1 TencentDB Agent Memory 是什么TencentDB Agent Memory 不是一个简单的数据库表存聊天记录它本质上是给 AI Agent 用的持久化记忆中间件。从源码划分来看它包含四层能力结构化记忆消息、任务、元数据、向量记忆embedding 检索、事务日志版本回溯、以及给外部 Agent 接入的 SDK 层。我读的这份源码里核心目录大概是这样的agent_memory/ ├── client/ # SDK 入口外部 Agent 只看到这一层 │ ├── AgentMemoryClient.py │ └── session.py ├── task/ # 任务管理与生命周期 │ ├── task_manager.py │ └── task_state.py ├── store/ # 底层存储 │ ├── message_store.py │ ├── version_log.py │ └── schema.sql ├── index/ # 向量索引 │ ├── embedding_service.py │ └── vector_index.py └── adapter/ # 外部 Agent 适配层 └── codex_adapter.py这份源码给的记忆模型有一个核心概念Task。它把一次有明确边界的人机协作过程当作记忆的基本单元Task 下面挂消息消息之间靠seq_no排序全局再给一个version_seq做版本管理。这个模型的假设是Agent 的工作像事务一样有开始、有提交、有回滚。问题就出在这个假设上。Codex 的工作流根本不是一个事务它是一个无休止的 Agent Loop。1.2 为什么非要把 Codex 接进去Codex 是 OpenAI 出的编码智能体能自主执行多步任务、调用工具、修改文件但它有个先天毛病会话结束就忘。它不会默认把用户偏好 A 文件不允许动之前已经验证过方案 B 不可行这类信息沉淀下来。我们希望的是Codex 跑完一个任务后下次再启动时记得这个项目的约束、偏好、历史决策以及哪些代码是它自己改过的。这就是 TencentDB Agent Memory 要补的位置。但接入这件事的关键不在能不能调通 API而在两边对记忆语义的定义是否一致。源码读得越深越能发现一致性的裂缝。1.3 读源码的三条主线我建议按三条主线程读这份源码不要从第一个文件顺序读到最后一个消息写入路径AgentMemoryClient.append_message()到message_store看一条消息是怎么被落库、怎么被编版本号的。消息读取路径AgentMemoryClient.recall()到vector_index看一次记忆检索到底走了哪几步。任务生命周期路径task_manager怎么 commit、abort 一个 Task这是和 Codex 冲突最密集的地方。我这次踩到的三个硬冲突分别落在这三条主线上。2. 从入口模块开始拆记忆服务的骨架2.1 先让人能跑起来的第一层SDK 与任务上下文AgentMemoryClient是外部 Agent 接入的唯一入口。它的方法不多核心就这几个begin_task(task_id, meta)开启一个记忆任务append_message(task_id, role, payload)往里追加消息recall(query, top_k, filters)按语义检索相关记忆commit_task(task_id)正式把任务落库abort_task(task_id)丢弃这个任务从代码注释和命名能看出这个客户端设计时的参照对象是传统数据库连接begin_task对标BEGIN TRANSACTIONcommit_task对标COMMIT。它假设 Agent 一定知道整个任务从哪开始、到哪结束。但 Codex 不是这么工作的。Codex 的循环是while not done: step agent_loop.step() # 推理一步 observation tool_executor.run(step.tool_call) # 执行一个工具 context_fragments.append(observation) # 观察结果塞回上下文 if user_interrupts: agent_loop.revise_plan() # 用户随时打断、改方案这个循环里没有commit这个概念只有继续跑和被打断两种状态。2.2 存储层消息表、元数据与版本号的关系存储层是最直白的地方。message_store.py里的建表语句核心字段长这样CREATE TABLE message_records ( id BIGINT PRIMARY KEY, task_id VARCHAR(64) NOT NULL, session_id VARCHAR(64) NOT NULL, role TINYINT NOT NULL, -- 0: user, 1: assistant, 2: tool channel VARCHAR(32) DEFAULT chat, payload MEDIUMBLOB NOT NULL, -- 序列化的消息内容 payload_size INT NOT NULL, seq_no INT NOT NULL, -- 任务内自增序号 version_seq BIGINT NOT NULL, -- 全局线性版本号 prev_version_seq BIGINT, -- 指向前一个版本 created_at DATETIME NOT NULL, UNIQUE KEY uk_task_seq (task_id, seq_no), UNIQUE KEY uk_version (version_seq) );这个表的设计思路很清晰task_id seq_no唯一确定一条会话内消息version_seq用来做全局回滚和审计。version_log表则记录每次提交commit对应的version_seq区间CREATE TABLE version_log ( version_seq BIGINT PRIMARY KEY, task_id VARCHAR(64), status ENUM(OPEN,COMMITTED,ROLLED_BACK), parent_version_seq BIGINT );到这里为止一切都是线性的一次任务一段区间区间内消息连续全局版本号单调递增。2.3 检索与索引层异步向量索引的读写路径向量索引的写入路径很有意思。EmbeddingService.embed()本身是同步的但vector_index的写入是异步的——消息先写进message_records再丢进一个后台队列由 worker 批量计算 embedding 后写入向量库。def append_message(task_id, role, payload): record insert_message_record(task_id, role, payload) index_queue.enqueue(record.id, payload) # 异步进入 embedding 队列 return record这样做有它的道理不能因为 embedding 服务抖动把核心消息写入拖垮。但代价在哪儿代价是记忆的读己之写read-your-writes得不到保证。这个细节会在第四章节引爆。3. 硬冲突之一任务边界的幻觉 vs 自由 Agent Loop3.1 代码里 Task 的概念是怎么定的我读task_manager.py的时候发现Task 状态机有四个状态PENDING、ACTIVE、COMMITTED、ABORTED。class TaskState(Enum): PENDING 0 ACTIVE 1 COMMITTED 2 ABORTED 3task_manager里有一段关键逻辑def commit_task(task_id): messages get_messages_by_task(task_id) if messages and messages[-1].role ! assistant: raise IncompleteTaskError(任务必须以 assistant 消息结束) write_version_log(task_id, statusCOMMITTED)注意这个判断任务必须以 assistant 消息结束。这个规则的潜台词是一个完整任务应该是用户发指令 → Agent 执行 → Agent 汇报结果Agent 的最后一句话是任务的收尾。但这个规则放在 Codex 上完全不成立。Codex 的交互流程里用户和 Agent 是来回穿插的用户帮我修复这个函数的边界条件Codex 分析代码改了文件用户等一下先别提交把另一个文件也改了Codex 继续改另一个文件用户算了还是回退到方案 ACodex 回滚代码重新实现整个过程里到底有几个 Task如果把整个对话算一个 Task那seq_no会一直累加commit_task会等到对话彻底结束才提交。但问题来了如果对话持续三个小时任务期间 Agent 崩溃了怎么办ACTIVE状态的 Task 没有超时机制所有未提交的消息全部悬在那里。3.2 Codex 的会话流为什么对不上反过来如果我们把每次工具调用都当作一个 Task又会遇到另一个问题。Codex 一次让我改个 bug的请求内部可能触发十几次工具调用搜索文件、读文件、改文件、跑测试、再改文件、再跑测试。每个 Task 只有一条用户消息加几条 assistant/tool 消息数据碎片化严重。后续做语义检索时top_k召回的片段跨了好几个 Task上下文拼起来完全不成句。我在codex_adapter.py里找到了最初团队写的映射逻辑大概是这样def map_codex_turn_to_task(turn_id, session_state): if session_state.is_complete(): return commit_task(turn_id) else: return keep_task_active(turn_id)这个映射最大的问题is_complete()在 Codex 的 Agent Loop 里根本无法精确判断。用户沉默五分钟不代表任务结束Codex 内部在思考也不代表任务结束。结果就是大量的 Task 被白白挂在ACTIVE状态永远是未完成。3.3 冲突现场一条幽灵 Task和一个过度切分问题我们用测试复现的时候出现了两个典型症状。第一个症状叫幽灵 Task。某个会话里有三条消息但task_manager因为 commit 前的最后一条消息是 tool 消息不是 assistant 消息直接拒绝了 commit。数据实际上已经写入message_records但version_log里没有对应记录。从记忆服务的视角看这些消息是不存在的从 Codex 的视角看这段对话明明发生了。第二个症状是过度切分。一次 10 分钟的 Codex 会话因为每次工具调用都被映射成独立 Task最后生成了 47 个 Task。每个 Task 里只有一两条消息retrieval 的时候按 Task 粒度召回片段选出来的内容鸡零狗碎甚至把用户确认 OK这样的短消息和修改了 300 行代码的完整决策拆在两个 Task 里语义完全割裂。我还特意去查了git log发现codex_adapter.py里曾经有人加过一段按时间间隔切割 Task的逻辑——如果两条消息间隔超过 5 分钟就切一个新 Task。这个逻辑在测试里跑得还行一上真实场景就崩Codex 跑测试可能卡 6 分钟于是同一个逻辑任务被硬生生切成两半。后来这段逻辑被注释掉了但注释掉之后又回到了整个会话一个 Task的状态。这个循环往复其实就是任务边界定义不清的直接体现。4. 硬冲突之二召回的一致性与 Codex 的 token 预算打架4.1 写后不读异步索引的墙第二个冲突出在检索路径上核心是异步索引和Codex 的实时推理之间的时序矛盾。读vector_index.py能看到索引流程消息写入后它要经历index_queue.enqueue()→ worker 消费 → 调 embedding 服务 → 计算向量 → 写入向量库 → 刷新索引。这个过程短则几百毫秒长则几秒甚至更久取决于队列积压情况。而 Codex 的推理节奏是每隔几步就可能调用一次recall()来翻记忆。比如它刚和用户确认保持现有日志格式不要动马上又需要回忆这个项目日志格式的原始要求是什么。对 Codex 来说这句保持现有日志格式不要动是它自己刚说出来的应该是最高置信度的记忆。但实际执行recall()的时候这条消息可能还在 embedding 队列里排队根本搜不到。我一开始以为是参数问题top_k不够、阈值太严。后来在源码里加了日志发现队列积压的副本还在索引里确实没有新增的数据。这种写后不读在传统数据库里是不可接受的但在异步索引系统里却很常见——它优化了写入吞吐牺牲了立即读的一致性。这还不是最坏的。最坏的情况发生在 Codex 生成过程中反复切换上下文时用户要求它同时维护代码修改和项目规范两条线索Codex 每次切换都要召回而每次召回都可能撞上刚写入未索引的消息导致它反复看到过时的记忆并基于过时记忆做出决策。4.2 top-k 召回选中的偏偏是最旧的内容第二个子冲突更隐蔽也更致命。Codex 本身有 token 预算管理超过窗口的消息会被截断或压缩。它的处理方式是保最近的丢最远的——因为最近的消息通常和当前工作最相关。但 TencentDB Agent Memory 的recall()不是这样工作的。recall()按语义相似度取top_k个 chunk它不关心消息新旧。一个项目早期用户写下的详细需求说明embedding 和当前任务相似度极高会被前排召回而 Codex 刚刚产生的中间决策因为内容比较短小、语义密度低反而排在后面。结果就形成了一个荒诞的配置Codex 侧把旧消息截断了记忆服务侧又把同一批旧消息以高相似度 chunk的形式塞回上下文。两边不仅没互补还在打架。上下文窗口很快被旧内容塞满Codex 的实际可用 token 反而变少了表现为越接记忆越蠢。我用一个表格对照过两边的行为差异维度Codex 自带上下文管理TencentDB Agent Memory recall时间偏好倾向于保留最近消息只按语义打分不看时间容量策略超出预算就截断/压缩top-k 固定取 N 条冲突代价旧内容被丢弃旧内容被重新注入最终影响上下文被召回内容挤占Codex 忽略关键近期指令recall()的实现里其实支持filters参数可以按created_at做时间过滤。但codex_adapter.py里根本没有传递这个 filter默认裸调recall()。这算是一个典型的接口提供了能力但没有接入的适配层缺失。4.3 如何用日志和 SQL 复现这个冲突复现这个冲突不需要复杂的链路。我当时做了三件事第一打开vector_index的 DEBUG 日志观察队列积压时间。如果enqueue到write index的间隔持续超过 2 秒就说明异步滞后已经存在。第二在recall()调用点打印返回结果的created_at分布。如果你发现召回结果里最早的 chunk 比 Codex 当前会话开始时间还早很多而最近的几条消息一条都没召回那基本就坐实了。第三直接查表对比SELECT COUNT(*), MIN(created_at), MAX(created_at) FROM message_records WHERE task_id 当前任务; SELECT COUNT(*) FROM message_records WHERE task_id 当前任务 AND id IN (SELECT message_id FROM vector_index WHERE status INDEXED);第一行和第二行数据差得越多说明索引越滞后。我在实测环境里见过INDEXED数量只有消息总量一半的情况。这里还牵扯出一个细节Codex 的 token 预算不是一个固定值它随模型上下文长度变化。记忆服务如果不感知这个预算top_k取太少则记忆没意义取太多则挤占推理空间。最合理的做法是把top_k改成动态的和当前剩余上下文成正比。这个我们后面改造时加了进去。5. 硬冲突之三线性版本号撞上 checkpoint 分支回滚5.1 记忆版本号的设计意图第三处硬冲突也是我认为最严重的一处出在版本管理上。version_log的设计是线性版本每次 commit 分配一个新的version_seqparent_version_seq指向上一次 commit。回滚操作就是把当前指针指到某个历史version_seq并把中间的版本标记为ROLLED_BACK。这层设计的意图是好的所有记忆有迹可循出问题能回到任意历史点。它假设的模型是一条时间线、顺序推进、偶尔回退。5.2 Codex checkpoint 的分支语义但 Codex 的 checkpoint 是分支式的。用户可以在任意时刻创建一个 checkpoint跑一段试验代码发现不行再回到 checkpoint 重新来。重新来之后可能走一条完全不同的路线。用版本控制的语言说Codex 的工作流天然会产生一个 DAG有向无环图而不是一条线checkpoint A ├── 分支 1: 修改 security.py → 验证失败 └── 分支 2: 修改 auth.py → 验证通过当分支 2 产生新的记忆时它的父版本应该指向 A而不是指向分支 1 的最后一个 commit。但version_log表里没有branch_id只有线性递增的version_seq。于是冲突就发生了。5.3 从覆盖写事故到 DAG 改造我们实际遇到了两种故障。第一种是版本号冲突Codex 从 checkpoint A 回滚后开始新分支新分支的消息写入时version_seq仍然按全局自增分配这没问题。但parent_version_seq指向哪里出了问题——如果指向回滚后的 A 的快照那么 A 的旧分支记录和新分支记录共享同一个父节点从时间线上看两条分支交织在一起后续recall()按version_seq排序时旧分支的消息会和新分支的消息交错出现语义混乱。第二种是静默覆盖某些会话级缓存直接把回滚前的分支消息从缓存里删掉了如果此时保存了新分支的关联记录旧分支的引用就成了悬空。读出来之后会出现Codex 明明回滚了但记忆里还残留着旧分支的执行结论。我在version_log的代码里找到了一个防范措施def rollback_to(version_seq): mark_range_rolled_back(version_seq 1, current_version_seq)它只是把[version_seq 1, current]区间标记成ROLLED_BACK并没有引入分支维度。这个区间模型在单线程线性工作流里足够用但在 Codex 这种随时产生 fork 的工作流里区间标记根本无法表达两个分支并行存在。到最后ROLLED_BACK的记录还是会被检索到因为它们只是状态变了语义内容还留在message_records里。这次的教训是版本管理如果做成区间回滚就只能支持一条时间线一旦面对多分支的 Agent 工作流必须升级为 DAG 模型让每条记录都带着branch_id和parent_branch_id。6. 基于源码冲突做的三层适配改造6.1 会话适配层把 Agent Loop 归一进 Task 框架第一个冲突的解法我们没有去改 TencentDB Agent Memory 的 Task 状态机那属于核心存储层改起来影响面太大而是在codex_adapter.py里做了一个会话适配层。适配层做的事情很简单把 Codex 的 Agent Loop 显式地映射成 Task 边界。映射规则不是按用户消息切也不是按固定时间间隔切而是按里程碑切每次工具调用执行完成生成一个milestone用户明确表达继续这样不行改个方向生成一个milestonecheckpoint 创建或回滚生成一个milestone。每个milestone落地为一个微型 Task它的生命周期由适配层管理。适配层不会在 tool 消息后尝试 commit 任务而是等下一个里程碑出现再关闭当前 Task。def handle_milestone(session_state, milestone_type, payload): if current_task_has_messages(): if milestone_type in (tool_done, checkpoint, user_redirect): commit_task(current_task_id) current_task_id new_task_id() append_message(current_task_id, roleassistant, payloadpayload)这个改动的核心价值是让存储层看到的依然是规范的 Task 序列但 Task 的边界真正对齐了 Codex 的语义而不是靠猜测。6.2 短程缓存先解决 read-your-writes针对第二个冲突里的写后不读我们加了一层短程缓存放在recall()前面。具体做法是每条消息 append 之后除了进异步索引队列还同时写入一个 Redis 短程缓存key 是session_idvalue 是最近 30 条消息内容TTL 十分钟。recall()的路径改成了两级def recall(query, top_k): fresh short_term_cache.get(session_id) if fresh: local_hits rank_local(query, fresh) # 本地小范围语义匹配 if local_hits and len(local_hits) top_k: return local_hits return vector_index.recall(query, top_k)效果很直接Codex 刚说过的话90% 以上的场景能直接从短程缓存命中不用等异步索引。短程缓存不命中时再走向量召回。与此同时我们把recall()的top_k改成和 Codex 剩余 token 预算联动def calculate_top_k(codex_remaining_tokens): return max(3, min(15, codex_remaining_tokens // 3000))token 越紧张召回数量越少避免旧 chunk 挤占解码空间。至于top-k 召回选中旧内容的问题我们在召回时加了时间权重因子score similarity * time_decay相似度差不多的前提下新消息优先。这个权重不改变向量索引本身只影响排序改动可控。6.3 让版本号表达分支而不是覆盖第三个冲突的改造最重我们动了version_log的表结构。改动思路很朴素从线性区间升级为分支 DAG。新表结构增加了两个字段ALTER TABLE version_log ADD COLUMN branch_id VARCHAR(64) NOT NULL DEFAULT main, ADD COLUMN parent_branch_id VARCHAR(64) NULL, ADD COLUMN parent_version_seq BIGINT NULL;核心逻辑变成version_seq依然全局自增但不再表示时间线上的顺序只表示写入顺序每次 checkpoint 回滚后创建新分支时branch_id重新生成回滚时不再把[version_seq1, current]整段标记ROLLED_BACK而是把旧分支整体归档新分支的parent_branch_id指向回滚点的分支recall()默认只检索当前分支链路上的记忆除非显式传入branch_id要求跨分支检索。def rollback_to(branch_id, version_seq): archive_branch(branch_id, reasoncheckpoint_rollback) new_branch_id f{branch_id}-fork-{uuid4()[:8]} create_branch(new_branch_id, parent_branch_idbranch_id, parent_version_seqversion_seq) return new_branch_id这个改动带来的收益是旧分支记忆不会干扰新分支决策但也保留在库里随时可以做审计或回溯。Codex 的 checkpoint 再频繁记忆层也能保持语义干净。7. 读源码排冲突时的三个实操心得如果你也要把 Codex 或其他 Agent 接进记忆服务我最后分享三个排错心得。第一先复现再读源码。我一开始是按目录结构从头读到尾效率很低。后来改成先写一个失败场景的测试再顺着报错信息逆查源码几乎每条调用链都是被迫读完的记忆反而更牢。比如版本号冲突的问题就是先撞上Duplicate entry错误再倒查version_log的写入逻辑很快就定位到了表结构缺陷。第二看适配层的git log比看代码本身更有价值。codex_adapter.py里有一堆注释掉的逻辑比如按时间间隔切 Task、按轮次切 Task、强制提交未完成 Task。每段注释背后都是一次失败的尝试。读懂这些历史能少走很多弯路。第三别把冲突全打成补丁。我一开始想着尽量不动 TencentDB Agent Memory 的源码全在适配层绕开结果绕到后面发现版本 DAG 这层根本绕不过去。后来果断改了version_log的表结构一切才顺起来。记忆服务的任务模型和 Agent 工作流的语义偏差有时候必须改到底层否则只是一层补丁摞一层补丁。这次梳理做完之后我的体会是所谓接记忆的难点从来不在 API 调不通而在两个系统对一次任务一条记忆一个版本的定义是否兼容。把这些定义对齐剩下的都是体力活。