做AI对话类应用大多数人一开始关心的是模型API怎么调、提示词怎么写总觉得数据层不就是存几条聊天记录吗有什么好设计的。直到用户量上来、会话多起来、流式输出一开才发现问题一个接一个消息回放顺序乱了、A用户看到了B的对话、同一条消息重复入库、断线重连后AI只回了一半。我这次给TinyRobot Kit重构对话数据层核心就围绕两个关键词展开流式消息和多会话。TinyRobot Kit是一个面向AI对话场景的轻量级开发套件之前它的数据层只是简单的一张消息表这次我把它重构成了接收-暂存-合并-回放的完整链路顺手把会话隔离、幂等写入、断线恢复这些坑都填了。如果你正在做AI聊天机器人、Agent应用或者用大模型API做产品但不确定消息怎么存、会话怎么管这篇文章可以帮你省掉几周的返工时间。1. 先想清楚对话数据层到底在解决什么1.1 一次正常对话背后藏着哪些状态很多人觉得一条对话消息就是一个请求一个响应但其实从数据层的角度看一条消息可以拆出好几个状态阶段。用户点发送消息进入已发送模型开始输出消息进入流式生成中模型输出完消息才变成已完成。如果中途断网、超时、进程重启这条消息可能是已中断甚至是失败。我第一次设计TinyRobot Kit时没把这件事想清楚就建了一张message表字段是id、session_id、role、content、created_at。用户输入和AI输出都往里插AI输出结束时再UPDATE一次content。单用户、单会话、非流式的情况下完全够用。但一旦改成SSE流式输出麻烦立刻来了前端每300毫秒收到一个增量每次收到增量我都不知道是直接拼到content里还是等结束了再写入。如果边收边拼那数据库里这条消息长时间处于半成品状态用户一旦刷新页面看到的就是一段残缺的文本。所以做对话数据层第一步不是建表而是梳理状态。一条消息至少要具备四个状态streaming代表正在生成completed代表已结束且内容完整interrupted代表流被中断failed代表调用失败。给消息增加状态位之后很多逻辑就顺了查询历史消息时completed的消息正常展示interrupted的消息可以展示已接收的部分并提示用户这条回答没有生成完前端如果想断点续传也可以根据状态决定是否重新请求。1.2 流式消息与普通HTTP请求的本质差异流式消息和普通HTTP请求最大的区别在于完成边界不一样。普通请求的响应是完整到达的你收到一个JSON解析完就能用数据的生命周期是一次性的。但SSE流式响应是分块的数据是一块一块到达的服务端直到接收完所有块才知道这条消息是不是完整更准确地说它还会收到一个显式的结束标记比如[DONE]或者流关闭事件你才知道这下是真的完了。这就产生了两个核心设计问题。第一写入时机是收完了再写入数据库还是收到一块就写一块收完再写逻辑简单但风险很高网络一旦抖动前面几秒收到的块全丢。收到一块写一块可靠性强了但要解决未完成消息的中间状态管理还要处理块与块之间的顺序。第二内容存储前端看到的是逐字输出的效果数据层如果只保存最终合并后的完整content那断线重连时前端想恢复逐字输出的过程就做不到了。TinyRobot Kit最终选择了边收边存的方案。每次收到模型返回的增量块就把它落进一张独立的chunk表。这样即使流中断最少也能保住已经接收到的所有增量文本。等收到结束标记再把chunk表里的增量块按顺序合并成一个完整content回写到message表同时把状态从streaming改为completed。整个过程是一个状态机驱动的增量写入配合幂等设计靠数据结构的清晰来对抗网络的不确定性。1.3 多会话带来的三个数据层挑战多会话不是简单地在前面加一个session_id字段就完事。我在实际重构TinyRobot Kit时明显感觉到有三个挑战是绕不开的。第一个是隔离。一个用户打开多个会话多个用户同时在使用系统底层表里所有消息是混在一起的。如果查询时只按session_id过滤不校验user_id那只要有人猜到或遍历到其他session_id就能看到别人的对话内容。这个隐患很隐蔽因为开发阶段你大概率只有一个测试账号根本暴露不出来。第二个是上下文的正确性。AI会话的上下文窗口是按会话隔离的用户在这个会话里聊了十轮生成第十一轮回答时数据层要能按会话把前面20条消息按正确顺序取出来拼给模型。听起来简单但一旦之前用了created_at排序、并发插入时时间戳一样顺序就可能错乱模型拿到的上下文就是乱的输出质量直接跳水。第三个是并发写入。同一个会话里用户连续发两条消息或者前端因为超时自动重试了同一请求数据层会同时面对多路写入。如果不做幂等控制就会出现同一条消息被插入两次、AI生成内容被覆盖、chunk内容和最终content不一致这些脏数据问题。这三个挑战环环相扣如果一开始不把数据模型设计好后面每个问题都要靠补丁去修会非常痛苦。2. 数据模型怎么设计三张表撑起整个对话层2.1 session表会话的最小租赁单位session表承担的是会话元信息的职责。它的核心字段不只有id和user_id我建议至少要包含title用于展示会话标题status标识会话是active还是archived方便做归档清理context_id用于绑定上下文窗口配置不同会话可以使用不同模型参数created_at和updated_at用于生命周期管理。这里有一个容易被忽略的点会话的所有查询必须带上user_id。这并不是说session_id不能作为唯一主键而是说会话属于谁这个归属关系必须被索引并且尽量在所有查询入口都强约束确保即使session_id泄漏也无法越权访问。建表时我给(user_id, id)做一个联合索引而不是只在id上建主键索引这样既能保证id唯一又能让按用户查会话走索引。session表不用设计得很复杂它本质上就是一棵树根所有消息都挂在它下面。它不需要存任何对话内容否则查询历史时会多一次大字段的读取。2.2 message表消息状态机落在哪message表是整个对话数据层最核心的表它记录一条条已经完结或正在生成的完整消息。我的设计里每一条消息都拥有session_id、roleuser/assistant/system、content、status、seq_no、version、created_at。这里最关键的是seq_no和version。seq_no是会话内的消息序号我坚持使用一个在会话内严格递增的整数。消息插入时取当前会话最大seq_no加一表面看起来比直接生成UUID多了两步但它在排序、回放、上下文构建时都极其有用。created_at在并发插入、时钟回拨时是不可靠的seq_no才是真正可靠的顺序依据。version字段用来做乐观锁控制。比如同一个assistant消息正在流式生成前端又发了一个暂停生成请求后台要把状态改成interrupted这时通过UPDATE ... WHERE id ? AND version ?可以防止两个请求互相覆盖。version字段不一定要参与业务逻辑但在并发操作高发的AI服务里它是一道便宜又有效的安全网。message表里的status字段我建议建立索引。理由是流式生成期间系统需要频繁查询有哪些消息还卡在streaming状态用于断线恢复和后台清理任务。没有索引的情况下一个用户历史消息上千条每次状态扫描都是一次全表扫描数据量大了之后耗时成倍增长。2.3 message_chunk表为什么必须存流式增量块message_chunk表是流式消息方案和传统CRUD方案最大的区别所在。它负责保存AI流式返回时每一个增量块表结构非常简单message_id、seq、content、created_at主键是(message_id, seq)。这张表存在的意义是在message表之外开辟一块临时拼装区。为什么不能直接在message.content上做追加更新因为每次追加都是一次UPDATE单行大字段的操作多个chunk并发写入时数据库行锁竞争会非常严重。而且一旦过程中断了你不能确定content字段里拼到哪一步了数据状态模糊不清。有独立的chunk表后每个chunk都是独立的INSERT互不干扰断不断流都无所谓数据是增量的、可恢复的。chunk表还有一个隐藏作用支持流式回放。前端断线重连时如果消息状态还是streaming服务端可以从chunk表把已经收到的增量块按序返回给前端前端可以继续在原来的流式效果上追加而不是重新开始或直接显示残文。合并完成后这些chunk要不要删除TinyRobot Kit目前的做法是保留。因为chunk表按主键(message_id, seq)组织磁盘占用相对可控而保留全量chunk可以方便做对话过程分析、数据回放和调试。如果你的存储成本敏感也可以只保留最近N天更早的chunk离线清理message表里的完整content不受影响。3. 实操用 Node.js Redis PostgreSQL 落地 TinyRobot Kit3.1 项目骨架和几个关键配置TinyRobot Kit的重构我用了Node.js TypeScript存储选了PostgreSQL缓存和锁用Redis。这套选型的理由很简单SSE解析和流式转发的生态很成熟TypeScript对异步流处理的类型提示比Python顺手PostgreSQL的string_agg和ON CONFLICT能把chunk合并和幂等插入写得很干净。项目目录结构我建议按数据流拆分而不是按传统MVC拆tinyrobot-kit/ src/ api/ # HTTP与SSE入口路由 streams/ # SSE接收、解析、转发逻辑 store/ # 数据访问层session/message/chunk services/ # 业务编排发送消息、生成、合并、回放 workers/ # 后台任务超时扫描、失败重试连接池配置是很多人会忽略的细节。流式场景下前端每收到一个块服务端就会往chunk表写一次一个完整的回答可能产生几十到几百条chunk写入。如果用默认连接池几十个并发会话一开连接数立刻打满。我在配置里把PostgreSQL连接池的最小连接数调到了10最大连接数根据CPU核数乘以4来设并且给chunk写入单独开了一个小连接池避免与普通查询互相抢占。环境变量方面Redis的key统一加tinyrobot:前缀避免与其他业务混在一起。PostgreSQL和Redis的密码都不要写死在代码里用环境变量注入本地开发再用.env。3.2 流式消息写入预生成消息ID与幂等追加流式写入最怕的一件事是前端重试。前端因为网络抖动重发了用户消息后端如果重新创建一条message记录数据库里就出现两条相同内容的用户消息。解决思路是预生成消息ID前端在发起会话前先向后端请求一个会话状态拿到唯一的sessionId和本次消息的messageId后续所有写入都绑定这个messageId。消息写入的核心流程分三步。第一步创建一条status为streaming的message记录content可以先填空。第二步接住模型的SSE流每个增量块解析出来后写入message_chunk表。第三步收到流的结束标记后执行合并UPDATE。chunk写入的幂等控制我直接用PostgreSQL的ON CONFLICT DO NOTHING// ChunkStore.append await sql INSERT INTO message_chunk (session_id, message_id, seq, content) VALUES (${sessionId}, ${messageId}, ${seq}, ${content}) ON CONFLICT (message_id, seq) DO NOTHING ;这里(message_id, seq)就是天然的唯一约束。同一个消息同一个序号重复写入不会产生新数据幂等性一句话就解决了。SSE接收端我封装成一个独立函数把流式解析和存储逻辑解耦import { createParser } from eventsource-parser; export async function consumeStream({ sessionId, messageId, upstream, chunkStore, }: { sessionId: string; messageId: string; upstream: ReadableStream; chunkStore: ChunkStore; }) { let seq 0; const parser createParser((event) { if (event.type ! event) return; if (event.data [DONE]) return; const payload JSON.parse(event.data); const delta payload.choices?.[0]?.delta?.content; if (!delta) return; // 每个增量块立即落库 await chunkStore.append(sessionId, messageId, seq, delta); }); const reader upstream.getReader(); const decoder new TextDecoder(); while (true) { const { done, value } await reader.read(); if (done) break; parser.feed(decoder.decode(value, { stream: true })); } // 流正常结束执行最终合并 }这段代码体现了流式数据层最重要的原则不要等到流结束才做IO每个块到达就立刻持久化即使进程在下一秒崩溃前面已接收的内容也不会丢。3.3 合并与回放重连后消息怎么补流结束后的合并操作我用PostgreSQL的聚合函数完成保证chunk按顺序拼成一个完整contentUPDATE message SET content ( SELECT string_agg(content, ORDER BY seq) FROM message_chunk WHERE message_id $1 ), status completed, completed_at NOW(), version version 1 WHERE id $1 AND status streaming;WHERE status streaming这个条件可以防止合并操作重复执行。如果后台任务或重放逻辑重复调用合并第二次执行时消息状态已经变成completedUPDATE影响行数为0天然避免了重复覆盖。回放功能是流式消息系统比较关键的体验细节。当用户刷新页面或重连时前端会请求某个会话的完整消息列表。服务端不能只返回message表里的content因为如果有消息还处于streaming状态message表里它的内容是空或者旧的前端拿到的是半截内容。正确做法是async function getMessagesForPlayback(sessionId: string, userId: string) { const list await getMessagesBySession(sessionId, userId); const result []; for (const msg of list) { if (msg.status streaming) { const chunks await chunkStore.listByMessage(msg.id); // 重放时把已收到的chunk拼起来给前端 msg.content chunks.map((c) c.content).join(); msg.streaming true; } result.push(msg); } return result; }这样前端刷新后可以看到普通消息正常展示未生成完的消息展示已接收部分内容 仍在生成中的状态等后台流结束前端可以通过轮询或WebSocket收到完成信号再刷新一次内容。3.4 多会话并发控制Redis锁与版本号怎么配合多会话场景下的并发控制核心目标是避免同一个会话同时被两个生成任务操作。TinyRobot Kit的方案是双保险Redis锁做互斥version字段做乐观锁兜底。Redis互斥锁的逻辑很直接const locked await redis.set( tinyrobot:lock:${sessionId}, 1, NX, EX, 30 ); if (!locked) { throw new Error(当前会话正在生成中请稍后再试); }用户在一个会话里连续发两条消息第二条请求进来时发现锁还在直接返回提示前端就可以引导用户等待。锁的过期时间设为30秒正常情况下一次生成不会超过30秒如果模型特别慢生成过程中可以续期避免任务还在跑、锁先过期导致另一个请求闯进来。但Redis锁不是万无一失的锁过期时恰好旧任务还没完新任务又进来了就会冲突。所以version字段是最后一道防线。生成任务在最终合并时会执行UPDATE ... WHERE id ? AND version ?如果版本号不匹配说明消息被另一个操作改过这次合并直接放弃保证数据不被乱覆盖。为什么不直接用数据库行锁因为流式写入是小事务高频操作chunk表每秒可能触发几十次INSERT行锁在高并发下很容易成为瓶颈。Redis的原子操作和version字段配合已经可以覆盖绝大多数冲突场景性能损耗也小得多。4. 踩坑实录流式会话数据层常见问题与排查4.1 消息凭空消失半截现象用户刷新页面后发现AI回复的内容只有前半段后半段没了。第一次遇到时我以为是大模型输出截断查了日志才发现流根本没有正常结束。原因模型API的流式响应在中途抛了异常服务端没有捕获到结束信号就退出了。当时代码里把流结束的标记写在了parser回调里但连接异常时回调不触发而message状态还停留在streamingcontent还是空的用户只能看到一条正在生成中的占位消息。解决把状态机做进每条消息的生命周期里。在consumeStream函数的finally块中判断消息当前状态如果还是streaming说明流异常终止了把它标记为interrupted如果已经收到[DONE]说明是一次正常结束。随后再用一个后台任务扫描所有长时间处于streaming状态的消息超过3分钟自动标记为interrupted并触发告警。4.2 会话串线A用户看到了B的对话现象联调时发现用户A登录后他的会话列表里出现了用户B创建的会话。原因初期接口设计时创建会话和查询消息都只传了session_id服务端以为只需要这一个参数就能定位数据没校验这个会话是否属于当前登录用户。后来运维排查发现前端为了图方便把session_id存到了localStorage不同用户在同一台机器上切换登录时旧session_id被新用户带上了。解决统一在服务端中间件解析登录态从token里取出user_id凡是按session_id查询的地方都强制追加user_id ?条件。同时后端接口不再信任客户端传的user_id只信任token解析出的身份。落地时在数据访问层加了一个强制校验函数所有写操作都必须先通过assertSessionOwnership漏掉这个校验的代码直接编译期报错。4.3 回放顺序错乱别再用created_at排序现象某个用户的历史对话里消息顺序偶尔会颠倒上一条回答出现在了下一条提问后面导致上下文构建时模型理解出错。原因消息表里没有seq_no查询历史时直接用ORDER BY created_at。并发插入时数据库时间戳精度不够两条消息的created_at完全相同排序结果不稳定有时先插入的排到了后面。解决给message表新增seq_no字段在会话内单调递增。所有查询历史、构建上下文的地方都改用ORDER BY seq_nocreated_at只作为展示层的时间显示。这里有一个配套操作为了兼容历史数据迁移脚本把已有的消息按created_at顺序补上seq_no后续新消息通过SELECT COALESCE(MAX(seq_no), 0) 1 FROM message WHERE session_id ?来生成。这个查询执行频繁我给sleep(session_id, seq_no)建了唯一索引保证同一会话内不会有重复序号。4.4 排查工具和日志设计清单流式会话问题比普通接口难排查因为数据是动态的出问题时消息往往处于中间状态。我总结了一套排查思路基本可以覆盖绝大多数问题场景。症状可能原因优先排查项解决方案消息只有半截流中断未标记状态message表status是否为streaming状态机 超时扫描任务会话串线只校验sessionId未校验userId日志对比sessionId与userId映射中间件统一鉴权强制归属校验顺序错乱按created_at排序对比seq_no与实际顺序统一改用seq_no排序消息重复入库前端重试未做幂等同一消息id是否存在多条预生成messageId ON CONFLICT内容与chunk不一致合并失败或重复合并对比message.content与chunk聚合结果UPDATE条件加status约束日志设计上我给每次会话请求生成一个traceId从HTTP入口到SSE流结束全程携带写入chunk时也带上这个traceId。排查问题时只需要在Redis里查tinyrobot:lock:{sessionId}是否存在再用SQL查SELECT id, status, seq_no, created_at FROM message WHERE session_id ? ORDER BY seq_no DESC LIMIT 10基本一眼就能定位问题出在哪个环节。还有一个我后来才加上的小工具定时跑一个探针脚本往一个专用测试会话里发一条固定消息检查它能否在30秒内正常完成流式落库。一旦脚本报警说明上游模型或者数据链路出问题了比用户反馈快得多。我在实际把TinyRobot Kit的对话数据层重构成流式消息多会话结构之后最深的体会是数据层不是简单的CRUD流式和并发逼迫你把状态管理提前想清楚。如果一开始就设计了消息状态机、seq_no、幂等写入后面那些数据错乱和串线问题根本不会出现。这个方案后续还可以继续扩展比如把chunk分析能力做成对话过程看板或者用Redis Stream做消息缓冲进一步降低数据库压力。对于正在做AI对话类产品的人来说先把这三张表和数据流理清楚比追求复杂的架构方案有用得多。