企业官网建设流程全解析

做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、role（user/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 Kit

3.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按顺序拼成一个完整content：

UPDATE 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'这个条件可以防止合并操作重复执行。如果后台任务或重放逻辑重复调用合并，第二次执行时消息状态已经变成completed，UPDATE影响行数为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状态还停留在streaming，content还是空的，用户只能看到一条"正在生成中"的占位消息。

解决：把状态机做进每条消息的生命周期里。在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_no，created_at只作为展示层的时间显示。这里有一个配套操作：为了兼容历史数据，迁移脚本把已有的消息按created_at顺序补上seq_no，后续新消息通过SELECT COALESCE(MAX(seq_no), 0) + 1 FROM message WHERE session_id = ?来生成。这个查询执行频繁，我给sleep(session_id, seq_no)建了唯一索引，保证同一会话内不会有重复序号。

4.4 排查工具和日志设计清单

流式会话问题比普通接口难排查，因为数据是动态的，出问题时消息往往处于中间状态。我总结了一套排查思路，基本可以覆盖绝大多数问题场景。

日志设计上，我给每次会话请求生成一个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对话类产品的人来说，先把这三张表和数据流理清楚，比追求复杂的架构方案有用得多。