大模型多租户公平计费方案：Token 不再均摊的成本分摊模型

一、"一个月烧了 8 万的 API 费用，内部 6 个团队都说不关我事"

公司部署了一个共享的 GPT-4 调用网关，6 个业务团队共用。财务对账单上只有一个数：本月 OpenAI API 费用 8.3 万。CTO 拉会讨论，客服团队说"我们一天才几百次调用"，营销团队说"我们用的是免费模型"，AI 团队说"我们只做测试，没有线上流量"。没有使用数据，就没有办法分账。

大模型 API 的成本分摊比云主机更复杂：云主机的计费维度是时间（EC2 按小时）或调用次数（Lambda 按次数），但大模型的计费维度是 Token。Token 的特殊性在于：同样的请求，不同模型的价格不同（GPT-4 vs GPT-3.5 差距可达 20 倍），同样的 prompt 不同长度，消耗也不同。

二、三层计费模型设计

三层计费模型：

1. Token 计数层：记录每次调用的 prompt tokens、completion tokens 和使用的模型名称。这是最底层的数据，精度决定分摊的公平性。
2. 成本计算层：根据模型定价表（可以定期同步各厂商的最新价格）将 Token 转换为金额。不同模型的单价差异巨大，必须分模型计算。
3. 费用分摊层：将成本分配到团队或个人。根据场景选择分摊策略。

三、分摊策略的核心矛盾：准确 vs 公平

按实际用量分摊是最简单的——记好每个 API Key 的 Token 消耗，按比例分摊。但这会导致"先到先得"的问题：如果一个团队发了 10 个高消耗请求，把本月预算（假设公司预存了 5 万）全用完了，其他团队就没得用了。

按加权共享分摊解决这个问题：所有共享模型的调用成本，先按团队"权重系数"分配，再用实际用量核销。权重可以根据团队业务优先级动态调整。例如：

• AI 研发团队权重 1.0（按比例分摊）
• 客服团队权重 0.3（只承担 30% 的共享成本）
• 营销团队权重 2.0（预算更高，承担 2 倍）

但如果某个团队用量低于权重应摊的比例，多缴的部分应该退还——这就是"先验分摊 + 后验结算"。

配额先行模式：每个团队在月初分配固定预算额度（如一万元），当消耗达到 80% 时告警，达到 100% 时截断。这种模式简单粗暴，适合成本敏感的团队。

四、Go 实现：多租户 Token 分摊引擎

package costallocation import ( "context" "fmt" "log" "sync" "time" ) // ========== 模型定价 ========== type ModelPricing struct { Model string PromptPricePer1K float64 CompletionPricePer1K float64 } var DefaultPricing = map[string]ModelPricing{ "gpt-4": {Model: "gpt-4", PromptPricePer1K: 0.03, CompletionPricePer1K: 0.06}, "gpt-4-turbo": {Model: "gpt-4-turbo", PromptPricePer1K: 0.01, CompletionPricePer1K: 0.03}, "gpt-3.5-turbo": {Model: "gpt-3.5-turbo", PromptPricePer1K: 0.0005, CompletionPricePer1K: 0.0015}, "claude-3-opus": {Model: "claude-3-opus", PromptPricePer1K: 0.015, CompletionPricePer1K: 0.075}, } // ========== 使用记录 ========== type UsageRecord struct { ID string TenantID string Model string PromptTokens int CompletionTokens int Timestamp time.Time Cost float64 } // ========== 分摊配置 ========== type AllocationConfig struct { Strategy string // usage-based / weighted-shared / quota-first // 加权分摊的权重系数 TenantWeights map[string]float64 // 配额先行模式的预算 TenantQuotas map[string]float64 // 结算周期 SettlementCycle time.Duration // 默认按月 } // ========== 分摊引擎 ========== type CostAllocator struct { config AllocationConfig pricing map[string]ModelPricing mu sync.RWMutex // 当前结算周期的使用量 usageByTenant map[string]*TenantUsage cycleStart time.Time } type TenantUsage struct { TenantID string TotalCost float64 PromptTokens int CompletionTokens int Records []UsageRecord } func NewCostAllocator(config AllocationConfig) *CostAllocator { return &CostAllocator{ config: config, pricing: DefaultPricing, usageByTenant: make(map[string]*TenantUsage), cycleStart: time.Now(), } } // RecordUsage 记录一次 API 调用 func (ca *CostAllocator) RecordUsage(ctx context.Context, record UsageRecord) error { ca.mu.Lock() defer ca.mu.Unlock() // 1. 计算本次调用成本 pricing, ok := ca.pricing[record.Model] if !ok { return fmt.Errorf("未知模型: %s", record.Model) } promptCost := float64(record.PromptTokens) / 1000.0 * pricing.PromptPricePer1K completionCost := float64(record.CompletionTokens) / 1000.0 * pricing.CompletionPricePer1K record.Cost = promptCost + completionCost // 2. 累加到租户 tu, ok := ca.usageByTenant[record.TenantID] if !ok { tu = &TenantUsage{TenantID: record.TenantID} ca.usageByTenant[record.TenantID] = tu } tu.TotalCost += record.Cost tu.PromptTokens += record.PromptTokens tu.CompletionTokens += record.CompletionTokens tu.Records = append(tu.Records, record) // 3. 配额检查（quota-first 模式） if ca.config.Strategy == "quota-first" { quota, ok := ca.config.TenantQuotas[record.TenantID] if ok && tu.TotalCost > quota*0.8 { log.Printf("[Cost] 租户 %s 已消耗 %.1f%% 配额 (%.4f/%.4f)", record.TenantID, tu.TotalCost/quota*100, tu.TotalCost, quota) } } return nil } // AllocateCosts 执行分摊结算 func (ca *CostAllocator) AllocateCosts(ctx context.Context) map[string]float64 { ca.mu.RLock() defer ca.mu.RUnlock() result := make(map[string]float64) switch ca.config.Strategy { case "usage-based": result = ca.usageBasedAllocation() case "weighted-shared": result = ca.weightedSharedAllocation() case "quota-first": result = ca.quotaFirstAllocation() default: result = ca.usageBasedAllocation() } return result } // ========== 策略1: 按实际用量分摊 ========== func (ca *CostAllocator) usageBasedAllocation() map[string]float64 { result := make(map[string]float64) var totalCost float64 for _, tu := range ca.usageByTenant { totalCost += tu.TotalCost } for _, tu := range ca.usageByTenant { if totalCost > 0 { result[tu.TenantID] = tu.TotalCost } } return result } // ========== 策略2: 加权分摊 ========== func (ca *CostAllocator) weightedSharedAllocation() map[string]float64 { result := make(map[string]float64) // 1. 计算总成本和总权重 var totalCost float64 var totalWeight float64 for _, tu := range ca.usageByTenant { totalCost += tu.TotalCost } for _, w := range ca.config.TenantWeights { totalWeight += w } if totalCost == 0 || totalWeight == 0 { return result } // 2. 按权重预先分摊 preAllocated := make(map[string]float64) for tenantID, weight := range ca.config.TenantWeights { preAllocated[tenantID] = totalCost * weight / totalWeight } // 3. 后验结算：多用多补，少用退款 for _, tu := range ca.usageByTenant { pre := preAllocated[tu.TenantID] actual := tu.TotalCost result[tu.TenantID] = max(pre, actual) } return result } // ========== 策略3: 配额先行 ========== func (ca *CostAllocator) quotaFirstAllocation() map[string]float64 { result := make(map[string]float64) for _, tu := range ca.usageByTenant { quota, ok := ca.config.TenantQuotas[tu.TenantID] if !ok { result[tu.TenantID] = tu.TotalCost continue } if tu.TotalCost <= quota { // 未超配额：按实际计费 result[tu.TenantID] = tu.TotalCost } else { // 超配额：配额内按正常费率，超额部分加收 50% overage := tu.TotalCost - quota result[tu.TenantID] = quota + overage*1.5 } } return result } // ========== 结算并重置 ========== func (ca *CostAllocator) Settle(ctx context.Context) (map[string]float64, error) { ca.mu.Lock() defer ca.mu.Unlock() // 1. 执行分摊计算 result := make(map[string]float64) for tenantID, tu := range ca.usageByTenant { result[tenantID] = tu.TotalCost } // 2. 生成结算报告（此处省略持久化到数据库逻辑） log.Printf("[Cost] 结算周期结束, 参与租户: %d", len(result)) for tenantID, cost := range result { log.Printf(" %s: $%.4f", tenantID, cost) } // 3. 重置 ca.usageByTenant = make(map[string]*TenantUsage) ca.cycleStart = time.Now() return result, nil } // ========== 查询当前使用量 ========== func (ca *CostAllocator) GetTenantUsage(tenantID string) (*TenantUsage, error) { ca.mu.RLock() defer ca.mu.RUnlock() tu, ok := ca.usageByTenant[tenantID] if !ok { return nil, fmt.Errorf("租户 %s 无使用记录", tenantID) } return tu, nil } // ========== 使用示例 ========== func ExampleAllocation() { config := AllocationConfig{ Strategy: "weighted-shared", TenantWeights: map[string]float64{ "team-ai": 1.0, "team-service": 0.3, "team-marketing": 2.0, }, } allocator := NewCostAllocator(config) // 模拟记录使用 _ = allocator.RecordUsage(context.Background(), UsageRecord{ TenantID: "team-ai", Model: "gpt-4", PromptTokens: 2000, CompletionTokens: 500, }) _ = allocator.RecordUsage(context.Background(), UsageRecord{ TenantID: "team-marketing", Model: "gpt-4-turbo", PromptTokens: 5000, CompletionTokens: 1500, }) // 月底结算 result, _ := allocator.Settle(context.Background()) for tenant, cost := range result { fmt.Printf("%s 本月费用: $%.4f\n", tenant, cost) } }

五、总结

大模型成本分摊的核心矛盾：共享模型的"共用性"和每个团队的"独立核算"需求之间的张力。三层计费模型（Token 计数 → 成本计算 → 费用分摊）提供了一条可操作的路径。

三种策略的适用场景：

• 按实际用量：适合团队间模型使用模式差异大的情况（客服团队用便宜模型，AI 团队用贵模型）
• 加权分摊：适合共享同一个模型实例、难以精确区分每次调用归属的场景
• 配额先行：适合成本预算严格、需要硬约束的场景

工程落地时最容易被忽略的细节：模型价格是变化的（厂商调价、新模型上线），定价表需要支持热更新。另外，prompt tokens 和 completion tokens 的分开计费很关键——有些团队 prompt 很长但期望短回复，另一些团队反之。混在一起计费会让"我觉得不公平"的团队无法追溯具体原因。