大模型调用的边际成本分析：每次对话都在花钱，如何优化

baronbool

523人浏览 · 2026-07-13 10:30:48

baronbool · 2026-07-13 10:30:48 发布

大模型调用的边际成本分析：每次对话都在花钱，如何优化

一、"免费试用"背后的成本黑洞

一个 AI 对话产品上线推广，头一个月"免费试用"，用户量暴增。月底收到云厂商账单：大模型 API 费用 14.7 万元。核算下来，平均每次对话成本 0.82 元，而产品的付费套餐是每月 29.9 元/无限次。这意味着一个付费用户每天聊 2 次就回本——聊 3 次就开始亏钱。

大模型的 Token 计费让成本变得"隐形"。不像服务器按时长收费，Token 费用随用户使用量线性增长。不把"每次对话花多少钱"算清楚，AI 产品的商业模式就是空中楼阁。

二、大模型调用的成本模型

一次模型调用的总成本 = 输入 Token 成本 + 输出 Token 成本。但实际应用中还有大量隐性成本：

flowchart LR
    A[用户发一条消息] --> B[输入 Token 计算]
    B --> B1[System Prompt: 固定 200-500 tokens]
    B --> B2[对话历史: 每轮 100-500 tokens]
    B --> B3[RAG 检索结果: 500-2000 tokens]
    
    B1 --> C[输入 Token 总成本]
    B2 --> C
    B3 --> C
    
    A --> D[输出 Token 成本]
    D --> D1[模型回复: 100-1000 tokens]
    D1 --> E[输出 Token 总成本]
    
    C --> F[单次对话成本 = 输入 + 输出]
    E --> F
    
    F --> G[日均成本 = 单次 × 日活 × 人均对话数]
    G --> H[月成本 = 日均 × 30]
    
    H --> I{月度预算能覆盖?}
    I -->|否| J[需要优化]
    I -->|是| K[继续监控]

以 GPT-4o 为例（输入 $2.50/1M tokens，输出 $10.00/1M tokens）：

一次典型的客服对话：System Prompt 200 tokens + 对话历史 800 tokens + RAG 检索结果 1000 tokens = 2000 输入 tokens。模型回复 300 输出 tokens。

单次成本 = (2000/1000000) × $2.50 + (300/1000000) × $10.00 = $0.005 + $0.003 = $0.008（约 0.06 元）

看起来便宜。但日活 1 万用户 × 人均 5 次对话 × 0.06 元 = 每天 3000 元 = 每月 9 万元。

三、Go 实现的 Token 用量追踪与成本计算

package costctl

import (
    "fmt"
    "sync"
    "sync/atomic"
    "time"
)

// ========== Token 定价模型 ==========

// ModelPricing 模型定价信息
type ModelPricing struct {
    ModelName     string
    InputPrice    float64 // 每 1K tokens 的价格（美元）
    OutputPrice   float64 // 每 1K tokens 的价格（美元）
    CurrencyRate  float64 // 汇率（美元→人民币）
}

var ModelPrices = map[string]ModelPricing{
    "gpt-4o": {
        ModelName:    "gpt-4o",
        InputPrice:   0.0025, // $2.50 / 1M tokens = $0.0025 / 1K tokens
        OutputPrice:  0.01,   // $10.00 / 1M tokens = $0.01 / 1K tokens
        CurrencyRate: 7.25,
    },
    "gpt-4o-mini": {
        ModelName:    "gpt-4o-mini",
        InputPrice:   0.00015,
        OutputPrice:  0.0006,
        CurrencyRate: 7.25,
    },
    "gpt-3.5-turbo": {
        ModelName:    "gpt-3.5-turbo",
        InputPrice:   0.0005,
        OutputPrice:  0.0015,
        CurrencyRate: 7.25,
    },
}

// ========== Token 用量追踪器 ==========

// TokenUsage Token 用量记录
type TokenUsage struct {
    InputTokens  int64
    OutputTokens int64
    Model        string
    Timestamp    time.Time
    UserID       string
    SessionID    string
}

// CostTracker 成本追踪器
type CostTracker struct {
    // 实时累计
    totalInputTokens  int64
    totalOutputTokens int64
    totalCostUSD      float64
    totalCostCNY      float64
    totalCalls        int64
    
    // 按用户统计
    userUsage map[string]*UserUsageStats
    mu        sync.RWMutex
    
    // 预算控制
    dailyBudget float64 // 每日预算上限（元）
    dailyCost   float64 // 当日已花费（元）
    dailyDate   string  // 日期标记
    
    // 告警回调
    onBudgetWarning func(usage float64, budget float64)
}

// UserUsageStats 用户用量统计
type UserUsageStats struct {
    UserID       string
    TotalCalls   int64
    TotalTokens  int64
    TotalCost    float64 // 元
    LastCallTime time.Time
}

func NewCostTracker(dailyBudget float64) *CostTracker {
    return &CostTracker{
        userUsage:   make(map[string]*UserUsageStats),
        dailyBudget: dailyBudget,
    }
}

// RecordUsage 记录一次模型调用
func (ct *CostTracker) RecordUsage(usage TokenUsage) {
    pricing, ok := ModelPrices[usage.Model]
    if !ok {
        return // 未知模型，跳过
    }
    
    // 计算成本
    inputCost := float64(usage.InputTokens) / 1000 * pricing.InputPrice
    outputCost := float64(usage.OutputTokens) / 1000 * pricing.OutputPrice
    costUSD := inputCost + outputCost
    costCNY := costUSD * pricing.CurrencyRate
    
    // 更新累计统计
    atomic.AddInt64(&ct.totalInputTokens, usage.InputTokens)
    atomic.AddInt64(&ct.totalOutputTokens, usage.OutputTokens)
    atomic.AddInt64(&ct.totalCalls, 1)
    
    ct.mu.Lock()
    defer ct.mu.Unlock()
    
    ct.totalCostUSD += costUSD
    ct.totalCostCNY += costCNY
    
    // 更新用户统计
    if _, ok := ct.userUsage[usage.UserID]; !ok {
        ct.userUsage[usage.UserID] = &UserUsageStats{UserID: usage.UserID}
    }
    stats := ct.userUsage[usage.UserID]
    stats.TotalCalls++
    stats.TotalTokens += usage.InputTokens + usage.OutputTokens
    stats.TotalCost += costCNY
    stats.LastCallTime = usage.Timestamp
    
    // 每日预算检查
    today := time.Now().Format("2006-01-02")
    if ct.dailyDate != today {
        ct.dailyDate = today
        ct.dailyCost = 0
    }
    ct.dailyCost += costCNY
    
    // 预算告警
    if ct.dailyBudget > 0 && ct.dailyCost > ct.dailyBudget*0.8 {
        if ct.onBudgetWarning != nil {
            ct.onBudgetWarning(ct.dailyCost, ct.dailyBudget)
        }
    }
}

// ========== 成本优化策略 ==========

// CostOptimizer 成本优化器
type CostOptimizer struct {
    tracker *CostTracker
    
    // 缓存配置
    semanticCache *SemanticCache
    
    // 模型路由
    modelRouter *ModelRouter
}

// ShouldUseCache 判断是否应该使用缓存（节省成本）
func (co *CostOptimizer) ShouldUseCache(query string) (string, bool) {
    if co.semanticCache == nil {
        return "", false
    }
    return co.semanticCache.FindSimilar(query, 0.92)
}

// RouteModel 根据查询复杂度选择合适的模型
func (co *CostOptimizer) RouteModel(query string, historyLength int) string {
    // 简单问候/常见问题 → 便宜模型
    if co.isSimpleQuery(query) {
        return "gpt-4o-mini"
    }
    
    // 长对话历史 → 大模型更好（上下文理解更强）
    if historyLength > 10 {
        return "gpt-4o"
    }
    
    // 默认
    return "gpt-4o-mini"
}

func (co *CostOptimizer) isSimpleQuery(query string) bool {
    simplePatterns := []string{"你好", "谢谢", "再见", "帮助"}
    for _, p := range simplePatterns {
        if len(query) < 20 && contains(query, p) {
            return true
        }
    }
    return false
}

// ========== 优化建议：减少 Token 消耗 ==========

// OptimizeHistory 优化对话历史：只保留最近 N 轮或有摘要
func OptimizeHistory(messages []Message, maxTokens int) []Message {
    // 策略 1：裁剪历史到最大 Token 数
    totalTokens := 0
    var result []Message
    
    // 保留 System Prompt（第一条）
    if len(messages) > 0 && messages[0].Role == "system" {
        result = append(result, messages[0])
        messages = messages[1:]
    }
    
    // 从最近的往历史方向取
    for i := len(messages) - 1; i >= 0; i-- {
        estimateTokens := len(messages[i].Content) / 4 // 粗略估算
        if totalTokens+estimateTokens > maxTokens {
            break
        }
        totalTokens += estimateTokens
        result = append([]Message{messages[i]}, result...)
    }
    
    return result
}

// EstimateTokens 估算文本的 Token 数（简单估算）
func EstimateTokens(text string) int {
    // 中文约 1.5 字符/token，英文约 4 字符/token
    chineseCount := 0
    englishCount := 0
    for _, r := range text {
        if r >= 0x4e00 && r <= 0x9fff {
            chineseCount++
        } else if (r >= 'a' && r <= 'z') || (r >= 'A' && r <= 'Z') {
            englishCount++
        }
    }
    return chineseCount*2/3 + englishCount/4 + len(text)/10
}

// ========== 类型定义 ==========

type Message struct {
    Role    string
    Content string
}

type SemanticCache struct{}

func (sc *SemanticCache) FindSimilar(query string, threshold float64) (string, bool) {
    return "", false
}

type ModelRouter struct{}

func contains(s, substr string) bool {
    for i := 0; i <= len(s)-len(substr); i++ {
        if s[i:i+len(substr)] == substr {
            return true
        }
    }
    return false
}