Grok Bot 成本优化：用 durable state 持久化状态降低 Token 消耗

最近一段时间，做 Grok Bot 相关的开发者越来越多。很多人花大量精力调 prompt、选模型、搭插件，结果一看账单，发现钱烧得比想象中快得多。问题往往不是模型本身太贵，而是我们的 Bot 在反复给模型提交完全相同的上下文。

如果你正在做这类聊天机器人、Agent 或自动化助手，并且已经开始被 token 消耗困扰，那么这篇文章值得看完。本文要聊的核心思路，是用 durable state（持久化状态）来减少 Grok Bot 的无效消耗，让每一次请求都尽量只携带“增量”和“必要摘要”，而不是把整个历史从头到尾重放一遍。

文章不会只停留在概念层面，我会给出一个可运行的最小工程示例，基于 SQLite 持久化状态，实现上下文摘要、状态恢复和 token 用量估算。你可以直接复制到自己的项目里去改造。

1. 先搞清楚：Grok Bot 的消耗到底烧在哪里

很多人对 LLM 对话成本的认知，停留在“输出 token 贵、输入 token 便宜”的层面。表面上没错，但实际跑一个 Bot 应用之后，你会发现消耗的大头往往不是输出，而是输入端的上下文重复。

原因很直接：大语言模型本身是无状态的。每次调用接口，模型都只看到你这一次请求里携带的 messages 数组。它不会记得上一次请求说过什么，也不会自动维护会话历史。所以一个对话 Bot 要在多轮对话里保持连贯，开发者就必须把历史消息重新传给模型。

我们来算一笔账。假设你的 Bot 每轮对话平均要携带 20 条历史消息，每条消息平均 300 token，那么在用户发起第 21 轮请求时，历史部分就有 6000 token。如果这个 Bot 每小时收到 1000 个请求，一小时仅历史输入就有 600 万 token。而这里面大部分内容，和当前请求真正相关可能只有几百 token。

再叠加一个容易忽略的问题：如果 Bot 有多个会话、多个用户，或者部署在多实例上，状态不在请求之间共享，那同样的历史会在每个请求里重复计算。更严重的是，很多 Bot 应用在进程重启后会丢失会话记忆，导致下一轮请求要么只能从空上下文开始，要么需要重新从数据库加载全量日志，再完整塞给模型。

所以，降低 Grok Bot 消耗的关键，不是在 prompt 里抠几个字，而是要把“无状态请求”改造成“有状态应用”。这就是 durable state 的价值所在。

2. durable state 到底是什么

durable state，直译是“持久化状态”。它的含义是：把应用运行过程中需要跨请求、跨会话、甚至跨重启保留的状态数据，写入可靠存储，而不是放在内存里随请求结束而消失。

在 Grok Bot 场景里，最常见的 durable state 包括：

用一句话概括：没有 durable state，Bot 是“金鱼记忆”，每次请求都要从零开始理解世界；有了 durable state，Bot 才能像正常人一样，把已经说过的话、做过的事记住，只把“新变化”如实同步给模型。

这里要区分一个概念：很多开发者在内存里维护一个messages列表，也把它叫“状态”。那是 ephemeral state（临时状态），进程一结束就没了。durable state 强调的是“跨生命周期存活”，所以它的核心是存储层。

为什么 durable state 能降低消耗？关键不在于存储本身，而在于它给了我们一个机会：可以对历史进行裁剪、压缩、筛选，而不是每次请求都原样重放。

如果没有 durable state，你只有两种选择：要么不传历史，Bot 失忆；要么传全部历史，token 爆炸。有了 durable state，你可以做到第三种选择：传“必要的记忆”。这第三种选择，才是省钱的核心。

3. 三种减少 Grok Bot 消耗的通用手段

在动手写代码之前，有必要先把方案想清楚。基于 durable state，减少消耗可以拆成三层手段。

3.1 持久化会话状态，避免重复重放

这是最基础的一层。把每一轮用户消息和 Bot 回复都写入数据库。下次请求时，不是从内存列表里拼上下文，而是从数据库加载最近 N 条消息。

这一步并没有直接减少总 token，但它让“上下文裁剪”成为可能。其次，它能避免因为进程重启导致 Bot 忘记之前的对话，从而被迫用更长的系统提示词去弥补上下文缺失。

3.2 上下文摘要与压缩，删掉低价值历史

当对话越来越长，最近 N 条消息可能仍然不够用，早期的重要信息必须保留。常见做法是：定期对历史消息生成摘要，然后把摘要作为系统提示的一部分传给模型。

例如，每累计 20 轮对话，调用一次 Grok 模型，把当前摘要 + 新增的 20 轮消息汇总成一份新的摘要，存回数据库。之后构建上下文时，不再加载全部历史，而是加载：

• 旧的全局摘要；
• 最近 10 轮原始消息；
• 当前用户问题。

这样，无论 Bot 运行多少轮，输入 token 都不会无限增长，而是稳定在一个可控范围内。

3.3 请求缓存与幂等设计，避免重复计算

有些 Bot 在回答用户问题时，会调用工具、查询数据或执行代码。如果用户连续问同一个问题，或者多个用户问了完全相同的问题，而答案又不需要实时变化，就可以考虑在 durable state 中保存计算结果。

这里的持久化状态可以是一张简单的缓存表，key 是请求内容的哈希，value 是历史输出。命中缓存时直接返回，不再调用模型。这种方式对查询型 Bot 特别有效。

4. 环境准备与项目结构

下面进入实战环节。我们会用 Python 3 写一个最小示例，存储层用 SQLite，模型接口用兼容 OpenAI SDK 的通用服务端点，方便你替换成 Grok 或其他兼容服务。

如果你用的是 Grok 官方提供的接口或第三方兼容网关，只需要替换base_url、api_key和model即可。本文示例不依赖具体的 Grok SDK 细节，因为不同接入方式的参数差异较大，但整体逻辑完全通用。

4.1 依赖清单

建议使用 Python 3.9 及以上版本。需要安装的依赖如下：

pip install openai

openai 这个 Python 包目前是社区事实上的标准客户端，很多模型服务都支持 OpenAI 兼容接口。如果不想引入 SDK，也可以用requests直接发 HTTP 请求，但用 SDK 代码会更简洁、更好维护。

4.2 项目结构

本文示例的目录结构如下：

grok_bot_durable_state/ ├── bot.py ├── state_manager.py ├── bot.db └── .env

其中：

• state_manager.py：持久化状态管理器，负责写入和读取消息、摘要、缓存。
• bot.py：Bot 主逻辑，负责加载上下文、调用模型、返回回复。
• bot.db：SQLite 数据库文件，运行后自动生成。
• .env：存放 API Key 和模型配置。

5. 核心代码实现：用 durable state 降低消耗

下面分三步实现。

5.1 初始化数据表

持久化状态的第一步是设计表结构。我们至少需要两张表：

• messages：保存每轮用户消息和 Bot 回复。
• session_state：保存会话摘要、缓存结果。

SQL 建表语句如下，可以直接在 SQLite 中执行：

CREATE TABLE IF NOT EXISTS messages ( id INTEGER PRIMARY KEY AUTOINCREMENT, session_id TEXT NOT NULL, role TEXT NOT NULL, content TEXT NOT NULL, created_at TEXT NOT NULL DEFAULT (datetime('now')) ); CREATE INDEX IF NOT EXISTS idx_messages_session_id ON messages(session_id, id); CREATE TABLE IF NOT EXISTS session_state ( session_id TEXT PRIMARY KEY, summary TEXT, last_compressed_msg_id INTEGER DEFAULT 0, updated_at TEXT NOT NULL DEFAULT (datetime('now')) );

设计逻辑很简单：messages表保存所有原始消息，用来追溯和压缩；session_state表保存该会话的最新摘要，以及上一次压缩到哪一条消息，避免每次压缩都从头扫描。

5.2 实现 DurableStateManager

下面编写state_manager.py，这个类负责所有持久化读写操作。

# 文件路径：state_manager.py import json import sqlite3 from typing import Optional class DurableStateManager: """基于 SQLite 的持久化状态管理器""" def __init__(self, db_path: str = "bot.db"): self.db_path = db_path self._init_tables() def _connect(self) -> sqlite3.Connection: conn = sqlite3.connect(self.db_path) conn.row_factory = sqlite3.Row return conn def _init_tables(self) -> None: with self._connect() as conn: conn.execute( """ CREATE TABLE IF NOT EXISTS messages ( id INTEGER PRIMARY KEY AUTOINCREMENT, session_id TEXT NOT NULL, role TEXT NOT NULL, content TEXT NOT NULL, created_at TEXT NOT NULL DEFAULT (datetime('now')) ) """ ) conn.execute( """ CREATE INDEX IF NOT EXISTS idx_messages_session_id ON messages(session_id, id) """ ) conn.execute( """ CREATE TABLE IF NOT EXISTS session_state ( session_id TEXT PRIMARY KEY, summary TEXT, last_compressed_msg_id INTEGER DEFAULT 0, updated_at TEXT NOT NULL DEFAULT (datetime('now')) ) """ ) def save_message(self, session_id: str, role: str, content: str) -> int: """保存一条消息，返回自增 ID""" with self._connect() as conn: cur = conn.execute( "INSERT INTO messages (session_id, role, content) VALUES (?, ?, ?)", (session_id, role, content), ) return cur.lastrowid def load_recent_messages( self, session_id: str, limit: int = 10, after_msg_id: int = 0 ) -> list[dict]: """加载某个会话的最近消息，按时间正序返回""" with self._connect() as conn: rows = conn.execute( """ SELECT id, role, content FROM messages WHERE session_id = ? AND id > ? ORDER BY id DESC LIMIT ? """, (session_id, after_msg_id, limit), ).fetchall() rows.reverse() return [dict(row) for row in rows] def load_message_count(self, session_id: str) -> int: """统计某个会话的消息总量""" with self._connect() as conn: row = conn.execute( "SELECT COUNT(*) AS cnt FROM messages WHERE session_id = ?", (session_id,), ).fetchone() return row["cnt"] if row else 0 def load_summary(self, session_id: str) -> tuple[Optional[str], int]: """读取会话摘要，返回 (摘要文本, 上次压缩到的消息 ID)""" with self._connect() as conn: row = conn.execute( "SELECT summary, last_compressed_msg_id FROM session_state WHERE session_id = ?", (session_id,), ).fetchone() if row is None: return None, 0 return row["summary"], row["last_compressed_msg_id"] def save_summary(self, session_id: str, summary: str, last_msg_id: int) -> None: """保存会话摘要和压缩进度""" with self._connect() as conn: conn.execute( """ INSERT INTO session_state (session_id, summary, last_compressed_msg_id, updated_at) VALUES (?, ?, ?, datetime('now')) ON CONFLICT(session_id) DO UPDATE SET summary = excluded.summary, last_compressed_msg_id = excluded.last_compressed_msg_id, updated_at = datetime('now') """, (session_id, summary, last_msg_id), ) def save_cache(self, key: str, value: str) -> None: """保存请求缓存，key 是请求参数的哈希值""" with self._connect() as conn: conn.execute( """ INSERT INTO session_state (session_id, summary, last_compressed_msg_id, updated_at) VALUES (?, ?, 0, datetime('now')) ON CONFLICT(session_id) DO UPDATE SET summary = excluded.summary, updated_at = datetime('now') """, (f"cache:{key}", value), ) def load_cache(self, key: str) -> Optional[str]: """读取请求缓存""" with self._connect() as conn: row = conn.execute( "SELECT summary FROM session_state WHERE session_id = ?", (f"cache:{key}",), ).fetchone() return row["summary"] if row else None

这里有几个设计细节需要解释。

第一，load_recent_messages使用了倒序查再反转的方式。SQLite 没有直接提供“取最后 N 条但保持正序”的窗口函数写法，用这个方案逻辑直观，数据量不大时性能足够。

第二，session_state表被复用来存放请求缓存，key 写成cache:{hash}。生产环境建议拆成独立缓存表，这里为了演示减少表数量，逻辑上会更便于理解。

第三，所有数据库操作都用了with self._connect()，确保事务自动提交，避免写一半崩溃导致数据损坏。

5.3 实现上下文构建与摘要压缩

现在实现 Bot 主逻辑。思路是：

1. 从持久化状态加载旧摘要；
2. 加载最近 N 条消息；
3. 如果历史消息数超过阈值，则调用模型生成最新摘要，并保存压缩进度；
4. 把摘要 + 最近消息 + 当前用户问题组装成 messages，发给模型；
5. 把用户消息和模型回复保存到数据库。

以下是bot.py的核心代码。

# 文件路径：bot.py import hashlib import os from openai import OpenAI from state_manager import DurableStateManager # 初始化模型客户端 client = OpenAI( api_key=os.getenv("GROK_API_KEY", "your-api-key"), base_url=os.getenv("GROK_BASE_URL", "https://api.example.com/v1"), ) MODEL_NAME = os.getenv("GROK_MODEL", "grok-4.6") RECENT_MESSAGE_LIMIT = 10 COMPRESSION_THRESHOLD = 20 COMPRESSION_PROMPT = "请用中文总结我们之前的对话，保留用户的关键需求和已经确定的信息，控制在200字以内。" def estimate_tokens(messages: list[dict]) -> int: """粗略估算 token 数：中文场景可按字符数约等于 token 数估算""" total = 0 for msg in messages: total += len(msg["content"]) return total def build_context( state: DurableStateManager, session_id: str, user_message: str, ) -> tuple[list[dict], dict]: """构建发送给模型的上下文，返回 (messages, 统计信息)""" summary, last_compressed_id = state.load_summary(session_id) recent_messages = state.load_recent_messages( session_id, limit=RECENT_MESSAGE_LIMIT ) # 构造基础消息列表 system_parts = [] if summary: system_parts.append( "以下是之前的对话摘要，请基于这些记忆回答当前问题。" ) system_parts.append(summary) system_message = { "role": "system", "content": "\n".join(system_parts), } context_messages = [system_message] if system_parts else [] for msg in recent_messages: context_messages.append( {"role": msg["role"], "content": msg["content"]} ) context_messages.append({"role": "user", "content": user_message}) stats = { "summary_len": len(summary) if summary else 0, "recent_msg_count": len(recent_messages), "total_msg_count": state.load_message_count(session_id), "estimated_input_tokens": estimate_tokens(context_messages), } return context_messages, stats def compress_history( state: DurableStateManager, session_id: str, last_compressed_id: int, ) -> str: """把历史消息压缩成摘要""" messages_to_compress = state.load_recent_messages( session_id, limit=500, after_msg_id=last_compressed_id ) if not messages_to_compress: return "" history_text = "\n".join( f"{msg['role']}: {msg['content']}" for msg in messages_to_compress ) compression_messages = [ {"role": "system", "content": COMPRESSION_PROMPT}, {"role": "user", "content": history_text}, ] response = client.chat.completions.create( model=MODEL_NAME, messages=compression_messages, temperature=0.3, ) summary = response.choices[0].message.content last_msg_id = messages_to_compress[-1]["id"] state.save_summary(session_id, summary, last_msg_id) return summary def ask_grok(state: DurableStateManager, session_id: str, user_message: str) -> str: """处理一轮用户消息""" # 1. 保存用户消息 state.save_message(session_id, "user", user_message) # 2. 如果历史太长，先做摘要压缩 total_msgs = state.load_message_count(session_id) if total_msgs % COMPRESSION_THRESHOLD == 0: _, last_compressed_id = state.load_summary(session_id) last_compressed_id = last_compressed_id or 0 compress_history(state, session_id, last_compressed_id) # 3. 构建上下文并调用模型 context_messages, stats = build_context(state, session_id, user_message) # 4. 可选请求缓存：同一问题在短期内直接返回历史结果 cache_key = hashlib.sha256( f"{session_id}:{user_message}".encode("utf-8") ).hexdigest() cached = state.load_cache(cache_key) if cached: print(f"[cache hit] session={session_id}") return cached response = client.chat.completions.create( model=MODEL_NAME, messages=context_messages, temperature=0.7, ) reply = response.choices[0].message.content # 5. 保存回复并写缓存 state.save_message(session_id, "assistant", reply) state.save_cache(cache_key, reply) print( f"[stats] session={session_id}, " f"recent={stats['recent_msg_count']}, total={stats['total_msg_count']}, " f"input_tokens≈{stats['estimated_input_tokens']}" ) return reply if __name__ == "__main__": state = DurableStateManager("bot.db") demo_session = "demo-user-001" while True: user_input = input("你：") if user_input.strip().lower() in {"exit", "quit", "q"}: break answer = ask_grok(state, demo_session, user_input) print(f"Bot：{answer}")

这段代码有几个关键点需要重点说明。

第一，compress_history并不是每一轮都执行，而是等消息总数达到阈值（COMPRESSION_THRESHOLD）时才触发。这样做的好处是避免频繁调用模型做摘要，因为摘要本身也要消耗 token。阈值的设置是成本与记忆粒度之间的权衡。

第二，build_context里加载的摘要和最近 N 条消息，都来自数据库，而不是内存。这意味着即使 Bot 进程重启，只要数据库还在，上下文依然可以恢复。这是 durable state 最重要的价值。

第三，请求缓存放在调用模型之前。用户重复问同一个问题时直接返回历史回复，能省下完整的一轮模型调用。不过生产环境需要注意缓存过期策略，不能所有问题都永久缓存。

5.4 环境变量配置示例

创建一个.env.example文件，方便团队协作时参考。实际运行时把变量写入环境变量，不要把真实 API Key 提交到代码仓库。

# 文件路径：.env.example GROK_API_KEY=your-api-key GROK_BASE_URL=https://api.example.com/v1 GROK_MODEL=grok-4.6

如果你的 Grok 服务端点是其他兼容地址，只需要替换GROK_BASE_URL。如果团队使用统一的 API 网关，还可以在这里加入路由前缀等配置。

6. 运行与效果验证

运行 Bot 很简单，先用命令行启动脚本：

python bot.py

启动后会进入交互式对话。连续输入多个问题，观察日志输出。下面是一次演示的预期输出（实际内容取决于模型回答）：

你：帮我整理一份关于 Python 异步编程的学习计划 Bot：可以，我们先从 asyncio 基础开始，再到协程、任务、事件循环…… [stats] session=demo-user-001, recent=0, total=1, input_tokens≈45 你：这个计划里哪部分最难？ Bot：对初学者来说，事件循环和并发思维转变通常是最难的…… [stats] session=demo-user-001, recent=2, total=3, input_tokens≈230 你：帮我总结一下我们聊过的内容 [stats] session=demo-user-001, recent=4, total=5, input_tokens≈410 Bot：我们刚才聊了 Python 异步编程的学习计划，并且讨论了事件循环和并发思维……

判断运行成功的标准有三个：

1. 数据库文件bot.db正常生成，且messages表有数据。
2. 第二轮开始，recent数值增加，说明上下文被正确加载。
3. total持续增长，但输入 token 不会无限增长。当total达到压缩阈值后，摘要会替代一部分历史。

你可以在命令行直接查看数据库内容：

sqlite3 bot.db "SELECT id, session_id, role, substr(content,1,30) FROM messages ORDER BY id DESC LIMIT 5;"

如果模型请求失败，第一个要看的是GROK_BASE_URL和GROK_API_KEY是否正确。这类接口报错通常会在client.chat.completions.create附近抛出异常，错误信息里会提示鉴权失败、模型不存在或网络超时。

7. 常见问题与排查思路

这里还有几个容易被忽略的点。

第一，SQLite 在本地开发和小并发场景表现非常好，但一旦有多个实例同时写入，很容易出现database is locked错误。生产环境建议把状态存储迁移到 Postgres 或 Redis，并保留同样的接口设计。

第二，摘要模型的调用本身也会消耗 token。如果会话非常短，没必要做摘要。这就解释了为什么压缩要设置阈值。在示例代码里，我写的是每 20 条消息触发一次，实际项目中可以根据平均对话长度调整。

第三，load_recent_messages只加载最近 N 条。如果你在摘要之外还需要更早的消息，比如用户明确询问“第一天我们说了什么”，就需要有一个检索模块，从messages表里按关键词或向量召回旧消息。这属于更进一步的设计。

8. 工程化最佳实践

前面示例代码能跑通流程，但要拿到生产环境，还需要补充几个工程细节。

8.1 存储层选型要匹配并发规模

如果 Bot 只是个人工具或内部系统，SQLite 完全够用，零运维、备份简单。如果是面向公网的多用户服务，建议把会话状态放到 Redis，把完整消息历史放到 Postgres。Redis 负责短期热数据，Postgres 负责长期可信存储。durable state 的设计核心是“状态不丢失”，所以写入操作不能只依赖内存缓存。

8.2 摘要策略要分层

不要试图用一个全局摘要保存所有信息。更稳妥的做法是分层摘要：

• 短期记忆：最近 10 轮原始消息；
• 中期记忆：按话题分段的摘要，可以按时间或关键词切分；
• 长期记忆：用户画像、固定偏好、业务规则。

越重要的信息，越应该以结构化字段保存。例如用户偏好是“喜欢简洁回答”，就不要写在自然语言摘要里，而是放到user_profile表的独立字段里。这样每次构建系统提示词时，结构化和非结构化信息可以分开组装。

8.3 所有状态写入都要考虑幂等

在示例代码中，缓存 key 是session_id + user_message的哈希。这个方案有几个问题：不同时间问同一句话，答案可能需要不同；同一个用户连续重复发送相同内容，也不一定应该触发重复请求。更合理的做法是为每次对话生成一个request_id，并在消息表里做唯一约束，避免网络重试导致消息重复保存。

例如，可以在messages表增加一列request_id TEXT UNIQUE，由调用方生成。这样即使客户端超时重试，也不会把同一条用户消息写入两次。

8.4 敏感信息不能进摘要

Grok Bot 在处理用户输入时，可能会接触到个人身份信息、密钥、内部系统地址等敏感数据。持久化状态下，这些数据会被写入数据库，也可能被摘要模型再次读取。生产环境必须做脱敏处理。

实践中至少做到三点：

1. 入库前过滤明显的敏感信息；
2. 摘要生成前，对消息中的手机号、邮箱、密码、Token 做正则替换；
3. 数据库文件设置文件权限，生产数据库使用加密或托管服务。

8.5 观测是省钱的前提

没有观测，就不知道 token 消耗在哪里。建议在每次模型调用前记录以下指标：

• session_id；
• 输入 token 估算值；
• 输出 token（从响应中获取）；
• 是否命中摘要；
• 是否命中缓存；
• 模型调用耗时。

这些数据可以写到日志系统，也可以直接落到一张usage_logs表。刚开始不必做得很复杂，但至少要能回答一个问题：如果今天账单涨了 20%，是哪个会话、哪个用户、哪类请求贡献的。

8.6 版本兼容与回滚

当你把 Bot 从“无状态”改造成“有状态”，要考虑老会话的数据兼容。比如历史会话里已经产生了一批消息，但没有摘要。上线新逻辑后，第一次加载这些会话时，摘要为空，上下文直接由最近 N 条消息构建。这种场景通常不会有问题，但如果你的系统提示词依赖摘要字段，就需要为“摘要缺失”写兜底逻辑。

回滚方面，durable state 的好处是消息原始数据都在，不管摘要策略怎么改，都可以从messages表重新生成新版摘要。上线前注意备份数据库，必要时写一个摘要重建脚本。

9. 总结与下一步实践

本文围绕“Reducing Grok Bot consumption with durable state”展开，核心判断很简单：Grok Bot 的消耗大头往往不是模型回答本身，而是每次请求重复携带的历史上下文。用 durable state 把会话状态、摘要状态、缓存状态持久化之后，Bot 从“每次重新自我介绍”变成了“带着记忆增量沟通”，输入 token 从随对话轮数线性增长，变成了稳定在固定窗口之内。

文中给出的示例是一个最小可用实现：SQLite 负责存储、摘要压缩控制历史长度、哈希缓存避免重复调用。这套结构可以直接扩展到 Redis、Postgres 或向量库方案，接口思路基本一致。

接下来你可以做三件事：

1. 把示例代码跑通，观察total和input_tokens两个指标的变化；
2. 根据自己 Bot 的对话长度调整RECENT_MESSAGE_LIMIT和COMPRESSION_THRESHOLD；
3. 加入真实模型调用，对比改造前后的 token 消耗和响应延迟。

如果你已经在做 Grok Bot，并且感觉 token 消耗涨得太快，不妨从今天起给 Bot 加上一张持久化状态表。这是投入产出比很高的一步：代码量不大，收益却直白可见。