企业内部 AI 聊天应用的聊天记录怎么存: 一张 DynamoDB 表, 不建索引¶
🌐 AI ReaderRead this document in English or ask questions: Open in Claude, ChatGPT, or Gemini ↗
ChatGPT 出来以后, 很多企业都想做一个自己内部的聊天工具. 说白了就是数据要隐私, 员工问的问题, 贴进去的内容, 不想离开公司自己的环境. 聊天这一块其实不难做, 调一下模型 API, 把回答流式吐给前端就完事了. 真正麻烦的是后面那件事: 每个用户登录以后的聊天数据存在哪, 怎么保证张三看不到李四的记录, 侧边栏那一串历史会话怎么按最近时间排好. 这个需求是实打实的, 看起来容易, 要做好其实没那么容易.
我的方案很轻: 一张 DynamoDB 表, 只用原生的 partition key (下文简称 pk) 和 sort key (sk), 不建任何 index. 每个会话存两条记录, 一条负责 "按 id 找到它", 一条负责 "按最近时间排序", 用户每发一条消息, 用一个事务把这几条记录一起改掉. 这篇讲清楚 key 怎么编码, 为什么这么编, 读写各要几次请求, 以及一个月 100 万轮对话数据库花多少钱.
要存的东西只有三样: 用户, 会话, 消息¶
先把范围划死. 这篇只讲 chat: 不讲 agent, 不讲文件上传和产出文件, 也不讲 ChatGPT 和 Claude.ai 里的 Project 那种把会话归组的功能. 剩下的实体只有三个.
用户 (user). 多租户是硬要求, 但不需要单独建一张用户表: 用户登录以后, 认证系统自然会给出一个 user_id, 数据按它隔离就够了.
会话 (session), 就是侧边栏里的一条聊天 thread. 字段:
字段 |
说明 |
|---|---|
|
会话 id |
|
会话属于谁 |
|
会话标题 |
|
创建时间 |
|
最近一条消息的时间, 侧边栏按它排序 |
|
消息条数 |
消息 (message), 是会话里一来一回的一轮: 用户问一句, AI 答一段, 存成同一条记录. 字段:
字段 |
说明 |
|---|---|
|
消息 id, 同时决定消息的先后顺序 |
|
属于哪个会话 |
|
用户发出问题的时间 |
|
AI 开始返回的时间 |
|
AI 返回结束的时间 |
|
用户的输入, 纯文本 |
|
AI 的回答, 纯文本 |
|
输入的 token 数 |
|
输出的 token 数 |
|
这一轮用的 system prompt 模板 |
|
AI 有没有返回: |
status 是这次新加的. 问题和回答放在同一条记录里, 但它们不是同时产生的: 用户一按发送, 问题就该落盘, 而回答要等几秒甚至几十秒. 有了 status, 就能先把问题存下来, AI 回完再补全, 中途失败了也看得出来.
还有一条规矩: 不给用户删除聊天记录的按钮. 删除是后台管理员的事, 不在这套读写流程里. 这一条省掉的麻烦比看上去多, 后面会说到.
两个查询决定了表怎么设计¶
DynamoDB 设计表的套路是先列查询, 再定 key. 一个聊天应用的界面, 真正高频的读只有两个:
打开应用, 侧边栏列出 "我的会话", 按最近一条消息的时间倒序, 一页 20 条, 往下滚再加载.
点开一个会话, 显示标题和最近的几十轮对话, 往上滚再加载更早的.
这两个查询有个共同点: 都是 "在某个范围里, 按某个顺序取一页". 这正好就是 DynamoDB 的 Query 干的事: 指定一个 pk, 按 sk 的字典序正着或倒着取, 带分页. 没有 join, 没有多条件筛选, 也没有跨用户的统计. 查询方式和 DynamoDB 的模型对得上, 这是我选它的根本原因. 按请求付费, 不用管服务器, 延迟稳定, 这些是附带的好处.
难点在侧边栏: 会话既要按 id 找, 又要按最近时间排¶
消息那个查询好办: 一个会话的消息放在同一个 pk 下, sk 用一个按时间递增的 id, 倒着 Query 就是最近的消息.
难的是会话. 侧边栏要按 last_message_time 排序, 那 last_message_time 就得在 sk 里. 可是一旦它在 sk 里, 你只拿着一个 session_id (比如用户从链接 /chat/<session_id> 直接打开) 就找不到这条记录了, 因为 GetItem 要完整的 sk, 而你不知道这个会话最后一次说话是什么时候.
反过来, sk 里先放 session_id 再放时间, 也就是 SES#<session_id>#<time>, 按 id 找是能找了, 可 Query 出来是按 session_id 的字典序排的, 时间只在 id 相同的时候才起作用, 侧边栏的顺序就乱了. 我原来那版设计就是这么写的, 这是个错.
把两种写法摆在一起看:
sk 的写法 |
只知道 |
Query 出来按最近时间排吗 |
|---|---|---|
|
能 |
不能, 按 id 排 |
|
不能, 不知道时间 |
不能, 还是按 id 排 |
|
不能, 不知道时间 |
能 |
没有一行两个都是 "能".
再加上 DynamoDB 的 sk 一旦写下就不能改, "更新最近时间" 只能删掉旧记录再写一条新的. 一条记录, 两种查法, 一个还天天在变, 怎么编都顾此失彼. 结论很直接: 别让一条记录干两份活.
每个会话存两条记录: 一条负责找, 一条负责排¶
整张表只有三种记录, 两种 pk:
记录 |
pk |
sk |
用来 |
|---|---|---|---|
会话列表项 |
|
|
侧边栏按最近时间排序 |
会话本体 |
|
|
按 id 找会话, 存当前的 |
消息 |
|
|
一个会话里的消息, 按时间排序 |
把它和界面对上就清楚了. 左边是用户看到的侧边栏和聊天窗口, 右边是表里的记录, 同一个编号的颜色是同一种记录:
侧边栏整个来自一个 pk 的一次 Query, 聊天窗口整个来自另一个 pk 的一次 Query, 两边互不打扰.
三种记录各管一件事¶
会话列表项是 "负责排" 的那条. 一个用户的所有列表项都在 USER#<user_id> 这个 pk 下, sk 以时间开头, 倒着 Query 就是侧边栏. 它身上冗余存一份 title, 这样侧边栏不用再去查会话本体. 时间一变, 它就删掉重写.
会话本体是 "负责找" 的那条. 它的 pk 只由 user_id 和 session_id 拼成, 拿着 id 就能直接定位. 它存着当前的 last_message_time, 这个值就是会话列表项 sk 里的那个时间: 想改列表项, 先从这里知道旧的 sk 是什么.
消息和会话本体放在同一个 pk 下. sk 选 SES#META 而不是更直观的 META, 是故意的: 按字典序 SES#META 排在所有 MSG#... 后面, 倒着 Query 时它排第一个. 于是打开一个会话只要一次 Query, 第一条是会话本体, 后面紧跟着最近的 20 轮对话.
时间和 id 都编码成能按字典序排的字符串¶
DynamoDB 的 sk 是按字符串的字典序排的, 所以 "能排序" 的前提是: 字符串的先后顺序和时间的先后顺序一致.
时间用 ISO 8601 的 UTC 毫秒格式, 比如 2026-11-06T08:30:45.123Z. 长度固定, 字典序和时间顺序一致, 放进 sk 里直接能排.
session_id 和 message_id 用 ULID 而不是 UUID. ULID 是 26 个字符, 前 10 个字符是毫秒时间戳, 后 16 个是随机数, 按字典序排就是按生成时间排. 所以 MSG#<message_id> 天然有序, 不用再单独存一个时间戳进 sk; 两条消息落在同一毫秒, 随机部分也会把它们分开, 不会撞. 原来担心的 "同一毫秒来两条消息, 排序不确定" 的问题, 换成 ULID 就没了.
多租户隔离靠 pk 前缀¶
所有 pk 都以 USER#<user_id> 开头, 这就是多租户隔离. 规矩只有一条: user_id 只从登录 token 里取, 永远不接受前端传过来的值. 后端拼 key 的时候, 张三的请求根本拼不出李四的 pk, 想越权读也读不到. 这是应用层的校验, 对内部工具来说够用了, 不用去搞 IAM 的 dynamodb:LeadingKeys 那一套. id 前面挂上 user_id, 顺带也让 id 冲突的可能性小到可以不管.
一个会话在表里的样子¶
下面就是上图里 "DynamoDB 表怎么设计" 这个会话的三种记录, 消息只列了一条:
# 会话列表项
{
"pk": "USER#u-123",
"sk": "SES#2026-11-06T08:35:22.456Z#01JC3Q8Z4F6M2V9K7T5R1N0B8D",
"session_id": "01JC3Q8Z4F6M2V9K7T5R1N0B8D",
"title": "DynamoDB 表怎么设计",
}
# 会话本体
{
"pk": "USER#u-123#SES#01JC3Q8Z4F6M2V9K7T5R1N0B8D",
"sk": "SES#META",
"session_id": "01JC3Q8Z4F6M2V9K7T5R1N0B8D",
"user_id": "u-123",
"title": "DynamoDB 表怎么设计",
"create_time": "2026-11-06T08:30:45.123Z",
"last_message_time": "2026-11-06T08:35:22.456Z",
"message_count": 2,
}
# 消息
{
"pk": "USER#u-123#SES#01JC3Q8Z4F6M2V9K7T5R1N0B8D",
"sk": "MSG#01JC3QHB2W8X5Y7Z9A1C3E5G7J",
"message_id": "01JC3QHB2W8X5Y7Z9A1C3E5G7J",
"session_id": "01JC3Q8Z4F6M2V9K7T5R1N0B8D",
"create_time": "2026-11-06T08:35:22.456Z",
"response_start_time": "2026-11-06T08:35:23.001Z",
"response_end_time": "2026-11-06T08:35:31.870Z",
"input_body": "聊天记录用 DynamoDB 怎么存?",
"output_body": "可以按用户和会话来设计 key ...",
"input_token": 12,
"output_token": 480,
"system_prompt_id": "default-v1",
"status": "done",
}
一轮对话写两次: 先占位, AI 回完再补全¶
用户按下发送的那一刻, 要同时做四件事: 写入这条消息, 更新会话本体的 last_message_time, 删掉旧的会话列表项, 写一个新的会话列表项. 这四件事必须一起成功或一起失败, 否则就会出现消息存了但侧边栏没动, 或者侧边栏里同一个会话出现两次. 所以用 TransactWriteItems 放进一个事务.
这时候 AI 还没回答, 消息先以 status=pending 写进去, 只有问题没有答案. 等 AI 回完, 再对这条消息单独做一次 Update, 补上回答, token 数和时间, status 改成 done; 调用模型出错就改成 failed. 第二次只动一条记录, 不需要事务.
一轮对话里, 用户, 后端, DynamoDB 和模型四方之间是这样来回的:
sequenceDiagram
autonumber
actor U as 用户
participant B as 后端
participant D as DynamoDB
participant M as 模型
U->>B: 发送问题
B->>D: 四项事务
D-->>B: 写入成功
Note over B,D: 问题先落盘, 再调模型
B->>M: 调用模型
M-->>B: 流式回答
B-->>U: 转发回答
B->>D: 补全消息
D-->>B: 更新成功
第 2 步. 事务成功之后才去调模型. 模型调用失败, 用户关掉页面, 后端进程挂掉, 问题都已经在表里了, 用户刷新页面能看到自己问过什么.
第 7 步. 只有这一步写回答. 它要是没发生, 消息就一直停在
pending, 后台可以把超时的pending当成失败处理.
这张图要说明的是: 一轮对话只碰数据库两次, 两次都在模型调用的外面, 模型那边慢也好, 出错也好, 都不会让数据库里留下一半的状态.
先把公共的部分准备好. 用 boto3 的低级 client, 配上 TypeSerializer 把普通的 Python 值转成 DynamoDB 的类型格式; ULID 用 python-ulid 这个包:
import datetime
import boto3
from boto3.dynamodb.types import TypeSerializer
from ulid import ULID
client = boto3.client("dynamodb")
TABLE = "chat"
_ser = TypeSerializer()
def ser(d: dict) -> dict:
"""把普通的 Python dict 转成 DynamoDB 的类型格式."""
return {k: _ser.serialize(v) for k, v in d.items()}
def now() -> str:
"""UTC 毫秒精度的 ISO 8601 时间, 比如 2026-11-06T08:30:45.123Z."""
dt = datetime.datetime.now(datetime.timezone.utc)
return dt.isoformat(timespec="milliseconds").replace("+00:00", "Z")
def list_key(user_id: str, last_message_time: str, session_id: str) -> dict:
return {"pk": f"USER#{user_id}", "sk": f"SES#{last_message_time}#{session_id}"}
def meta_key(user_id: str, session_id: str) -> dict:
return {"pk": f"USER#{user_id}#SES#{session_id}", "sk": "SES#META"}
def msg_key(user_id: str, session_id: str, message_id: str) -> dict:
return {"pk": f"USER#{user_id}#SES#{session_id}", "sk": f"MSG#{message_id}"}
在已有的会话里发消息¶
这是最常见的写. 下图是用户在 "周报怎么写" 这个会话里发一条新消息, 事务里的 4 个操作分别改了哪条记录:
旧的 last_message_time 从哪来? 用户打开这个会话时, 那次 Query 已经把会话本体读出来了, 前端或后端手上本来就有这个值, 不用额外再读一次. 代码就是把上图的 4 个操作原样写出来:
def send_message(
user_id: str,
session_id: str,
old_last_message_time: str,
title: str,
input_body: str,
system_prompt_id: str,
) -> tuple[str, str]:
"""写入一条 pending 消息, 并把会话挪到侧边栏最上面. 返回 (message_id, 新时间)."""
message_id = str(ULID())
t = now()
client.transact_write_items(
TransactItems=[
{ # 1. 写入消息, 先只有问题
"Put": {
"TableName": TABLE,
"Item": ser({
**msg_key(user_id, session_id, message_id),
"message_id": message_id,
"session_id": session_id,
"create_time": t,
"input_body": input_body,
"system_prompt_id": system_prompt_id,
"status": "pending",
}),
"ConditionExpression": "attribute_not_exists(pk)",
}
},
{ # 2. 更新会话本体的时间, 前提是它还是我读到的那个旧值
"Update": {
"TableName": TABLE,
"Key": ser(meta_key(user_id, session_id)),
"UpdateExpression": "SET last_message_time = :new ADD message_count :one",
"ConditionExpression": "last_message_time = :old",
"ExpressionAttributeValues": ser({
":new": t,
":old": old_last_message_time,
":one": 1,
}),
}
},
{ # 3. 删掉旧的会话列表项
"Delete": {
"TableName": TABLE,
"Key": ser(list_key(user_id, old_last_message_time, session_id)),
"ConditionExpression": "attribute_exists(pk)",
}
},
{ # 4. 写入新的会话列表项
"Put": {
"TableName": TABLE,
"Item": ser({
**list_key(user_id, t, session_id),
"session_id": session_id,
"title": title,
}),
}
},
]
)
return message_id, t
第 2 步的条件表达式是这一段的关键. 同一个用户在两个浏览器窗口里同时往一个会话里发消息, 两个请求手上拿的是同一个旧时间. 先到的那个事务成功, 后到的那个在第 2 步条件不成立, 整个事务被取消, 抛出 TransactionCanceledException. 这时候重新读一次会话本体, 拿到最新的 last_message_time, 再发一次就行. 没有这个条件, 后到的那个会去删一条已经不存在的列表项, 然后再写一条, 侧边栏里就多出一个重复的会话.
AI 回完以后补全消息¶
def finish_message(
user_id: str,
session_id: str,
message_id: str,
output_body: str,
input_token: int,
output_token: int,
response_start_time: str,
response_end_time: str,
):
client.update_item(
TableName=TABLE,
Key=ser(msg_key(user_id, session_id, message_id)),
UpdateExpression=(
"SET output_body = :out, input_token = :it, output_token = :ot, "
"response_start_time = :rs, response_end_time = :re, #s = :done"
),
ConditionExpression="#s = :pending",
ExpressionAttributeNames={"#s": "status"}, # status 是 DynamoDB 的保留字
ExpressionAttributeValues=ser({
":out": output_body,
":it": input_token,
":ot": output_token,
":rs": response_start_time,
":re": response_end_time,
":done": "done",
":pending": "pending",
}),
)
调用模型失败时, 同样的写法把 status 改成 failed, 前端据此显示一个 "重新生成" 按钮. 补全这一步不碰会话的 last_message_time: 侧边栏的顺序按用户什么时候说话来排, 不按 AI 什么时候说完, 这样一轮对话只挪一次侧边栏.
新建会话和改标题¶
新建会话是三个 Put 放进一个事务: 会话本体 (message_count 为 1), 会话列表项, 第一条 pending 消息. 会话本体加上 attribute_not_exists(pk) 条件, 防止重复创建. 标题一开始可以先用用户问题的前几十个字, 等第一轮回答完, 再让模型起一个正经标题.
改标题要改两条记录: 会话本体和会话列表项上的 title, 都只是 Update 一个普通属性, 不涉及 sk, 放进一个两项的事务里. 会话本体同样带上 last_message_time = :old 的条件, 保证改的列表项就是当前那一条.
不给用户删除, 在这里又省了一笔. 要是能删, 删一个会话就得把它下面几百条消息一条条删掉, 还得处理 "删到一半又来了新消息" 的情况. 现在这些都不用管, 后台要清理的时候, 按 pk Query 出来批量删就是了.
读只需要两个请求¶
侧边栏: 对 USER#<user_id> 倒序 Query, 一页 20 条. 返回结果里有 LastEvaluatedKey 就说明还有下一页, 把它当成下一次请求的 ExclusiveStartKey 传回去.
def list_sessions(user_id: str, start_key: dict | None = None) -> dict:
kwargs = dict(
TableName=TABLE,
KeyConditionExpression="pk = :pk AND begins_with(sk, :prefix)",
ExpressionAttributeValues=ser({":pk": f"USER#{user_id}", ":prefix": "SES#"}),
ScanIndexForward=False, # 倒序, 最近的在前
Limit=20,
ConsistentRead=True,
)
if start_key:
kwargs["ExclusiveStartKey"] = start_key
return client.query(**kwargs)
打开一个会话: 对 USER#<user_id>#SES#<session_id> 倒序 Query, Limit=21. 第一条是 SES#META, 也就是会话本体, 后面 20 条是最近的 20 轮对话. 拿到的消息是从新到旧的, 前端显示前反转一下. 往上翻更早的消息, 同样用 LastEvaluatedKey 翻页, 后面的页里不会再出现会话本体.
def open_session(user_id: str, session_id: str, start_key: dict | None = None) -> dict:
kwargs = dict(
TableName=TABLE,
KeyConditionExpression="pk = :pk",
ExpressionAttributeValues=ser({":pk": f"USER#{user_id}#SES#{session_id}"}),
ScanIndexForward=False,
Limit=21,
ConsistentRead=True,
)
if start_key:
kwargs["ExclusiveStartKey"] = start_key
kwargs["Limit"] = 20
return client.query(**kwargs)
两个查询都用了 ConsistentRead=True. 强一致读的费用是默认读的两倍, 但聊天应用里读本来就便宜, 换来的是用户刚发完消息, 一刷新侧边栏, 这个会话一定在最上面.
为什么不用 GSI, LSI 和 Streams¶
最常见的写法是: 会话只存一条记录, sk 固定为 session_id, last_message_time 当成普通属性原地 Update, 再建一个 GSI 按 last_message_time 排序. 写起来确实省事, 但 GSI 是异步复制的, 只支持最终一致读. 通常延迟很短, 可官方不保证, 用户发完消息看到侧边栏没动, 这种 bug 很难复现也很难解释. GSI 还是另一份要付钱的存储和写入, 而它干的活, 上面那条会话列表项已经干了.
LSI 支持强一致读, 看起来正好合适, 但它只能在建表时定义, 以后想改就只能重建表; 而且用了 LSI, 同一个 pk 下所有数据加起来不能超过 10 GB. 为一个用两条记录就能解决的问题, 背上一个建表时就写死的约束, 不划算.
DynamoDB Streams 加 Lambda 异步更新会话时间也能做, 但它是事后补的: 用户连发两条消息, 两次更新的先后顺序不一定对, 出错了还要处理重试. 写的时候一个事务就能把事情做完, 没必要把它拆成异步的.
说到底, 这个场景只需要按一个维度排序, key 编码得合理, 主表的 pk 和 sk 就够用了. 哪天要按标题搜, 按 token 用量排, 那时候再加 GSI 也不迟.
100 万轮对话, 数据库每月十来美元¶
按 2026-11 的 on-demand 价格 (us-east-1, Standard 表) 算: 写 \(0.625 / 百万 WRU, 读 \)0.125 / 百万 RRU, 存储 $0.25 / GB-月. 有两条计费规则对这套设计影响最大: 事务写每个 item 每 KB 收 2 个 WRU, 是普通写的两倍; 强一致读每 4 KB 收 1 个 RRU, 也是默认读的两倍. 价格以 AWS 官方定价页 为准.
假设一轮对话的消息记录在 AI 回完后是 4 KB, 会话本体和列表项都在 1 KB 以内:
项 |
每轮 |
100 万轮 / 月 |
|---|---|---|
发送事务: 4 个 item, 每个 2 WRU |
8 WRU |
$5.00 |
补全消息: 普通写, 按 4 KB 算 |
4 WRU |
$2.50 |
打开会话: 假设 100 万次, 每次读 META 加 20 轮约 80 KB |
20 RRU |
$2.50 |
侧边栏: 假设 200 万次, 每次 20 个列表项不到 4 KB |
1 RRU |
$0.25 |
存储: 100 万条消息约 4 GB |
$1.00 |
|
合计 |
约 $11 |
这笔账要和模型的钱放在一起看. 哪怕每轮对话的模型调用只花 1 美分, 100 万轮就是 1 万美元, 数据库的钱连它的零头都算不上. 所以这套设计里, 为了强一致和代码简单多花的那些 WRU, 我一点都不心疼. 真正要省钱, 得去省 token.
这套方案不管什么¶
这套设计只管一件事: 多用户的聊天记录, 按会话存, 按时间读, 侧边栏按最近排序. 它刻意不管的东西有这几样.
全文搜索. "在我所有会话里搜某个关键词", DynamoDB 做不了. 需要的话, 把消息同步到 OpenSearch 之类的搜索引擎里去, 那是另一套系统.
超长的消息. DynamoDB 单个 item 最大 400 KB, 一问一答存一条, 普通聊天远远到不了. 如果用户会贴进来整本文档, 或者回答特别长, 就把正文放到 S3, item 里只存 key.
删除. 用户界面上不给删, 后台的清理和数据保留策略另外做. 想按时间自动清理, 可以给消息加 TTL 属性.
Project, agent, 文件. 不在这篇的范围里. Project 这种 "按组排序" 的需求, 可以照着会话列表项的思路, 再加一层 "项目列表项", 道理是一样的.
如果你的需求就是 "一个公司内部的 ChatGPT, 每个人看自己的聊天记录", 这一张表就够了: 不建 index, 不上 Streams, 每发一条消息一个事务加一次 Update, 侧边栏和会话各一次 Query. 什么时候该换方案? 当你开始需要按好几个不同的维度排序或筛选会话的时候, 那就是加 GSI 的时候了.