AI应用的成本归因：精确追踪每分钱花在哪里
2026/9/28大约 20 分钟成本管理Token成本成本归因Spring AIJava
AI应用的成本归因：精确追踪每分钱花在哪里
月底账单20万，老板问你花在哪了，你答得上来吗？
2026年3月的最后一天，下午5点47分，字节跳动某业务线的后端负责人陈磊盯着屏幕上的账单，手指悬在键盘上动不了。
OpenAI的月账单：19.8万元。
他的老板在旁边站着，语气平静，但每个字都像钉子："这19.8万，智能客服花了多少？代码补全花了多少？文档生成花了多少？"
陈磊打开监控系统，打开日志平台，打开数据库——全是gpt-4调用记录，全是total_tokens: 2847这样的数字，加在一起就是19.8万。但哪个功能花了多少，没有人知道。
那个下午，陈磊花了4个小时，用Excel手动估算，误差率超过40%。老板当场决定：暂停所有AI功能扩展，直到能说清楚每分钱花在哪里。
这一暂停，就是整整6周。
6周后，他们建立了一套完整的AI成本归因系统。现在每天早上，成本看板自动更新，精确到每个用户、每个功能、每个租户的AI花费，误差率不超过0.3%。
这篇文章，就是那套系统的完整实现。
一、AI成本的真实构成：你以为只是Token费，其实差远了
在建立归因系统之前，我们首先要搞清楚AI应用的成本到底由哪些部分组成。很多团队只盯着Token费用，但实际上完整的AI成本有4个维度。
1.1 Token费用（通常占总成本70-85%）
Token费用是最显眼的成本，但也最容易被低估：
GPT-4o：input $2.5/1M tokens，output $10/1M tokens
GPT-4o-mini：input $0.15/1M tokens，output $0.6/1M tokens
Claude 3.5 Sonnet：input $3/1M tokens，output $15/1M tokens
Claude 3 Haiku：input $0.25/1M tokens，output $1.25/1M tokens
Gemini 1.5 Pro：input $1.25/1M tokens，output $5/1M tokens注意：同样的功能，用GPT-4o比用GPT-4o-mini贵16倍以上。
1.2 向量数据库费用（通常占10-20%）
Pinecone Serverless：$0.096/1M vectors/月存储 + $0.08/1M read units
Weaviate Cloud：$0.05/1M objects/月
Milvus Cloud：按节点计费，入门约$65/月对于RAG密集型应用，向量检索费用可能超过Token费。
1.3 嵌入模型费用（通常占3-8%）
OpenAI text-embedding-3-small：$0.02/1M tokens
OpenAI text-embedding-3-large：$0.13/1M tokens每次RAG检索都要embedding，高频应用这块不可忽视。
1.4 计算资源费用（通常占5-15%）
- 私有化部署的GPU成本
- 推理服务器的CPU/内存成本
- 数据预处理的计算成本
1.5 成本归因的核心挑战
成本归因之所以难，是因为AI调用和业务逻辑之间有多层隔离：
当账单到来时，API只知道total_tokens，不知道这些Token是哪个功能消耗的。归因必须在调用时就埋点，事后补救几乎不可能做到精确。
二、成本归因数据模型设计
2.1 核心实体设计
// 成本归因上下文 - 每次AI调用必须携带
@Data
@Builder
public class CostContext {
// 业务维度
private String tenantId; // 租户ID（多租户场景）
private String userId; // 用户ID
private String featureCode; // 功能代码：customer_service/code_assist/doc_gen
private String scenarioCode; // 场景代码：chat/search/summarize
// 技术维度
private String modelId; // 模型标识：gpt-4o/claude-3-5-sonnet
private String requestId; // 请求追踪ID
private String sessionId; // 会话ID
// 元数据
private Map<String, String> tags; // 自定义标签
private Instant timestamp;
}2.2 成本记录数据模型
@Entity
@Table(name = "ai_cost_records", indexes = {
@Index(name = "idx_tenant_feature_time", columnList = "tenant_id,feature_code,created_at"),
@Index(name = "idx_user_time", columnList = "user_id,created_at"),
@Index(name = "idx_model_time", columnList = "model_id,created_at"),
@Index(name = "idx_request_id", columnList = "request_id", unique = true)
})
@Data
@Builder
@NoArgsConstructor
@AllArgsConstructor
public class AiCostRecord {
@Id
@GeneratedValue(strategy = GenerationType.IDENTITY)
private Long id;
// 归因维度
@Column(name = "tenant_id", length = 64)
private String tenantId;
@Column(name = "user_id", length = 64)
private String userId;
@Column(name = "feature_code", length = 64, nullable = false)
private String featureCode;
@Column(name = "scenario_code", length = 64)
private String scenarioCode;
@Column(name = "request_id", length = 128, nullable = false)
private String requestId;
@Column(name = "session_id", length = 128)
private String sessionId;
// 模型信息
@Column(name = "model_id", length = 64, nullable = false)
private String modelId;
@Column(name = "model_provider", length = 32)
private String modelProvider; // openai/anthropic/google
// Token消耗
@Column(name = "input_tokens", nullable = false)
private Integer inputTokens;
@Column(name = "output_tokens", nullable = false)
private Integer outputTokens;
@Column(name = "total_tokens", nullable = false)
private Integer totalTokens;
// 成本计算（单位：美分，避免浮点精度问题）
@Column(name = "input_cost_cents", nullable = false)
private Long inputCostCents;
@Column(name = "output_cost_cents", nullable = false)
private Long outputCostCents;
@Column(name = "total_cost_cents", nullable = false)
private Long totalCostCents;
// 性能数据
@Column(name = "latency_ms")
private Integer latencyMs;
@Column(name = "first_token_latency_ms")
private Integer firstTokenLatencyMs;
// 状态
@Enumerated(EnumType.STRING)
@Column(name = "status", length = 16)
private CallStatus status; // SUCCESS/FAILED/TIMEOUT
@Column(name = "error_code", length = 64)
private String errorCode;
// 自定义标签（JSON格式）
@Column(name = "tags", columnDefinition = "JSON")
@Convert(converter = JsonMapConverter.class)
private Map<String, String> tags;
@Column(name = "created_at", nullable = false)
private Instant createdAt;
public enum CallStatus {
SUCCESS, FAILED, TIMEOUT, PARTIAL
}
}2.3 成本聚合表（预聚合，提升查询性能）
@Entity
@Table(name = "ai_cost_daily_summary", indexes = {
@Index(name = "idx_summary_date_tenant", columnList = "summary_date,tenant_id"),
@Index(name = "idx_summary_date_feature", columnList = "summary_date,feature_code"),
@Index(name = "idx_summary_unique",
columnList = "summary_date,tenant_id,feature_code,model_id",
unique = true)
})
@Data
@Builder
@NoArgsConstructor
@AllArgsConstructor
public class AiCostDailySummary {
@Id
@GeneratedValue(strategy = GenerationType.IDENTITY)
private Long id;
@Column(name = "summary_date", nullable = false)
private LocalDate summaryDate;
@Column(name = "tenant_id", length = 64)
private String tenantId;
@Column(name = "feature_code", length = 64)
private String featureCode;
@Column(name = "model_id", length = 64)
private String modelId;
// 聚合指标
@Column(name = "total_calls")
private Long totalCalls;
@Column(name = "success_calls")
private Long successCalls;
@Column(name = "failed_calls")
private Long failedCalls;
@Column(name = "total_input_tokens")
private Long totalInputTokens;
@Column(name = "total_output_tokens")
private Long totalOutputTokens;
@Column(name = "total_cost_cents")
private Long totalCostCents;
@Column(name = "avg_latency_ms")
private Integer avgLatencyMs;
@Column(name = "p95_latency_ms")
private Integer p95LatencyMs;
@Column(name = "unique_users")
private Long uniqueUsers;
@Column(name = "updated_at")
private Instant updatedAt;
}2.4 MySQL Schema（完整DDL）
-- 成本明细表（按月分区）
CREATE TABLE ai_cost_records (
id BIGINT NOT NULL AUTO_INCREMENT,
tenant_id VARCHAR(64),
user_id VARCHAR(64),
feature_code VARCHAR(64) NOT NULL,
scenario_code VARCHAR(64),
request_id VARCHAR(128) NOT NULL,
session_id VARCHAR(128),
model_id VARCHAR(64) NOT NULL,
model_provider VARCHAR(32),
input_tokens INT NOT NULL DEFAULT 0,
output_tokens INT NOT NULL DEFAULT 0,
total_tokens INT NOT NULL DEFAULT 0,
input_cost_cents BIGINT NOT NULL DEFAULT 0,
output_cost_cents BIGINT NOT NULL DEFAULT 0,
total_cost_cents BIGINT NOT NULL DEFAULT 0,
latency_ms INT,
first_token_latency_ms INT,
status VARCHAR(16) NOT NULL DEFAULT 'SUCCESS',
error_code VARCHAR(64),
tags JSON,
created_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3),
PRIMARY KEY (id, created_at),
UNIQUE KEY uk_request_id (request_id),
KEY idx_tenant_feature_time (tenant_id, feature_code, created_at),
KEY idx_user_time (user_id, created_at),
KEY idx_model_time (model_id, created_at),
KEY idx_feature_time (feature_code, created_at)
) ENGINE=InnoDB
PARTITION BY RANGE (UNIX_TIMESTAMP(created_at)) (
PARTITION p202601 VALUES LESS THAN (UNIX_TIMESTAMP('2026-02-01')),
PARTITION p202602 VALUES LESS THAN (UNIX_TIMESTAMP('2026-03-01')),
PARTITION p202603 VALUES LESS THAN (UNIX_TIMESTAMP('2026-04-01')),
-- 以此类推，每月一个分区
PARTITION p_future VALUES LESS THAN MAXVALUE
);
-- 每日汇总表
CREATE TABLE ai_cost_daily_summary (
id BIGINT NOT NULL AUTO_INCREMENT PRIMARY KEY,
summary_date DATE NOT NULL,
tenant_id VARCHAR(64),
feature_code VARCHAR(64),
model_id VARCHAR(64),
total_calls BIGINT DEFAULT 0,
success_calls BIGINT DEFAULT 0,
failed_calls BIGINT DEFAULT 0,
total_input_tokens BIGINT DEFAULT 0,
total_output_tokens BIGINT DEFAULT 0,
total_cost_cents BIGINT DEFAULT 0,
avg_latency_ms INT,
p95_latency_ms INT,
unique_users BIGINT DEFAULT 0,
updated_at DATETIME(3) DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3),
UNIQUE KEY uk_daily_summary (summary_date, tenant_id, feature_code, model_id),
KEY idx_date_tenant (summary_date, tenant_id),
KEY idx_date_feature (summary_date, feature_code)
) ENGINE=InnoDB;
-- 预算配置表
CREATE TABLE ai_budget_config (
id BIGINT NOT NULL AUTO_INCREMENT PRIMARY KEY,
tenant_id VARCHAR(64),
feature_code VARCHAR(64),
budget_period VARCHAR(16) NOT NULL DEFAULT 'MONTHLY', -- DAILY/WEEKLY/MONTHLY
budget_cents BIGINT NOT NULL,
alert_threshold_pct INT NOT NULL DEFAULT 80, -- 80%时告警
hard_limit_pct INT NOT NULL DEFAULT 100, -- 100%时限流
enabled TINYINT NOT NULL DEFAULT 1,
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
updated_at DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
UNIQUE KEY uk_budget (tenant_id, feature_code, budget_period)
) ENGINE=InnoDB;三、Spring AI Advisor实现透明成本追踪
Spring AI的Advisor机制是实现透明追踪的关键。通过Advisor，我们可以在不修改业务代码的前提下，拦截所有AI调用并记录成本。
3.1 成本追踪Advisor核心实现
@Component
@Slf4j
public class CostTrackingAdvisor implements CallAroundAdvisor, StreamAroundAdvisor {
private final CostRecordService costRecordService;
private final ModelPricingService pricingService;
private final MeterRegistry meterRegistry;
// 上下文Key，用于在请求中传递归因信息
public static final String COST_CONTEXT_KEY = "cost_context";
public CostTrackingAdvisor(
CostRecordService costRecordService,
ModelPricingService pricingService,
MeterRegistry meterRegistry) {
this.costRecordService = costRecordService;
this.pricingService = pricingService;
this.meterRegistry = meterRegistry;
}
@Override
public String getName() {
return "CostTrackingAdvisor";
}
@Override
public int getOrder() {
return Ordered.LOWEST_PRECEDENCE - 100; // 在最后执行，确保能获取完整的usage信息
}
@Override
public AdvisedResponse aroundCall(AdvisedRequest advisedRequest, CallAroundAdvisorChain chain) {
long startTime = System.currentTimeMillis();
CostContext costContext = extractCostContext(advisedRequest);
AdvisedResponse response = null;
CallStatus status = CallStatus.SUCCESS;
String errorCode = null;
try {
response = chain.nextAroundCall(advisedRequest);
return response;
} catch (Exception e) {
status = CallStatus.FAILED;
errorCode = e.getClass().getSimpleName();
throw e;
} finally {
long latencyMs = System.currentTimeMillis() - startTime;
if (response != null || status == CallStatus.FAILED) {
recordCost(costContext, advisedRequest, response, status, errorCode, latencyMs);
}
}
}
@Override
public Flux<AdvisedResponse> aroundStream(AdvisedRequest advisedRequest, StreamAroundAdvisorChain chain) {
long startTime = System.currentTimeMillis();
CostContext costContext = extractCostContext(advisedRequest);
// 使用AtomicReference收集流式响应的最终usage
AtomicReference<Usage> usageRef = new AtomicReference<>();
return chain.nextAroundStream(advisedRequest)
.doOnNext(response -> {
// 流式响应中，最后一条消息通常包含完整的usage信息
if (response.response() != null && response.response().getMetadata() != null) {
Usage usage = response.response().getMetadata().getUsage();
if (usage != null && usage.getTotalTokens() > 0) {
usageRef.set(usage);
}
}
})
.doOnComplete(() -> {
long latencyMs = System.currentTimeMillis() - startTime;
Usage finalUsage = usageRef.get();
if (finalUsage != null) {
recordCostFromUsage(costContext, advisedRequest, finalUsage,
CallStatus.SUCCESS, null, latencyMs);
}
})
.doOnError(e -> {
long latencyMs = System.currentTimeMillis() - startTime;
recordCostFromUsage(costContext, advisedRequest, null,
CallStatus.FAILED, e.getClass().getSimpleName(), latencyMs);
});
}
private CostContext extractCostContext(AdvisedRequest request) {
// 从请求上下文中提取归因信息
Map<String, Object> context = request.adviseContext();
CostContext costContext = (CostContext) context.get(COST_CONTEXT_KEY);
if (costContext == null) {
// 如果没有设置成本上下文，使用默认值并告警
log.warn("No CostContext found in request, using defaults. " +
"Please set CostContext for accurate cost attribution. " +
"Request context keys: {}", context.keySet());
costContext = CostContext.builder()
.featureCode("unknown")
.tenantId("unknown")
.userId("anonymous")
.timestamp(Instant.now())
.build();
}
return costContext;
}
private void recordCost(CostContext costContext, AdvisedRequest request,
AdvisedResponse response, CallStatus status,
String errorCode, long latencyMs) {
Usage usage = null;
if (response != null && response.response() != null) {
ChatResponseMetadata metadata = response.response().getMetadata();
if (metadata != null) {
usage = metadata.getUsage();
}
}
recordCostFromUsage(costContext, request, usage, status, errorCode, latencyMs);
}
private void recordCostFromUsage(CostContext costContext, AdvisedRequest request,
Usage usage, CallStatus status,
String errorCode, long latencyMs) {
try {
String modelId = extractModelId(request);
int inputTokens = usage != null ? (int) usage.getPromptTokens() : 0;
int outputTokens = usage != null ? (int) usage.getGenerationTokens() : 0;
// 计算成本（单位：美分的1/100，即0.01美分）
ModelPricing pricing = pricingService.getPricing(modelId);
long inputCostCents = pricing.calculateInputCost(inputTokens);
long outputCostCents = pricing.calculateOutputCost(outputTokens);
AiCostRecord record = AiCostRecord.builder()
.tenantId(costContext.getTenantId())
.userId(costContext.getUserId())
.featureCode(costContext.getFeatureCode())
.scenarioCode(costContext.getScenarioCode())
.requestId(costContext.getRequestId() != null ?
costContext.getRequestId() : UUID.randomUUID().toString())
.sessionId(costContext.getSessionId())
.modelId(modelId)
.modelProvider(pricing.getProvider())
.inputTokens(inputTokens)
.outputTokens(outputTokens)
.totalTokens(inputTokens + outputTokens)
.inputCostCents(inputCostCents)
.outputCostCents(outputCostCents)
.totalCostCents(inputCostCents + outputCostCents)
.latencyMs((int) latencyMs)
.status(status)
.errorCode(errorCode)
.tags(costContext.getTags())
.createdAt(Instant.now())
.build();
// 异步保存，不阻塞主流程
costRecordService.saveAsync(record);
// 上报Prometheus指标
recordMetrics(record);
} catch (Exception e) {
log.error("Failed to record AI cost, this will cause cost tracking inaccuracy", e);
}
}
private String extractModelId(AdvisedRequest request) {
// 从请求中提取模型ID
if (request.chatOptions() instanceof OpenAiChatOptions openAiOptions) {
return openAiOptions.getModel();
}
if (request.chatOptions() instanceof AnthropicChatOptions anthropicOptions) {
return anthropicOptions.getModel();
}
return "unknown-model";
}
private void recordMetrics(AiCostRecord record) {
Tags tags = Tags.of(
"feature", record.getFeatureCode(),
"model", record.getModelId(),
"tenant", record.getTenantId() != null ? record.getTenantId() : "default",
"status", record.getStatus().name()
);
meterRegistry.counter("ai.cost.total_cents", tags)
.increment(record.getTotalCostCents());
meterRegistry.counter("ai.cost.input_tokens", tags)
.increment(record.getInputTokens());
meterRegistry.counter("ai.cost.output_tokens", tags)
.increment(record.getOutputTokens());
meterRegistry.counter("ai.cost.calls", tags).increment();
if (record.getLatencyMs() != null) {
meterRegistry.timer("ai.cost.latency", tags)
.record(record.getLatencyMs(), TimeUnit.MILLISECONDS);
}
}
}3.2 模型定价服务
@Service
@Slf4j
public class ModelPricingService {
// 所有价格单位：每1000 tokens的美分数（0.01美分精度）
// 避免浮点数精度问题
private static final Map<String, ModelPricing> PRICING_TABLE = new HashMap<>();
static {
// OpenAI GPT-4o
PRICING_TABLE.put("gpt-4o", ModelPricing.builder()
.modelId("gpt-4o")
.provider("openai")
.inputCostPerKTokenCents(25L) // $0.0025/1K = 0.25美分/1K
.outputCostPerKTokenCents(100L) // $0.01/1K = 1美分/1K
.build());
// OpenAI GPT-4o-mini
PRICING_TABLE.put("gpt-4o-mini", ModelPricing.builder()
.modelId("gpt-4o-mini")
.provider("openai")
.inputCostPerKTokenCents(2L) // $0.00015/1K
.outputCostPerKTokenCents(6L) // $0.0006/1K
.build());
// Claude 3.5 Sonnet
PRICING_TABLE.put("claude-3-5-sonnet-20241022", ModelPricing.builder()
.modelId("claude-3-5-sonnet-20241022")
.provider("anthropic")
.inputCostPerKTokenCents(30L) // $0.003/1K
.outputCostPerKTokenCents(150L) // $0.015/1K
.build());
// Claude 3 Haiku
PRICING_TABLE.put("claude-3-haiku-20240307", ModelPricing.builder()
.modelId("claude-3-haiku-20240307")
.provider("anthropic")
.inputCostPerKTokenCents(3L) // $0.00025/1K
.outputCostPerKTokenCents(13L) // $0.00125/1K
.build());
// OpenAI Embeddings
PRICING_TABLE.put("text-embedding-3-small", ModelPricing.builder()
.modelId("text-embedding-3-small")
.provider("openai")
.inputCostPerKTokenCents(0L) // $0.00002/1K ≈ 0
.outputCostPerKTokenCents(0L)
.build());
}
public ModelPricing getPricing(String modelId) {
ModelPricing pricing = PRICING_TABLE.get(modelId);
if (pricing == null) {
log.warn("Unknown model pricing for: {}, using zero cost", modelId);
return ModelPricing.builder()
.modelId(modelId)
.provider("unknown")
.inputCostPerKTokenCents(0L)
.outputCostPerKTokenCents(0L)
.build();
}
return pricing;
}
@Data
@Builder
public static class ModelPricing {
private String modelId;
private String provider;
private Long inputCostPerKTokenCents; // 每1000 tokens的成本（美分）
private Long outputCostPerKTokenCents;
public long calculateInputCost(int tokens) {
return (long) tokens * inputCostPerKTokenCents / 1000;
}
public long calculateOutputCost(int tokens) {
return (long) tokens * outputCostPerKTokenCents / 1000;
}
}
}3.3 在业务代码中传递成本上下文
@Service
@RequiredArgsConstructor
public class CustomerServiceAI {
private final ChatClient chatClient;
private final HttpServletRequest httpRequest;
public String handleCustomerQuery(String userId, String tenantId, String query) {
// 构建成本归因上下文
CostContext costContext = CostContext.builder()
.tenantId(tenantId)
.userId(userId)
.featureCode("customer_service")
.scenarioCode("chat")
.requestId(UUID.randomUUID().toString())
.sessionId(getSessionId())
.timestamp(Instant.now())
.tags(Map.of(
"channel", "web",
"query_type", classifyQuery(query)
))
.build();
// 通过advisorContext传递成本上下文
return chatClient.prompt()
.user(query)
.advisors(advisor -> advisor.param(
CostTrackingAdvisor.COST_CONTEXT_KEY, costContext))
.call()
.content();
}
private String getSessionId() {
return Optional.ofNullable(httpRequest.getSession(false))
.map(session -> session.getId())
.orElse(null);
}
private String classifyQuery(String query) {
// 简单分类，不调AI，避免递归成本追踪
if (query.contains("退款") || query.contains("退货")) return "refund";
if (query.contains("物流") || query.contains("快递")) return "logistics";
return "general";
}
}3.4 异步成本记录服务（高性能批量写入）
@Service
@Slf4j
public class CostRecordService {
private final AiCostRecordRepository repository;
private final BlockingQueue<AiCostRecord> buffer;
private final ScheduledExecutorService flushScheduler;
private static final int BUFFER_SIZE = 10000;
private static final int BATCH_SIZE = 500;
private static final long FLUSH_INTERVAL_MS = 5000; // 5秒批量写入一次
public CostRecordService(AiCostRecordRepository repository) {
this.repository = repository;
this.buffer = new LinkedBlockingQueue<>(BUFFER_SIZE);
this.flushScheduler = Executors.newSingleThreadScheduledExecutor(
r -> new Thread(r, "cost-record-flusher"));
// 定期批量写入
flushScheduler.scheduleAtFixedRate(
this::flushBuffer,
FLUSH_INTERVAL_MS,
FLUSH_INTERVAL_MS,
TimeUnit.MILLISECONDS
);
}
public void saveAsync(AiCostRecord record) {
boolean offered = buffer.offer(record);
if (!offered) {
// 缓冲区满，直接丢弃并告警（成本数据允许少量丢失）
log.warn("Cost record buffer full, dropping record for feature: {}",
record.getFeatureCode());
Metrics.counter("ai.cost.records_dropped").increment();
}
}
@Scheduled(fixedDelay = 5000)
public void flushBuffer() {
if (buffer.isEmpty()) return;
List<AiCostRecord> batch = new ArrayList<>(BATCH_SIZE);
buffer.drainTo(batch, BATCH_SIZE);
if (!batch.isEmpty()) {
try {
repository.saveAll(batch);
log.debug("Flushed {} cost records to database", batch.size());
Metrics.counter("ai.cost.records_saved").increment(batch.size());
} catch (Exception e) {
log.error("Failed to save {} cost records, re-queuing", batch.size(), e);
// 写入失败，重新入队（可能导致重复，需要幂等处理）
batch.forEach(r -> buffer.offer(r));
}
}
}
@PreDestroy
public void shutdown() {
// 应用关闭时，确保所有缓冲数据都写入
log.info("Flushing remaining {} cost records before shutdown", buffer.size());
flushBuffer();
flushScheduler.shutdown();
}
}四、成本查询API与聚合逻辑
4.1 成本查询服务
@Service
@RequiredArgsConstructor
@Slf4j
public class CostQueryService {
private final AiCostRecordRepository recordRepository;
private final AiCostDailySummaryRepository summaryRepository;
private final RedisTemplate<String, Object> redisTemplate;
/**
* 按功能查询成本（使用预聚合表，毫秒级响应）
*/
public List<FeatureCostSummary> getFeatureCostByDateRange(
String tenantId, LocalDate startDate, LocalDate endDate) {
String cacheKey = String.format("cost:feature:%s:%s:%s", tenantId, startDate, endDate);
@SuppressWarnings("unchecked")
List<FeatureCostSummary> cached =
(List<FeatureCostSummary>) redisTemplate.opsForValue().get(cacheKey);
if (cached != null) return cached;
List<FeatureCostSummary> result = summaryRepository
.findFeatureCostSummary(tenantId, startDate, endDate);
redisTemplate.opsForValue().set(cacheKey, result, Duration.ofMinutes(10));
return result;
}
/**
* 按用户查询Top N成本用户
*/
public List<UserCostSummary> getTopCostUsers(
String tenantId, LocalDate startDate, LocalDate endDate, int topN) {
return recordRepository.findTopUsersByCost(tenantId, startDate, endDate, topN);
}
/**
* 获取实时成本（当日，从明细表查询）
*/
public TodayCostSummary getTodayCost(String tenantId) {
LocalDate today = LocalDate.now();
Instant startOfDay = today.atStartOfDay(ZoneId.systemDefault()).toInstant();
return recordRepository.aggregateCostSince(tenantId, startOfDay);
}
/**
* 成本趋势分析（同比/环比）
*/
public CostTrendAnalysis analyzeCostTrend(String tenantId, String featureCode, int months) {
LocalDate endDate = LocalDate.now();
LocalDate startDate = endDate.minusMonths(months);
List<MonthlyCostData> monthlyCosts = summaryRepository
.findMonthlyCostByFeature(tenantId, featureCode, startDate, endDate);
return CostTrendAnalysis.builder()
.tenantId(tenantId)
.featureCode(featureCode)
.monthlyCosts(monthlyCosts)
.monthOverMonthGrowth(calculateMoMGrowth(monthlyCosts))
.projectedNextMonthCost(projectNextMonth(monthlyCosts))
.build();
}
}4.2 Repository层关键查询
@Repository
public interface AiCostRecordRepository extends JpaRepository<AiCostRecord, Long> {
@Query(value = """
SELECT
r.user_id AS userId,
SUM(r.total_cost_cents) AS totalCostCents,
SUM(r.total_tokens) AS totalTokens,
COUNT(*) AS totalCalls,
COUNT(DISTINCT DATE(r.created_at)) AS activeDays
FROM ai_cost_records r
WHERE r.tenant_id = :tenantId
AND r.created_at >= :startDate
AND r.created_at < :endDate
GROUP BY r.user_id
ORDER BY totalCostCents DESC
LIMIT :topN
""", nativeQuery = true)
List<UserCostSummary> findTopUsersByCost(
@Param("tenantId") String tenantId,
@Param("startDate") LocalDate startDate,
@Param("endDate") LocalDate endDate,
@Param("topN") int topN
);
@Query(value = """
SELECT
SUM(total_cost_cents) AS totalCostCents,
SUM(input_tokens) AS inputTokens,
SUM(output_tokens) AS outputTokens,
COUNT(*) AS totalCalls,
COUNT(DISTINCT user_id) AS uniqueUsers,
AVG(latency_ms) AS avgLatencyMs
FROM ai_cost_records
WHERE tenant_id = :tenantId
AND created_at >= :since
""", nativeQuery = true)
TodayCostSummary aggregateCostSince(
@Param("tenantId") String tenantId,
@Param("since") Instant since
);
}@Repository
public interface AiCostDailySummaryRepository
extends JpaRepository<AiCostDailySummary, Long> {
@Query(value = """
SELECT
s.feature_code AS featureCode,
SUM(s.total_cost_cents) AS totalCostCents,
SUM(s.total_calls) AS totalCalls,
SUM(s.total_input_tokens) AS totalInputTokens,
SUM(s.total_output_tokens) AS totalOutputTokens,
AVG(s.avg_latency_ms) AS avgLatencyMs
FROM ai_cost_daily_summary s
WHERE (s.tenant_id = :tenantId OR :tenantId IS NULL)
AND s.summary_date BETWEEN :startDate AND :endDate
GROUP BY s.feature_code
ORDER BY totalCostCents DESC
""", nativeQuery = true)
List<FeatureCostSummary> findFeatureCostSummary(
@Param("tenantId") String tenantId,
@Param("startDate") LocalDate startDate,
@Param("endDate") LocalDate endDate
);
}五、Redis实现预算告警与限流
5.1 预算告警服务
@Service
@RequiredArgsConstructor
@Slf4j
public class BudgetAlertService {
private final RedisTemplate<String, String> redisTemplate;
private final BudgetConfigRepository budgetConfigRepository;
private final AlertNotificationService alertService;
private static final String BUDGET_USAGE_KEY = "budget:usage:%s:%s:%s";
// budget:usage:{tenantId}:{featureCode}:{period}
/**
* 检查并累计预算使用量
* 返回：是否允许继续调用
*/
public BudgetCheckResult checkAndAccumulate(
String tenantId, String featureCode, long estimatedCostCents) {
List<BudgetConfig> configs = budgetConfigRepository
.findByTenantIdAndFeatureCode(tenantId, featureCode);
if (configs.isEmpty()) {
return BudgetCheckResult.allowed();
}
for (BudgetConfig config : configs) {
if (!config.isEnabled()) continue;
String period = getCurrentPeriod(config.getBudgetPeriod());
String key = String.format(BUDGET_USAGE_KEY, tenantId, featureCode, period);
// 原子性地检查和更新预算使用量
Long currentUsage = redisTemplate.opsForValue().increment(key, estimatedCostCents);
// 设置过期时间（根据预算周期）
if (currentUsage != null && currentUsage == estimatedCostCents) {
// 第一次设置，设置过期时间
Duration ttl = getPeriodTTL(config.getBudgetPeriod());
redisTemplate.expire(key, ttl);
}
if (currentUsage == null) continue;
long budget = config.getBudgetCents();
double usageRate = (double) currentUsage / budget;
// 超过硬限制，拒绝调用
if (usageRate >= config.getHardLimitPct() / 100.0) {
log.warn("Budget hard limit reached for tenant={}, feature={}, " +
"usage={}/{} cents", tenantId, featureCode, currentUsage, budget);
return BudgetCheckResult.denied(
String.format("预算已用%.1f%%，当前周期已达上限", usageRate * 100));
}
// 超过告警阈值，发送告警
if (usageRate >= config.getAlertThresholdPct() / 100.0) {
sendBudgetAlert(tenantId, featureCode, currentUsage, budget, usageRate);
}
}
return BudgetCheckResult.allowed();
}
private void sendBudgetAlert(String tenantId, String featureCode,
long currentUsage, long budget, double usageRate) {
String alertKey = String.format("budget:alerted:%s:%s", tenantId, featureCode);
Boolean alreadyAlerted = redisTemplate.opsForValue()
.setIfAbsent(alertKey, "1", Duration.ofHours(1));
if (Boolean.TRUE.equals(alreadyAlerted)) {
// 1小时内只发一次告警
alertService.sendBudgetAlert(BudgetAlert.builder()
.tenantId(tenantId)
.featureCode(featureCode)
.currentUsageCents(currentUsage)
.budgetCents(budget)
.usageRate(usageRate)
.build());
}
}
private String getCurrentPeriod(String budgetPeriod) {
LocalDate now = LocalDate.now();
return switch (budgetPeriod) {
case "DAILY" -> now.format(DateTimeFormatter.BASIC_ISO_DATE);
case "WEEKLY" -> now.getYear() + "W" + now.get(WeekFields.ISO.weekOfWeekBasedYear());
case "MONTHLY" -> now.format(DateTimeFormatter.ofPattern("yyyyMM"));
default -> now.format(DateTimeFormatter.ofPattern("yyyyMM"));
};
}
private Duration getPeriodTTL(String budgetPeriod) {
return switch (budgetPeriod) {
case "DAILY" -> Duration.ofDays(2);
case "WEEKLY" -> Duration.ofDays(8);
case "MONTHLY" -> Duration.ofDays(32);
default -> Duration.ofDays(32);
};
}
@Data
@Builder
public static class BudgetCheckResult {
private boolean allowed;
private String denyReason;
public static BudgetCheckResult allowed() {
return BudgetCheckResult.builder().allowed(true).build();
}
public static BudgetCheckResult denied(String reason) {
return BudgetCheckResult.builder().allowed(false).denyReason(reason).build();
}
}
}5.2 将预算检查集成到Advisor
@Component
@RequiredArgsConstructor
@Slf4j
public class BudgetGuardAdvisor implements CallAroundAdvisor {
private final BudgetAlertService budgetAlertService;
private final ModelPricingService pricingService;
@Override
public String getName() {
return "BudgetGuardAdvisor";
}
@Override
public int getOrder() {
return Ordered.HIGHEST_PRECEDENCE; // 最先执行，在调用之前检查预算
}
@Override
public AdvisedResponse aroundCall(AdvisedRequest advisedRequest, CallAroundAdvisorChain chain) {
CostContext costContext = (CostContext) advisedRequest.adviseContext()
.get(CostTrackingAdvisor.COST_CONTEXT_KEY);
if (costContext != null) {
// 估算本次调用成本（基于输入token数）
int estimatedInputTokens = estimateInputTokens(advisedRequest);
ModelPricingService.ModelPricing pricing =
pricingService.getPricing(extractModelId(advisedRequest));
long estimatedCost = pricing.calculateInputCost(estimatedInputTokens) * 3; // 预估3倍余量
BudgetAlertService.BudgetCheckResult result =
budgetAlertService.checkAndAccumulate(
costContext.getTenantId(),
costContext.getFeatureCode(),
estimatedCost
);
if (!result.isAllowed()) {
throw new BudgetExceededException(result.getDenyReason());
}
}
return chain.nextAroundCall(advisedRequest);
}
private int estimateInputTokens(AdvisedRequest request) {
// 粗略估算：字符数 / 4
String userText = request.userText() != null ? request.userText() : "";
return Math.max(100, userText.length() / 4);
}
private String extractModelId(AdvisedRequest request) {
if (request.chatOptions() instanceof OpenAiChatOptions opts) {
return opts.getModel();
}
return "gpt-4o-mini";
}
}六、Grafana成本看板配置
6.1 关键Grafana Panel配置（JSON）
{
"title": "AI Cost Attribution Dashboard",
"panels": [
{
"id": 1,
"title": "当月总成本趋势",
"type": "timeseries",
"gridPos": {"x": 0, "y": 0, "w": 24, "h": 8},
"targets": [
{
"expr": "sum(increase(ai_cost_total_cents_total[1d])) by (feature) / 100",
"legendFormat": "{{feature}}"
}
],
"fieldConfig": {
"defaults": {
"unit": "currencyUSD",
"thresholds": {
"steps": [
{"color": "green", "value": 0},
{"color": "yellow", "value": 5000},
{"color": "red", "value": 10000}
]
}
}
}
},
{
"id": 2,
"title": "按功能成本分布",
"type": "piechart",
"gridPos": {"x": 0, "y": 8, "w": 12, "h": 8},
"targets": [
{
"expr": "sum(increase(ai_cost_total_cents_total[30d])) by (feature) / 100",
"legendFormat": "{{feature}}"
}
]
},
{
"id": 3,
"title": "模型成本对比",
"type": "bargauge",
"gridPos": {"x": 12, "y": 8, "w": 12, "h": 8},
"targets": [
{
"expr": "sum(increase(ai_cost_total_cents_total[30d])) by (model) / 100",
"legendFormat": "{{model}}"
}
]
},
{
"id": 4,
"title": "预算使用率告警",
"type": "stat",
"gridPos": {"x": 0, "y": 16, "w": 8, "h": 4},
"targets": [
{
"expr": "ai_budget_usage_percent",
"legendFormat": "{{feature}}"
}
],
"fieldConfig": {
"defaults": {
"thresholds": {
"steps": [
{"color": "green", "value": 0},
{"color": "yellow", "value": 80},
{"color": "red", "value": 100}
]
},
"unit": "percent"
}
}
},
{
"id": 5,
"title": "每请求平均成本（美分）",
"type": "timeseries",
"gridPos": {"x": 8, "y": 16, "w": 16, "h": 4},
"targets": [
{
"expr": "sum(rate(ai_cost_total_cents_total[5m])) by (feature) / sum(rate(ai_cost_calls_total[5m])) by (feature)",
"legendFormat": "{{feature}} 平均成本"
}
]
}
]
}七、成本优化建议引擎
7.1 自动发现优化机会
@Service
@RequiredArgsConstructor
@Slf4j
public class CostOptimizationEngine {
private final AiCostDailySummaryRepository summaryRepository;
private final ChatClient chatClient;
/**
* 分析最近30天的成本数据，自动生成优化建议
*/
public List<OptimizationSuggestion> analyzeAndSuggest(String tenantId) {
List<OptimizationSuggestion> suggestions = new ArrayList<>();
LocalDate endDate = LocalDate.now();
LocalDate startDate = endDate.minusDays(30);
List<FeatureCostSummary> featureCosts = summaryRepository
.findFeatureCostSummary(tenantId, startDate, endDate);
// 规则1：检测是否有功能使用了过贵的模型
suggestions.addAll(detectOverPricedModelUsage(featureCosts));
// 规则2：检测重复查询（相似度高的query，可以用缓存）
suggestions.addAll(detectCacheablePatterns(tenantId, startDate, endDate));
// 规则3：检测输入Token过长的功能
suggestions.addAll(detectVerbosePrompts(featureCosts));
// 规则4：检测失败率高的功能（失败的Token也付费）
suggestions.addAll(detectHighFailureRateCosts(featureCosts));
// 按预计节省金额排序
suggestions.sort(Comparator.comparing(OptimizationSuggestion::getEstimatedMonthlySavingCents)
.reversed());
return suggestions;
}
private List<OptimizationSuggestion> detectOverPricedModelUsage(
List<FeatureCostSummary> featureCosts) {
List<OptimizationSuggestion> suggestions = new ArrayList<>();
for (FeatureCostSummary feature : featureCosts) {
// 如果功能使用的是GPT-4o，但平均输入token较少（简单任务），建议降级
if ("gpt-4o".equals(feature.getPrimaryModel()) &&
feature.getAvgInputTokens() < 500 &&
feature.getTotalCostCents() > 10000) { // 超过$1
long potentialSaving = feature.getTotalCostCents() * 85 / 100; // 降级可省85%
suggestions.add(OptimizationSuggestion.builder()
.featureCode(feature.getFeatureCode())
.type(SuggestionType.MODEL_DOWNGRADE)
.title("建议将 " + feature.getFeatureCode() + " 从 GPT-4o 降级到 GPT-4o-mini")
.description(String.format(
"该功能平均输入仅%d tokens，属于简单任务，使用GPT-4o过于昂贵。" +
"GPT-4o-mini成本低16倍，建议先A/B测试质量是否满足要求。",
feature.getAvgInputTokens()
))
.estimatedMonthlySavingCents(potentialSaving)
.implementationDifficulty("LOW")
.build());
}
}
return suggestions;
}
private List<OptimizationSuggestion> detectVerbosePrompts(
List<FeatureCostSummary> featureCosts) {
List<OptimizationSuggestion> suggestions = new ArrayList<>();
for (FeatureCostSummary feature : featureCosts) {
double inputOutputRatio = (double) feature.getTotalInputTokens() /
Math.max(1, feature.getTotalOutputTokens());
// 如果输入/输出比超过10:1，说明有大量系统提示词或上下文
if (inputOutputRatio > 10.0 && feature.getTotalCostCents() > 5000) {
long potentialSaving = feature.getTotalInputTokens() > 0 ?
(long) (feature.getTotalCostCents() * 0.3) : 0; // 压缩30%
suggestions.add(OptimizationSuggestion.builder()
.featureCode(feature.getFeatureCode())
.type(SuggestionType.PROMPT_COMPRESSION)
.title("优化 " + feature.getFeatureCode() + " 的系统提示词")
.description(String.format(
"该功能的输入/输出Token比为%.1f:1（正常应为3-5:1），" +
"说明系统提示词过长或注入了过多上下文。建议精简提示词或引入提示词压缩技术。",
inputOutputRatio
))
.estimatedMonthlySavingCents(potentialSaving)
.implementationDifficulty("MEDIUM")
.build());
}
}
return suggestions;
}
@Data
@Builder
public static class OptimizationSuggestion {
private String featureCode;
private SuggestionType type;
private String title;
private String description;
private long estimatedMonthlySavingCents;
private String implementationDifficulty; // LOW/MEDIUM/HIGH
}
public enum SuggestionType {
MODEL_DOWNGRADE, PROMPT_COMPRESSION, CACHING, FAILURE_REDUCTION, BATCHING
}
}八、Spring Batch月度成本报告
8.1 月度报告Job配置
@Configuration
@RequiredArgsConstructor
public class CostReportBatchConfig {
private final JobRepository jobRepository;
private final PlatformTransactionManager transactionManager;
private final CostQueryService costQueryService;
private final EmailService emailService;
private final TenantRepository tenantRepository;
@Bean
public Job monthlyCostReportJob() {
return new JobBuilder("monthlyCostReportJob", jobRepository)
.start(generateReportStep())
.next(sendReportStep())
.build();
}
@Bean
public Step generateReportStep() {
return new StepBuilder("generateReportStep", jobRepository)
.<Tenant, MonthlyCostReport>chunk(10, transactionManager)
.reader(tenantReader())
.processor(reportProcessor())
.writer(reportWriter())
.build();
}
@Bean
public ItemReader<Tenant> tenantReader() {
return new RepositoryItemReaderBuilder<Tenant>()
.repository(tenantRepository)
.methodName("findAllActive")
.sorts(Map.of("id", Sort.Direction.ASC))
.build();
}
@Bean
public ItemProcessor<Tenant, MonthlyCostReport> reportProcessor() {
return tenant -> {
LocalDate lastMonth = LocalDate.now().minusMonths(1);
LocalDate startDate = lastMonth.withDayOfMonth(1);
LocalDate endDate = lastMonth.withDayOfMonth(lastMonth.lengthOfMonth());
List<FeatureCostSummary> featureCosts = costQueryService
.getFeatureCostByDateRange(tenant.getId(), startDate, endDate);
List<OptimizationSuggestion> suggestions = costOptimizationEngine
.analyzeAndSuggest(tenant.getId());
return MonthlyCostReport.builder()
.tenantId(tenant.getId())
.tenantName(tenant.getName())
.reportMonth(lastMonth.format(DateTimeFormatter.ofPattern("yyyy年MM月")))
.featureCosts(featureCosts)
.totalCostCents(featureCosts.stream()
.mapToLong(FeatureCostSummary::getTotalCostCents).sum())
.topOptimizationSuggestions(suggestions.stream().limit(5).toList())
.build();
};
}
@Bean
public Step sendReportStep() {
return new StepBuilder("sendReportStep", jobRepository)
.<MonthlyCostReport, MonthlyCostReport>chunk(10, transactionManager)
.reader(reportReader())
.writer(emailReportWriter())
.build();
}
// 定时：每月1日9:00执行
@Scheduled(cron = "0 0 9 1 * ?")
public void triggerMonthlyReport() throws Exception {
JobParameters params = new JobParametersBuilder()
.addLocalDate("reportDate", LocalDate.now())
.toJobParameters();
jobLauncher.run(monthlyCostReportJob(), params);
}
}九、ROI计算：AI功能的投入产出比
9.1 ROI计算模型
@Service
@RequiredArgsConstructor
public class AiRoiCalculator {
private final CostQueryService costQueryService;
private final BusinessMetricsService businessMetrics;
/**
* 计算AI功能的ROI
* ROI = (收益 - 成本) / 成本 × 100%
*/
public RoiResult calculateRoi(String featureCode, LocalDate startDate, LocalDate endDate) {
// 1. 计算AI直接成本
long aiCostCents = costQueryService
.getFeatureTotalCost(featureCode, startDate, endDate);
// 2. 计算人力成本节省
BusinessMetrics metrics = businessMetrics.getMetrics(featureCode, startDate, endDate);
// 客服场景：每次AI处理节省5分钟人工，人工成本$0.5/分钟
long laborSavingCents = 0;
if ("customer_service".equals(featureCode)) {
laborSavingCents = metrics.getAiHandledCount() * 5 * 50; // 5分钟 * 50美分/分钟
}
// 代码补全场景：提升30%开发效率，按人力成本折算
if ("code_assist".equals(featureCode)) {
long developerHourCostCents = 5000; // $50/小时
laborSavingCents = (long) (metrics.getActiveUsers() *
metrics.getAvgDailyUsageHours() *
0.3 * developerHourCostCents *
daysBetween(startDate, endDate));
}
// 3. 计算收入影响（需要业务数据支持）
long revenueImpactCents = businessMetrics
.getAiAttributedRevenueCents(featureCode, startDate, endDate);
// 4. 计算ROI
long totalBenefitCents = laborSavingCents + revenueImpactCents;
double roi = aiCostCents > 0 ?
(double)(totalBenefitCents - aiCostCents) / aiCostCents * 100 : 0;
return RoiResult.builder()
.featureCode(featureCode)
.periodDays(daysBetween(startDate, endDate))
.aiCostCents(aiCostCents)
.laborSavingCents(laborSavingCents)
.revenueImpactCents(revenueImpactCents)
.totalBenefitCents(totalBenefitCents)
.roiPercent(roi)
.paybackDays(calculatePaybackDays(aiCostCents, totalBenefitCents,
daysBetween(startDate, endDate)))
.build();
}
@Data
@Builder
public static class RoiResult {
private String featureCode;
private int periodDays;
private long aiCostCents; // AI成本
private long laborSavingCents; // 人力节省
private long revenueImpactCents; // 收入影响
private long totalBenefitCents; // 总收益
private double roiPercent; // ROI百分比
private int paybackDays; // 回本天数
public String getFormattedRoi() {
if (roiPercent >= 0) {
return String.format("+%.1f%%（每投入$1，产出$%.2f）",
roiPercent, 1 + roiPercent/100);
}
return String.format("%.1f%%（尚未回本）", roiPercent);
}
}
}十、性能基准数据
在陈磊的团队实际运行中，这套成本归因系统的性能表现：
| 指标 | 数值 |
|---|---|
| Advisor性能开销 | < 0.5ms（异步写入，不阻塞主流程） |
| 批量写入吞吐 | 5000条/秒（500条/批次，10批次/秒） |
| 查询响应时间（预聚合） | < 50ms（命中缓存）/ < 200ms（数据库） |
| 成本数据延迟 | 最大5秒（批量写入间隔） |
| 归因准确率 | 99.7%（0.3%因缓冲区溢出丢失） |
| 月报生成时间 | < 2分钟（Spring Batch并行处理） |
系统上线后的实际效果：
- 从"不知道钱花哪了"到"精确到功能级别的成本归因"
- 通过优化建议，3个月内将AI月均成本从19.8万降至11.2万（降低43%）
- 最大的节省来源：将5个功能的模型从GPT-4o降级到GPT-4o-mini（占总节省的67%）
FAQ
Q1：如果AI调用是在消息队列中异步处理的，成本上下文如何传递？
A：使用消息头（Header）传递CostContext的序列化JSON。消费者端在处理消息时，从Header中重建CostContext并放入ThreadLocal，Advisor从ThreadLocal中读取。需要注意的是，跨线程时ThreadLocal不会自动传递，需要使用TransmittableThreadLocal（阿里巴巴TTL库）。
Q2：同一个请求中调用了多次AI（链式调用），如何归因？
A：每次调用都会被Advisor独立记录，但它们会共享同一个requestId。查询时可以通过requestId聚合，也可以按sessionId聚合整个会话的成本。
Q3：估算成本和实际账单差多少？
A：主要误差来源是Token计数。OpenAI的Token计数使用tiktoken，我们用字符数/4估算，误差通常在5-15%。建议用tiktoken4j库做精确计数，但这会增加少量CPU开销。
Q4：向量数据库的成本怎么归因？
A：向量检索的成本需要在RAG的Service层手动记录。每次检索操作，创建一个类型为VECTOR_SEARCH的CostContext，手动调用CostRecordService.saveAsync()，成本金额根据Pinecone/Weaviate的计费规则手动计算。
Q5：多租户场景下，租户能看到自己的成本数据吗？
A：可以。在查询接口加上租户ID过滤，再加上行级数据隔离（RLS）或在DAO层强制注入租户条件。建议每个租户都有独立的成本看板，可以在Grafana中通过变量实现。