做 AI 应用的人都有一个共同体验第一版 Demo 跑通特别快把大模型接口一调、前端一个流式渲染加上去看起来就能用了。但一旦想把它变成真正能落地的产品比如一个支持多会话、流式输出的 AI 机器人套件你会发现最花时间的根本不是调模型接口而是怎么把对话数据这一层组织好。TinyRobot Kit 就是在这个背景下折腾出来的东西。它不是又一个 LLM 封装库而是一个偏数据层的对话机器人套件核心就解决三件事流式消息怎么可靠落库、多会话之间怎么隔离与切换、模型需要的上下文档怎么从库里高效组装。这篇文章按我实际设计和迭代这套 Kit 的过程把数据层的组织思路、核心表结构、流式写入方案、多会话并发处理、以及我在生产环境踩过的坑完整复盘一遍。正在做对话类 AI 应用的开发者尤其是从调通接口往做产品过渡阶段的同学应该能从里面找到不少可以直接抄走的方案。1. 对话数据层AI 应用里最容易被低估的一层1.1 从调用模型到组织对话的转折点大多数 AI 项目的演进路径是高度相似的先调通一个 Chat Completion 接口用流式把 token 吐到前端做一个打字机效果Demo 就成立了。但 Demo 和产品之间隔着一整层东西——数据。普通 CRUD 应用的数据是结构化的、静止的用户表、订单表、商品表一条记录就是一个完整的事实。对话数据完全不是这样。一条消息从生成到最终定稿中间经历了正在生成、生成了一部分、生成完毕、生成失败等多个状态这些状态还是高频率变化的模型每秒钟可能吐出来几十个 token聊到第三十轮的时候用户想回看第一轮说了什么数据还得按原样还原出来。TinyRobot Kit 要做的第一件事就是把对话这个概念从内存里搬到持久化存储里。模型调用是可以无状态的但产品不能无状态。用户关掉页面再打开会话还在网络断了一下已经生成的内容不能丢两个会话轮流切换每个会话的上下文不能串。1.2 数据层必须回答的三个问题我在设计 Kit 的数据层时把所有需求收敛成了三个问题一条消息的一生是怎样的从用户按下发送到模型返回完整内容这条消息经历了哪些阶段每个阶段的数据要存成什么样多个会话如何并存一个用户可能同时开着五个会话后台可能还有一个定时任务在往会话里写数据。如何保证互不干扰、切换无损模型的上下文从哪里来每次请求模型前Kit 都要把历史消息取出来拼成 prompt。这块取数逻辑如果做得糙轮数一多就会出大问题。这三个问题分别对应了消息生命周期管理、会话隔离与并发控制、上下文组装策略。下面的章节就按这条线展开。2. TinyRobot Kit 的会话模型Session 与 Message 的边界到底怎么划2.1 为什么不能只存一张消息表最早我的想法特别朴素一张 message 表字段是 id、会话 id、角色、内容、时间完事了。但用起来很快就发现不够。最直接的问题是——会话本身也是需要存东西的。每个会话都有标题、创建时间、最后活跃时间、关联的用户、甚至还有模型参数temperature、max_tokens。这些字段如果冗余到每条消息里浪费且容易不一致如果完全不带会话列表页就没法做了。所以第一步就是把数据拆成两层sessions和messages。sessions表存的是会话的骨架它回答的是有哪些会话、每个会话属于谁、处于什么状态CREATE TABLE sessions ( id TEXT PRIMARY KEY, user_id TEXT NOT NULL, title TEXT DEFAULT 新会话, status TEXT NOT NULL DEFAULT active, -- active / archived / deleted system_prompt TEXT, model_config JSONB, -- temperature、max_tokens 等参数 last_message_at TIMESTAMPTZ, message_count INTEGER DEFAULT 0, created_at TIMESTAMPTZ NOT NULL DEFAULT now(), updated_at TIMESTAMPTZ NOT NULL DEFAULT now() ); CREATE INDEX idx_sessions_user_time ON sessions (user_id, updated_at DESC);messages表存的是对话的血肉回答的是这个会话里聊了什么、每条消息现在怎么样了CREATE TABLE messages ( id TEXT PRIMARY KEY, session_id TEXT NOT NULL REFERENCES sessions(id), role TEXT NOT NULL, -- system / user / assistant / tool content TEXT, -- 消息最终内容 content_delta TEXT, -- 流式增量写入的临时区域 status TEXT NOT NULL DEFAULT pending, -- pending / streaming / completed / failed seq INTEGER NOT NULL, -- 会话内递增序号 token_count INTEGER DEFAULT 0, parent_id TEXT, -- 用于将来支持分支会话 meta JSONB, -- 扩展字段比如工具调用参数 created_at TIMESTAMPTZ NOT NULL DEFAULT now(), updated_at TIMESTAMPTZ NOT NULL DEFAULT now() ); CREATE INDEX idx_messages_session_seq ON messages (session_id, seq);2.2 消息状态机一条消息的一生如果只需要存最终结论content一个字段就够了。但要支持流式你得先把生成中这个状态显式建模出来。我把消息的生命周期定义成了四个状态pending等待中请求刚创建消息记录已插入但模型还没有返回任何内容。streaming生成中模型已经开始返回增量content_delta不断追加。completed完成流结束content_delta合并进contenttoken 计数写入。failed失败请求中断或异常保留已生成的部分打上失败标记。这个状态机是整个数据层的核心。它的价值在于任何时刻你都能准确回答这条消息现在到底怎么样了。前端要显示 loading 还是完整内容、后台要决定是否重试、日志要记录失败原因全靠它。还有一个容易被忽略的点seq字段。同一会话内的消息如果按created_at排序在并发写入、时钟回拨、批量导入的场景下顺序是不可靠的。用session_id seq作为排序键和唯一键才能保证顺序的确定性。我把seq设计成会话内的单调递增序列插入新消息时用max(seq) 1计算宁可多一条 SQL 查询也不在排序上埋雷。2.3 会话元数据与上下文的关系设计完表结构我意识到一个更深的问题会话不只是消息的容器它还隐含了这个对话的背景。同一个 session 里system_prompt是全局设定的、history是需要动态截取的、model_config是固定附带的。这三类数据如果散落各处组装上下文的时候就要到处拼。所以我把system_prompt和model_config直接冗余进了sessions表而不是做成关联表。冗余在这里不是坏味道而是刻意的取舍。理由是会话表的变化频率极低但读取频率极高——每次请求模型、每次打开会话列表都要读。与其每次都 JOIN 一张配置表不如在写入会话时一次性冗余好读取时一条 SQL 全拿出来。这个取舍在规模变大之后收益非常明显。3. 流式消息落库打字机效果的每一帧都不能丢3.1 流式过程的数据视角拆解大模型的流式返回从数据视角看是这样一个过程请求发出后服务端在几秒到几十秒的时间里陆续吐出一批又一批的文本增量。每批增量都是一个小的文本块把它们按顺序拼接起来才是一条完整的 assistant 消息。实现流式落库最笨的办法是每收到一个增量就 UPDATE 一次数据库。在模型返回慢、增量频率低的时候这样做没问题。但现在的模型返回速度越来越快某些情况下每秒钟会有几十帧数据过来。如果每一帧都触发一次 SQL UPDATE数据库连接会被迅速打满而且大量写入是多余的——用户根本不在乎中间那几十个中间态。TinyRobot Kit 采用的方案是前端走 SSE 实时收每一帧后端落库走批量合并。SSEServer-Sent Events负责把增量即时推给前端渲染保证打字机效果数据库落库则用一个增量缓冲区按时间窗口批量合并写入。3.2 增量缓冲区的实现我定义了一个流式写入器 StreamingMessageWriter它的职责只有一个接收增量合并写入控制落库频率。class StreamingMessageWriter: def __init__(self, session_id: str, message_id: str, repo, flush_interval2.0): self.session_id session_id self.message_id message_id self.repo repo self.buffer [] self.flush_interval flush_interval self._last_flush time.time() async def append(self, delta: str): self.buffer.append(delta) # 缓冲达到阈值立即落库 if len(.join(self.buffer)) 512: await self.flush() async def flush(self): if not self.buffer: return text .join(self.buffer) self.buffer.clear() await self.repo.append_delta(self.session_id, self.message_id, text) self._last_flush time.time()这里有两个触发落库的条件一是缓冲区文本达到 512 字符二是定时器到了 2 秒。前者保证了大段文本不会长期滞留内存后者保证了小碎块增量也会被周期性地持久化。也就是说不论增量大小最迟 2 秒内的数据一定已经进库。为什么是 512 和 2 秒这两个数没有绝对标准是我在实测中调出来的。设太小比如 128 字符写入频率还是偏高设太大比如 2048在模型输出慢的场景下用户关页面时容易丢失较多数据。2 秒的定时刷新配合流结束时的强制 flush可以做到最多丢 2 秒的增量——这个损失在已生成内容的视角下是完全可以接受的。3.3 流结束时的定稿操作流式返回结束之后不能直接拍屁股走人。缓冲区里可能还有没落库的增量消息状态还是 streamingtoken 计数还没更新。我专门做了一个 finalize 流程async def finalize(self, token_count: int, extra_meta: dict None): await self.flush() content await self.repo.get_full_content(self.message_id) await self.repo.complete_message( message_idself.message_id, contentcontent, token_counttoken_count, metaextra_meta, ) # 顺带更新会话的 last_message_at 和 message_count await self.repo.bump_session(self.session_id)注意complete_message这一步它做的不只是 UPDATE 状态还把之前累积在content_delta里的所有碎片合并成完整文本写入content。我选择用一次读再写的方式而不是在内存里维护完整内容原因是在分布式部署下负责流式接收的进程和负责定稿的进程可能不是同一个从库里拿全量最稳妥。3.4 中断与恢复让残片数据也能自洽流式请求最讨厌的地方在于它随时可能断。网络抖动、用户关页面、模型服务超时随便一种情况都会让消息停在生成了一半的状态。TinyRobot Kit 的处理策略是这样的流中断时写入器先尝试执行一次 flush把已收到的增量落库。消息状态置为 failed但保留content_delta里已有的部分文本。读取会话时遇到 failed 的 assistant 消息前端可以展示已中断以及已经生成的内容而不是整个消息消失。重试时插入一条新的 assistant 消息而不是覆盖失败的旧消息避免并发重试互相污染。这个策略的本质是不要试图让失败的数据变得完美而是让失败本身可观察、可恢复。用户和管理员都能清楚地看到哪条消息是失败的、失败前生成了什么这比默默吞掉错误好得多。4. 多会话并发隔离、切换与上下文窗口管理4.1 会话隔离按 session_id 做分区多会话的核心矛盾在于一个进程里同时跑着多个用户、多个会话的流式请求数据绝不能串。我见过不少新手项目把所有消息放在一个全局数组里session 用内存里的一个变量标记一换会话就乱了。TinyRobot Kit 的隔离策略简单但有效一切数据操作都以session_id为第一维度。SQL 查询强制带 session_id缓存 key 强制带 session_id内存中的流式写入器实例也按 session_id 分桶存放。class SessionRepository: def list_messages(self, session_id: str, after_seq: int 0, limit: int 50): return self.db.query( SELECT * FROM messages WHERE session_id ? AND seq ? ORDER BY seq LIMIT ?, session_id, after_seq, limit, ) def get_context_messages(self, session_id: str, window_size: int): # 只取该 session 内的消息天然隔离 return self.db.query( SELECT * FROM messages WHERE session_id ? AND status IN (completed, user) ORDER BY seq DESC LIMIT ?, session_id, window_size, )在内存层面每个会话的写入器单独用一个asyncio.Lock保护。用户同时往同一个会话发两条消息的情况锁会保证串行处理避免两条流式响应在同一会话里交错写入。4.2 上下文组装滑动窗口 token 预算模型不是无限上下文组装 prompt 必须有取舍。我的方案是从尾部向前按 token 预算截取。具体算法先固定放 system prompt消耗一部分 token。从会话最新的消息开始倒序往历史走逐条累加 token。累加到剩余预算不足时停止剩下的更早消息被截断。把截取到的消息按正序排列作为上下文送入模型。TOKEN_BUDGET 8000 # 目标模型上下文的一半留余量给本次回复 def assemble_context(system_prompt: str, history, budgetTOKEN_BUDGET): used estimate_tokens(system_prompt) selected [] for msg in reversed(history): cost estimate_tokens(msg.content) 4 # 每条消息的 overhead if used cost budget: break selected.append(msg) used cost selected.reverse() return [{role: system, content: system_prompt}] [ {role: m.role, content: m.content} for m in selected ]几个值得注意的细节budget 不要设满。目标模型的上下文如果是 16k我一般只给历史分配 8k剩下的留给 system prompt 和本次回复的生成空间。消息级截断优于 token 级截断。宁可整体丢弃更早的一轮对话也不要在一轮的中间切断否则模型看到的语义是残缺的。failed 状态的消息进不了上下文。组装之前必须过滤掉没有完成的消息否则模型会被半截文本干扰。4.3 会话切换与懒加载多会话产品里最常见的交互是用户在一个列表里点来点去切换会话。我的做法是消息列表懒加载而不是一次性把所有会话的消息全查出来。会话列表只查sessions表用updated_at DESC排序展示标题和最后消息时间。点击某个会话时才去查该会话最近的消息。再配合加载更多的分页查询初始只取 50 条下拉时再取更早的 50 条。这样即使某个会话聊了几百轮打开列表页的性能也不会被拖垮。懒加载还有一个附带好处内存里的活跃会话数量可控。因为每次只渲染当前会话系统维护的流式写入器、上下文缓存都只跟当前活跃的少数会话有关不是全量。4.4 同会话并发写冲突的处理最容易被忽略的场景是用户在同一个会话里趁上一条回复还没生成完就发了第二条消息。这会在数据层引发两个问题一是上下文组装时读到半成品 assistant 消息二是两条流式写入同时更新同一条 session 记录message_count可能算错。处理方案分两步。第一步组装上下文时只取statuscompleted或roleuser的消息streaming 中的半成品直接跳过。第二步对sessions表的message_count、last_message_at更新改成定稿时重算而不是追加时自增。async def bump_session(self, session_id: str): await self.db.execute( UPDATE sessions SET message_count (SELECT count(*) FROM messages WHERE session_id ?), last_message_at now(), updated_at now() WHERE id ? , session_id, session_id, )有人会觉得 count(*) 慢但一个会话的消息数量撑死在几千条加上session_id上的索引子查询代价非常小。用重算替代自增从根上避免了并发下计数器漂移的问题。5. 踩坑实录数据层设计里我交过的学费5.1 坑一created_at 排序导致消息顺序错乱第一版代码我用created_at给消息排序测试时一切正常。直到有一次给消息表做批量导入导入脚本用的时间戳是当前时间批量写入结果同一会话里的消息顺序全乱了因为一批导入的消息 created_at 几乎相同排序结果不确定。后来我把seq作为排序键created_at只做展示时间。所有新消息的seq都从会话当前最大值加一导入脚本也必须显式指定 seq。这个改动之后再没出现过顺序问题。5.2 坑二流式中断留下的 content_delta 残片某次线上故障排查时发现有大量 assistant 消息的content是空的但content_delta里有一段完整的文本。原因是流式请求在处理过程中被强制 killfinalize 流程没跑content_delta里的内容就永久停留在临时区了。修复方案是双管齐下。一是让读取侧兼容读取消息时如果content为空但content_delta不为空把 delta 当做已生成内容返回。二是加一个定时巡检任务找出状态为 streaming 但超过 10 分钟没有更新的消息把它们强制标记为 failed 并合并残片。这套机制上线后再也没有出现过消息消失了的用户反馈。5.3 坑三token 统计口径不一致上下文组装的时候我用的是字符数除以 4的粗略估算计费模块用的却是模型返回的usage字段。两个口径对不上导致后台统计的 token 消耗和实际账单差了不少。统一口径之后messages.token_count一律以模型返回的 usage 为准assistant 消息、user 消息分别记录上下文组装时的估算只用于做截断判断不进入任何计费逻辑。估算值和精确值的用途分开问题自然消解。5.4 坑四同会话并发请求把上下文搞乱还有一次压测时发现同会话并发发两条消息模型收到的历史里出现了未来的消息。追了半天发现两个请求几乎同时组装上下文第二个请求组装时第一条 assistant 消息还没定稿入库所以上下文里没有它等第二条请求生成完第一条才定稿顺序看起来就像时间线错位了。解决办法就是 4.4 里说的上下文组装只认 seq 小于当前请求的已定稿消息并且同一个会话的请求生成过程加锁串行化。宁可让用户的第二条消息排队等一秒也不能让模型拿到一份逻辑错乱的历史。5.5 踩坑小结问题根因解法消息顺序错乱created_at 排序不稳定引入会话内单调递增 seq 作为排序键流式残片不可见finalize 流程未执行读取侧兼容 巡检任务兜底token 统计偏差估算与计费口径混用估算只用于截断计费以 usage 为准上下文出现未来消息并发请求未串行化同会话加锁 只组装已定稿消息计数器漂移自增更新存在竞态改为定稿时 count(*) 重算6. 走向生产索引、缓存与整个数据层的下一步6.1 索引设计查询模式决定索引数据层跑稳之后我开始盯性能。对话类应用的查询模式其实很固定查会话列表、查某会话的消息分页、查某会话用于组装上下文。针对这三个模式索引设计就三张sessions(user_id, updated_at DESC)支撑会话列表页。messages(session_id, seq)支撑消息分页与上下文取数。messages(session_id, status, seq)支撑过滤未完成消息的上下文组装。不要试图给消息的 content 建全文索引对话检索是另一个专题把它和主链路的数据层混在一起会让索引膨胀失控。生产环境跑下来TinyRobot Kit 在有几千个活跃会话时最频繁的三条查询都在 10ms 以内。6.2 缓存只缓存稳定的数据对话数据的缓存策略我踩过不少弯路之后总结出一条原则缓存只给稳定数据流式数据不进缓存。会话列表可以缓存因为它的变更频率低、读频率高TTL 设为 30 秒用户感觉不到延迟数据库压力小一大截。消息的完整内容不缓存因为消息一旦进入流式阶段每帧都在变缓存刷新跟不上的话只会带来一致性问题。最终实现是只有定稿的 completed 消息才在读取时做一次短 TTL比如 5 分钟的缓存且只缓存消息完整内容不缓存列表页。6.3 演进方向分支会话、长期记忆与事件溯源数据层稳定交付之后TinyRobot Kit 的下一步我有三个方向一是分支会话。messages.parent_id字段已经预留后续可以让用户在某个历史节点上重新开始生成一个不同走向的会话分支。二是长期记忆层。会话之间存在跨会话的稳定信息比如用户偏好、未完成事项。光靠数据库表存聊天记录表达不了这种关系需要单独提炼一个 memory 存储与会话数据解耦。三是事件溯源。目前 messages 表的 status 流转是通过 UPDATE 完成的如果未来要做完整的审计和重放需要把状态变更本身作为事件流存下来。这个改造比较大属于等到真有需求再动的范畴。就个人实践而言TinyRobot Kit 到目前为止最有价值的不是某个单一的亮点而是那一套把对话当数据管理的完整思维每条消息有状态机、每个会话有边界、每次写入有缓冲、每次组装有预算。对话类 AI 应用模型能力决定下限数据层决定上限。把数据层想清楚产品规模再大心里也是踏实的。