Python构建AI产品计费系统：Token成本模型与Stripe支付集成

1. 为什么“能跑通的 AI 产品”和“能赚钱的 AI 产品”是两回事

我见过太多团队，模型调得飞起，Demo 演示惊艳全场，上线三个月后一算账，每服务一个用户净亏两毛。问题从来不在模型本身，而在于从第一天起就没把成本模型和计费系统当成产品的一等公民来设计。你调用一次大模型 API，消耗的是真金白银的 Token；用户点一次生成按钮，背后可能是几厘到几块钱的成本波动。如果计费逻辑是拍脑袋定的，或者干脆后补，那这个产品从架构层面就是亏损的。

这篇内容要聊的，就是怎么用 Python 把“AI 能力”包装成一个可计量、可计费、可盈利的产品。核心链路包括三块：第一，搞清楚你的成本到底花在哪里，建立 Token 级别的成本模型；第二，设计一套能扛住并发、对账清晰的计费系统，这里会重点用到 Stripe 做支付闭环；第三，把用户认证和额度控制串起来，让每一次 API 调用都有迹可循、有钱可收。适合谁看？如果你正在做 AI 应用、AI Agent、或者任何按量付费的 SaaS 工具，并且希望它最终能自负盈亏，那这篇就是写给你的。我会给出可直接复用的代码结构和参数计算过程，不是泛泛而谈的架构图。

2. 成本模型：先把账算清楚，再谈赚钱

2.1 Token 成本拆解：输入、输出与隐藏开销

大模型 API 的计费单位是 Token，但很多人只盯着“输出 Token 单价”，忽略了输入 Token 往往才是大头。以常见的对话场景为例，每次请求你都要把系统提示词、历史对话、用户当前问题一起塞进去，这些全是输入 Token。假设系统提示词 500 Token，历史对话保留最近 10 轮，每轮平均 200 Token，那单次请求的输入就是 500 + 2000 + 用户问题 ≈ 2600 Token。输出可能只有 300 Token。如果输入单价是输出单价的 1/3，那输入成本反而是输出的近 3 倍。

我一般会建一个成本计算表，把每个模型的输入单价、输出单价、上下文窗口限制列清楚。下面是一个简化版的 Python 成本估算函数，你可以直接拿去改：

# 成本模型：按 Token 用量估算单次请求成本（单位：元） MODEL_PRICING = { "model-a": {"input": 0.0008, "output": 0.002}, # 每 1K Token 价格 "model-b": {"input": 0.0012, "output": 0.0036}, } def estimate_cost(model: str, input_tokens: int, output_tokens: int) -> float: pricing = MODEL_PRICING.get(model) if not pricing: raise ValueError(f"未知模型: {model}") input_cost = (input_tokens / 1000) * pricing["input"] output_cost = (output_tokens / 1000) * pricing["output"] return round(input_cost + output_cost, 6)

这个函数看起来简单，但它是整个计费系统的地基。你需要在每次 API 调用返回后，从响应里拿到真实的usage字段，把prompt_tokens和completion_tokens分别传入，算出这次调用的实际成本。注意，有些模型还会对“缓存命中”的输入 Token 给折扣，如果你的业务有大量重复前缀，这个折扣要单独建模，否则成本会算高。

2.2 定价策略：成本加成还是价值定价

算出成本之后，定价就是下一个决策点。我见过两种极端：一种是成本加成，比如成本 0.01 元，加价 30% 卖 0.013 元，结果用户用一次付一次，客单价低到支付通道费都不够；另一种是拍脑袋定个“9.9 元包月”，结果重度用户一个月跑出 200 元成本，直接亏穿。

我的经验是：按量计费的产品，定价必须覆盖“最坏情况成本”加上支付损耗和运营分摊。具体做法是，先统计你用户群体的 Token 消耗分布，找到 P95 或 P99 的用量，确保你的单价在这个用量下仍有毛利。同时，设置一个“公平使用上限”，比如包月套餐里包含 100 万 Token，超出部分按量续费。这样既给了用户确定性，又防止了极端滥用。

这里有个关键参数：支付通道费率。Stripe 的标准费率大约是 2.9% + 0.3 美元每笔。如果你做的是小额高频计费，比如每次生成收 0.05 元，那支付费率占比会高得离谱。所以计费系统通常采用“预充值 + 按量扣减”模式，用户先充 10 元、50 元，后续调用从余额里扣，这样支付笔数大幅减少，费率摊薄。

2.3 成本监控：别等月底账单来了才后悔

成本模型不是算一次就完事，它需要实时监控。我会在每次 API 调用后，把user_id、model、input_tokens、output_tokens、cost、timestamp写进一张流水表。然后每天跑一个聚合任务，看三个指标：单用户日均成本、单次调用平均成本、成本最高的前 10 个用户。如果发现某个用户的成本突然飙升，要么是他在滥用，要么是你的提示词设计有问题导致输入 Token 暴涨。

注意：不要等到月底对账才发现亏损。实时成本监控应该和计费系统共用同一套 Token 计量逻辑，避免“算成本的是一套，扣费的是另一套”这种经典事故。

3. 计费系统设计：从余额扣减到 Stripe 支付闭环

3.1 账户余额模型：预充值 + 实时扣减

计费系统的核心是一张用户余额表和一张扣费流水表。余额表记录当前可用额度，流水表记录每一笔扣减的明细。每次 API 调用前，先检查余额是否大于预估成本；调用后，根据实际 Token 用量扣减。这里有个并发问题：如果同一个用户同时发起多个请求，余额检查可能会通过多次，导致透支。解决办法是用数据库的行级锁或者乐观锁版本号。

我通常用 PostgreSQL 实现，余额扣减的 SQL 大概长这样：

UPDATE user_balance SET balance = balance - :cost, version = version + 1 WHERE user_id = :user_id AND balance >= :cost AND version = :version;

如果affected_rows为 0，说明余额不足或版本冲突，直接拒绝请求。这个逻辑放在 API 网关层，比放在业务代码里更可靠。

3.2 Stripe 支付集成：从 Checkout 到 Webhook

Stripe 的集成分两步：前端创建 Checkout Session，后端处理 Webhook 事件。前端只需要一个按钮，调用后端接口拿到checkout_url，跳转过去完成支付。后端在收到checkout.session.completed事件后，给用户余额充值。

import stripe from flask import Flask, request, jsonify app = Flask(__name__) stripe.api_key = "sk_test_..." @app.route("/create-checkout-session", methods=["POST"]) def create_checkout_session(): data = request.json session = stripe.checkout.Session.create( payment_method_types=["card"], line_items=[{ "price_data": { "currency": "cny", "product_data": {"name": f"充值 {data['amount']} 元"}, "unit_amount": int(data["amount"] * 100), # 单位：分 }, "quantity": 1, }], mode="payment", success_url="https://yourdomain.com/success", cancel_url="https://yourdomain.com/cancel", metadata={"user_id": data["user_id"], "amount": data["amount"]}, ) return jsonify({"checkout_url": session.url})

Webhook 处理是重点，必须验证签名，防止伪造请求：

@app.route("/stripe-webhook", methods=["POST"]) def stripe_webhook(): payload = request.data sig_header = request.headers.get("Stripe-Signature") try: event = stripe.Webhook.construct_event( payload, sig_header, "whsec_..." ) except (ValueError, stripe.error.SignatureVerificationError): return "", 400 if event["type"] == "checkout.session.completed": session = event["data"]["object"] user_id = session["metadata"]["user_id"] amount = float(session["metadata"]["amount"]) # 给用户余额充值，写入流水 credit_balance(user_id, amount, reference=session["id"]) return "", 200

提示：Webhook 必须做幂等处理。Stripe 可能会重复发送同一个事件，如果你不检查session["id"]是否已经处理过，用户就会被重复充值。我一般会在流水表里对reference字段加唯一索引，插入冲突就忽略。

3.3 用量计量与对账：Token 流水怎么记

计费系统最怕的是“扣了钱但不知道扣在哪”。所以每一笔扣费都必须关联到具体的 API 调用记录。我的做法是：每次调用大模型后，生成一个usage_record，包含request_id、user_id、model、input_tokens、output_tokens、cost、balance_after。然后异步写入流水表，同时更新余额。

对账的时候，把 Stripe 的收款记录和余额充值流水对一遍，再把余额扣减流水和 API 调用日志对一遍。如果两边对不上，优先查 Webhook 是否漏处理、余额扣减是否因为并发导致少扣。我踩过的坑是：有一次 Webhook 处理超时，Stripe 重试了三次，但我的代码没有幂等，结果用户被充了三次钱。后来加了唯一索引才解决。

4. 用户认证与额度控制：Token 续签与权限校验

4.1 JWT 实现 Token 登录与续签

AI 产品的用户认证通常用 JWT。登录成功后签发一个access_token（短期，比如 15 分钟）和一个refresh_token（长期，比如 7 天）。access_token用于每次 API 请求的鉴权，过期后用refresh_token换新的。这里的关键是：refresh_token 必须存库，并且可以撤销。否则用户登出后，旧的 refresh_token 还能用，存在安全风险。

import jwt from datetime import datetime, timedelta SECRET_KEY = "your-secret-key" def create_tokens(user_id: str): access_payload = { "user_id": user_id, "exp": datetime.utcnow() + timedelta(minutes=15), "type": "access", } refresh_payload = { "user_id": user_id, "exp": datetime.utcnow() + timedelta(days=7), "type": "refresh", } access_token = jwt.encode(access_payload, SECRET_KEY, algorithm="HS256") refresh_token = jwt.encode(refresh_payload, SECRET_KEY, algorithm="HS256") # 将 refresh_token 存入数据库，关联 user_id save_refresh_token(user_id, refresh_token) return access_token, refresh_token

续签接口收到refresh_token后，先验证签名和过期时间，再查库确认这个 token 没有被撤销，然后签发新的access_token。如果用户登出，直接把库里的refresh_token标记为失效即可。

4.2 额度校验中间件：每次调用前先查余额

认证通过后，下一步是额度校验。我习惯把它做成一个中间件，在业务逻辑之前执行。中间件从 JWT 里拿到user_id，查余额表，如果余额低于某个阈值（比如 0.01 元），直接返回 402 Payment Required，并提示用户充值。如果余额充足，放行到业务逻辑，同时在请求上下文里注入user_id和estimated_cost，方便后续扣费。

这个中间件还要处理“预估成本”和“实际成本”的差异。预估成本可以按最大输出 Token 数来算，实际成本按 API 返回的usage来算。如果实际成本超过预估，且余额不够扣，那就把用户余额扣到 0，并记录一笔欠款，下次充值时优先补扣。

4.3 防滥用：速率限制与异常检测

AI 产品很容易被滥用，尤其是按量计费时，有人会写脚本疯狂调用。除了余额限制，还需要加一层速率限制。我一般用 Redis 做滑动窗口限流，比如每个用户每分钟最多 20 次调用，每天最多 500 次。超过就返回 429 Too Many Requests。

另外，异常检测也很重要。如果某个用户的输入 Token 突然从平均 500 涨到 5000，可能是他在尝试注入攻击或者滥用长上下文。这时候可以触发人工审核，或者临时降低他的速率限制。这些策略不需要太复杂，但必须有，否则你的成本模型会被少数人打穿。

5. 常见问题与排查技巧实录

5.1 Token 计量不准：为什么账单和实际用量对不上

最常见的原因是流式响应。如果你用流式输出，API 返回的usage可能只在最后一个 chunk 里，或者根本不返回。这时候你需要自己估算输出 Token，或者改用非流式调用后再拼接。另一个原因是重试机制：如果调用失败后自动重试，第一次失败的请求可能已经产生了输入 Token 消耗，但你没有记录。解决办法是在重试逻辑里也记录usage，或者把重试的请求单独标记。

5.2 Stripe Webhook 失败：签名验证与幂等处理

Webhook 失败通常有三个原因：签名密钥不对、请求体被中间件修改、没有做幂等。签名验证必须用原始请求体，不能先用 JSON 解析再验证。如果你用了 Nginx 或 Flask 的request.json，请求体可能已经被消费了，导致签名验证失败。正确做法是用request.data拿原始字节。幂等处理就是前面说的唯一索引，简单但有效。

5.3 余额扣减并发问题：行级锁与乐观锁的选择

高并发下，余额扣减必须保证原子性。行级锁（SELECT ... FOR UPDATE）简单直接，但会阻塞其他请求，影响吞吐。乐观锁（版本号）性能更好，但冲突时需要重试。我的建议是：如果单用户并发不高，用行级锁就够了；如果单用户可能同时发起几十个请求，用乐观锁 + 重试队列。无论哪种，都要在扣减失败时返回明确错误，而不是静默失败。

5.4 常见问题速查表

6. 从成本模型到计费系统的完整代码结构

6.1 项目目录与模块划分

我习惯把这类项目分成四个模块：auth（认证）、billing（计费）、usage（用量计量）、api（业务接口）。每个模块独立测试，通过接口调用。目录结构大概是这样：

project/ ├── auth/ │ ├── jwt_handler.py │ └── middleware.py ├── billing/ │ ├── balance.py │ ├── stripe_integration.py │ └── webhook.py ├── usage/ │ ├── cost_model.py │ └── recorder.py ├── api/ │ └── generate.py └── config.py

这样划分的好处是，计费逻辑和业务逻辑解耦，换模型或换支付通道时，只需要改对应模块，不会牵一发动全身。

6.2 关键配置参数与计算过程

在config.py里，我会把模型单价、支付费率、最低充值金额、速率限制阈值都做成配置项。比如：

MODEL_PRICING = { "model-a": {"input": 0.0008, "output": 0.002}, } STRIPE_FEE_RATE = 0.029 STRIPE_FEE_FIXED = 0.30 # 美元 MIN_RECHARGE = 10.0 # 元 RATE_LIMIT_PER_MINUTE = 20

定价时，我会用这个公式算最低售价：售价 = (成本 / (1 - 费率)) + 固定费分摊。假设单次成本 0.01 元，费率 2.9%，固定费 0.3 美元约 2 元，如果每次调用都走支付，那单次售价至少要 2.1 元才能不亏。所以必须预充值，把固定费分摊到多次调用上。

6.3 部署与监控建议

部署时，Webhook 接口必须暴露在公网，并且用 HTTPS。余额扣减和流水记录建议放在同一个数据库事务里，保证一致性。监控方面，我会用 Prometheus 采集三个指标：api_calls_total、token_usage_total、balance_deduction_total。然后配一个告警：如果某小时的成本比收入高 20%，就发通知。这个告警帮我提前发现过好几次定价漏洞。

最后分享一个小技巧：在用户充值成功后的第一个请求里，返回一个“本次调用消耗 X Token，折合 Y 元，余额剩余 Z 元”的提示。用户看到这个，对扣费的感知会清晰很多，客诉率明显下降。这个提示不需要很显眼，放在响应头或者响应体的meta字段里就行。