尧图网页设计 
最近一段时间做 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 包括状态类型存储内容典型存储介质会话状态当前会话 ID、用户 ID、会话元数据SQLite、Redis、Postgres消息历史多轮对话的原始消息记录SQLite、MongoDB、对象存储摘要状态历史对话压缩后的摘要SQLite、Redis、向量数据库业务状态任务进度、待办事项、工具调用结果Redis、Postgres、本地文件用一句话概括没有 durable stateBot 是“金鱼记忆”每次请求都要从零开始理解世界有了 durable stateBot 才能像正常人一样把已经说过的话、做过的事记住只把“新变化”如实同步给模型。这里要区分一个概念很多开发者在内存里维护一个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 特别有效。手段解决的核心问题适合场景会话状态持久化历史丢失、上下文中断多轮对话、客服 Bot、Agent上下文摘要压缩历史无限增长、token 爆炸长对话、复杂任务、记忆型 Bot请求缓存与幂等重复请求、重复计算问答 Bot、知识库助手、报告生成4. 环境准备与项目结构下面进入实战环节。我们会用 Python 3 写一个最小示例存储层用 SQLite模型接口用兼容 OpenAI SDK 的通用服务端点方便你替换成 Grok 或其他兼容服务。如果你用的是 Grok 官方提供的接口或第三方兼容网关只需要替换base_url、api_key和model即可。本文示例不依赖具体的 Grok SDK 细节因为不同接入方式的参数差异较大但整体逻辑完全通用。4.1 依赖清单建议使用 Python 3.9 及以上版本。需要安装的依赖如下pip install openaiopenai 这个 Python 包目前是社区事实上的标准客户端很多模型服务都支持 OpenAI 兼容接口。如果不想引入 SDK也可以用requests直接发 HTTP 请求但用 SDK 代码会更简洁、更好维护。4.2 项目结构本文示例的目录结构如下grok_bot_durable_state/ ├── bot.py ├── state_manager.py ├── bot.db └── .env其中state_manager.py持久化状态管理器负责写入和读取消息、摘要、缓存。bot.pyBot 主逻辑负责加载上下文、调用模型、返回回复。bot.dbSQLite 数据库文件运行后自动生成。.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) , (fcache:{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 ?, (fcache:{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 主逻辑。思路是从持久化状态加载旧摘要加载最近 N 条消息如果历史消息数超过阈值则调用模型生成最新摘要并保存压缩进度把摘要 最近消息 当前用户问题组装成 messages发给模型把用户消息和模型回复保存到数据库。以下是bot.py的核心代码。# 文件路径bot.py import hashlib import os from openai import OpenAI from state_manager import DurableStateManager # 初始化模型客户端 client OpenAI( api_keyos.getenv(GROK_API_KEY, your-api-key), base_urlos.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, limitRECENT_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, limit500, after_msg_idlast_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( modelMODEL_NAME, messagescompression_messages, temperature0.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( modelMODEL_NAME, messagescontext_messages, temperature0.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}, frecent{stats[recent_msg_count]}, total{stats[total_msg_count]}, finput_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(fBot{answer})这段代码有几个关键点需要重点说明。第一compress_history并不是每一轮都执行而是等消息总数达到阈值COMPRESSION_THRESHOLD时才触发。这样做的好处是避免频繁调用模型做摘要因为摘要本身也要消耗 token。阈值的设置是成本与记忆粒度之间的权衡。第二build_context里加载的摘要和最近 N 条消息都来自数据库而不是内存。这意味着即使 Bot 进程重启只要数据库还在上下文依然可以恢复。这是 durable state 最重要的价值。第三请求缓存放在调用模型之前。用户重复问同一个问题时直接返回历史回复能省下完整的一轮模型调用。不过生产环境需要注意缓存过期策略不能所有问题都永久缓存。5.4 环境变量配置示例创建一个.env.example文件方便团队协作时参考。实际运行时把变量写入环境变量不要把真实 API Key 提交到代码仓库。# 文件路径.env.example GROK_API_KEYyour-api-key GROK_BASE_URLhttps://api.example.com/v1 GROK_MODELgrok-4.6如果你的 Grok 服务端点是其他兼容地址只需要替换GROK_BASE_URL。如果团队使用统一的 API 网关还可以在这里加入路由前缀等配置。6. 运行与效果验证运行 Bot 很简单先用命令行启动脚本python bot.py启动后会进入交互式对话。连续输入多个问题观察日志输出。下面是一次演示的预期输出实际内容取决于模型回答你帮我整理一份关于 Python 异步编程的学习计划 Bot可以我们先从 asyncio 基础开始再到协程、任务、事件循环…… [stats] sessiondemo-user-001, recent0, total1, input_tokens≈45 你这个计划里哪部分最难 Bot对初学者来说事件循环和并发思维转变通常是最难的…… [stats] sessiondemo-user-001, recent2, total3, input_tokens≈230 你帮我总结一下我们聊过的内容 [stats] sessiondemo-user-001, recent4, total5, input_tokens≈410 Bot我们刚才聊了 Python 异步编程的学习计划并且讨论了事件循环和并发思维……判断运行成功的标准有三个数据库文件bot.db正常生成且messages表有数据。第二轮开始recent数值增加说明上下文被正确加载。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. 常见问题与排查思路问题现象可能原因排查方式解决方案启动后提示数据库表不存在数据库路径或初始化顺序错误检查bot.db文件是否生成调用_init_tables()确保DurableStateManager实例化后再操作数据库模型返回 401 鉴权失败API Key 未设置或填错检查环境变量和请求日志中的错误信息确认 Key 是否有效并检查是否添加了正确的 Bearer 头模型返回 404 模型不存在GROK_MODEL模型名写错或网关不支持该模型查看服务端错误提示中的 model 字段换成网关支持的模型名一般以版本发布公告为准上下文压缩后 Bot 反而“失忆”摘要信息丢失或压缩时机太早查看session_state表中的摘要内容调整COMPRESSION_THRESHOLD并检查摘要 prompt 是否足够清晰输入 token 仍然持续增长上下文裁剪没有生效或摘要未加载成功打印build_context的统计信息观察summary_len是否为 0检查load_summary是否读取到数据确认save_summary后进度已更新多实例部署时状态冲突多个进程并发写同一个 SQLite 数据库开启 SQLite WAL 模式处理database is locked错误生产环境换用 Redis、Postgres或用 SQLite 的单写多读方案这里还有几个容易被忽略的点。第一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 在处理用户输入时可能会接触到个人身份信息、密钥、内部系统地址等敏感数据。持久化状态下这些数据会被写入数据库也可能被摘要模型再次读取。生产环境必须做脱敏处理。实践中至少做到三点入库前过滤明显的敏感信息摘要生成前对消息中的手机号、邮箱、密码、Token 做正则替换数据库文件设置文件权限生产数据库使用加密或托管服务。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 或向量库方案接口思路基本一致。接下来你可以做三件事把示例代码跑通观察total和input_tokens两个指标的变化根据自己 Bot 的对话长度调整RECENT_MESSAGE_LIMIT和COMPRESSION_THRESHOLD加入真实模型调用对比改造前后的 token 消耗和响应延迟。如果你已经在做 Grok Bot并且感觉 token 消耗涨得太快不妨从今天起给 Bot 加上一张持久化状态表。这是投入产出比很高的一步代码量不大收益却直白可见。 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 包括状态类型存储内容典型存储介质会话状态当前会话 ID、用户 ID、会话元数据SQLite、Redis、Postgres消息历史多轮对话的原始消息记录SQLite、MongoDB、对象存储摘要状态历史对话压缩后的摘要SQLite、Redis、向量数据库业务状态任务进度、待办事项、工具调用结果Redis、Postgres、本地文件用一句话概括没有 durable stateBot 是“金鱼记忆”每次请求都要从零开始理解世界有了 durable stateBot 才能像正常人一样把已经说过的话、做过的事记住只把“新变化”如实同步给模型。这里要区分一个概念很多开发者在内存里维护一个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 特别有效。手段解决的核心问题适合场景会话状态持久化历史丢失、上下文中断多轮对话、客服 Bot、Agent上下文摘要压缩历史无限增长、token 爆炸长对话、复杂任务、记忆型 Bot请求缓存与幂等重复请求、重复计算问答 Bot、知识库助手、报告生成4. 环境准备与项目结构下面进入实战环节。我们会用 Python 3 写一个最小示例存储层用 SQLite模型接口用兼容 OpenAI SDK 的通用服务端点方便你替换成 Grok 或其他兼容服务。如果你用的是 Grok 官方提供的接口或第三方兼容网关只需要替换base_url、api_key和model即可。本文示例不依赖具体的 Grok SDK 细节因为不同接入方式的参数差异较大但整体逻辑完全通用。4.1 依赖清单建议使用 Python 3.9 及以上版本。需要安装的依赖如下pip install openaiopenai 这个 Python 包目前是社区事实上的标准客户端很多模型服务都支持 OpenAI 兼容接口。如果不想引入 SDK也可以用requests直接发 HTTP 请求但用 SDK 代码会更简洁、更好维护。4.2 项目结构本文示例的目录结构如下grok_bot_durable_state/ ├── bot.py ├── state_manager.py ├── bot.db └── .env其中state_manager.py持久化状态管理器负责写入和读取消息、摘要、缓存。bot.pyBot 主逻辑负责加载上下文、调用模型、返回回复。bot.dbSQLite 数据库文件运行后自动生成。.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) , (fcache:{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 ?, (fcache:{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 主逻辑。思路是从持久化状态加载旧摘要加载最近 N 条消息如果历史消息数超过阈值则调用模型生成最新摘要并保存压缩进度把摘要 最近消息 当前用户问题组装成 messages发给模型把用户消息和模型回复保存到数据库。以下是bot.py的核心代码。# 文件路径bot.py import hashlib import os from openai import OpenAI from state_manager import DurableStateManager # 初始化模型客户端 client OpenAI( api_keyos.getenv(GROK_API_KEY, your-api-key), base_urlos.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, limitRECENT_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, limit500, after_msg_idlast_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( modelMODEL_NAME, messagescompression_messages, temperature0.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( modelMODEL_NAME, messagescontext_messages, temperature0.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}, frecent{stats[recent_msg_count]}, total{stats[total_msg_count]}, finput_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(fBot{answer})这段代码有几个关键点需要重点说明。第一compress_history并不是每一轮都执行而是等消息总数达到阈值COMPRESSION_THRESHOLD时才触发。这样做的好处是避免频繁调用模型做摘要因为摘要本身也要消耗 token。阈值的设置是成本与记忆粒度之间的权衡。第二build_context里加载的摘要和最近 N 条消息都来自数据库而不是内存。这意味着即使 Bot 进程重启只要数据库还在上下文依然可以恢复。这是 durable state 最重要的价值。第三请求缓存放在调用模型之前。用户重复问同一个问题时直接返回历史回复能省下完整的一轮模型调用。不过生产环境需要注意缓存过期策略不能所有问题都永久缓存。5.4 环境变量配置示例创建一个.env.example文件方便团队协作时参考。实际运行时把变量写入环境变量不要把真实 API Key 提交到代码仓库。# 文件路径.env.example GROK_API_KEYyour-api-key GROK_BASE_URLhttps://api.example.com/v1 GROK_MODELgrok-4.6如果你的 Grok 服务端点是其他兼容地址只需要替换GROK_BASE_URL。如果团队使用统一的 API 网关还可以在这里加入路由前缀等配置。6. 运行与效果验证运行 Bot 很简单先用命令行启动脚本python bot.py启动后会进入交互式对话。连续输入多个问题观察日志输出。下面是一次演示的预期输出实际内容取决于模型回答你帮我整理一份关于 Python 异步编程的学习计划 Bot可以我们先从 asyncio 基础开始再到协程、任务、事件循环…… [stats] sessiondemo-user-001, recent0, total1, input_tokens≈45 你这个计划里哪部分最难 Bot对初学者来说事件循环和并发思维转变通常是最难的…… [stats] sessiondemo-user-001, recent2, total3, input_tokens≈230 你帮我总结一下我们聊过的内容 [stats] sessiondemo-user-001, recent4, total5, input_tokens≈410 Bot我们刚才聊了 Python 异步编程的学习计划并且讨论了事件循环和并发思维……判断运行成功的标准有三个数据库文件bot.db正常生成且messages表有数据。第二轮开始recent数值增加说明上下文被正确加载。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. 常见问题与排查思路问题现象可能原因排查方式解决方案启动后提示数据库表不存在数据库路径或初始化顺序错误检查bot.db文件是否生成调用_init_tables()确保DurableStateManager实例化后再操作数据库模型返回 401 鉴权失败API Key 未设置或填错检查环境变量和请求日志中的错误信息确认 Key 是否有效并检查是否添加了正确的 Bearer 头模型返回 404 模型不存在GROK_MODEL模型名写错或网关不支持该模型查看服务端错误提示中的 model 字段换成网关支持的模型名一般以版本发布公告为准上下文压缩后 Bot 反而“失忆”摘要信息丢失或压缩时机太早查看session_state表中的摘要内容调整COMPRESSION_THRESHOLD并检查摘要 prompt 是否足够清晰输入 token 仍然持续增长上下文裁剪没有生效或摘要未加载成功打印build_context的统计信息观察summary_len是否为 0检查load_summary是否读取到数据确认save_summary后进度已更新多实例部署时状态冲突多个进程并发写同一个 SQLite 数据库开启 SQLite WAL 模式处理database is locked错误生产环境换用 Redis、Postgres或用 SQLite 的单写多读方案这里还有几个容易被忽略的点。第一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 在处理用户输入时可能会接触到个人身份信息、密钥、内部系统地址等敏感数据。持久化状态下这些数据会被写入数据库也可能被摘要模型再次读取。生产环境必须做脱敏处理。实践中至少做到三点入库前过滤明显的敏感信息摘要生成前对消息中的手机号、邮箱、密码、Token 做正则替换数据库文件设置文件权限生产数据库使用加密或托管服务。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 或向量库方案接口思路基本一致。接下来你可以做三件事把示例代码跑通观察total和input_tokens两个指标的变化根据自己 Bot 的对话长度调整RECENT_MESSAGE_LIMIT和COMPRESSION_THRESHOLD加入真实模型调用对比改造前后的 token 消耗和响应延迟。如果你已经在做 Grok Bot并且感觉 token 消耗涨得太快不妨从今天起给 Bot 加上一张持久化状态表。这是投入产出比很高的一步代码量不大收益却直白可见。