尧图建网站 尧图建网站 YAOTU WEB BUILD 免费咨询
ARTICLE DETAIL

资讯详情

深耕网站建设与建站编程的一线实战洞察。

Tokenmaxxing应对:LLM应用预算控制与工程实践

Tokenmaxxing应对:LLM应用预算控制与工程实践 最近在开发群里又聊到一个让不少团队头疼的话题Tokenmaxxing。很多人一开始觉得这个词很酷结果一句话没聊完预算爆了、接口被封、账单翻倍。更现实的一种情况是云平台侧开始对这类高消耗的调用方式亮出红线典型的表现就是预算一旦卡死超限的部分直接由调用方自己承担。这篇文章不讨论某个公司的内部决策而是从工程角度拆解一个问题当平台开始严格限制 token 消耗时我们自己搭建的 LLM 应用应该怎么做才能不被打个措手不及。这篇文章适合正在接 Azure OpenAI、OpenAI API 以及各类国内大模型服务的后端开发、架构师和 AI 应用负责人。读完你会得到一套完整的前置预算控制方案包含 token 预估、请求前扣减、请求后校准、失败回滚、常见报错排查和工程最佳实践。即使你的接口不是微软系的这套思路也可以直接迁移到其他模型服务上。1. 背景Tokenmaxxing 为什么会被叫停1.1 什么是 Tokenmaxxing先不说术语我们看一个非常常见的场景。某个新上线的小功能本来只需要在用户提问时把最近 3 条聊天记录一起带过去结果开发为了方便直接把整个历史会话、系统文档、示例数据全部塞进了 prompt。调用量不高的时候还好一旦用户量上来每天消耗的 token 数量可能会比预估高出几十倍。Tokenmaxxing 这个词本身带有“极客式”的意味它指的是把 token 消耗当作目标本身来推进的一种写法。具体表现包括故意构造超长 prompt把不需要的上下文全部塞进去。为了“防止输出不完整”每轮都把 max_tokens 调到最大。不做任何消息截断或历史摘要无限堆叠多轮对话。批量任务中频繁重试重试时重新发送同样的大 prompt。同时调用多个模型服务却不评估每个模型的 token 单价。从短期看Tokenmaxxing 也许能换来稍微稳定一点的回答质量但它背后意味着成本失控。对个人开发者来说可能只是这个月多花几十美元对团队或公司来说可能就是某天突然收到信用卡扣款提醒或者是云端的预算告警。1.2 平台收紧预算带来的工程信号现在很多大语言模型服务商都已经意识到这个问题。除了对不同付费等级设置速率配额也开始把消耗型资源纳入预算管理。常见的平台侧限制手段包括Token 每分钟速率限制即 TPM 限制。请求每分钟次数限制即 RPM 限制。按模型维度设置配额比如某个部署只允许消耗多少 TPM。在控制台上设置成本预算超出预算后禁止继续调用。一旦平台对 Tokenmaxxing 这类高消耗模式启动限制开发者的直接感受就不再是“最多响应慢一点”而是请求被直接拒绝甚至出现 429、403、400 报错。很多团队在公司预算审批流程还没走完的时候应用已经不可用了。这就引出标题里后半句的含义预算卡死超限自负。所谓“卡死”是指预算上限一旦到达就不再放行所谓“自负”是指业务方必须自己承担由于成本评估缺失导致的可用性风险。对技术团队来说不能只依赖平台侧的配额必须在自己的应用层做预算控制器。1.3 适合哪些读者本文不是一篇纯产品分析文章而是偏实战的教程。在做以下事情的开发者最适合继续读下去正在接入大模型 API担心成本失控。需要为自己的应用设计 token 配额和预算机制。遇到 429、超限、账单异常等问题想知道怎么排查。想搭建一个能商用的 LLM 调用服务而不只是 demo。如果你有 1 到 2 年开发经验可以直接跳到第 4 节看代码。如果你刚接触大模型 API建议从第 2 节开始按顺序看每一步都有解释。2. 环境准备与版本说明2.1 使用 Azure OpenAI 还是 OpenAI API本文示例以 Azure OpenAI 为主因为很多企业开发团队实际使用的是微软云服务。Azure OpenAI 与 OpenAI 官方 API 的调用方式非常接近只是 endpoint、api_key 和 model 参数略有不同。如果你使用的是 OpenAI 官方 API可以直接把客户端初始化部分替换一下其他预算控制逻辑完全一致。如果你使用的是国内的大模型网关服务通常也支持 OpenAI 兼容格式所以下面的思路仍然适用。版本的快速变化是这类文章最容易误导人的地方。本文示例中的 Python 代码基于常见版本编写但你在实际项目中要以当前 installed 的 openai 版本为准。如果你的模型部署版本比较老可能 usage 字段、api_version 参数会有所不同需要查阅对应 SDK 的文档。2.2 运行环境与依赖本文示例的运行环境如下操作系统Windows 10/11、macOS 或 Linux 均可。Python 版本3.9 及以上。大模型服务Azure OpenAI 中已经创建好的模型部署。Python 库openai、tiktoken、python-dotenv、sqlite3。安装命令如下pip install openai1.30.0,2.0.0 pip install tiktoken0.6.0 pip install python-dotenvsqlite3 是 Python 标准库自带模块不需要额外安装。如果你在生产环境使用 Redis 代替 SQLite 做共享预算存储那么还需要安装 redis 库这个在后面的最佳实践里会提到。2.3 本文项目的准备步骤在使用 Azure OpenAI 服务之前你需要确保已经在 Azure 门户中完成了下列操作创建 Azure OpenAI 资源。部署一个模型例如 gpt-4o 或 gpt-4o-mini。获取 endpoint 和 API Key。确认该部署的配额足够你的测试流量。对于还没有开通服务的开发者可以使用 OpenAI 官方 API 进行本地测试。两者的差异只在客户端构造部分核心预算控制逻辑不变。版本需要根据你的项目实际情况调整本文示例以常见环境为例重点演示配置思路。3. 核心原理Token 消耗与预算控制3.1 Token 是如何参与计费的要理解预算控制先要理解 token 是怎么计量的。大语言模型的输入和输出都会按 token 数量计费不同类型的 token 单价不同。一个简单的公式如下单次请求成本 输入 token 数 / 1000 * 输入单价 输出 token 数 / 1000 * 输出单价大多数模型 API 会在返回结果中带上 usage 字段{ usage: { prompt_tokens: 120, completion_tokens: 45, total_tokens: 165 } }这里有个容易被忽略的点prompt_tokens 是请求中所有消息的 token 总量包括 system prompt、历史消息、用户问题、工具调用定义等。即使服务端返回的内容很短只要输入侧构造了一个很长的上下文成本也会明显上升。很多 Tokenmaxxing 场景的问题就出现在这里开发者只知道设置 max_tokens却忽略了输入 token 已经成千上万。max_tokens 只约束输出长度不会帮你限制输入长度。3.2 服务端配额与应用层预算的区别服务端配额和预算控制听起来差不多实际是两件事。服务端配额例如 Azure OpenAI 里的 TPM、RPM控制的是单位时间内的请求吞吐。它的作用是防止某个应用把某个区域的资源瞬间打满。当你超过配额时平台会返回 429 错误这时候所有请求都会失败而不会继续累计消耗。预算控制更像是一道财务闸门。它负责设置“今天这个项目最多可以花多少钱”一旦超过就停止调用。很多云平台在控制台层面提供成本预算和告警但是在 API 调用这个粒度上平台很难精确告诉你的业务逻辑“这次请求是否可以发出去”。原因很简单一次请求的最终 token 消耗在请求发出前只能估算不能精确确定。所以“预算卡死超限自负”的工程含义非常明确如果你想真正把成本控制在计划范围内就不能只依赖平台侧检查还要在应用层实现一个预算逻辑在请求前算账在请求后对账在失败后冲正。3.3 预算控制的设计模式应用层预算控制通常采用“三阶段”设计。第一阶段是事前预估。构造好 messages 之后在调用模型 API 之前先使用 tiktoken 等工具估算输入 token再加上可能的最大输出 token得到一个预估成本。如果预估成本加上今天已经消耗的成本超过了预算就直接拒绝请求根本不发出去。第二阶段是事中锁定。这里有两种做法一种是直接向预算存储中记录预扣金额另一种是在 Redis 或数据库里生成一条待确认账单标记当前请求已占用预算。推荐后面的做法因为模型返回的实际 token 数可能与预估不一致太早把金额写进去会导致账目失真。第三阶段是事后校准。请求成功返回后解析 usage 字段计算出实际成本把待确认账单更新为最终金额并在当日累计用量上追加实际成本。如果请求异常就取消预扣把预算空间释放出来。这种设计的好处是即使有多个线程并发调用也不会因为每一个请求都先“全额扣款”而导致预算很快被浪费。它保证预算数据最终和实际账单基本一致。4. 实战实现一个自带预算控制的 LLM 调用服务4.1 项目结构设计先创建一个项目目录我把它命名为 llm-budget-demo。结构如下llm-budget-demo/ ├── .env ├── config.py ├── token_helper.py ├── budget_store.py ├── llm_service.py ├── main.py └── requirements.txtconfig.py读取环境变量和配置。token_helper.py负责 token 预估。budget_store.py负责预算存储和账目管理使用 SQLite。llm_service.py封装大模型调用逻辑集成预算控制。main.py演示入口。4.2 配置文件在项目根目录创建 .env 文件AZURE_OPENAI_ENDPOINThttps://your-resource.openai.azure.com/ AZURE_OPENAI_API_KEYyour-api-key DEPLOYMENT_NAMEgpt-4o DAILY_BUDGET10.0 INPUT_PRICE_PER_1K0.005 OUTPUT_PRICE_PER_1K0.015 MAX_OUTPUT_TOKENS1024注意上面的价格仅为演示占位。不同模型、不同区域、不同付费方式的单价相差很大一定要以你实际账单中的单价为准。建议把价格配置放到外部配置中心或数据库便于后续调整。config.py 的内容如下import os from dotenv import load_dotenv load_dotenv() AZURE_OPENAI_ENDPOINT os.getenv(AZURE_OPENAI_ENDPOINT, ) AZURE_OPENAI_API_KEY os.getenv(AZURE_OPENAI_API_KEY, ) DEPLOYMENT_NAME os.getenv(DEPLOYMENT_NAME, gpt-4o) DAILY_BUDGET float(os.getenv(DAILY_BUDGET, 10.0)) INPUT_PRICE_PER_1K float(os.getenv(INPUT_PRICE_PER_1K, 0.005)) OUTPUT_PRICE_PER_1K float(os.getenv(OUTPUT_PRICE_PER_1K, 0.015)) MAX_OUTPUT_TOKENS int(os.getenv(MAX_OUTPUT_TOKENS, 1024))4.3 Token 预估模块token_helper.py 的作用是在请求发出前估算输入 token。import tiktoken MODEL_ENCODER_MAP { gpt-4o: o200k_base, gpt-4o-mini: o200k_base, gpt-4: cl100k_base, gpt-4-turbo: cl100k_base, gpt-3.5-turbo: cl100k_base, } def get_encoder(model_name: str): encoding_name MODEL_ENCODER_MAP.get(model_name, cl100k_base) return tiktoken.get_encoding(encoding_name) def count_message_tokens(model_name: str, messages: list) - int: 估算 messages 列表的 token 数。 encoder get_encoder(model_name) total 0 for message in messages: content message.get(content, ) if isinstance(content, str): total len(encoder.encode(content)) elif isinstance(content, list): for part in content: text part.get(text, ) total len(encoder.encode(text)) return total这里为了演示简化了对图片、工具调用等复杂内容的计算。如果模型支持图片输入图片部分也会消耗 token而且通常按图片尺寸和细节等级计算。这部分更准确的方式是参考官方文档或者先查看模型返回的 usage 字段再反推估算误差。4.4 预算存储模块预算存储是整个方案的核心。我使用 SQLite 来记录当天已用金额和请求账单。这样可以在多个线程之间安全地共享状态并且重启后数据不会丢失。budget_store.py 的代码如下import sqlite3 import threading from datetime import date class BudgetExceededError(Exception): 当预算不足时抛出。 pass class BudgetStore: def __init__(self, db_path: str budget.db, daily_limit: float 10.0): self.conn sqlite3.connect(db_path, check_same_threadFalse) self.lock threading.Lock() self.daily_limit daily_limit self._init_db() def _init_db(self): with self.lock: self.conn.execute( CREATE TABLE IF NOT EXISTS daily_usage ( day TEXT PRIMARY KEY, used_amount REAL NOT NULL DEFAULT 0 ) ) self.conn.execute( CREATE TABLE IF NOT EXISTS charge_log ( id INTEGER PRIMARY KEY AUTOINCREMENT, request_id TEXT, day TEXT, pre_charge REAL, final_cost REAL, status TEXT, created_at TEXT DEFAULT CURRENT_TIMESTAMP ) ) self.conn.commit() def get_used(self, day: str None) - float: day day or date.today().isoformat() row self.conn.execute( SELECT used_amount FROM daily_usage WHERE day ?, (day,) ).fetchone() return row[0] if row else 0.0 def try_precharge(self, request_id: str, amount: float, day: str None) - bool: 请求前预扣预算返回是否允许调用。 day day or date.today().isoformat() with self.lock: used self.get_used(day) if used amount self.daily_limit: return False self.conn.execute( INSERT INTO charge_log(request_id, day, pre_charge, status) VALUES (?, ?, ?, PRE_CHARGE), (request_id, day, amount), ) self.conn.commit() return True def confirm_charge(self, request_id: str, final_cost: float, day: str None): 请求成功后用实际成本更新账本。 day day or date.today().isoformat() with self.lock: self.conn.execute( UPDATE charge_log SET final_cost ?, status CONFIRMED WHERE request_id ? AND status PRE_CHARGE, (final_cost, request_id), ) self.conn.execute( INSERT INTO daily_usage(day, used_amount) VALUES (?, ?) ON CONFLICT(day) DO UPDATE SET used_amount used_amount excluded.used_amount , (day, final_cost), ) self.conn.commit() def rollback_charge(self, request_id: str, day: str None): 请求失败时取消预扣释放预算空间。 day day or date.today().isoformat() with self.lock: self.conn.execute( UPDATE charge_log SET status ROLLED_BACK WHERE request_id ? AND status PRE_CHARGE, (request_id,), ) self.conn.commit() def close(self): self.conn.close()这里用到 SQLite 的 upsert 语法可以在记录不存在时插入存在时累加。字段 status 有三个状态PRE_CHARGE、CONFIRMED、ROLLED_BACK分别表示待确认、已确认、已回滚。4.5 大模型调用服务llm_service.py 负责把预算存储和大模型 API 串起来。import uuid from openai import AzureOpenAI from config import ( AZURE_OPENAI_ENDPOINT, AZURE_OPENAI_API_KEY, DEPLOYMENT_NAME, DAILY_BUDGET, INPUT_PRICE_PER_1K, OUTPUT_PRICE_PER_1K, ) from budget_store import BudgetStore, BudgetExceededError from token_helper import count_message_tokens class LLMService: def __init__(self): self.client AzureOpenAI( api_keyAZURE_OPENAI_API_KEY, api_version2024-06-01, azure_endpointAZURE_OPENAI_ENDPOINT, ) self.store BudgetStore(budget.db, daily_limitDAILY_BUDGET) def estimate_cost(self, model_name: str, messages: list, max_output_tokens: int) - float: input_tokens count_message_tokens(model_name, messages) input_cost input_tokens / 1000 * INPUT_PRICE_PER_1K output_cost max_output_tokens / 1000 * OUTPUT_PRICE_PER_1K return input_cost output_cost def chat(self, messages: list, max_output_tokens: int 1024, request_id: str None): request_id request_id or uuid.uuid4().hex estimated_cost self.estimate_cost(DEPLOYMENT_NAME, messages, max_output_tokens) # 请求前预算检查 if not self.store.try_precharge(request_id, estimated_cost): raise BudgetExceededError(今日预算已耗尽请求被拒绝) try: response self.client.chat.completions.create( modelDEPLOYMENT_NAME, messagesmessages, max_tokensmax_output_tokens, ) usage response.usage final_cost ( usage.prompt_tokens / 1000 * INPUT_PRICE_PER_1K usage.completion_tokens / 1000 * OUTPUT_PRICE_PER_1K ) # 请求成功后校准账本 self.store.confirm_charge(request_id, final_cost) return response except Exception: # 请求失败释放预占预算 self.store.rollback_charge(request_id) raise def today_usage(self) - float: return self.store.get_used()这段代码的逻辑是这样每次调用前先估算本次请求的预估成本然后尝试预扣。如果今天已经用了 9 元而本次预估需要 2 元预算上限是 10 元那么本次请求就会被直接拒绝。这个行为的本质就是“预算卡死”不是等账单出来以后再处理而是请求前就拦住。请求成功后用模型返回的实际 usage 重算成本再把预扣状态更新成最终金额。这样账本不会因为预估偏差而失真。4.6 运行入口与测试main.py 是一个最简单的测试入口from llm_service import LLMService from budget_store import BudgetExceededError def main(): service LLMService() messages [ {role: system, content: 你是一个简洁的助手请用很少的话回答问题。}, {role: user, content: 用一句话解释什么是 Tokenmaxxing。}, ] try: response service.chat(messages, max_output_tokens200) print(模型回复, response.choices[0].message.content) except BudgetExceededError as e: print(预算拦截, e) except Exception as e: print(调用失败, e) print(今日已用金额, service.today_usage()) if __name__ __main__: main()运行命令python main.py如果你的 Azure OpenAI 配置正确会看到类似下面的输出模型回复Tokenmaxxing 是一种在设计提示词或调用模型时过度追求 token 消耗上限的极端做法。 今日已用金额 0.000925当然实际金额取决于你的模型单价和你发送的 messages。4.7 结果说明从结果可以看到一次简单的调用会被记入当日账本。如果你反复调用当累计金额超过 DAILY_BUDGET 时后续请求就会被 BudgetExceededError 拦截不再发送到模型服务。这种做法的意义在于即使平台侧的配额没有生效应用侧也能兜底。它把预算问题从“事后看账单”变成了“事前拦截”从根上防止 Tokenmaxxing 式消耗把成本拉爆。5. 常见问题与排查思路5.1 常用报错清单在实际开发中你可能会遇到下面这些现象。我整理了一个排查表格。问题现象常见原因解决思路调用返回 429 Too Many Requests超过了 Azure OpenAI 的 TPM 或 RPM 配额查看部署配额申请提高配额或降低并发预算明明没到上限却报 BudgetExceededError预估成本过高例如 max_output_tokens 设置过大合理设置 max_output_tokens对输入做截断预算到了但请求仍能发出没有在应用层做预算控制只依赖平台引入本文的预算控制器或使用 Redis 计数重试后账单翻倍对 429、5xx 等错误无条件重试区分错误类型采用指数退避并限制最大重试次数预估 token 和实际 token 相差很大图片输入、工具调用、消息格式差异用 usage 回填校准定期校正预估逻辑SQLite 报 database is locked多线程并发写入同一个 SQLite 文件使用连接池或迁移到 Redis 作为共享存储5.2 429 错误排查步骤如果你遇到 429不要直接调大重试次数。按下面顺序排查查看是 RPM 超限还是 TPM 超限响应头或错误信息里通常会说明。检查当前 Deployment 的配额设置确认是否需要提高。检查应用是否在并发场景下发送了大量请求可以先做本地限流。如果只是偶尔超限可以使用指数退避例如第一次等待 1 秒第二次 2 秒第三次 4 秒。如果重试之后仍然 429考虑临时切换到其他部署或模型。注意429 时盲目重试不会帮助消耗降低反而会推高 request 次数而且如果每次重试都发送同一个很大的 prompt还会白白增加 token 计量。5.3 预算账目不一致如何排查当你发现账本里的金额与账单不一致时可以这样排查查看 charge_log 中 status 为 PRE_CHARGE 的残留记录这些可能是在请求进行中进程崩溃导致没有确认或回滚。检查是否有多实例部署每个实例都连了同一个 SQLite 文件且并发写入产生了覆盖。检查价格配置是否已按最新模型单价更新。检查是否遗漏了流式请求的 usage 处理。在流式场景中usage 通常最后一个 chunk 里返回需要特殊处理。在预算控制中一致性比实时性更重要。宁可账目稍滞后也不要出现重复记账或漏记账。6. 最佳实践与工程建议6.1 分层预算控制前面演示的是单机版预算控制适合个人项目或小型团队。在正式生产环境推荐采用分层预算组织层级设置总额度例如部门每天最多 500 元。项目层级每个应用有自己的预算例如客服机器人每天 200 元。用户层级每个用户或每个 API 调用方有独立额度防止单个调用方拖垮整体。实现分层时可以在 BudgetStore 里增加 user_id、project_id 字段把预算检查变成多层校验。第一层检查用户额度第二层检查项目额度第三层检查全局额度。任何一层的余额不足都拒绝请求。6.2 使用 Redis 做跨实例共享预算SQLite 适合单机或小规模部署但如果在多台服务器上部署同一个服务就需要一个集中式的计数器。Redis 是常见选择。基本思路是使用 Redis 的 INCRBY 和过期时间实现按日统计import redis from datetime import date r redis.Redis(hostlocalhost, port6379, db0) def get_day_key(): return fbudget:used:{date.today().isoformat()} def trial_charge(amount): current float(r.get(get_day_key()) or 0) if current amount DAILY_BUDGET: return False r.incrbyfloat(get_day_key(), amount) r.expire(get_day_key(), 48 * 3600) return True用 Redis 的 incrbyfloat 可以确保多个实例并发扣减时不会互相覆盖。同时设置过期时间避免次日数据残留。6.3 请求前预扣与最终校准结合在预算控制中我不建议只做事前预估因为预估很容易偏小或偏大。正确做法是请求前用乐观值预扣或者标记待确认账单。请求完成后用实际 usage 更新。对长时间未确认的账单做超时清理。如果预估偏差很小可以简化成“直接扣预估金额不平账”。但如果你的服务面向大量并发请求还是应该预留校准机制否则账本会在一天内逐渐失真。6.4 消息压缩与历史截断Tokenmaxxing 的根源往往是 prompt 过长。真正解决成本问题除了预算拦截还要减少不必要的 token。固定窗口截断只保留最近 N 条消息。时间窗口过滤只取最近 30 分钟内的消息。摘要压缩把历史对话先用模型或规则压缩成摘要再拼入下一轮请求。去除重复内容检查 messages 中是否有重复的 system prompt 或工具定义。这些手段配合预算控制能让应用在有限的成本预算下服务更多真实请求。6.5 重试策略与熔断超时、5xx 错误需要重试但重试必须在预算和速率上都做限制。建议采用最大重试次数为 3。使用指数退避基础等待时间 1 秒起步。针对 429、503 区分处理。如果连续失败超过阈值触发熔断不再调用模型改走缓存或降级文案。重试时还要重新检查预算。因为原请求失败后预扣已经被回滚如果重试请求顶着新的预算空间发出可能造成并发拥挤。更稳妥的做法是重试时重新走一遍预算检查逻辑而不是直接绕过。6.6 语义缓存很多用户会问重复问题。如果每次都调用大模型不仅浪费预算响应也不稳定。可以引入语义缓存把历史问题和答案记录在向量数据库中。大致流程如下用户提问后先用 embedding 模型生成向量。在向量数据库中检索历史问题计算相似度。如果相似度大于阈值直接返回缓存答案不调用大模型。如果低于阈值走正常模型调用并把结果回写缓存。语义缓存能显著降低 token 消耗尤其是在客服场景中效果非常明显。6.7 日志与审计成本控制不只是“拒绝请求”更需要可观测性。建议为每次调用记录以下字段request_iduser_id / project_id模型名称prompt_tokenscompletion_tokenstotal_tokens预估成本与实际成本请求状态包括成功、失败、预算拦截耗时日志写完以后按天汇总到成本报表。每当预算告警就能知道是哪类请求、哪个用户或哪个功能超出了预期。7. 结语与后续方向Tokenmaxxing 被叫停并不是大模型应用不能做了而是提醒我们调用大模型 API 和调用普通接口不一样每一次请求都有明确的成本。如果把 token 当空气最终只能是预算卡死、超限自负。与其等平台来限制不如先在自己的应用层实现一套预算控制方案。本文从概念讲解到完整代码演示了一个最小可运行的预算控制服务。你可以把它用在自己的项目中也可以在此基础上扩展出 Redis 版、多租户版、可视化报表版。后续建议你重点做三件事一是把价格配置和预算配置放到外部配置中心方便动态调整二是对接真实的成本日报验证预估与实际账单的偏差三是给核心链路加上语义缓存和模型降级降低对单一模型服务的过度依赖。如果你最近正准备把大模型能力接入项目不妨先按本文的方案给服务加上一把成本锁。以后再遇到预算超限的问题至少不会再毫无防备。
返回列表