跳至主要內容

AI 系统的成本归因——谁用了多少钱，怎么分摊

老张2026/4/30大约 8 分钟

AI 系统的成本归因——谁用了多少钱，怎么分摊

我们产品上线三个月后，老板问我一个问题：每个月 API 费用大概多少？

我说大概两万块。他又问：哪些功能花的？哪个团队用的？成本趋势是涨还是降？

我答不上来。

我只知道一个总数，是信用卡账单上的数字。但这个数字背后，是客服功能还是搜索功能消耗的？是 A 产品线还是 B 产品线？是因为用户增长导致的合理增长，还是某个 Prompt 写得太臭导致 Token 爆炸？

说不清楚。

这个问题困扰了我一段时间，然后我意识到这不是一个工程问题，而是一个成本管理问题——任何规模化的系统都需要成本归因，AI 系统尤其如此，因为 AI 的成本弹性非常大：同样的功能，Prompt 写得好和写得差可以差 5 倍的 Token 消耗。

为什么 AI 成本必须做归因

先说清楚为什么需要归因，而不是只看总数。

原因一：成本优化没有抓手。 如果你只知道总成本是 2 万，你不知道去哪里优化。但如果你知道 80% 的成本来自"文档智能分析"功能，而这个功能的 Token 消耗里有 60% 是无效的重复上下文，你就有明确的优化方向了。

原因二：成本分摊。 在企业内部，如果 AI 基础设施是公共资源，不同业务团队都在用，成本分摊是合理的。没有归因，你没办法跟财务说某个团队用了多少钱。

原因三：异常检测。 如果某天某个功能的成本突然暴涨，你需要立刻知道，而不是月底看账单才发现。成本归因是异常检测的前提。

原因四：ROI 分析。 哪个功能花的钱和带来的用户价值是匹配的？这个分析必须建立在成本归因的基础上。

成本归因的层次模型

成本归因要分层次，不同层次的粒度服务不同的目的：

这个层次从上到下：环境 → 业务线 → 功能 → 用户层级 → Token 类型。每一层都有其意义：

  • 环境层：区分生产和非生产，避免测试费用污染生产成本数据
  • 业务线层：支持部门间成本分摊
  • 功能层：找到成本大头，指导优化方向
  • 用户层级：支持差异化定价，VIP 用户用贵的模型，免费用户用便宜的
  • Token 类型：Prompt Token 和 Completion Token 的单价不同，且优化方向也不同（前者压缩 Prompt，后者限制 max_tokens）

Tag-based 成本追踪的实现

实现成本归因的核心技术手段是在每次 API 调用时记录元数据 Tag，然后基于 Tag 做聚合统计。

Step 1：定义成本上下文

// 成本追踪上下文，在整个请求链路中传递
@Data
@Builder
public class CostTrackingContext {
    
    private String requestId;          // 请求唯一 ID
    private String environment;        // 环境：prod/staging/dev
    private String productLine;        // 产品线：product-a/product-b/internal
    private String feature;            // 功能：customer-service/doc-analysis/search
    private String userId;             // 用户 ID（可 hash 处理隐私）
    private String userTier;           // 用户层级：vip/standard/free
    private String teamId;             // 调用团队 ID
    private String modelId;            // 使用的模型
    private String taskType;           // 任务类型：chat/embedding/completion
    private Instant requestTime;       // 请求时间
    
    // 方便在日志里输出
    public Map<String, String> toTagMap() {
        Map<String, String> tags = new LinkedHashMap<>();
        tags.put("env", environment);
        tags.put("product", productLine);
        tags.put("feature", feature);
        tags.put("user_tier", userTier);
        tags.put("team", teamId);
        tags.put("model", modelId);
        tags.put("task_type", taskType);
        return tags;
    }
}

Step 2：ThreadLocal 传递上下文

@Component
public class CostTrackingContextHolder {
    
    private static final ThreadLocal<CostTrackingContext> CONTEXT = new ThreadLocal<>();
    
    public static void setContext(CostTrackingContext context) {
        CONTEXT.set(context);
    }
    
    public static CostTrackingContext getContext() {
        CostTrackingContext ctx = CONTEXT.get();
        if (ctx == null) {
            // 没有设置上下文时，返回默认值而不是 null，避免 NPE
            return CostTrackingContext.builder()
                .environment("unknown")
                .feature("unknown")
                .userTier("unknown")
                .build();
        }
        return ctx;
    }
    
    public static void clear() {
        CONTEXT.remove();
    }
}

Step 3：拦截 AI API 调用，记录成本数据

@Component
@Slf4j
public class AICallCostInterceptor implements AdvisorSpec {
    
    private final MeterRegistry meterRegistry;
    private final CostEventPublisher eventPublisher;
    
    // 各模型的定价（美元/1M Token），定期更新
    private static final Map<String, ModelPricing> MODEL_PRICING = Map.of(
        "gpt-4o", new ModelPricing(5.0, 15.0),          // input: $5/1M, output: $15/1M
        "gpt-4o-mini", new ModelPricing(0.15, 0.6),
        "text-embedding-3-small", new ModelPricing(0.02, 0.0)
    );
    
    public void recordCost(String model, int promptTokens, int completionTokens) {
        CostTrackingContext ctx = CostTrackingContextHolder.getContext();
        ModelPricing pricing = MODEL_PRICING.getOrDefault(model, ModelPricing.DEFAULT);
        
        // 计算成本（单位：美分，避免浮点精度问题）
        long promptCostCents = Math.round(promptTokens * pricing.inputPricePerMillion() / 1_000_000 * 100);
        long completionCostCents = Math.round(completionTokens * pricing.outputPricePerMillion() / 1_000_000 * 100);
        long totalCostCents = promptCostCents + completionCostCents;
        
        // 记录指标到 Prometheus
        Map<String, String> tags = ctx.toTagMap();
        
        Counter.builder("ai.cost.cents")
            .description("AI API 调用成本（美分）")
            .tags(toTagArray(tags))
            .register(meterRegistry)
            .increment(totalCostCents);
            
        Counter.builder("ai.tokens.prompt")
            .tags(toTagArray(tags))
            .register(meterRegistry)
            .increment(promptTokens);
            
        Counter.builder("ai.tokens.completion")
            .tags(toTagArray(tags))
            .register(meterRegistry)
            .increment(completionTokens);
        
        // 发布成本事件到消息队列，用于实时告警和详细报表
        eventPublisher.publish(CostEvent.builder()
            .requestId(ctx.getRequestId())
            .tags(tags)
            .model(model)
            .promptTokens(promptTokens)
            .completionTokens(completionTokens)
            .totalCostCents(totalCostCents)
            .timestamp(Instant.now())
            .build());
        
        // 结构化日志，方便 ELK 聚合
        log.info("AI cost recorded: model={}, prompt_tokens={}, completion_tokens={}, cost_cents={}, tags={}",
            model, promptTokens, completionTokens, totalCostCents, tags);
    }
    
    private String[] toTagArray(Map<String, String> tags) {
        List<String> tagList = new ArrayList<>();
        tags.forEach((k, v) -> {
            tagList.add(k);
            tagList.add(v != null ? v : "unknown");
        });
        return tagList.toArray(new String[0]);
    }
}

Step 4：在 Spring AI 调用前后注入拦截逻辑

@Component
public class AIServiceWithCostTracking {
    
    private final ChatClient chatClient;
    private final AICallCostInterceptor costInterceptor;
    
    public String chat(String feature, String userId, String userTier, 
                       String systemPrompt, String userMessage) {
        // 设置成本追踪上下文
        CostTrackingContextHolder.setContext(
            CostTrackingContext.builder()
                .requestId(UUID.randomUUID().toString())
                .environment(System.getenv("APP_ENV"))
                .feature(feature)
                .userId(hashUserId(userId))  // hash 处理，保护隐私
                .userTier(userTier)
                .teamId(resolveTeamId(feature))
                .taskType("chat")
                .build()
        );
        
        try {
            ChatResponse response = chatClient.prompt()
                .system(systemPrompt)
                .user(userMessage)
                .call()
                .chatResponse();
            
            // 从响应里提取 Token 使用量
            Usage usage = response.getMetadata().getUsage();
            costInterceptor.recordCost(
                "gpt-4o",
                usage.getPromptTokens().intValue(),
                usage.getGenerationTokens().intValue()
            );
            
            return response.getResult().getOutput().getContent();
            
        } finally {
            // 务必清理，否则 ThreadLocal 泄漏
            CostTrackingContextHolder.clear();
        }
    }
    
    private String hashUserId(String userId) {
        // 简单 hash，不存原始 userId
        return DigestUtils.md5DigestAsHex(userId.getBytes()).substring(0, 8);
    }
    
    private String resolveTeamId(String feature) {
        Map<String, String> featureTeamMap = Map.of(
            "customer-service", "team-cs",
            "doc-analysis", "team-product",
            "search", "team-search"
        );
        return featureTeamMap.getOrDefault(feature, "unknown");
    }
}

基于 Prometheus 的成本看板

数据有了，需要可视化。Grafana 的 Prometheus 查询：

# 按功能维度的每日成本（美元）
sum by (feature) (
  increase(ai_cost_cents_total{env="prod"}[24h])
) / 100

# 成本 Top 5 功能（过去 7 天）
topk(5,
  sum by (feature) (
    increase(ai_cost_cents_total{env="prod"}[7d])
  )
) / 100

# 各团队的本月成本
sum by (team) (
  increase(ai_cost_cents_total{env="prod"}[30d])
) / 100

# 每用户层级的平均成本（每次调用）
sum by (user_tier) (rate(ai_cost_cents_total[1h]))
/ 
sum by (user_tier) (rate(ai_calls_total[1h]))

# 成本异常检测：当前小时成本 vs 过去 7 天同小时均值的比例
(
  sum(increase(ai_cost_cents_total{env="prod"}[1h]))
)
/
(
  sum(increase(ai_cost_cents_total{env="prod"}[1h] offset 7d))
)

告警规则：

# prometheus-alerts.yaml
groups:
- name: ai-cost-alerts
  rules:
  # 成本突增告警
  - alert: AICostSurge
    expr: |
      (
        sum(rate(ai_cost_cents_total{env="prod"}[1h])) * 3600
      ) > (
        sum(rate(ai_cost_cents_total{env="prod"}[1h] offset 7d)) * 3600 * 2
      )
    for: 15m
    labels:
      severity: warning
      team: ai-platform
    annotations:
      summary: "AI 成本突增"
      description: "过去 1 小时成本是上周同时段的 {{ $value | printf \"%.1f\" }} 倍"
  
  # 单功能成本超限
  - alert: FeatureCostOverBudget
    expr: |
      sum by (feature) (
        increase(ai_cost_cents_total{env="prod"}[24h])
      ) / 100 > 500
    for: 5m
    labels:
      severity: warning
    annotations:
      summary: "功能 {{ $labels.feature }} 今日成本超过 $500"
      description: "当前成本 ${{ $value | printf \"%.2f\" }}"
  
  # Token 使用率异常（某功能 completion/prompt 比例异常高，说明输出太长）
  - alert: HighCompletionTokenRatio
    expr: |
      sum by (feature) (rate(ai_tokens_completion[1h]))
      /
      sum by (feature) (rate(ai_tokens_prompt[1h]))
      > 2
    for: 30m
    labels:
      severity: info
    annotations:
      summary: "功能 {{ $labels.feature }} 的 completion/prompt Token 比例过高"
      description: "比例为 {{ $value | printf \"%.1f\" }}x，可能需要限制 max_tokens"

成本优化的落地

有了成本归因之后，我们做了几个优化，效果很显著：

优化一：发现 Prompt 膨胀问题。 通过 Token 明细，我们发现"文档分析"功能的 Prompt Token 异常高，平均每次调用消耗 6000 Token 的 Prompt。排查后发现，代码里把整个文档都塞进了 Prompt，而实际上只需要相关段落。改成 RAG 后，Prompt Token 降到了 1200，成本下降 80%。

优化二：用户分层定价。 数据显示免费用户的人均 Token 消耗只有 VIP 用户的 1/10，但免费用户和付费用户使用的是同一套模型。把免费用户改成 gpt-4o-mini，VIP 用户保留 gpt-4o，成本减少 40%，但 VIP 用户体验不变。

优化三：发现测试污染生产成本。 环境分层之后发现，有 30% 的 API 调用来自 env=unknown，追踪下去是有几个脚本在生产环境用生产 API Key 跑测试。这部分成本完全可以避免。

总结

AI 系统的成本归因，本质上是在每次 API 调用时打 Tag，然后基于 Tag 做聚合分析。

实现这套体系需要：一个贯穿请求链路的上下文对象（ThreadLocal），一个在 AI 调用后记录 Token 和成本的拦截器，一个 Prometheus 成本指标体系，一套基于成本异常的告警规则。

成本归因做好了，你才能回答老板的问题，才能指导成本优化，才能做功能 ROI 分析。这不是锦上添花，是 AI 应用走向规模化必须要有的基础能力。

编辑此页
最近更新:
贡献者: xiaozhangge