多租户 AI 服务最容易死在两件事：一个租户打爆全局配额，以及月底账单炸了才发现。

这篇给你一套可直接落地的 Go 方案：令牌桶限流 + 预算熔断 + 账单归因，目标是“先活下来，再精细化”。

先定三个硬目标（保守但有效）

  1. 隔离性：单租户异常流量不会拖垮其他租户。
  2. 可控性：超预算时自动降级/熔断，而不是人工救火。
  3. 可归因：每次请求都能追到租户、模型、成本。

如果你的系统现在没有这三个目标，先别谈“智能路由优化”，先把止血做好。

总体架构：三道闸门

请求进入后按顺序过三道闸：

  1. 租户令牌桶（QPS/TPS）：防流量尖峰。
  2. 预算闸门（小时/日/月）：防成本失控。
  3. 供应商调用 + 成本回写：防“调用成功但账务失真”。
Client -> API Gateway -> Tenant Middleware
                     -> Token Bucket (tenant)
                     -> Budget Guard (tenant/model/time-window)
                     -> OpenAI Responses API
                     -> Usage Collector -> Cost Ledger

1) 令牌桶限流：按租户 + 按模型双层控制

不要只做全局限流。正确做法是：

  • 租户总桶：限制租户整体吞吐
  • 模型子桶：限制高价模型（比如 gpt-5.x）
// /Users/wow/dev/book/mengboy/examples/quota/token_bucket.go
package quota

import (
    "context"
    "fmt"
    "time"

    "golang.org/x/time/rate"
)

type BucketSet struct {
    TenantLimiter *rate.Limiter
    ModelLimiter  map[string]*rate.Limiter
}

func (b *BucketSet) Allow(ctx context.Context, model string) error {
    if !b.TenantLimiter.Allow() {
        return fmt.Errorf("tenant_rate_limited")
    }
    lim, ok := b.ModelLimiter[model]
    if ok && !lim.Allow() {
        return fmt.Errorf("model_rate_limited")
    }
    return nil
}

func NewDefaultBucketSet() *BucketSet {
    return &BucketSet{
        TenantLimiter: rate.NewLimiter(rate.Every(50*time.Millisecond), 20), // 20 burst, ~20 rps
        ModelLimiter: map[string]*rate.Limiter{
            "gpt-5.3": rate.NewLimiter(rate.Every(100*time.Millisecond), 8),
            "gpt-5.3-mini": rate.NewLimiter(rate.Every(40*time.Millisecond), 30),
        },
    }
}

实战建议

  • 高频租户优先做分层套餐，不同层级给不同 burst。
  • 模型升级时同步调整子桶，否则会出现“新模型上线即全红”。

2) 预算熔断：不是等超了再报警，而是提前收口

预算至少分 3 层：

  • hourly_soft_limit：触发降级（大模型→小模型）
  • daily_hard_limit：拒绝非关键请求
  • monthly_hard_limit：仅保留白名单业务
// /Users/wow/dev/book/mengboy/examples/quota/budget_guard.go
package quota

import "fmt"

type BudgetSnapshot struct {
    TenantID      string
    HourlyUsedUSD float64
    DailyUsedUSD  float64
    MonthlyUsedUSD float64

    HourlySoftUSD float64
    DailyHardUSD  float64
    MonthlyHardUSD float64
}

type Decision struct {
    Action string // allow | degrade | block
    Reason string
}

func EvaluateBudget(b BudgetSnapshot) Decision {
    if b.MonthlyUsedUSD >= b.MonthlyHardUSD {
        return Decision{Action: "block", Reason: "monthly_budget_exceeded"}
    }
    if b.DailyUsedUSD >= b.DailyHardUSD {
        return Decision{Action: "block", Reason: "daily_budget_exceeded"}
    }
    if b.HourlyUsedUSD >= b.HourlySoftUSD {
        return Decision{Action: "degrade", Reason: "hourly_soft_limit_reached"}
    }
    return Decision{Action: "allow", Reason: "within_budget"}
}

func MustAllow(d Decision) error {
    if d.Action == "block" {
        return fmt.Errorf(d.Reason)
    }
    return nil
}

降级策略别搞玄学

按固定顺序就行：

  1. 降模型规格（如 gpt-5.3 -> gpt-5.3-mini）
  2. 缩短上下文窗口（限制输入 token）
  3. 关闭非关键工具调用

可预测，比“智能策略”更稳定。

3) 账单归因：每次请求都写成本台账

做归因的关键不是“月底导出 CSV”，而是请求级事件日志。

建议记录字段：

  • request_id, tenant_id, user_id
  • model, input_tokens, output_tokens
  • latency_ms, status, cost_usd
  • route（是否降级）
-- /Users/wow/dev/book/mengboy/examples/quota/cost_ledger.sql
CREATE TABLE ai_cost_ledger (
  id BIGSERIAL PRIMARY KEY,
  request_id TEXT NOT NULL,
  tenant_id TEXT NOT NULL,
  model TEXT NOT NULL,
  input_tokens INT NOT NULL,
  output_tokens INT NOT NULL,
  cost_usd NUMERIC(12,6) NOT NULL,
  route TEXT NOT NULL,
  status TEXT NOT NULL,
  created_at TIMESTAMPTZ NOT NULL DEFAULT now()
);
CREATE INDEX idx_ai_cost_ledger_tenant_time ON ai_cost_ledger(tenant_id, created_at DESC);

4) OpenAI Responses 调用中间件顺序（Go）

func HandleResponsesRequest(ctx context.Context, req Req) (Resp, error) {
    bucket := limiterStore.Get(req.TenantID)
    if err := bucket.Allow(ctx, req.Model); err != nil {
        return Resp{}, wrap429(err)
    }

    snap := budgetStore.Snapshot(req.TenantID)
    decision := EvaluateBudget(snap)
    if decision.Action == "degrade" {
        req.Model = "gpt-5.3-mini"
        req.Route = "degraded_hourly_budget"
    }
    if err := MustAllow(decision); err != nil {
        return Resp{}, wrap402(err)
    }

    resp, usage, err := openaiClient.Responses(req)
    cost := pricing.Calc(req.Model, usage.InputTokens, usage.OutputTokens)
    ledger.Write(req, usage, cost, decision)
    budgetStore.Accumulate(req.TenantID, cost)

    return resp, err
}

常见坑（以及避免方式）

坑 1：只按请求数限流，不按 token 限流

结果：短请求没问题，长请求把账单打穿。
修复：增加 TPM（tokens per minute）桶，至少对高价模型启用。

坑 2：预算判断与成本回写不同步

结果：明明已经超预算，仍然放行。
修复：预算快照和写回必须同一个一致性域（同库事务或原子更新）。

坑 3：错误码全是 500

结果：业务方没法重试和降级。
修复：

  • 限流：429 + tenant_rate_limited
  • 预算：402/429 + daily_budget_exceeded
  • 上游失败：503 + provider_unavailable

监控看板（最低配置）

至少放这 6 个指标：

  1. 租户 QPS / TPM（P95）
  2. 限流命中率（按租户）
  3. 小时/日/月预算消耗率
  4. 降级触发率
  5. 单请求成本分布（P50/P95）
  6. 账单归因缺失率（应接近 0）

经验值：归因缺失率 > 0.5% 时，财务对账会非常痛苦。

一套保守可上线参数（起步版）

  • 租户总限流：20 rps burst 20
  • 高价模型限流：8 rps burst 8
  • 小时软预算：月预算的 0.8%
  • 日硬预算：月预算的 8%
  • 月硬预算：合同预算 100%
  • 降级优先级：模型降级 > 缩窗口 > 关工具

先用这组参数跑两周，再按真实业务调。

总结

多租户 AI 成本治理不靠“神奇算法”，靠三件朴素但硬核的事：

  • 令牌桶隔离流量
  • 预算熔断控制损失
  • 请求级台账保证可归因

把这三层打实，你的 OpenAI Responses 服务才有资格谈“规模化”。