下面我把 **Phase 1 最关键的「计费系统」** 完整展开,并给你一份**可直接下载/复制的 Markdown 文档**,你可以直接放进项目 `docs/` 目录,也可以直接粘贴给 Qoder Quest 生成 Spec 和代码骨架。 --- # AI PaaS 计费系统 · 技术规格文档(Qoder 投喂版) > **文档定位**:本文档是 Phase 1 MVP 中「计费系统」的完整技术规格,供 Qoder Quest 生成 Story / Spec / 代码骨架使用,也供人工 Review 时对照。 > > **安全等级**:🔴 **人 Design + 人写核心,AI 只补测试和胶水代码** > > **文件路径建议**:`docs/billing-spec.md` --- ## 一、计费系统总览 ### 1.1 业务目标 - 平台向用户(租户)提供 AI 能力(Agent / MCP / Skill / LLM 调用) - 每次调用按 **Token 消耗量** 实时扣费 - 支持 **订阅套餐(月度/年度)+ 按量超出(Pay-as-you-go)** 混合模式 - 支持 **Skill 交易分成**(平台抽成 20%~30%,Phase 2 再做,本期预留字段) - 所有金额/余额以 **Credit(积分)** 为内部统一单位,1 Credit = 1000 tokens(不同模型有系数折算) ### 1.2 核心设计原则 | 原则 | 说明 | |---|---| | **不超卖** | 先扣费后服务,原子操作,绝不允许"调用成功但余额为负" | | **不丢数** | 即使服务重启、网络抖动,计量数据最终一致,对账可修复 | | **幂等** | 同一请求多次回调/重试,绝不重复记账 | | **可审计** | 每一笔 Credit 变动都有 Trace_ID + 原因 + 关联资源 | | **实时可见** | Dashboard 上 Token 消耗和余额接近实时(秒级延迟可接受) | ### 1.3 计费架构图 ``` ┌─────────────────────────────────────────────────────────────┐ │ API Gateway (Higress) │ │ 注入 tenant_id / user_id / trace_id │ └──────────────────────────┬──────────────────────────────────┘ │ 转发请求 ┌──────────────────────────▼──────────────────────────────────┐ │ LLM Router (LiteLLM / 自研适配层) │ │ 调用完成后返回 usage: {prompt, completion, total} │ └──────────────────────────┬──────────────────────────────────┘ │ 上报用量事件 ┌──────────────────────────▼──────────────────────────────────┐ │ 计费服务 (billing-service) │ │ ┌─────────────┐ ┌──────────────┐ ┌──────────────────┐ │ │ │ Redis Lua │ │ Kafka Topic │ │ 日终对账 Cron │ │ │ │ 实时计量 │─▶│ usage.events│─▶│ 修复不一致数据 │ │ │ └─────────────┘ └──────────────┘ └──────────────────┘ │ │ │ │ │ │ ▼ ▼ │ │ ┌─────────────┐ ┌──────────────────┐ │ │ │ Redis 余额 │ │ PostgreSQL │ │ │ │ + 流水缓存 │ │ 永久账单 + 对账 │ │ │ └─────────────┘ └──────────────────┘ │ └─────────────────────────────────────────────────────────────┘ ``` --- ## 二、Credit 模型与定价 ### 2.1 统一内部单位:Credit ```yaml base_unit: "Credit" conversion: default: "1 Credit = 1000 tokens" per_model_coefficient: deepseek-chat: 1.0 # 1 Credit / 1000 tokens deepseek-reasoner: 2.0 # 2 Credits / 1000 tokens qwen-plus: 1.5 gpt-4o: 10.0 claude-sonnet: 8.0 ``` ### 2.2 订阅套餐(示例) | 套餐 | 月费 | 包含 Credits | 超出单价 | QPS 限制 | |---|---|---|---|---| | Free | ¥0 | 100 | 不可用 | 2 | | Pro | ¥99 | 50,000 | ¥0.01/Credit | 20 | | Team | ¥499 | 300,000 | ¥0.008/Credit | 100 | ### 2.3 Skill 交易定价(Phase 2 预留) ```yaml skill_pricing: model: "per_call" | "per_token" | "subscription" platform_fee_rate: 0.25 # 平台抽成 25% developer_share: 0.75 ``` --- ## 三、实时计量:Redis + Lua ### 3.1 为什么用 Redis Lua - **原子性**:扣余额 + 记流水在一个 Lua 脚本里完成,不存在"扣了余额没记流水"的中间态 - **高性能**:单次调用 < 1ms,支撑高并发 - **可重入**:通过 `idempotency_key` 实现幂等,重复请求直接返回之前的结果 ### 3.2 Redis Key 设计规范 ``` # 租户余额(实时) billing:balance:{tenant_id} → String, Credit 数量(整数,单位:毫 Credit 避免浮点) # 幂等去重(24h TTL) billing:idempotency:{tenant_id}:{idempotency_key} → Hash {result, credit_change, remaining} # 今日用量统计(按天聚合,TTL 48h) billing:daily:{tenant_id}:{YYYY-MM-DD} → Hash {total_tokens, total_credits, call_count, by_model:json} # 订阅配额(如有) billing:subscription:{tenant_id} → Hash {plan, credits_included, credits_used, period_end} ``` ### 3.3 核心 Lua 脚本:`metering.lua` ```lua -- KEYS[1]: balance_key -- KEYS[2]: idempotency_key -- KEYS[3]: daily_stats_key -- KEYS[4]: subscription_key (可选,为空则跳过) -- -- ARGV[1]: tenant_id -- ARGV[2]: idempotency_key -- ARGV[3]: credit_cost (毫 Credit) -- ARGV[4]: token_count -- ARGV[5]: model_name -- ARGV[6]: current_timestamp -- ARGV[7]: subscription_check ("1" = check, "0" = skip) -- 1. 幂等检查 if redis.call("EXISTS", KEYS[2]) == 1 then return redis.call("GET", KEYS[2]) end -- 2. 订阅配额检查(如果需要) if ARGV[7] == "1" and KEYS[4] ~= "" then local sub = redis.call("HGETALL", KEYS[4]) -- 简化逻辑:检查剩余订阅额度是否足够 -- 实际实现需解析 hash 字段 end -- 3. 余额检查 local current_balance = tonumber(redis.call("GET", KEYS[1]) or "0") local cost = tonumber(ARGV[3]) if current_balance < cost then return cjson.encode({error = "INSUFFICIENT_BALANCE", balance = current_balance}) end -- 4. 扣减余额 redis.call("DECRBY", KEYS[1], cost) -- 5. 记录幂等结果 local result = cjson.encode({ success = true, credit_cost = cost, remaining_balance = current_balance - cost, timestamp = ARGV[6] }) redis.call("SET", KEYS[2], result, "EX", 86400) -- 6. 更新今日统计 redis.call("HINCRBY", KEYS[3], "total_tokens", tonumber(ARGV[4])) redis.call("HINCRBY", KEYS[3], "total_credits", cost) redis.call("HINCRBY", KEYS[3], "call_count", 1) redis.call("EXPIRE", KEYS[3], 172800) -- 48h TTL -- 7. 按模型统计 local by_model = redis.call("HGET", KEYS[3], "by_model") or "{}" local model_stats = cjson.decode(by_model) model_stats[ARGV[5]] = (model_stats[ARGV[5]] or 0) + tonumber(ARGV[4]) redis.call("HSET", KEYS[3], "by_model", cjson.encode(model_stats)) return result ``` ### 3.4 Go 调用侧伪代码 ```go func (s *BillingService) MeterUsage(ctx context.Context, req MeterRequest) (*MeterResult, error) { // 构造 Redis Keys keys := []string{ fmt.Sprintf("billing:balance:%s", req.TenantID), fmt.Sprintf("billing:idempotency:%s:%s", req.TenantID, req.IdempotencyKey), fmt.Sprintf("billing:daily:%s:%s", req.TenantID, time.Now().Format("2006-01-02")), fmt.Sprintf("billing:subscription:%s", req.TenantID), } args := []interface{}{ req.TenantID, req.IdempotencyKey, req.CreditCostMilli, // 毫 Credit req.TokenCount, req.ModelName, time.Now().Unix(), "1", // subscription_check } result, err := s.redis.Eval(ctx, meteringLuaScript, keys, args...).Result() if err != nil { return nil, fmt.Errorf("redis eval failed: %w", err) } // 解析结果,发送 Kafka 事件供异步落库 s.kafkaProducer.Publish(ctx, "usage.events", UsageEvent{ TenantID: req.TenantID, TraceID: req.TraceID, CreditCost: req.CreditCostMilli, TokenCount: req.TokenCount, Model: req.ModelName, Timestamp: time.Now(), IdempotencyKey: req.IdempotencyKey, }) return parseResult(result), nil } ``` --- ## 四、异步持久化:Kafka → PostgreSQL ### 4.1 Kafka Topic 设计 ```yaml topic: usage.events partition_key: tenant_id # 保证同一租户的事件顺序 retention: 7d replication_factor: 3 ``` ### 4.2 PostgreSQL 表结构 ```sql -- 租户余额表(实时从 Redis 同步,也可作为兜底) CREATE TABLE tenant_balances ( tenant_id UUID PRIMARY KEY, credit_balance BIGINT NOT NULL DEFAULT 0, -- 毫 Credit updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), version BIGINT NOT NULL DEFAULT 0 -- 乐观锁 ); -- 账单流水表(不可变,追加写) CREATE TABLE credit_transactions ( id UUID PRIMARY KEY DEFAULT gen_random_uuid(), tenant_id UUID NOT NULL REFERENCES tenants(id), transaction_type TEXT NOT NULL, -- 'usage', 'topup', 'refund', 'subscription_renew' amount BIGINT NOT NULL, -- 正=收入,负=扣费(毫 Credit) balance_after BIGINT NOT NULL, reason TEXT NOT NULL, ref_type TEXT, -- 'capability_invocation', 'subscription', 'skill_purchase' ref_id UUID, trace_id TEXT, idempotency_key TEXT UNIQUE NOT NULL, created_at TIMESTAMPTZ NOT NULL DEFAULT NOW() ); CREATE INDEX idx_credit_txns_tenant_time ON credit_transactions(tenant_id, created_at DESC); CREATE INDEX idx_credit_txns_trace ON credit_transactions(trace_id); -- Token 用量明细表(按月分区) CREATE TABLE token_usage ( id UUID NOT NULL DEFAULT gen_random_uuid(), tenant_id UUID NOT NULL, capability_id UUID, capability_type TEXT NOT NULL, -- 'agent', 'mcp', 'skill', 'llm_direct' model TEXT NOT NULL, prompt_tokens INT NOT NULL, completion_tokens INT NOT NULL, total_tokens INT NOT NULL, cost_credits BIGINT NOT NULL, -- 毫 Credit trace_id TEXT NOT NULL, invoked_by UUID, -- user_id created_at TIMESTAMPTZ NOT NULL DEFAULT NOW() ) PARTITION BY RANGE (created_at); -- 按月分区示例 CREATE TABLE token_usage_y2026m01 PARTITION OF token_usage FOR VALUES FROM ('2026-01-01') TO ('2026-02-01'); CREATE INDEX idx_token_usage_tenant ON token_usage(tenant_id, created_at DESC); CREATE INDEX idx_token_usage_capability ON token_usage(capability_id, created_at DESC); -- RLS 策略 ALTER TABLE credit_transactions ENABLE ROW LEVEL SECURITY; ALTER TABLE token_usage ENABLE ROW LEVEL SECURITY; CREATE POLICY tenant_isolation ON credit_transactions USING (tenant_id = current_setting('app.current_tenant')::UUID); CREATE POLICY tenant_isolation ON token_usage USING (tenant_id = current_setting('app.current_tenant')::UUID); ``` ### 4.3 Consumer 写入逻辑(Go) ```go func (c *UsageConsumer) Consume(event UsageEvent) error { tx := c.db.Begin() // 1. 幂等检查(双保险,Redis 已有但 DB 也要防) var exists bool tx.Raw("SELECT EXISTS(SELECT 1 FROM credit_transactions WHERE idempotency_key = ?)", event.IdempotencyKey).Scan(&exists) if exists { return nil // 已处理,跳过 } // 2. 更新余额 tx.Exec("UPDATE tenant_balances SET credit_balance = credit_balance - ?, updated_at = NOW() WHERE tenant_id = ?", event.CreditCost, event.TenantID) // 3. 插入流水 tx.Exec("INSERT INTO credit_transactions (tenant_id, transaction_type, amount, reason, trace_id, idempotency_key) VALUES (?, 'usage', ?, ?, ?, ?)", event.TenantID, -event.CreditCost, "API invocation", event.TraceID, event.IdempotencyKey) // 4. 插入用量明细 tx.Exec("INSERT INTO token_usage (tenant_id, model, total_tokens, cost_credits, trace_id) VALUES (?, ?, ?, ?, ?)", event.TenantID, event.Model, event.TokenCount, event.CreditCost, event.TraceID) return tx.Commit().Error } ``` --- ## 五、日终对账(Daily Reconciliation) ### 5.1 为什么需要对账 - Redis Lua 扣费成功,但 Kafka 消息丢失 → DB 少记 - Consumer 崩溃 → 部分事件未落库 - 网络分区 → 双写不一致 ### 5.2 对账流程 ``` 每日 02:00 AM(低峰期) │ ├── 1. 快照 Redis 当日汇总(billing:daily:{tenant}:{yesterday}) │ ├── 2. 查询 PG 昨日流水(credit_transactions WHERE DATE(created_at) = yesterday) │ ├── 3. 逐租户比对: │ Redis.total_credits ≈ SUM(PG.amount WHERE type='usage') │ ├── 4. 差异处理: │ ├── Redis > PG → 补写 PG 流水 + 用量明细(标记为 reconciled) │ ├── Redis < PG → 告警人工介入(不应发生,可能是 Redis 过期) │ └── 余额不一致 → 以 PG 为准修复 Redis(SET balance = PG.balance) │ └── 5. 生成对账报告(存 S3 / 发邮件给管理员) ``` ### 5.3 对账 SQL 示例 ```sql -- 找出 Redis 和 PG 不一致的租户 WITH redis_summary AS ( -- 从 Redis 导出的 CSV 临时表(通过外部脚本导入) SELECT tenant_id, total_credits, total_tokens FROM redis_daily_snapshot WHERE date = '2026-01-15' ), pg_summary AS ( SELECT tenant_id, ABS(SUM(amount)) as total_credits, SUM(total_tokens) as total_tokens FROM credit_transactions ct LEFT JOIN token_usage tu ON ct.trace_id = tu.trace_id WHERE DATE(ct.created_at) = '2026-01-15' AND ct.transaction_type = 'usage' GROUP BY tenant_id ) SELECT r.tenant_id, r.total_credits as redis_credits, COALESCE(p.total_credits, 0) as pg_credits, r.total_credits - COALESCE(p.total_credits, 0) as diff FROM redis_summary r LEFT JOIN pg_summary p ON r.tenant_id = p.tenant_id WHERE ABS(r.total_credits - COALESCE(p.total_credits, 0)) > 0; ``` ### 5.4 修复脚本(Go) ```go func (r *Reconciler) FixDiscrepancy(ctx context.Context, diff Discrepancy) error { // Redis > PG:补写 PG if diff.RedisCredits > diff.PGCredits { missing := diff.RedisCredits - diff.PGCredits r.db.Exec("INSERT INTO credit_transactions (...) VALUES (?, 'usage_reconciled', ?, 'Reconciliation fix', ?, ?)", diff.TenantID, -missing, diff.TraceID, fmt.Sprintf("reconcile_%s", uuid.New())) // 修复余额 r.db.Exec("UPDATE tenant_balances SET credit_balance = credit_balance - ? WHERE tenant_id = ?", missing, diff.TenantID) } return nil } ``` --- ## 六、支付对接(Stripe + 支付宝) ### 6.1 充值流程 ``` 用户点击充值 → 选择金额 → 创建 PaymentIntent(Stripe)/ 预下单(支付宝) │ ├── Stripe: payment_intents.create → client_secret → 前端确认支付 │ ├── 支付宝: alipay.trade.precreate → QR Code → 用户扫码 │ └── Webhook 回调(🔴 幂等关键) │ ├── idempotency_key = "payment_{provider}_{payment_id}" │ ├── 验证签名(Stripe signature / 支付宝公钥) │ ├── 状态机校验(pending → succeeded,不允许倒退) │ ├── Redis INCRBY balance │ └── INSERT credit_transactions (type='topup', amount=+, idempotency_key=...) ``` ### 6.2 Webhook 幂等代码骨架 ```go func (h *WebhookHandler) HandleStripe(w http.ResponseWriter, r *http.Request) { payload, _ := io.ReadAll(r.Body) sig := r.Header.Get("Stripe-Signature") event, err := webhook.ConstructEvent(payload, sig, h.webhookSecret) if err != nil { ... } // 幂等:同一 event.ID 绝不处理两次 idempotencyKey := fmt.Sprintf("stripe_%s", event.ID) if h.isProcessed(idempotencyKey) { w.WriteHeader(http.StatusOK) return } switch event.Type { case "payment_intent.succeeded": pi := parsePaymentIntent(event.Data) h.billingService.TopUp(context.Background(), TopUpRequest{ TenantID: pi.Metadata["tenant_id"], Amount: pi.Amount, // 注意货币单位转换 Provider: "stripe", ProviderTxID: pi.ID, IdempotencyKey: idempotencyKey, }) } w.WriteHeader(http.StatusOK) } ``` --- ## 七、Qoder Quest Story 拆分(直接可用) ``` Epic: mvp:billing-system │ ├── Story 1: 🔴 billing:redis-lua-metering │ ├── Task: 设计 Redis Key 规范(见第三节) │ ├── Task: 编写 metering.lua 脚本(见 3.3) │ ├── Task: Go 调用侧封装 + 单元测试 │ └── Task: 压力测试(10k QPS 下余额正确性) │ ├── Story 2: 🔴 billing:kafka-persistence │ ├── Task: Kafka topic 创建 + Go Producer │ ├── Task: Go Consumer + PG 写入(见第四节) │ ├── Task: 幂等双保险(DB 层 idempotency_key UNIQUE) │ └── Task: 消费延迟监控 │ ├── Story 3: 🔴 billing:daily-reconciliation │ ├── Task: 对账 SQL 编写(见第五节) │ ├── Task: Go Cron Job(每日 02:00) │ ├── Task: 差异修复逻辑 │ └── Task: 对账报告生成 + 告警 │ ├── Story 4: 🔴 billing:stripe-alipay-webhook │ ├── Task: Stripe PaymentIntent 创建 + 前端确认 │ ├── Task: 支付宝预下单 + 轮询/回调 │ ├── Task: Webhook Handler(签名验证 + 幂等) │ └── Task: 充值 → Redis 余额 + PG 流水 │ └── Story 5: 🟡 billing:dashboard-analytics ├── Task: 用量聚合 API(按天/模型/能力) ├── Task: Dashboard 图表组件(消耗曲线、延迟 P99) └── Task: 余额预警(低于阈值邮件通知) ``` --- ## 八、验收标准(Definition of Done) | 场景 | 预期结果 | |---|---| | 正常调用 | Redis 余额扣减,PG 流水 + 用量明细均写入,Dashboard 实时可见 | | 重复请求(同 idempotency_key) | 第二次返回缓存结果,不重复扣费 | | Redis 宕机 | 服务返回 503,不执行调用(不超卖) | | Kafka 积压 | 消费者追上后数据最终一致,对账修复微小偏差 | | 日终对账 | 差异 < 0.01% 自动修复,> 0.01% 告警人工 | | 支付回调重放 | 同一 payment_id 只记账一次 | | 掉电恢复 | 重启后从 PG 恢复 Redis 余额(启动时 sync) |