AI 聊天数据如何写入 Cassandra：从一次对话到可持久化的消息记录

做 AI 聊天应用时，用户发出的提问、模型生成的回答、工具调用记录和流式输出状态，都需要被可靠保存下来。

如果系统用户量不大，用 MySQL 或 PostgreSQL 保存聊天记录完全没有问题。但当聊天消息量持续增长、写入并发很高、服务还需要多机房高可用时，Apache Cassandra 是一个值得考虑的选择。

这篇文章以“将 AI 聊天数据存入 Cassandra”为例，讲清楚它适合解决什么问题、表应该怎么设计，以及一次消息从产生到落库的基本流程。

一、为什么 AI 聊天应用需要持久化数据？

一条 AI 对话看起来只是用户问一句、模型答一句，但实际往往会产生多类数据：

用户消息，例如“帮我写一封邮件”；
AI 回复，例如模型最终生成的内容；
流式输出片段，例如逐字返回的 Token；
会话元信息，例如标题、创建时间、使用的模型；
工具调用，例如搜索、数据库查询、代码执行；
审计和行为记录，例如点赞、踩、复制、重新生成；
用量数据，例如 Token 消耗、响应时长、调用次数。

这些数据不仅用于展示聊天历史，也用于问题排查、统计分析、模型评估和合规审计。

当系统规模变大后，聊天记录通常有两个特点：

写入很多：每次对话至少写入用户消息和 AI 回复。
查询模式固定：最常见的查询是“读取某个会话的历史消息”。

这正是 Cassandra 擅长的场景。

二、整体数据流长什么样？

一次典型的 AI 对话流程可以理解为：

用户发送问题
   ↓
后端创建用户消息并写入 Cassandra
   ↓
后端调用大模型
   ↓
模型以流式方式返回内容
   ↓
后端将最终回复或分段内容写入 Cassandra
   ↓
前端显示回复；后续可从 Cassandra 恢复历史会话

重点在于：Cassandra 负责保存高频、海量的聊天消息；模型推理仍由大模型服务完成。

三、先明确查询需求，再设计表

使用 Cassandra 时，不要先想着“我要存哪些字段”，而要先想“我要怎么查”。

一个 AI 聊天系统最常见的需求是：

给定 conversation_id，按时间顺序读取该会话的消息。

因此，conversation_id 很适合作为分区键（Partition Key），消息时间或消息 ID 适合作为排序字段。

四、创建 Keyspace 和消息表

先创建一个 Keyspace。开发环境可以使用简单复制策略：

CREATE KEYSPACE ai_chat
WITH replication = {
  'class': 'SimpleStrategy',
  'replication_factor': 1
};

进入 Keyspace：

USE ai_chat;

接着创建聊天消息表：

CREATE TABLE chat_messages (
  conversation_id text,
  message_id timeuuid,
  role text,
  content text,
  model text,
  status text,
  created_at timestamp,
  token_count int,
  PRIMARY KEY (conversation_id, message_id)
) WITH CLUSTERING ORDER BY (message_id ASC);

这张表中：

字段	用途
`conversation_id`	会话 ID，也是分区键
`message_id`	基于时间生成的唯一 ID，用于排序
`role`	消息角色，如 `user`、`assistant`、`system`、`tool`
`content`	消息正文
`model`	产生回复的模型名称
`status`	消息状态，如 `streaming`、`completed`、`failed`
`created_at`	创建时间
`token_count`	Token 使用量，可用于计费或统计

主键 PRIMARY KEY (conversation_id, message_id) 的含义是：

同一个会话的数据聚集在一起；
同一个会话内，消息会按照 message_id 的时间顺序排序；
查询某个会话的历史消息会很快。

五、写入用户消息和 AI 回复

当用户发送消息时，后端可以先写入一条 user 类型的消息：

INSERT INTO chat_messages (
  conversation_id,
  message_id,
  role,
  content,
  status,
  created_at
) VALUES (
  'conv_10001',
  now(),
  'user',
  '帮我解释一下什么是 Cassandra',
  'completed',
  toTimestamp(now())
);

模型生成回复后，再写入一条 assistant 消息：

INSERT INTO chat_messages (
  conversation_id,
  message_id,
  role,
  content,
  model,
  status,
  created_at,
  token_count
) VALUES (
  'conv_10001',
  now(),
  'assistant',
  'Cassandra 是一个分布式 NoSQL 数据库……',
  'gpt-5',
  'completed',
  toTimestamp(now()),
  128
);

读取整个会话的历史记录：

SELECT * FROM chat_messages
WHERE conversation_id = 'conv_10001';

因为查询条件中包含了分区键 conversation_id，这是 Cassandra 推荐的高效查询方式。

六、如何处理流式输出？

AI 聊天通常是流式返回的：模型不是一次性给出完整回答，而是不断推送一小段内容。

这里有两种常见保存方式。

方式一：完成后再保存完整回复

这是最简单、最推荐的入门方案。

流程是：

前端实时收到模型输出；
后端在内存中累积完整内容；
模型结束后，把完整回复写入 Cassandra。

优点是表结构简单、查询方便、写入次数少。

缺点是如果服务在生成过程中异常退出，尚未完成的内容可能丢失。

方式二：分段保存流式内容

如果业务要求断线恢复或保留完整生成过程，可以保存每个内容片段：

CREATE TABLE chat_message_chunks (
  conversation_id text,
  message_id timeuuid,
  chunk_index int,
  content text,
  created_at timestamp,
  PRIMARY KEY ((conversation_id, message_id), chunk_index)
) WITH CLUSTERING ORDER BY (chunk_index ASC);

每生成一段内容，就写入一条 chunk：

INSERT INTO chat_message_chunks (
  conversation_id,
  message_id,
  chunk_index,
  content,
  created_at
) VALUES (
  'conv_10001',
  8f52b8b0-0000-11f0-8000-000000000001,
  1,
  'Cassandra 是',
  toTimestamp(now())
);

恢复消息时，按 chunk_index 读取并拼接即可。

不过不要为了每一个 Token 都立刻写数据库。更实用的做法是按几十到几百个字符、或按固定时间间隔进行批量写入，避免产生过多微小写操作。

七、会话元信息应该放在哪里？

除了消息内容，系统通常还需要保存会话标题、所属用户、创建时间和最后活跃时间。

可以单独创建一张会话表：

CREATE TABLE conversations_by_user (
  user_id text,
  updated_at timestamp,
  conversation_id text,
  title text,
  model text,
  created_at timestamp,
  PRIMARY KEY (user_id, updated_at, conversation_id)
) WITH CLUSTERING ORDER BY (updated_at DESC);

这样就可以支持另一个常见查询：

查询某个用户最近的会话列表。

查询示例：

SELECT * FROM conversations_by_user
WHERE user_id = 'user_001'
LIMIT 20;

注意：这张表是为“按用户读取会话列表”设计的；消息表则是为“按会话读取消息历史”设计的。它们虽然可能保存部分重复信息，但这种冗余在 Cassandra 中是正常且常见的。

八、一个容易踩坑的问题：超大分区

如果一个热门会话永远只用同一个 conversation_id 作为分区键，长期来看可能产生一个非常大的分区。

对于普通聊天产品，这通常不是最先出现的问题；但对于群聊、客服会话、长时间 AI Agent 任务，需要提前规划。

一种常见办法是增加时间桶：

PRIMARY KEY ((conversation_id, month), message_id)

例如：

conversation_id = conv_10001
month = 2026-09

这样，同一个会话每个月的数据会放到不同分区中，避免单个分区无限膨胀。

读取历史消息时，根据月份依次查询并合并结果即可。

九、Cassandra 在 AI 应用中的位置

一个成熟的 AI 聊天系统通常不会只使用 Cassandra：

数据类型	更适合的存储
用户、权限、订阅、支付	PostgreSQL 或 MySQL
聊天消息、事件记录、访问日志	Cassandra
图片、文件、音频	对象存储
知识库向量与相似度检索	向量数据库
缓存、会话短状态、限流	Redis

也就是说，Cassandra 很适合作为 AI 聊天系统中的“消息历史与事件数据层”，但不需要承担所有数据存储职责。

十、总结

Cassandra 可以帮助 AI 聊天系统稳定保存大量对话数据，尤其适合以下模式：

高并发写入聊天消息；
按会话读取消息历史；
保存模型调用事件和工具调用日志；
通过多节点部署提高可用性；
随着用户增长，增加服务器进行横向扩容。

入门时，最重要的原则只有一个：

先确定查询方式，再为每种查询方式设计表。

如果你的核心需求是“按会话查看聊天记录”和“按用户查看最近会话”，Cassandra 可以成为一个非常合适的持久化方案。