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

baronbool

1505人浏览 · 2026-07-16 16:30:20

baronbool · 2026-07-16 16:30:20 发布

大模型多租户公平计费方案：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 不同长度，消耗也不同。

二、三层计费模型设计

三层计费模型：

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

三、分摊策略的核心矛盾：准确 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 很长但期望短回复，另一些团队反之。混在一起计费会让"我觉得不公平"的团队无法追溯具体原因。