cost-accrual-tracker

成本累计跟踪器

在大模型执行期间实时跟踪 API 成本，并支持中止时记录部分成本。

适用场景

✅ 适用于：

• 实现执行期间的实时成本跟踪
• 执行中止时记录部分成本
• 为执行界面构建成本显示组件
• 将 Token 计数集成到执行流水线中
• 添加超出预算阈值时自动停止的功能

❌ 不适用于：

• 执行前的成本估算（使用定价计算器）
• 计费系统设计（使用 billing-system skill）
• 价格层级管理或折扣
• 历史成本分析仪表板

核心模式

1. 基于 Token 的成本计算

interface TokenUsage {
  inputTokens: number;
  outputTokens: number;
  cacheReadTokens?: number;   // Prompt caching hits
  cacheWriteTokens?: number;  // Prompt caching misses
}

interface CostCalculation {
  inputCostUsd: number;
  outputCostUsd: number;
  cacheSavingsUsd?: number;
  totalCostUsd: number;
}

function calculateCost(usage: TokenUsage, model: string): CostCalculation {
  const pricing = MODEL_PRICING[model];

  const inputCostUsd = (usage.inputTokens / 1_000_000) * pricing.inputPerMTok;
  const outputCostUsd = (usage.outputTokens / 1_000_000) * pricing.outputPerMTok;

  return {
    inputCostUsd,
    outputCostUsd,
    totalCostUsd: inputCostUsd + outputCostUsd,
  };
}

2. 增量累计模式

随着成本产生而跟踪，而不只是等到完成时：

class CostAccrualTracker {
  private totalInputTokens = 0;
  private totalOutputTokens = 0;
  private accruedCostUsd = 0;
  private readonly model: string;

  constructor(model: string) {
    this.model = model;
  }

  /**
   * Called after each API response (streaming or complete)
   */
  recordUsage(usage: TokenUsage): void {
    this.totalInputTokens += usage.inputTokens;
    this.totalOutputTokens += usage.outputTokens;

    const cost = calculateCost(usage, this.model);
    this.accruedCostUsd += cost.totalCostUsd;
  }

  /**
   * Get current accrued cost (for real-time display)
   */
  getCurrentCost(): number {
    return this.accruedCostUsd;
  }

  /**
   * Finalize on completion or abort
   */
  finalize(reason: 'completed' | 'aborted' | 'failed'): CostReport {
    return {
      totalInputTokens: this.totalInputTokens,
      totalOutputTokens: this.totalOutputTokens,
      totalCostUsd: this.accruedCostUsd,
      completionReason: reason,
      finalizedAt: Date.now(),
    };
  }
}

3. 感知中止的成本捕获

至关重要：始终在中止时捕获部分成本：

// In execution handler
const tracker = new CostAccrualTracker(model);

try {
  for await (const chunk of executeStream(request)) {
    if (abortSignal.aborted) {
      // CRITICAL: Capture cost BEFORE throwing
      const partialCost = tracker.finalize('aborted');
      onCostUpdate(partialCost);
      throw new AbortError('Execution aborted');
    }

    tracker.recordUsage(chunk.usage);
    onCostUpdate(tracker.getCurrentCost());
  }

  return tracker.finalize('completed');
} catch (error) {
  if (error instanceof AbortError) {
    throw error; // Already handled
  }
  return tracker.finalize('failed');
}

4. 预算阈值模式

超出预算时自动停止执行：

interface BudgetConfig {
  maxCostUsd: number;
  warnAtPercentage: number;  // e.g., 0.8 for 80%
  onWarn?: (current: number, max: number) => void;
  onExceed?: (current: number, max: number) => void;
}

function createBudgetGuard(config: BudgetConfig) {
  return {
    check(currentCostUsd: number): 'ok' | 'warn' | 'exceed' {
      const percentage = currentCostUsd / config.maxCostUsd;

      if (percentage >= 1.0) {
        config.onExceed?.(currentCostUsd, config.maxCostUsd);
        return 'exceed';
      }

      if (percentage >= config.warnAtPercentage) {
        config.onWarn?.(currentCostUsd, config.maxCostUsd);
        return 'warn';
      }

      return 'ok';
    }
  };
}

反模式

中止时丢失成本

新手想法：“中止时直接抛出错误就行。”

实际情况：如果中止前不捕获成本，就会丢失：

• 部分执行的 Token 使用数据
• 用于计费的准确成本报告
• 用于调试的审计追踪

时间线：这一直是个问题，但随着高价模型（GPT-4、Claude Opus）的出现，问题变得至关重要。

正确做法：抛出中止错误之前，始终使用部分数据调用 finalize()。

未防抖的轮询

新手想法：“每 100ms 轮询一次成本端点，以便实时更新。”

实际情况：

• 浪费带宽和 CPU
• 只有在 API 响应后才会更新成本
• 轮询频率快于响应速率毫无意义

正确做法：每隔 1-2 秒轮询一次，或使用执行流提供的事件驱动更新。

忽略提示词缓存

新手想法：“直接用 Token 数量乘以每个 Token 的价格就行。”

实际情况：Claude 的提示词缓存会改变成本模型：

• 缓存读取的成本低 90%
• 首次使用时，缓存写入需要额外付费
• 忽略缓存会导致成本不准确

时间线：

• 2024 年之前：没有缓存，计算简单
• 2024 年及以后：Claude 提示词缓存需要单独跟踪

正确做法：分别跟踪 cache_read_input_tokens 和 cache_creation_input_tokens。

每个请求使用独立的成本对象

新手想法：“为每个请求创建新的跟踪器。”

实际情况：对于包含多个节点的 DAG 执行：

• 需要汇总所有节点的成本
• 需要将成本归属到特定节点
• 需要汇总到父级执行

正确做法：采用分层跟踪方式——每个节点使用独立跟踪器，并将成本汇总到执行级别。

状态流转

                    ┌─────────────────────────────────────────┐
                    │           CostAccrualTracker            │
                    └─────────────────────────────────────────┘
                                        │
              ┌─────────────────────────┼─────────────────────────┐
              │                         │                         │
              ▼                         ▼                         ▼
    ┌─────────────────┐     ┌─────────────────┐     ┌─────────────────┐
    │  recordUsage()  │     │ getCurrentCost()│     │   finalize()    │
    │                 │     │                 │     │                 │
    │ After each API  │     │ For real-time   │     │ On completion,  │
    │ response        │     │ display         │     │ abort, or fail  │
    └─────────────────┘     └─────────────────┘     └─────────────────┘
              │                         │                         │
              │                         │                         │
              ▼                         ▼                         ▼
    ┌─────────────────────────────────────────────────────────────────┐
    │                        CostReport                               │
    │  { inputTokens, outputTokens, totalCostUsd, completionReason }  │
    └─────────────────────────────────────────────────────────────────┘

UI 显示模式

在执行界面中实时显示成本：

// Poll every 2 seconds while executing
useEffect(() => {
  if (status !== 'running') return;

  const interval = setInterval(async () => {
    const response = await fetch(`/api/execute/${executionId}`);
    const data = await response.json();
    setAccruedCost(data.cost.accruedUsd);
    setTokens({
      input: data.cost.inputTokens,
      output: data.cost.outputTokens,
    });
  }, 2000);

  return () => clearInterval(interval);
}, [executionId, status]);

// Display format
<div className="cost-display">
  <span className="cost-amount">${accruedCost.toFixed(4)}</span>
  <span className="token-count">
    {tokens.input.toLocaleString()} in / {tokens.output.toLocaleString()} out
  </span>
</div>

集成点

组件	职责
CostAccrualTracker	单次执行的 Token 计数与成本计算
ExecutionManager	汇总 DAG 执行中的成本
BudgetGuard	阈值监控与自动停止
/api/execute/:id	通过轮询提供当前成本
成本显示组件	实时 UI 渲染

参考资料

有关当前 Claude API 定价，请参阅 /references/claude-api-pricing.md。