SuperAPI AI 工具集
合作ENAI 行业资讯收藏
Skill 广场ai-persistence/build-custom-adapter

ai-persistence/build-custom-adapter

智能体与模型

查看项目

更新于 10-16 10:42

技能介绍

用途

当应用需要在没有专用配方（recipe）的数据库上实现 TanStack AI 聊天持久化时使用本技能。适用的数据库包括原生 Postgres（pg / postgres.js）、Kysely、node:sqlite、MongoDB 等。

交付物是应用内的一个文件——src/lib/chat-persistence.ts——从应用已有的数据库客户端导出 ChatPersistence，外加该数据库所需的 DDL（通过应用现有的迁移流程添加）。不要创建包、第二个客户端或迁移运行器。

路由优先

如果应用已经在运行以下方案之一，请停止并改用对应技能，它包含驱动专用代码：

  • 应用运行 Drizzle ORM（任何方言）→ ai-persistence/build-drizzle-adapter
  • 应用运行 Prisma → ai-persistence/build-prisma-adapter
  • 应用运行 Cloudflare Workers + D1 → ai-persistence/build-cloudflare-adapter

其余情况都归入本技能。完整契约及其不变量见 ai-persistence/stores 和 docs/persistence/store-reference.md；完整的 node:sqlite 演练见 docs/persistence/build-your-own-chat-adapter.md 和 examples/ts-react-chat/src/lib/sqlite-persistence.ts。

1. 写任何代码前先阅读应用

需要确认的信息包括：客户端在何处导入（src/db.ts、src/lib/db.ts、src/server/db.ts 等），文件导入什么——绝不要构造第二个连接池；客户端生命周期（模块单例还是按请求工厂），决定导出 const chatPersistence 还是 function；迁移流程（migrations/、drizzle/、supabase/migrations/ 或 ORM CLI），用于应用 DDL；命名规范，需加前缀（chat_*）避免冲突；JSON 支持情况（jsonb、json、text），决定映射器是否 stringify/parse；导入别名（tsconfig.json paths），如 @/db、~/db、#/db 或相对路径。

2. 塑造存储结构

四类逻辑记录，无论使用何种引擎都保持以下键，因为存储方法正是按这些键查找记录：

  • thread：键为 threadId，字段 messages（完整对话数组）
  • run：键为 runId，字段 threadId、status、startedAt、finishedAt?、error?、usage?、parentRunId?、subagentRunId?、name?、sandboxKey?、detachedSince?、cancelRequested?、driverEpoch?
  • interrupt：键为 interruptId，字段 runId、threadId、status、requestedAt、resolvedAt?、payload、response?
  • metadata：键为 (namespace, key)，字段 value

记录中时间戳为 epoch 毫秒（number）。以引擎偏好的方式存储，在映射器中转换。(namespace, key) 是复合键，绝不要用分隔符拼接，('a:b','c') 与 ('a','b:c') 必须保持为不同记录。

为列表路径建立索引：runs(threadId, status)、runs(threadId, startedAt)、interrupts(threadId, requestedAt)。如实现 listReclaimable，还需索引 runs(status, detachedSince)；如实现 listByParentRun，还需索引 runs(parentRunId, startedAt)。

run.error 是结构化 RunError（{ message: string, code?: string }），不是裸字符串。在 SQL 表中存为两列（error、error_code），并在 update 中一起移动，避免残留过期的 code。run.status 取值为 'running' | 'interrupted' | 'completed' | 'failed' | 'aborted'；'interrupted' 是暂停而非终态，只有 'completed' | 'failed' | 'aborted' 是终态。

应用自有的额外列（如 userId、审计列）只要可为空或有默认值即可。

3. 不变量

这些不变量与引擎无关，写错任何一条通常会导致审批卡住或历史被清空：

  1. saveThread 是全量覆盖，绝不追加。参数是完整权威的对话记录。
  2. loadThread 对未知 thread 返回 []，绝不返回 null。
  3. createOrResume 是插入或返回已有——已存在的 runId 原样返回，忽略新字段值。
  4. runs.update 对未知 id 是静默无操作——不得抛出，也不得插入。
  5. runs.update 对耐用运行字段（sandboxKey、detachedSince、cancelRequested、driverEpoch）需区分“字段省略”与“字段显式清空”。必须检查 'detachedSince' in patch，而非 patch.detachedSince !== undefined。
  6. interrupts.create 是插入或返回已有——绝不把已解决的 interrupt 覆盖回 pending。所有 list* 按 requestedAt 升序排列。
  7. runs.listReclaimable 使用包含式截止时间（如实现）：status === 'running' AND detachedSince <= now - ttlMs。它是查询而非自动回收。

runs.findActiveRun 是必需的；runs.listByThread、runs.listByParentRun、runs.listReclaimable 是可选的。行映射器应省略不存在的可选字段，以便记录能干净地与参考内存后端比较。

4. 编写 src/lib/chat-persistence.ts

四个工厂加一个装配。文档给出以 pg（Postgres）为例的代码骨架，包括 createMessageStore、createRunStore、createInterruptStore、createMetadataStore 以及最终的 defineAIPersistence 装配。需注解 ChatPersistence 类型——裸 AIPersistence 是全可选容器，会被 withPersistence 拒绝。没有 locks 存储：stores 只接受 messages、runs、interrupts、metadata。协调通过 withLocks 单独接线（见 ai-core/locks）。

如果客户端是按请求的（Workers bindings、请求作用域事务），则导出 chatPersistence() 工厂而非 const，并在 handler 内调用。

引擎注意事项

  • Postgres（pg、postgres.js、Neon、Supabase）：jsonb 列可往返对象，读取路径可跳过 JSON.stringify；bigint 列在 pg 中返回字符串，需在映射器中用显式 Number() 转换，或将 epoch ms 存为 double precision 并转换一次。复合键为 PRIMARY KEY (namespace, key)。
  • Kysely：在应用的 Database 接口中定义四张表，然后使用 onConflict(...).doNothing() 和 executeTakeFirst()。updateTable(...).execute() 对零匹配已是无操作。
  • node:sqlite / better-sqlite3：完整实现在指南和示例文件中。在工厂作用域准备语句，使用 INSERT ... ON CONFLICT，JSON 存为文本，epoch ms 存为整数。将同步调用包装为 async 方法。
  • MongoDB：每种记录类型一个集合，_id 设为自然键（threadId、runId、interruptId）。metadata 使用 _id: { namespace, key } 复合子文档或唯一索引，绝不用分隔符拼接的字符串。createOrResume 使用 updateOne({ _id }, { $setOnInsert: doc }, { upsert: true }) 再 findOne。
  • Redis / Upstash：四种存储都适用，Redis 也很适合 LockStore。将每条记录存为一个 JSON 文档，通过有序集合索引列出 runs 和 interrupts。这些索引只有在记录及其加入的每个索引都由同一个 Lua 脚本（EVAL）写入时才安全。@upstash/agentkit-tanstack-ai（upstashPersistence()）是已发布的实现，通过一致性测试。
  • 其他：只需满足上述七条不变量即可，核心从不检查你的存储。

只采用一部分

很少需要同时使用全部四种存储。可以只实现自己拥有的，其余从其他基础填充，使用 composePersistence 和 memoryPersistence。只有列出的键会被替换。不存在跨存储事务，幂等不变量正是使重试安全的原因。

接入聊天路由

文档给出使用 chat、chatParamsFromRequest、toServerSentEventsResponse、openaiText、withPersistence、chatPersistence 的 POST handler 示例。threadId 对存储而言是裸字符串。应在路由处授权线程访问——从会话派生用户，绝不信任客户端提供的 id。

验证（必需）

这一点在此处比其他地方更重要：没有参考驱动可供对比，测试套件是唯一能阻止微妙幂等性 bug 进入生产的手段。使用 @tanstack/ai-persistence/testkit 的 runPersistenceConformance，指向一次性数据库并在运行间重置。套件覆盖全部七个存储，因此需声明每一处有意的省略。skip 从不接受 'locks'。如未实现 listByThread 或 listReclaimable，需用 skipMethods 单独声明，例如 { skipMethods: ['runs.listByThread'] }。未声明的省略方法会导致失败。