添加 yuanbao-aiflow

2026-06-27 15:31:52 +08:00
1 parent 2052066dde
commit 5476fc3069
@@ -0,0 +1,518 @@
 下面我把 **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） |