最近一两年大模型从“能聊天的演示工具”快速变成了“可被应用调用的基础设施”。很多大厂 AI 平台陆续开放 API按 Token 计费、按调用量结算开发者只要申请一个 Key就能在自己的应用里接入对话、推理、向量化等能力。这个模式看起来和普通云服务很像但如果从后端视角去看它其实相当于给每个接入 AI 的应用装了一个“新收银台”谁来调用、用了多少、该扣多少钱、钱够不够、会不会被刷爆全部要在毫秒级完成判断。这篇文章我想从工程实现的角度拆解大厂 AI 背后的“收银台”逻辑。我们会先梳理核心概念再用 Python FastAPI 从零实现一个最小可运行的 AI 计费网关包含 API Key 鉴权、Token 统计、配额扣减、账单落库等关键能力。最后再聊一聊生产环境里的流式计费、幂等、预算保护、安全边界等进阶问题。内容适合有 Python 基础、想了解 AI 应用后端实现或者正在做 AI 平台相关开发的读者。跟着文章跑完一遍你会对“大厂 AI 是如何按量收费的”有一个非常具体的认识。1. 背景大厂 AI 与新“收银台”到底是什么1.1 从模型能力到可计费服务最早接触大模型时多数人的印象是“一个网页对话框”。你输入一句话它回你一段话看起来和搜索引擎差不多。但真正的商业级大模型并不是以网页为核心而是以 API 为核心。开发者拿到一个 API Key把对话请求发送到模型服务模型返回结果服务端在返回时附带本次请求消耗了多少 Token。Token 就是大模型世界的“字数”也是计费的依据。这种模式带来的变化非常明显模型能力变成了可以被任意业务调用的函数而不是一个单机工具。比如客服系统可以调对话模型搜索系统可以调向量模型内容平台可以调生成模型。每一次调用本质上都是一次“结账”模型的算力成本被折算成 Token 单价再乘以调用量就变成了大厂 AI 平台的收入。这个过程非常像线下超市的收银台你拿商品请求收银员扫码计费结账扣费小票账单。1.2 AI 收银台的四个核心层从系统架构上看一个大厂 AI 平台的“收银台”通常包含四层第一层是接入层。开发者通过 API 网关把请求送到平台网关负责 SSL、路由、限流、鉴权。开发者拿到的 API Key 在这一层被识别。第二层是模型服务层。请求经过网关后被转发给不同的模型推理服务。这里会涉及模型选择、上下文拼接、流式输出、超时控制等逻辑。第三层是计费层。模型服务返回后系统需要拿到本次请求的 Token 用量根据模型单价计算金额并更新开发者的配额或余额。这是“收银台”最核心的部分。第四层是数据观测层。每笔调用要有记录包括用户 ID、模型、Token 用量、金额、耗时、返回状态。这些数据既用于账单查询也用于成本分析、告警和风控。很多开发者以为接入大模型 API 只是发一个 HTTP 请求那么简单其实平台侧要保证计费准确、并发可控、数据可靠远不是写一个接口转发就够了。1.3 从本文能学到什么本文不会只停留在概念层面。后面我会带着你实现一个简化版 AI 计费网关重点展示三块内容API Key 如何生成和校验Token 用量如何统计并转换成费用配额如何扣减以及如何防止超用。这三块理解后再看大厂 AI 平台的官方文档和计费规则你会有一种“原来是这么实现的”感觉。2. 核心概念API Key、Token 与配额2.1 API Key 是收银台的会员卡用过支付宝、微信支付的开发者都知道每次支付需要有账号、有支付凭证。在 AI 平台里API Key 承担了类似“支付凭证”的角色。调用方在请求头里带上自己的 Key平台根据 Key 判断你是谁、你有没有权限、你账户里还有没有额度。从安全角度讲API Key 本质上是敏感信息。真实生产系统中数据库不会明文保存 Key而是保存 Key 的哈希值。当平台生成 API Key 时明文只展示一次之后再也无法从后台查看。这样即使数据库泄露攻击者拿到的也只是一串不可逆的哈希值。在我们的示例系统中为了先跑通流程会先使用明文 Key但在最佳实践章节会重点说明哈希存储的必要性。2.2 Token 是 AI 世界的计价单位Token 可以通俗理解为“模型看文本的最小单位”。英文文本中 1 个 Token 大约对应 0.75 个单词中文文本中 1 个 Token 可能对应半个到一个汉字。大模型在处理文本时不是按字符数计算的而是按 Token 数计算的。为什么按 Token 计费因为 Token 数量直接对应模型的计算量。模型生成的 Token 越多占用的算力越大推理时间越长成本也越高。所以平台在返回结果时通常会附带类似这样的信息{ usage: { prompt_tokens: 32, completion_tokens: 45, total_tokens: 77 } }其中 prompt_tokens 是输入文本的 Token 数completion_tokens 是模型生成的 Token 数total_tokens 是两者之和。需要说明的是Token 数的真正计算并不是简单用字符串长度除以 2而是由模型的 Tokenizer 完成。不同模型有不同 Tokenizer同一段文本在不同模型下 Token 数也可能不同。生产环境中计费应当以模型服务返回的 usage 字段为准而不是自己在网关层用 len(text) 估算。本文示例为了简化会采用一个近似估算方法但你一定要清楚这两者的区别。2.3 配额与限流防止“刷爆”收银台配额指的是一个 API Key 最大可以消耗多少 Token或者最大可以花多少钱。它的作用类似于手机套餐里的剩余流量。每次请求完成后系统把本次消耗的 Token 累加到已用额度上然后判断是否超过总配额。限流则是一个前置保护。限流可以在请求进入模型服务前拦截超高 QPS避免一个 Key 突发大量请求把服务打满。常见的限流算法有固定窗口、滑动窗口、令牌桶等。计费网关中配额和限流是两套独立但又需要配合的机制配额管总量限流管速率。3. 整体架构与项目准备3.1 技术选型与运行环境本文示例采用 Python 语言原因是大模型生态中 Python 最为通用代码表达也直观。Web 框架选择 FastAPI它支持异步、自动生成接口文档并且和第三方生态配合非常方便。数据库选择 SQLiteSQLite 不需要额外安装服务端适合做最小演示。生产环境可以平滑替换成 MySQL 或 PostgreSQL。运行环境建议如下Python 3.9 或以上版本FastAPIUvicorn系统自带 sqlite3依赖可以按下面方式安装pip install fastapi uvicorn pydantic不要刻意追求最新版本安装时以 pip 自动解析出的稳定版本为准。3.2 项目结构为了保持代码易读我把系统拆成几个文件ai-billing-gateway/ ├── main.py # FastAPI 应用与接口路由 ├── db.py # SQLite 初始化和连接工具 ├── auth.py # API Key 生成与校验 └── requirements.txt # 依赖如果你在练习时想改名文件完全没问题只要保证模块导入路径一致即可。3.3 数据库设计数据库一共需要两张核心表。第一张是 API Key 表记录每个用户或应用的 Key、配额和使用量。第二张是账单记录表每笔调用生成一条记录。字段设计如下api_keys 表字段类型说明idINTEGER主键api_keyTEXT唯一 Keyuser_idTEXT所属用户quota_tokensINTEGER总配额used_tokensINTEGER已用 Token 数statusTEXT状态active / disabledbilling_records 表字段类型说明idINTEGER主键api_key_idINTEGER关联的 Key IDuser_idTEXT所属用户modelTEXT模型名称request_idTEXT请求幂等 IDprompt_tokensINTEGER输入 Tokencompletion_tokensINTEGER输出 Tokentotal_tokensINTEGER总 TokenamountINTEGER扣费金额单位自定义两张表的关系很简单一个 API Key 会产生多条账单记录。我们通过 api_key_id 关联查询就能统计某个 Key 某一段时间内的总消耗。4. 完整实战从零实现一个 AI 计费网关4.1 初始化数据库先写 db.py负责创建数据库和连接。代码里使用 sqlite3 自带的 executescript 创建表结构并设置 row_factory方便后续把行对象直接转成字典使用。# db.py import os import sqlite3 DB_PATH os.environ.get(AI_BILLING_DB, ai_billing.db) def get_conn(): conn sqlite3.connect(DB_PATH) conn.row_factory sqlite3.Row return conn def init_db(): conn get_conn() conn.executescript( CREATE TABLE IF NOT EXISTS api_keys ( id INTEGER PRIMARY KEY AUTOINCREMENT, api_key TEXT NOT NULL UNIQUE, user_id TEXT NOT NULL, quota_tokens INTEGER NOT NULL DEFAULT 1000000, used_tokens INTEGER NOT NULL DEFAULT 0, status TEXT NOT NULL DEFAULT active, created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP ); CREATE TABLE IF NOT EXISTS billing_records ( id INTEGER PRIMARY KEY AUTOINCREMENT, api_key_id INTEGER NOT NULL, user_id TEXT NOT NULL, model TEXT NOT NULL, request_id TEXT NOT NULL, prompt_tokens INTEGER NOT NULL DEFAULT 0, completion_tokens INTEGER NOT NULL DEFAULT 0, total_tokens INTEGER NOT NULL DEFAULT 0, amount INTEGER NOT NULL DEFAULT 0, created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP ); ) conn.commit() conn.close()这里需要注意SQLite 默认把整张数据库文件当成自己的“数据库”不需要额外创建 database 名。设置 DB_PATH 为环境变量是为了方便你在测试时切换数据库文件。4.2 生成 API Key接着写 auth.py。生成 Key 时我们用 secrets.token_urlsafe 生成一段随机字符串并且加一个 ak_ 前缀做标识。真实生产环境里这里应该只保存 Key 的哈希值后面我会再提。# auth.py import secrets from fastapi import Request, HTTPException from db import get_conn def generate_api_key(): return ak_ secrets.token_urlsafe(32) def save_api_key(api_key, user_id, quota_tokens): conn get_conn() conn.execute( INSERT INTO api_keys (api_key, user_id, quota_tokens) VALUES (?, ?, ?), (api_key, user_id, quota_tokens), ) conn.commit() conn.close() def verify_api_key(request: Request): auth_header request.headers.get(Authorization, ) if not auth_header.startswith(Bearer ): raise HTTPException(status_code401, detail缺少 API Key) api_key auth_header[len(Bearer ):].strip() conn get_conn() row conn.execute( SELECT * FROM api_keys WHERE api_key ? AND status active, (api_key,), ).fetchone() conn.close() if row is None: raise HTTPException(status_code401, detailAPI Key 无效或已禁用) return dict(row)verify_api_key 是核心鉴权函数它会从请求头中取出 API Key然后查询数据库。如果 Key 不存在或状态不是 active直接返回 401。这样每个需要登录的接口都可以复用这个函数。4.3 实现模型调用与计费main.py 是整个网关的主体。为了不依赖真实的外部模型服务我们内置一个模拟模型客户端。真实场景中这里应该替换成对官方模型的 HTTP 调用。计费部分使用一个 mock 单价表单位为“分 / 千 Token”。例如PRICING { demo-chat: {price_per_1k_tokens: 2}, demo-embedding: {price_per_1k_tokens: 1}, }金额计算逻辑是 total_tokens * 单价 / 1000这样总 Token 越大费用越高。# main.py import uuid from typing import Optional from fastapi import FastAPI, HTTPException, Request from pydantic import BaseModel from auth import generate_api_key, save_api_key, verify_api_key from db import init_db, get_conn app FastAPI(titleAI Billing Gateway) PRICING { demo-chat: {price_per_1k_tokens: 2}, demo-embedding: {price_per_1k_tokens: 1}, } class ChatRequest(BaseModel): model: str messages: list stream: Optional[bool] False class CreateKeyRequest(BaseModel): user_id: str quota_tokens: int 1000000 def call_mock_model(messages): 模拟模型推理。真实场景替换为模型服务调用。 prompt_text for msg in messages: prompt_text msg.get(content, ) reply_text 你好这是一段模拟的大模型回复。 prompt_tokens max(1, len(prompt_text) // 2) completion_tokens max(1, len(reply_text) // 2) return { content: reply_text, prompt_tokens: prompt_tokens, completion_tokens: completion_tokens, } def calc_amount(model: str, total_tokens: int) - int: price PRICING.get(model, {}).get(price_per_1k_tokens, 1) return max(1, total_tokens * price // 1000) app.on_event(startup) def on_startup(): init_db() app.post(/admin/api_keys, response_modeldict) def create_api_key_route(req: CreateKeyRequest): api_key generate_api_key() save_api_key(api_key, req.user_id, req.quota_tokens) return { api_key: api_key, user_id: req.user_id, quota_tokens: req.quota_tokens, } app.post(/v1/chat/completions, response_modeldict) def chat_completions(req: ChatRequest, request: Request): key_info verify_api_key(request) result call_mock_model(req.messages) total_tokens result[prompt_tokens] result[completion_tokens] if key_info[used_tokens] total_tokens key_info[quota_tokens]: raise HTTPException(status_code402, detail配额不足请充值) amount calc_amount(req.model, total_tokens) request_id str(uuid.uuid4()) conn get_conn() try: conn.execute( UPDATE api_keys SET used_tokens used_tokens ? WHERE id ?, (total_tokens, key_info[id]), ) conn.execute( INSERT INTO billing_records (api_key_id, user_id, model, request_id, prompt_tokens, completion_tokens, total_tokens, amount) VALUES (?, ?, ?, ?, ?, ?, ?, ?), ( key_info[id], key_info[user_id], req.model, request_id, result[prompt_tokens], result[completion_tokens], total_tokens, amount, ), ) conn.commit() except Exception as e: conn.rollback() raise HTTPException(status_code500, detailf账单写入失败: {e}) finally: conn.close() return { id: request_id, model: req.model, choices: [ { message: { role: assistant, content: result[content], } } ], usage: { prompt_tokens: result[prompt_tokens], completion_tokens: result[completion_tokens], total_tokens: total_tokens, }, amount: amount, }这个接口的流程可以理解为先鉴权再调用模型统计 Token检查配额算费用写账单最后返回结果。其中最关键的是“配额检查和扣减放在同一个事务里”这样可以避免高并发下多个请求同时查到剩余配额仍足够最后却把配额打穿的问题。SQLite 在并发性能上有短板但事务逻辑在生产数据库上是一样的。4.4 启动服务并运行验证在项目目录下执行uvicorn main:app --host 0.0.0.0 --port 8000启动成功后可以看到类似下面的日志INFO: Uvicorn running on http://0.0.0.0:8000此时你可以打开 http://127.0.0.1:8000/docs 查看 FastAPI 自动生成的 Swagger 文档。第一步先创建一个 API Key。可以在 Swagger 里调用 /admin/api_keys或者直接用 curlcurl -X POST http://127.0.0.1:8000/admin/api_keys \ -H Content-Type: application/json \ -d {user_id: user001, quota_tokens: 1000}返回示例{ api_key: ak_xxxxx, user_id: user001, quota_tokens: 1000 }拿到 Key 后调用对话接口curl -X POST http://127.0.0.1:8000/v1/chat/completions \ -H Content-Type: application/json \ -H Authorization: Bearer ak_xxxxx \ -d {model: demo-chat, messages: [{role: user, content: 你好}]}返回示例{ id: xxx-xxx-xxx, model: demo-chat, choices: [ { message: { role: assistant, content: 你好这是一段模拟的大模型回复。 } } ], usage: { prompt_tokens: 3, completion_tokens: 11, total_tokens: 14 }, amount: 1 }此时再次请求同一个接口会发现 used_tokens 累加。如果总消耗超过 quota_tokens服务会返回 402 配额不足。4.5 结果说明通过这个最小实现我们已经把“收银台”的主要流程跑通了。你可以看到API Key 是进入收银台的凭证。每次请求都会产生一次 Token 统计。配额扣减发生在返回结果之前。每笔调用都有独立账单记录。这套逻辑和大厂 AI 平台的计费网关在核心流程上是一致的。差别主要在规模、并发、精度和容错层面。5. 进阶设计让收银台更接近生产环境5.1 流式响应的计费处理真实大模型接口通常支持 stream 模式。模型不是一次性返回完整结果而是像打字机一样一个个 Token 输出。流式模式的好处是首字延迟低用户体验好。但对计费来说流式响应带来了一个麻烦什么时候算钱常见做法是客户端在请求中带上 streamtrue服务端使用 SSE 或流式 HTTP 返回。在流结束时模型服务会返回最终的 usage 统计。此时网关再更新配额和写入账单。如果你的业务不要求实时扣减可以在流结束时统一结算。但为了避免请求结束后客户端已经断开、结算消息丢失网关侧往往还会设计一个兜底任务定期扫描未结算的流重新计算。简单项目中更稳妥的做法是网关在接到完整请求时就快速估算一个预授权 Token 数先从配额中冻结待真实用量返回后再进行调整。5.2 多模型路由与动态计价大厂 AI 平台往往有多个模型不同模型的 Token 单价不一样。如果项目未来要接入更多模型建议把“价格表”做成可配置项甚至放到数据库或配置中心里而不是硬编码在 Python 文件中。一个简单的模型注册表可以这样设计MODEL_REGISTRY { demo-chat: { client: call_mock_model, price_per_1k_tokens: 2, }, demo-embedding: { client: call_mock_embedding, price_per_1k_tokens: 1, }, }调用时先从 MODL_REGISTRY 中取出对应的处理函数和价格再做逻辑分发。这样新增模型时只需要注册一个配置不需要改动主流程。5.3 幂等与防重支付系统最怕重复扣款AI 计费也一样。当客户端因为网络超时重试同一个请求时服务端不能把它当成两笔新调用否则用户会被多扣费。解决办法是引入幂等键。客户端在请求头里传一个 X-Request-ID网关在处理前先查账单表如果这个 ID 已经存在就直接返回上一次结果不重复扣费。在我们的示例中request_id 是服务端生成的真实生产环境应该允许客户端传入并做唯一索引约束。如果在 billing_records 表上给 request_id 加唯一索引那么即使在并发情况下重复插入也会被数据库拒绝从根源上避免重复扣费。5.4 成本失控保护预算保护是很多企业接入大模型 API 时最关心的问题。某个用户代码里写了一个死循环或者一个 for 循环无限调用模型几个小时就能把预算烧光。生产环境通常有三种保护手段第一种是单请求限制。单次请求消耗 Token 数不能超过阈值超过直接拒绝。第二种是单 Key 每日预算。按天累计费用达到设定金额后自动禁用 Key 或进入限流状态。第三种是全局熔断。当整个平台的错误率或费用异常上升时网关自动开启熔断停止转发新的请求。这些机制都属于“前置控制”比事后看账单再补救要有效得多。6. 常见问题与排查思路6.1 调用成功但账单没记录如果模型返回了结果但 billing_records 表里查不到记录最常见的原因是计费逻辑放在了模型调用之后而模型调用抛了异常函数提前返回。也可能是账单写入和扣减不在同一个事务里导致只有一部分操作成功。排查时可以重点看日志中是否出现“账单写入失败”或类似的异常信息。修复思路是确保配额扣减和账单插入在同一个数据库事务内任何一步失败都整体回滚。问题现象常见原因解决思路调用成功但账单没记录计费逻辑被异常中断检查异常日志统一事务重复请求被多次扣费缺少幂等控制增加请求唯一 ID 和唯一索引用户配额被超额消耗配额检查和扣减非原子使用UPDATE原子扣减或事务锁流式请求结束后费用未更新结算时机不对在流结束回调或兜底任务中结算6.2 并发下配额扣减不准在高并发场景下如果先 SELECT 查出 used_tokens再把计算结果 UPDATE 回去会出现竞态条件。两个请求同时读到 used_tokens100同时写入 110结果少扣了 20。正确的做法是使用原子 UPDATE例如UPDATE api_keys SET used_tokens used_tokens ? WHERE id ? AND used_tokens ? quota_tokens如果更新影响的行数为 0说明配额不足可以返回 402。这种方式把检查与扣减合并成一条 SQL避免了并发导致的多扣或超扣。6.3 模型超时导致误扣费模型服务可能因为负载过高出现响应超时。如果请求最终失败了就不应该向用户收费。实现上需要注意只有模型成功返回 usage 信息或者已经产生输出内容时才执行计费。如果模型调用抛超时异常应当直接返回 5xx不写账单。如果要更精细地处理“部分输出成功”的场景还需要根据模型返回的 completion_tokens 精度来判断。总之不能让网关在业务失败时仍然扣费否则用户信任会很快崩塌。6.4 API Key 泄露与滥用API Key 泄露是 AI 平台最常见的风险之一。开发者习惯把 Key 写在前端代码、Git 仓库或环境变量里一旦泄露攻击者就可以消耗你的配额和余额。平台侧可以做的有限制 Key 的访问 IP、设置每日消费上限、支持 Key 的即时禁用和轮换、推送异常消费告警。更推荐在存储层使用哈希保存 Key这样即使数据库被拖走攻击者也无法直接使用原文 Key 刷接口。问题现象常见原因解决思路Key 被陌生人使用明文写在客户端或仓库立即禁用并轮换 Key某天费用突然飙升没有预算告警设置单日消费阈值并通知数据库泄露但 Key 仍可用明文存储 Key使用哈希存储 Key某个模型被大量调用权限控制过粗按模型拆分权限和配额7. 最佳实践与工程建议7.1 数据安全是收银台的底线AI 计费网关涉及资金和用户数据安全优先级非常高。API Key 必须哈希存储建议用带随机盐的哈希算法账单中的用户 ID 和请求内容需要做脱敏处理对外接口要通过网关层加 Web 应用防火墙防止恶意请求直接打到业务逻辑。如果平台允许用户传入合法内容还需要考虑内容合规审核。虽然这不是计费系统本身的职责但它会影响请求是否能被放行。7.2 计费与业务解耦在实际项目中AI 计费网关不应该和具体业务逻辑耦合太深。模型调用、配额扣减、账单写入可以抽象成独立的服务或模块。这样即使后续更换模型服务商或者增加新的计费维度都不需要改动主业务代码。模块之间通过明确的接口通信例如网关只负责接收请求、鉴权、转发、计费不关心上层应用是客服机器人还是内容生成工具。7.3 可观测性建设一个没有日志和监控的收银台是在裸奔。至少需要监控以下指标每秒请求数QPS请求成功率平均响应时间Token 消耗速率配额不足触发次数账单写入失败次数这些指标可以通过 Prometheus 采集配合 Grafana 展示。日志方面要有全局 Request ID从请求进入网关到模型调用、计费完成全程携带同一个 ID便于问题追踪。7.4 生产环境替换与扩展本文使用了 SQLite 做演示生产环境建议替换为 MySQL 或 PostgreSQL。替换时注意三点表结构中的 TEXT 长度要符合业务需要对 api_key、request_id 增加唯一索引账单表建议按月或按天分表避免单表数据量过大。模型调用部分建议封装成独立 SDK 或客户端库把请求重试、超时、流式解析等逻辑收敛到一处。每个模型服务的超时设置也要单独配置因为不同模型的响应速度差异很大。更进一步你可以把配额管理从简单的 Token 总数扩展为余额扣费即用户先充值每个请求按金额扣款。底层逻辑和 Token 计费类似只是把“检查 quota_tokens”换成“检查 balance”把“扣除 Token”换成“扣除金额”。理解了本文的示例再去做余额系统会容易很多。如果你想继续深入可以先从本文第 4 节的最小网关跑起来再逐步加入第 5 节的流式计费和幂等机制。你会发现所谓大厂 AI 收银台本质上就是一套严谨的后端计费系统把模型能力像商品一样安全、准确地卖给每一个调用方。
用Python实现AI计费网关：API Key鉴权与Token配额扣减全解析
最近一两年大模型从“能聊天的演示工具”快速变成了“可被应用调用的基础设施”。很多大厂 AI 平台陆续开放 API按 Token 计费、按调用量结算开发者只要申请一个 Key就能在自己的应用里接入对话、推理、向量化等能力。这个模式看起来和普通云服务很像但如果从后端视角去看它其实相当于给每个接入 AI 的应用装了一个“新收银台”谁来调用、用了多少、该扣多少钱、钱够不够、会不会被刷爆全部要在毫秒级完成判断。这篇文章我想从工程实现的角度拆解大厂 AI 背后的“收银台”逻辑。我们会先梳理核心概念再用 Python FastAPI 从零实现一个最小可运行的 AI 计费网关包含 API Key 鉴权、Token 统计、配额扣减、账单落库等关键能力。最后再聊一聊生产环境里的流式计费、幂等、预算保护、安全边界等进阶问题。内容适合有 Python 基础、想了解 AI 应用后端实现或者正在做 AI 平台相关开发的读者。跟着文章跑完一遍你会对“大厂 AI 是如何按量收费的”有一个非常具体的认识。1. 背景大厂 AI 与新“收银台”到底是什么1.1 从模型能力到可计费服务最早接触大模型时多数人的印象是“一个网页对话框”。你输入一句话它回你一段话看起来和搜索引擎差不多。但真正的商业级大模型并不是以网页为核心而是以 API 为核心。开发者拿到一个 API Key把对话请求发送到模型服务模型返回结果服务端在返回时附带本次请求消耗了多少 Token。Token 就是大模型世界的“字数”也是计费的依据。这种模式带来的变化非常明显模型能力变成了可以被任意业务调用的函数而不是一个单机工具。比如客服系统可以调对话模型搜索系统可以调向量模型内容平台可以调生成模型。每一次调用本质上都是一次“结账”模型的算力成本被折算成 Token 单价再乘以调用量就变成了大厂 AI 平台的收入。这个过程非常像线下超市的收银台你拿商品请求收银员扫码计费结账扣费小票账单。1.2 AI 收银台的四个核心层从系统架构上看一个大厂 AI 平台的“收银台”通常包含四层第一层是接入层。开发者通过 API 网关把请求送到平台网关负责 SSL、路由、限流、鉴权。开发者拿到的 API Key 在这一层被识别。第二层是模型服务层。请求经过网关后被转发给不同的模型推理服务。这里会涉及模型选择、上下文拼接、流式输出、超时控制等逻辑。第三层是计费层。模型服务返回后系统需要拿到本次请求的 Token 用量根据模型单价计算金额并更新开发者的配额或余额。这是“收银台”最核心的部分。第四层是数据观测层。每笔调用要有记录包括用户 ID、模型、Token 用量、金额、耗时、返回状态。这些数据既用于账单查询也用于成本分析、告警和风控。很多开发者以为接入大模型 API 只是发一个 HTTP 请求那么简单其实平台侧要保证计费准确、并发可控、数据可靠远不是写一个接口转发就够了。1.3 从本文能学到什么本文不会只停留在概念层面。后面我会带着你实现一个简化版 AI 计费网关重点展示三块内容API Key 如何生成和校验Token 用量如何统计并转换成费用配额如何扣减以及如何防止超用。这三块理解后再看大厂 AI 平台的官方文档和计费规则你会有一种“原来是这么实现的”感觉。2. 核心概念API Key、Token 与配额2.1 API Key 是收银台的会员卡用过支付宝、微信支付的开发者都知道每次支付需要有账号、有支付凭证。在 AI 平台里API Key 承担了类似“支付凭证”的角色。调用方在请求头里带上自己的 Key平台根据 Key 判断你是谁、你有没有权限、你账户里还有没有额度。从安全角度讲API Key 本质上是敏感信息。真实生产系统中数据库不会明文保存 Key而是保存 Key 的哈希值。当平台生成 API Key 时明文只展示一次之后再也无法从后台查看。这样即使数据库泄露攻击者拿到的也只是一串不可逆的哈希值。在我们的示例系统中为了先跑通流程会先使用明文 Key但在最佳实践章节会重点说明哈希存储的必要性。2.2 Token 是 AI 世界的计价单位Token 可以通俗理解为“模型看文本的最小单位”。英文文本中 1 个 Token 大约对应 0.75 个单词中文文本中 1 个 Token 可能对应半个到一个汉字。大模型在处理文本时不是按字符数计算的而是按 Token 数计算的。为什么按 Token 计费因为 Token 数量直接对应模型的计算量。模型生成的 Token 越多占用的算力越大推理时间越长成本也越高。所以平台在返回结果时通常会附带类似这样的信息{ usage: { prompt_tokens: 32, completion_tokens: 45, total_tokens: 77 } }其中 prompt_tokens 是输入文本的 Token 数completion_tokens 是模型生成的 Token 数total_tokens 是两者之和。需要说明的是Token 数的真正计算并不是简单用字符串长度除以 2而是由模型的 Tokenizer 完成。不同模型有不同 Tokenizer同一段文本在不同模型下 Token 数也可能不同。生产环境中计费应当以模型服务返回的 usage 字段为准而不是自己在网关层用 len(text) 估算。本文示例为了简化会采用一个近似估算方法但你一定要清楚这两者的区别。2.3 配额与限流防止“刷爆”收银台配额指的是一个 API Key 最大可以消耗多少 Token或者最大可以花多少钱。它的作用类似于手机套餐里的剩余流量。每次请求完成后系统把本次消耗的 Token 累加到已用额度上然后判断是否超过总配额。限流则是一个前置保护。限流可以在请求进入模型服务前拦截超高 QPS避免一个 Key 突发大量请求把服务打满。常见的限流算法有固定窗口、滑动窗口、令牌桶等。计费网关中配额和限流是两套独立但又需要配合的机制配额管总量限流管速率。3. 整体架构与项目准备3.1 技术选型与运行环境本文示例采用 Python 语言原因是大模型生态中 Python 最为通用代码表达也直观。Web 框架选择 FastAPI它支持异步、自动生成接口文档并且和第三方生态配合非常方便。数据库选择 SQLiteSQLite 不需要额外安装服务端适合做最小演示。生产环境可以平滑替换成 MySQL 或 PostgreSQL。运行环境建议如下Python 3.9 或以上版本FastAPIUvicorn系统自带 sqlite3依赖可以按下面方式安装pip install fastapi uvicorn pydantic不要刻意追求最新版本安装时以 pip 自动解析出的稳定版本为准。3.2 项目结构为了保持代码易读我把系统拆成几个文件ai-billing-gateway/ ├── main.py # FastAPI 应用与接口路由 ├── db.py # SQLite 初始化和连接工具 ├── auth.py # API Key 生成与校验 └── requirements.txt # 依赖如果你在练习时想改名文件完全没问题只要保证模块导入路径一致即可。3.3 数据库设计数据库一共需要两张核心表。第一张是 API Key 表记录每个用户或应用的 Key、配额和使用量。第二张是账单记录表每笔调用生成一条记录。字段设计如下api_keys 表字段类型说明idINTEGER主键api_keyTEXT唯一 Keyuser_idTEXT所属用户quota_tokensINTEGER总配额used_tokensINTEGER已用 Token 数statusTEXT状态active / disabledbilling_records 表字段类型说明idINTEGER主键api_key_idINTEGER关联的 Key IDuser_idTEXT所属用户modelTEXT模型名称request_idTEXT请求幂等 IDprompt_tokensINTEGER输入 Tokencompletion_tokensINTEGER输出 Tokentotal_tokensINTEGER总 TokenamountINTEGER扣费金额单位自定义两张表的关系很简单一个 API Key 会产生多条账单记录。我们通过 api_key_id 关联查询就能统计某个 Key 某一段时间内的总消耗。4. 完整实战从零实现一个 AI 计费网关4.1 初始化数据库先写 db.py负责创建数据库和连接。代码里使用 sqlite3 自带的 executescript 创建表结构并设置 row_factory方便后续把行对象直接转成字典使用。# db.py import os import sqlite3 DB_PATH os.environ.get(AI_BILLING_DB, ai_billing.db) def get_conn(): conn sqlite3.connect(DB_PATH) conn.row_factory sqlite3.Row return conn def init_db(): conn get_conn() conn.executescript( CREATE TABLE IF NOT EXISTS api_keys ( id INTEGER PRIMARY KEY AUTOINCREMENT, api_key TEXT NOT NULL UNIQUE, user_id TEXT NOT NULL, quota_tokens INTEGER NOT NULL DEFAULT 1000000, used_tokens INTEGER NOT NULL DEFAULT 0, status TEXT NOT NULL DEFAULT active, created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP ); CREATE TABLE IF NOT EXISTS billing_records ( id INTEGER PRIMARY KEY AUTOINCREMENT, api_key_id INTEGER NOT NULL, user_id TEXT NOT NULL, model TEXT NOT NULL, request_id TEXT NOT NULL, prompt_tokens INTEGER NOT NULL DEFAULT 0, completion_tokens INTEGER NOT NULL DEFAULT 0, total_tokens INTEGER NOT NULL DEFAULT 0, amount INTEGER NOT NULL DEFAULT 0, created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP ); ) conn.commit() conn.close()这里需要注意SQLite 默认把整张数据库文件当成自己的“数据库”不需要额外创建 database 名。设置 DB_PATH 为环境变量是为了方便你在测试时切换数据库文件。4.2 生成 API Key接着写 auth.py。生成 Key 时我们用 secrets.token_urlsafe 生成一段随机字符串并且加一个 ak_ 前缀做标识。真实生产环境里这里应该只保存 Key 的哈希值后面我会再提。# auth.py import secrets from fastapi import Request, HTTPException from db import get_conn def generate_api_key(): return ak_ secrets.token_urlsafe(32) def save_api_key(api_key, user_id, quota_tokens): conn get_conn() conn.execute( INSERT INTO api_keys (api_key, user_id, quota_tokens) VALUES (?, ?, ?), (api_key, user_id, quota_tokens), ) conn.commit() conn.close() def verify_api_key(request: Request): auth_header request.headers.get(Authorization, ) if not auth_header.startswith(Bearer ): raise HTTPException(status_code401, detail缺少 API Key) api_key auth_header[len(Bearer ):].strip() conn get_conn() row conn.execute( SELECT * FROM api_keys WHERE api_key ? AND status active, (api_key,), ).fetchone() conn.close() if row is None: raise HTTPException(status_code401, detailAPI Key 无效或已禁用) return dict(row)verify_api_key 是核心鉴权函数它会从请求头中取出 API Key然后查询数据库。如果 Key 不存在或状态不是 active直接返回 401。这样每个需要登录的接口都可以复用这个函数。4.3 实现模型调用与计费main.py 是整个网关的主体。为了不依赖真实的外部模型服务我们内置一个模拟模型客户端。真实场景中这里应该替换成对官方模型的 HTTP 调用。计费部分使用一个 mock 单价表单位为“分 / 千 Token”。例如PRICING { demo-chat: {price_per_1k_tokens: 2}, demo-embedding: {price_per_1k_tokens: 1}, }金额计算逻辑是 total_tokens * 单价 / 1000这样总 Token 越大费用越高。# main.py import uuid from typing import Optional from fastapi import FastAPI, HTTPException, Request from pydantic import BaseModel from auth import generate_api_key, save_api_key, verify_api_key from db import init_db, get_conn app FastAPI(titleAI Billing Gateway) PRICING { demo-chat: {price_per_1k_tokens: 2}, demo-embedding: {price_per_1k_tokens: 1}, } class ChatRequest(BaseModel): model: str messages: list stream: Optional[bool] False class CreateKeyRequest(BaseModel): user_id: str quota_tokens: int 1000000 def call_mock_model(messages): 模拟模型推理。真实场景替换为模型服务调用。 prompt_text for msg in messages: prompt_text msg.get(content, ) reply_text 你好这是一段模拟的大模型回复。 prompt_tokens max(1, len(prompt_text) // 2) completion_tokens max(1, len(reply_text) // 2) return { content: reply_text, prompt_tokens: prompt_tokens, completion_tokens: completion_tokens, } def calc_amount(model: str, total_tokens: int) - int: price PRICING.get(model, {}).get(price_per_1k_tokens, 1) return max(1, total_tokens * price // 1000) app.on_event(startup) def on_startup(): init_db() app.post(/admin/api_keys, response_modeldict) def create_api_key_route(req: CreateKeyRequest): api_key generate_api_key() save_api_key(api_key, req.user_id, req.quota_tokens) return { api_key: api_key, user_id: req.user_id, quota_tokens: req.quota_tokens, } app.post(/v1/chat/completions, response_modeldict) def chat_completions(req: ChatRequest, request: Request): key_info verify_api_key(request) result call_mock_model(req.messages) total_tokens result[prompt_tokens] result[completion_tokens] if key_info[used_tokens] total_tokens key_info[quota_tokens]: raise HTTPException(status_code402, detail配额不足请充值) amount calc_amount(req.model, total_tokens) request_id str(uuid.uuid4()) conn get_conn() try: conn.execute( UPDATE api_keys SET used_tokens used_tokens ? WHERE id ?, (total_tokens, key_info[id]), ) conn.execute( INSERT INTO billing_records (api_key_id, user_id, model, request_id, prompt_tokens, completion_tokens, total_tokens, amount) VALUES (?, ?, ?, ?, ?, ?, ?, ?), ( key_info[id], key_info[user_id], req.model, request_id, result[prompt_tokens], result[completion_tokens], total_tokens, amount, ), ) conn.commit() except Exception as e: conn.rollback() raise HTTPException(status_code500, detailf账单写入失败: {e}) finally: conn.close() return { id: request_id, model: req.model, choices: [ { message: { role: assistant, content: result[content], } } ], usage: { prompt_tokens: result[prompt_tokens], completion_tokens: result[completion_tokens], total_tokens: total_tokens, }, amount: amount, }这个接口的流程可以理解为先鉴权再调用模型统计 Token检查配额算费用写账单最后返回结果。其中最关键的是“配额检查和扣减放在同一个事务里”这样可以避免高并发下多个请求同时查到剩余配额仍足够最后却把配额打穿的问题。SQLite 在并发性能上有短板但事务逻辑在生产数据库上是一样的。4.4 启动服务并运行验证在项目目录下执行uvicorn main:app --host 0.0.0.0 --port 8000启动成功后可以看到类似下面的日志INFO: Uvicorn running on http://0.0.0.0:8000此时你可以打开 http://127.0.0.1:8000/docs 查看 FastAPI 自动生成的 Swagger 文档。第一步先创建一个 API Key。可以在 Swagger 里调用 /admin/api_keys或者直接用 curlcurl -X POST http://127.0.0.1:8000/admin/api_keys \ -H Content-Type: application/json \ -d {user_id: user001, quota_tokens: 1000}返回示例{ api_key: ak_xxxxx, user_id: user001, quota_tokens: 1000 }拿到 Key 后调用对话接口curl -X POST http://127.0.0.1:8000/v1/chat/completions \ -H Content-Type: application/json \ -H Authorization: Bearer ak_xxxxx \ -d {model: demo-chat, messages: [{role: user, content: 你好}]}返回示例{ id: xxx-xxx-xxx, model: demo-chat, choices: [ { message: { role: assistant, content: 你好这是一段模拟的大模型回复。 } } ], usage: { prompt_tokens: 3, completion_tokens: 11, total_tokens: 14 }, amount: 1 }此时再次请求同一个接口会发现 used_tokens 累加。如果总消耗超过 quota_tokens服务会返回 402 配额不足。4.5 结果说明通过这个最小实现我们已经把“收银台”的主要流程跑通了。你可以看到API Key 是进入收银台的凭证。每次请求都会产生一次 Token 统计。配额扣减发生在返回结果之前。每笔调用都有独立账单记录。这套逻辑和大厂 AI 平台的计费网关在核心流程上是一致的。差别主要在规模、并发、精度和容错层面。5. 进阶设计让收银台更接近生产环境5.1 流式响应的计费处理真实大模型接口通常支持 stream 模式。模型不是一次性返回完整结果而是像打字机一样一个个 Token 输出。流式模式的好处是首字延迟低用户体验好。但对计费来说流式响应带来了一个麻烦什么时候算钱常见做法是客户端在请求中带上 streamtrue服务端使用 SSE 或流式 HTTP 返回。在流结束时模型服务会返回最终的 usage 统计。此时网关再更新配额和写入账单。如果你的业务不要求实时扣减可以在流结束时统一结算。但为了避免请求结束后客户端已经断开、结算消息丢失网关侧往往还会设计一个兜底任务定期扫描未结算的流重新计算。简单项目中更稳妥的做法是网关在接到完整请求时就快速估算一个预授权 Token 数先从配额中冻结待真实用量返回后再进行调整。5.2 多模型路由与动态计价大厂 AI 平台往往有多个模型不同模型的 Token 单价不一样。如果项目未来要接入更多模型建议把“价格表”做成可配置项甚至放到数据库或配置中心里而不是硬编码在 Python 文件中。一个简单的模型注册表可以这样设计MODEL_REGISTRY { demo-chat: { client: call_mock_model, price_per_1k_tokens: 2, }, demo-embedding: { client: call_mock_embedding, price_per_1k_tokens: 1, }, }调用时先从 MODL_REGISTRY 中取出对应的处理函数和价格再做逻辑分发。这样新增模型时只需要注册一个配置不需要改动主流程。5.3 幂等与防重支付系统最怕重复扣款AI 计费也一样。当客户端因为网络超时重试同一个请求时服务端不能把它当成两笔新调用否则用户会被多扣费。解决办法是引入幂等键。客户端在请求头里传一个 X-Request-ID网关在处理前先查账单表如果这个 ID 已经存在就直接返回上一次结果不重复扣费。在我们的示例中request_id 是服务端生成的真实生产环境应该允许客户端传入并做唯一索引约束。如果在 billing_records 表上给 request_id 加唯一索引那么即使在并发情况下重复插入也会被数据库拒绝从根源上避免重复扣费。5.4 成本失控保护预算保护是很多企业接入大模型 API 时最关心的问题。某个用户代码里写了一个死循环或者一个 for 循环无限调用模型几个小时就能把预算烧光。生产环境通常有三种保护手段第一种是单请求限制。单次请求消耗 Token 数不能超过阈值超过直接拒绝。第二种是单 Key 每日预算。按天累计费用达到设定金额后自动禁用 Key 或进入限流状态。第三种是全局熔断。当整个平台的错误率或费用异常上升时网关自动开启熔断停止转发新的请求。这些机制都属于“前置控制”比事后看账单再补救要有效得多。6. 常见问题与排查思路6.1 调用成功但账单没记录如果模型返回了结果但 billing_records 表里查不到记录最常见的原因是计费逻辑放在了模型调用之后而模型调用抛了异常函数提前返回。也可能是账单写入和扣减不在同一个事务里导致只有一部分操作成功。排查时可以重点看日志中是否出现“账单写入失败”或类似的异常信息。修复思路是确保配额扣减和账单插入在同一个数据库事务内任何一步失败都整体回滚。问题现象常见原因解决思路调用成功但账单没记录计费逻辑被异常中断检查异常日志统一事务重复请求被多次扣费缺少幂等控制增加请求唯一 ID 和唯一索引用户配额被超额消耗配额检查和扣减非原子使用UPDATE原子扣减或事务锁流式请求结束后费用未更新结算时机不对在流结束回调或兜底任务中结算6.2 并发下配额扣减不准在高并发场景下如果先 SELECT 查出 used_tokens再把计算结果 UPDATE 回去会出现竞态条件。两个请求同时读到 used_tokens100同时写入 110结果少扣了 20。正确的做法是使用原子 UPDATE例如UPDATE api_keys SET used_tokens used_tokens ? WHERE id ? AND used_tokens ? quota_tokens如果更新影响的行数为 0说明配额不足可以返回 402。这种方式把检查与扣减合并成一条 SQL避免了并发导致的多扣或超扣。6.3 模型超时导致误扣费模型服务可能因为负载过高出现响应超时。如果请求最终失败了就不应该向用户收费。实现上需要注意只有模型成功返回 usage 信息或者已经产生输出内容时才执行计费。如果模型调用抛超时异常应当直接返回 5xx不写账单。如果要更精细地处理“部分输出成功”的场景还需要根据模型返回的 completion_tokens 精度来判断。总之不能让网关在业务失败时仍然扣费否则用户信任会很快崩塌。6.4 API Key 泄露与滥用API Key 泄露是 AI 平台最常见的风险之一。开发者习惯把 Key 写在前端代码、Git 仓库或环境变量里一旦泄露攻击者就可以消耗你的配额和余额。平台侧可以做的有限制 Key 的访问 IP、设置每日消费上限、支持 Key 的即时禁用和轮换、推送异常消费告警。更推荐在存储层使用哈希保存 Key这样即使数据库被拖走攻击者也无法直接使用原文 Key 刷接口。问题现象常见原因解决思路Key 被陌生人使用明文写在客户端或仓库立即禁用并轮换 Key某天费用突然飙升没有预算告警设置单日消费阈值并通知数据库泄露但 Key 仍可用明文存储 Key使用哈希存储 Key某个模型被大量调用权限控制过粗按模型拆分权限和配额7. 最佳实践与工程建议7.1 数据安全是收银台的底线AI 计费网关涉及资金和用户数据安全优先级非常高。API Key 必须哈希存储建议用带随机盐的哈希算法账单中的用户 ID 和请求内容需要做脱敏处理对外接口要通过网关层加 Web 应用防火墙防止恶意请求直接打到业务逻辑。如果平台允许用户传入合法内容还需要考虑内容合规审核。虽然这不是计费系统本身的职责但它会影响请求是否能被放行。7.2 计费与业务解耦在实际项目中AI 计费网关不应该和具体业务逻辑耦合太深。模型调用、配额扣减、账单写入可以抽象成独立的服务或模块。这样即使后续更换模型服务商或者增加新的计费维度都不需要改动主业务代码。模块之间通过明确的接口通信例如网关只负责接收请求、鉴权、转发、计费不关心上层应用是客服机器人还是内容生成工具。7.3 可观测性建设一个没有日志和监控的收银台是在裸奔。至少需要监控以下指标每秒请求数QPS请求成功率平均响应时间Token 消耗速率配额不足触发次数账单写入失败次数这些指标可以通过 Prometheus 采集配合 Grafana 展示。日志方面要有全局 Request ID从请求进入网关到模型调用、计费完成全程携带同一个 ID便于问题追踪。7.4 生产环境替换与扩展本文使用了 SQLite 做演示生产环境建议替换为 MySQL 或 PostgreSQL。替换时注意三点表结构中的 TEXT 长度要符合业务需要对 api_key、request_id 增加唯一索引账单表建议按月或按天分表避免单表数据量过大。模型调用部分建议封装成独立 SDK 或客户端库把请求重试、超时、流式解析等逻辑收敛到一处。每个模型服务的超时设置也要单独配置因为不同模型的响应速度差异很大。更进一步你可以把配额管理从简单的 Token 总数扩展为余额扣费即用户先充值每个请求按金额扣款。底层逻辑和 Token 计费类似只是把“检查 quota_tokens”换成“检查 balance”把“扣除 Token”换成“扣除金额”。理解了本文的示例再去做余额系统会容易很多。如果你想继续深入可以先从本文第 4 节的最小网关跑起来再逐步加入第 5 节的流式计费和幂等机制。你会发现所谓大厂 AI 收银台本质上就是一套严谨的后端计费系统把模型能力像商品一样安全、准确地卖给每一个调用方。
