--- title: 批量导入 1000 条对话：成本拆分与性能验收 description: 从固定导入链路区分解析、事务、WASM 导出与摘要队列，并用临时合成输入检查流式保留范围与数据等价性。 image: "https://zengliangyi.com/og/blog/bulk-import-performance-plan.png" --- 批量导入的成本来自多个阶段：发现来源、读取和解析、归一化、SQL 写入、数据库导出；摘要、嵌入和关系发现又是另一条链路。先分开计时，才能知道该减少解析对象、导出次数，还是模型请求。本文保留“1000 条”作为合成样本规模，不把原稿的个人机器耗时、OOM 和优化倍数当作已验证结果。 ## 固定实现已经做了什么 [server/src/services/import.ts](https://github.com/ZengLiangYi/ChatCrystal/blob/764276814e5c5a33ecbfeae63754f74b9261a3a8/server/src/services/import.ts) 先对启用的适配器依次 detect/scan，把全部元数据收集到 allMetas，然后逐条检查、解析和写入。大小与修改时间相同可在解析前跳过；元数据变化但内容哈希相同只更新元数据。它不会因为已经记录 parserVersion 就强制重解析，见[适配器内部链路](/blog/source-adapter-internals/)。 每篇解析结果仍是完整 messages 数组。写入用单篇事务和预编译语句插入消息；替换会失效旧的自动笔记并安排向量清理，同时保留符合条件的用户编辑笔记或拒绝状态。批次结束调用 saveDatabase。[server/src/db/index.ts](https://github.com/ZengLiangYi/ChatCrystal/blob/764276814e5c5a33ecbfeae63754f74b9261a3a8/server/src/db/index.ts) 的自动保存仍由服务启动，没有在 importAll 中临时关闭；“每条都导出、导入时自动停保存、已经流式 parseAndInsert”都不是这份提交的行为。 导入返回 imported、replaced、skipped、errors 以及具体 ID 和 summarizationCandidateIds。候选 ID 不是已完成摘要。[server/src/services/summarize.ts](https://github.com/ZengLiangYi/ChatCrystal/blob/764276814e5c5a33ecbfeae63754f74b9261a3a8/server/src/services/summarize.ts) 才会执行经验筛选、摘要、嵌入和关系发现；摘要可被 filtered，嵌入失败也不一定使摘要失败。 ## 给各阶段定义可核对的成本 | 阶段 | 同时记录 | 先验证的条件 | | --- | --- | --- | | detect/scan | 来源、文件数、元数据数、耗时 | 可见文件与预期来源一致 | | parse/hash | 消息数、输入字节、最大单篇大小 | 内容字段和顺序保持一致 | | SQL/事务 | 新增、替换、跳过、失败 ID | 失败篇不留半条会话 | | export/save | 导出字节、调用次数、重开结果 | 内存提交与文件落盘分开验证 | | 摘要/嵌入/关系 | 入队与最终状态、请求数、错误 | 队列收据不等于完成或质量 | Node 的 [memoryUsage 文档](https://nodejs.org/download/release/v24.21.0/docs/api/process.html#processmemoryusage)区分 RSS、V8 heap 和 external；arrayBuffers 已包含在 external 中，不要重复相加。定时采样可能漏掉同步 export 的短峰值，末尾快照更不是整次峰值。记录采样方法、Node/依赖版本、硬件、样本和冷暖缓存条件；不要用对话数量推导固定内存门槛。 ## 用 1000 条合成输入比较保留范围 在独立目录用 Node 24 保存下面文件为 performance.mjs，先运行 `npm install --ignore-scripts --save-exact sql.js@1.14.2`，再运行 `node performance.mjs`。程序只创建临时归一化 JSONL 和独立 WASM 数据库：一条路径保留全部输入对象，另一条用 [readline](https://nodejs.org/download/release/v24.21.0/docs/api/readline.html) 逐行插入；两者都是每篇事务、每篇预编译消息语句、最后一次导出并重开检查。 它比较教学输入的对象保留范围和最终数据，不运行 ChatCrystal 的 detect/scan、厂商解析器、自动保存或模型；更不表示当前适配器已经支持边读边插入。计数中的 1000 与 1 是显式保留的归一化对象数，不是缓冲区大小、整个数据库内存或进程峰值。计时及内存快照会随机器和运行顺序变化。 ```js import assert from 'node:assert/strict' ; import { mkdtemp, writeFile, readFile, rm } from 'node:fs/promises' ; import { createReadStream } from 'node:fs' ; import { tmpdir } from 'node:os' ; import { join } from 'node:path' ; import { createInterface } from 'node:readline' ; import { performance } from 'node:perf_hooks' ; import initSqlJs from 'sql.js' ; const SQL = await initSqlJs (); const dir = await mkdtemp ( join ( tmpdir (), 'import-fixture-' )); const file = join (dir, 'normalized.jsonl' ); const count = 1000 ; const rows = Array. from ({ length: count }, ( _ , i ) => ({ id: `fixture-${ i }` , messages: [ { role: 'user' , content: `question ${ i }` }, { role: 'assistant' , content: `answer ${ i }` }, ], })); await writeFile (file, rows. map ( JSON .stringify). join ( ' \n ' ) + ' \n ' ); rows. length = 0 ; async function run ( retainAll ) { const db = new SQL . Database (); db. run ( 'CREATE TABLE messages (conversation TEXT, position INTEGER, role TEXT, content TEXT, PRIMARY KEY (conversation, position))' ); let peakRetained = 0 ; let imported = 0 ; const started = performance. now (); const insert = ( item ) => { assert. equal (item.messages. length , 2 ); db. run ( 'BEGIN' ); const stmt = db. prepare ( 'INSERT INTO messages VALUES (?, ?, ?, ?)' ); try { item.messages. forEach (( m , i ) => stmt. run ([item.id, i, m.role, m.content])); db. run ( 'COMMIT' ); imported ++ ; } catch (error) { db. run ( 'ROLLBACK' ); throw error; } finally { stmt. free (); } }; try { if (retainAll) { const items = ( await readFile (file, 'utf8' )). trim (). split ( ' \n ' ). map ( JSON .parse); peakRetained = items. length ; items. forEach (insert); } else { const lines = createInterface ({ input: createReadStream (file), crlfDelay: Infinity }); try { for await ( const line of lines) { if ( ! line. trim ()) continue ; peakRetained = Math. max (peakRetained, 1 ); insert ( JSON . parse (line)); } } finally { lines. close (); } } const values = db. exec ( 'SELECT conversation, position, role, content FROM messages ORDER BY conversation, position' )[ 0 ].values; assert. equal (imported, count); assert. equal (values. length , count * 2 ); const bytes = db. export (); const reopened = new SQL . Database (bytes); try { assert. equal (reopened. exec ( 'SELECT count(*) FROM messages' )[ 0 ].values[ 0 ][ 0 ], count * 2 ); } finally { reopened. close (); } return { values, peakRetained, elapsedMs: performance. now () - started, exportedBytes: bytes. length , memory: process. memoryUsage (), }; } finally { db. close (); } } try { const retained = await run ( true ); const streamed = await run ( false ); assert. deepEqual (streamed.values, retained.values); assert. equal (retained.peakRetained, count); assert. equal (streamed.peakRetained, 1 ); const { values : a , ... first } = retained; const { values : b , ... second } = streamed; console. log ( JSON . stringify ( { scope: 'synthetic normalized JSONL; not ChatCrystal importAll' , retained: first, streamed: second, }, null , 2 ) ); } finally { await rm (dir, { recursive: true , force: true }); } ``` ## 优化前先让验收失败一次 将逐行路径的 peakRetained 改成 count，保留范围断言应失败；只插入 user 消息，消息数量和等价性验收应失败。恢复代码再运行。程序输出 elapsedMs 和末尾 memory，仅作本次观察；两条路径在同一进程顺序执行，不能据此宣布速度或 RSS 获胜。 对真实导入另做故障样本：坏 JSON、超长单篇、重复/改写内容、不可写输出和落盘重开。流式读取减少某类输入对象，不会让 WASM 数据库变成外存数据库。先保证事务、内容哈希、替换语义和可恢复性，再决定是否值得增加异步流式接口，参见[WASM 持久化边界](/blog/sqljs-wasm-persistence/)。 ## 摘要队列和监听器分别验收 [server/src/queue/index.ts](https://github.com/ZengLiangYi/ChatCrystal/blob/764276814e5c5a33ecbfeae63754f74b9261a3a8/server/src/queue/index.ts) 的并发为 1，intervalCap 为 1、interval 为 1000ms；这是队列任务启动节奏，不是所有 Provider HTTP 请求的统一限速。内部 SDK 重试和任务重试可能增加请求；外层只在错误字符串包含 429 且仍有机会时指数等待，默认最多四次任务尝试。不能用入队数估算真实请求数或完成时间。 [server/src/watcher/index.ts](https://github.com/ZengLiangYi/ChatCrystal/blob/764276814e5c5a33ecbfeae63754f74b9261a3a8/server/src/watcher/index.ts) 在文件稳定等待之外再防抖 3 秒，同一 watcher 内已有导入时置 pending，结束后延迟重跑。该布尔量属于 watcher，不是所有 HTTP/CLI 导入共享的全局锁；云模式由 watcherPolicy 禁止监听。批量测试应固定入口并观察任务状态，避免把文件事件、解析次数和模型请求混成一个吞吐数字。 分享： ## 相关文章 [## AI 对话格式差异：事件、快照与跨库消息定位 对照五个固定适配器，区分 JSONL 容器、事件语义与 SQLite 键布局，用两个内存数据库观察消息索引、孤立数据和未知版本。 曾亮懿 2026年10月7日 阅读约 4 分钟 #ChatCrystal #Node.js #SQLite](/blog/ai-conversation-format-divergence) [## AI 对话管理方案：四种存储路径与迁移验收 比较 Claude Code、Cursor、Copilot、Trae 的固定适配器输入，区分原始副本、可读导出与知识库，并用合成清单验证身份和信息保留。 曾亮懿 2026年10月7日 阅读约 4 分钟 #ChatCrystal #Node.js #SQLite](/blog/ai-conversation-management-options) [## sql.js WASM 持久化：内存、导出、外键与保存点 对照 ChatCrystal 的数据库生命周期，运行 sql.js 1.14.2 样本，区分 SQL 提交、二进制导出、文件写入、外键恢复和嵌套回滚。 曾亮懿 2026年10月7日 阅读约 5 分钟 #ChatCrystal #Node.js #SQLite](/blog/sqljs-wasm-persistence) ## 从文章到代码 到我的 GitHub 看看 ChatCrystal、开发者工具和日常学习记录，欢迎交流实现细节。 [访问我的 GitHub](https://github.com/ZengLiangYi) ```json { "@context": "https://schema.org", "@type": "WebSite", "name": "ZengLiangYi Blog", "alternateName": [ "Zeng Liangyi", "ZengLiangYi", "zengliangyi.com" ], "url": "https://zengliangyi.com", "description": "曾亮懿的个人网站。记录 AI Native 探索、开发者工具与全栈产品实践。" } { "@context": "https://schema.org", "@type": "BlogPosting", "headline": "批量导入 1000 条对话：成本拆分与性能验收", "description": "从固定导入链路区分解析、事务、WASM 导出与摘要队列，并用临时合成输入检查流式保留范围与数据等价性。", "url": "https://zengliangyi.com/blog/bulk-import-performance-plan/", "image": "https://zengliangyi.com/images/blog/bulk-import-performance-plan.svg", "datePublished": "2026-10-07T00:00:00.000Z", "dateModified": "2026-10-07T00:00:00.000Z", "author": { "@type": "Person", "name": "曾亮懿", "url": "https://zengliangyi.com/about" }, "publisher": { "@type": "Organization", "name": "ZengLiangYi Blog", "logo": { "@type": "ImageObject", "url": "https://zengliangyi.com/favicon-96x96.png" } }, "mainEntityOfPage": { "@type": "WebPage", "@id": "https://zengliangyi.com/blog/bulk-import-performance-plan/" } } { "@context": "https://schema.org", "@type": "BreadcrumbList", "itemListElement": [ { "@type": "ListItem", "position": 1, "name": "首页", "item": "https://zengliangyi.com/" }, { "@type": "ListItem", "position": 2, "name": "文章", "item": "https://zengliangyi.com/blog" }, { "@type": "ListItem", "position": 3, "name": "批量导入 1000 条对话：成本拆分与性能验收", "item": "https://zengliangyi.com/blog/bulk-import-performance-plan/" } ] } ```