成本与限流

职责概述

解决的问题：用大模型烧钱很快——一个复杂任务可能消耗数美元的 token。用户和组织需要知道花了多少钱、还剩多少预算，以及在预算快耗尽时优雅降级而不是直接报错。

应用场景：① 实时显示当前对话已消耗的 token 数和费用 ② Pro 用户达到每日限额时提示升级而不是直接中断 ③ 组织管理员设置月度预算上限 ④ 自动压缩上下文以减少 token 消耗 ⑤ 按模型分类统计成本，帮助用户选择性价比更高的模型。

一句话理解：就像手机的流量管理——实时显示用了多少流量、还剩多少、快超了就提醒你，超了就限速但不断网。

架构设计

核心数据流

成本追踪流程

策略限制执行流程

Token 预算多级阈值

关键类型与接口

Token 警告状态

// services/compact/autoCompact.ts:93-145
export function calculateTokenWarningState(
  tokenUsage: number,
  model: string,
): {
  percentLeft: number                   // 剩余百分比
  isAboveWarningThreshold: boolean      // 超过警告阈值 (threshold - 20K)
  isAboveErrorThreshold: boolean        // 超过错误阈值 (threshold - 20K)
  isAboveAutoCompactThreshold: boolean  // 超过自动压缩阈值
  isAtBlockingLimit: boolean            // 达到阻塞限制
}

成本累加核心函数

// cost-tracker.ts:278-323
export function addToTotalSessionCost(
  cost: number,
  usage: Usage,     // BetaUsage from Anthropic SDK
  model: string,
): number {
  const modelUsage = addToTotalModelUsage(cost, usage, model)
  addToTotalCostState(cost, modelUsage, model)

  // OpenTelemetry counters
  getCostCounter()?.add(cost, { model })
  getTokenCounter()?.add(usage.input_tokens, { model, type: 'input' })
  getTokenCounter()?.add(usage.output_tokens, { model, type: 'output' })
  getTokenCounter()?.add(usage.cache_read_input_tokens ?? 0,
    { model, type: 'cacheRead' })
  getTokenCounter()?.add(usage.cache_creation_input_tokens ?? 0,
    { model, type: 'cacheCreation' })

  // Advisor（嵌入式小模型）成本
  for (const advisorUsage of getAdvisorUsage(usage)) {
    const advisorCost = calculateUSDCost(advisorUsage.model, advisorUsage)
    totalCost += addToTotalSessionCost(advisorCost, advisorUsage, advisorUsage.model)
  }
  return totalCost
}

存储的成本状态

// cost-tracker.ts:71-80
type StoredCostState = {
  totalCostUSD: number
  totalAPIDuration: number
  totalAPIDurationWithoutRetries: number
  totalToolDuration: number
  totalLinesAdded: number
  totalLinesRemoved: number
  lastDuration: number | undefined
  modelUsage: { [modelName: string]: ModelUsage } | undefined
}

策略限制响应 Schema

// services/policyLimits/types.ts:8-16
export const PolicyLimitsResponseSchema = lazySchema(() =>
  z.object({
    restrictions: z.record(
      z.string(),
      z.object({ allowed: z.boolean() })
    ),
  })
)

export type PolicyLimitsFetchResult = {
  success: boolean
  restrictions?: PolicyLimitsResponse['restrictions'] | null
  etag?: string
  error?: string
  skipRetry?: boolean   // true = 不重试（如认证错误）
}

有效上下文窗口计算

// services/compact/autoCompact.ts:33-49
export function getEffectiveContextWindowSize(model: string): number {
  const reservedTokensForSummary = Math.min(
    getMaxOutputTokensForModel(model),
    MAX_OUTPUT_TOKENS_FOR_SUMMARY   // 20,000
  )
  let contextWindow = getContextWindowForModel(model, getSdkBetas())

  // 支持环境变量覆盖
  const autoCompactWindow = process.env.CLAUDE_CODE_AUTO_COMPACT_WINDOW
  if (autoCompactWindow) {
    const parsed = parseInt(autoCompactWindow, 10)
    if (!isNaN(parsed) && parsed > 0) {
      contextWindow = Math.min(contextWindow, parsed)
    }
  }

  return contextWindow - reservedTokensForSummary
}

设计模式与亮点

1. 递归成本计算（Advisor 嵌套）

addToTotalSessionCost() 中处理 advisor 用量时递归调用自身。一次 API 响应可能包含主模型 + advisor 模型的混合用量，递归确保每层模型的成本都被正确计算和归属。这种设计支持无限嵌套的 advisor 链，虽然实践中通常只有一层。

2. 成本持久化与 Session 恢复

成本状态通过 saveCurrentSessionCosts() 写入项目配置文件（.claude/settings.json 的 lastCost/lastModelUsage 字段），在 --resume 恢复会话时通过 restoreCostStateForSession() 读回。Session ID 校验确保不会加载过期会话的成本数据。

// cost-tracker.ts:130-137
export function restoreCostStateForSession(sessionId: string): boolean {
  const data = getStoredSessionCosts(sessionId)
  if (!data) return false  // session ID 不匹配
  setCostStateForRestore(data)
  return true
}

3. 多级缓冲区设计

系统使用三个不同大小的缓冲区控制上下文使用率：

• AUTOCOMPACT_BUFFER_TOKENS (13,000)：自动压缩触发前的安全余量
• WARNING_THRESHOLD_BUFFER_TOKENS (20,000)：警告用户即将耗尽
• MANUAL_COMPACT_BUFFER_TOKENS (3,000)：手动压缩时的更小余量

4. 策略限制的 Checksum 缓存

Policy Limits 使用 SHA-256 checksum 而非简单 ETag 做缓存验证：computeChecksum() 递归排序所有 key 后计算哈希，确保不同请求顺序产生相同的 checksum。这避免了服务端 ETag 实现差异导致的缓存失效。

// services/policyLimits/index.ts:133-159
function sortKeysDeep(obj: unknown): unknown {
  if (Array.isArray(obj)) return obj.map(sortKeysDeep)
  if (obj !== null && typeof obj === 'object') {
    const sorted: Record<string, unknown> = {}
    for (const [key, value] of Object.entries(obj).sort(([a], [b]) =>
      a.localeCompare(b)
    )) sorted[key] = sortKeysDeep(value)
    return sorted
  }
  return obj
}

function computeChecksum(restrictions): string {
  const sorted = sortKeysDeep(restrictions)
  const hash = createHash('sha256').update(jsonStringify(sorted)).digest('hex')
  return `sha256:${hash}`
}

5. HIPAA 场景的 Fail Closed

一般策略限制遵循 fail open 原则，但 ESSENTIAL_TRAFFIC_DENY_ON_MISS 集合中的策略（如 allow_product_feedback）在 HIPAA 模式下 fail closed——当缓存不可用时直接拒绝。这是合规要求的硬性约束，优先级高于可用性。

6. 进程退出时的成本快照

useCostSummary() Hook 在 process.on('exit') 时同时做两件事：输出格式化的成本摘要到 stdout，以及将当前成本状态持久化到项目配置。这确保即使异常退出，下次 session 也能看到准确的累计成本。

◈ 可视化处理拓扑图

Phase 1 — 成本追踪管线

每次 API 响应到达后，成本追踪管线按模型分类累计 token 用量和美元成本，同时递归处理嵌套的 advisor 调用。

Phase 2 — 策略限制：资格检查 + 三级降级

Phase 3 — Token 预算与自动压缩多层守卫

⇉ 核心处理流程详解

成本与限流系统由三条独立但协同的流程组成：成本追踪（每次 API 调用后累计）、策略限制（启动时加载 + 后台轮询）、自动压缩（token 用量达到阈值时触发）。三条流程共同确保 Claude Code 在成本可控的前提下持续运行。

★ 设计精华

1. Advisor 递归计费——透明成本归属

API 响应中的 usage 字段可能包含主模型和 advisor 模型的混合用量。getAdvisorUsage() 将 advisor 用量从主用量中提取出来，然后递归调用 addToTotalSessionCost() 计费。这意味着 advisor 的成本会出现在模型分类统计和 OTel 指标中，用户可以在 /cost 命令中看到每个 advisor 模型的独立成本。

// cost-tracker.ts:312-322
for (const advisorUsage of getAdvisorUsage(usage)) {
  const advisorCost = calculateUSDCost(advisorUsage.model, advisorUsage)
  logEvent('tengu_advisor_tool_token_usage', {
    advisor_model: advisorUsage.model,
    input_tokens: advisorUsage.input_tokens,
    output_tokens: advisorUsage.output_tokens,
    cost_usd_micros: Math.round(advisorCost * 1_000_000),
  })
  totalCost += addToTotalSessionCost(advisorCost, advisorUsage, advisorUsage.model)
}

2. 策略限制的三级降级——失败开放

策略限制的获取采用 "fails open" 策略：fetchAndLoadPolicyLimits()（policyLimits/index.ts:432-495）在获取失败时不会阻止用户使用 CLI。三级降级路径：服务端返回最新限制 → 使用本地文件缓存（即使过期）→ 返回 null（不施加任何限制）。304 Not Modified 复用缓存，404 删除缓存文件。

// services/policyLimits/index.ts:432-495
async function fetchAndLoadPolicyLimits() {
  const cachedRestrictions = loadCachedRestrictions()
  const cachedChecksum = cachedRestrictions ? computeChecksum(cachedRestrictions) : undefined
  try {
    const result = await fetchWithRetry(cachedChecksum)
    if (!result.success) {
      if (cachedRestrictions) { return cachedRestrictions }  // 降级到缓存
      return null  // 无限制
    }
    if (result.restrictions === null && cachedRestrictions) {
      return cachedRestrictions  // 304: 缓存仍有效
    }
    // 成功: 保存并使用新限制
    await saveCachedRestrictions(newRestrictions)
    return newRestrictions
  } catch {
    return cachedRestrictions ?? null  // 异常降级
  }
}

3. 自动压缩的多层守卫——防止自毁式压缩

shouldAutoCompact()（autoCompact.ts:160-239）的设计目标是防止压缩在不恰当的上下文中触发。四层守卫：(1) session_memory/compact 来源不压缩（否则会死锁——压缩本身就是一次 API 调用）；(2) context-collapse 模式下不压缩（collapse 有自己的 90%/95% 阈值）；(3) reactive-only 模式下不主动压缩（等 API 返回 prompt-too-long 再处理）；(4) 只有 token 用量真正超过阈值才触发。

// services/compact/autoCompact.ts:160-239
export async function shouldAutoCompact(messages, model, querySource?, snipTokensFreed = 0) {
  // 守卫1: 压缩来源不触发压缩（防止死锁）
  if (querySource === 'session_memory' || querySource === 'compact') return false
  // 守卫2: context-collapse 模式自己管理上下文
  if (feature('CONTEXT_COLLAPSE') && isContextCollapseEnabled()) return false
  // 守卫3: reactive-only 模式等 API 报错再处理
  if (feature('REACTIVE_COMPACT') && getFeatureValue_CACHED_MAY_BE_STALE('tengu_cobalt_raccoon', false)) return false
  // 守卫4: 真正检查 token 用量
  const { isAboveAutoCompactThreshold } = calculateTokenWarningState(tokenCount, model)
  return isAboveAutoCompactThreshold
}

4. Session 隔离的成本持久化——多实例安全

saveCurrentSessionCosts()（cost-tracker.ts:143-175）将 session ID 与成本数据一起写入项目配置。恢复时通过 projectConfig.lastSessionId !== sessionId 校验（在 restoreCostStateForSession() L130-137 中），防止读取错误 session 的成本。这在多 session 并行场景（多个 Claude Code 实例同时运行）中确保了成本数据的隔离性。

// cost-tracker.ts:143-175 + 130-137
export function saveCurrentSessionCosts(fpsMetrics?) {
  const sessionId = getSessionId()
  // 将 sessionId + 成本数据 + 模型用量写入 projectConfig
  updateProjectConfig({ lastSessionId: sessionId, costs: ... })
}

export function restoreCostStateForSession(sessionId: string): boolean {
  const stored = getStoredSessionCosts(sessionId)
  if (!stored) return false
  // 仅当 sessionId 匹配时才恢复——防止串扰
  setCostStateForRestore(stored)
  return true
}

◈ Agent 实践借鉴 — 客服 Agent 成本控制设计

一、场景映射：客服 agent 的成本治理挑战

每个 LLM 调用都要花钱。假设一个客服平台有 10 万个客户同时在线，每个客户平均 5 轮对话，每轮消耗 2000 token，使用 GPT-4o 每百万 token 约 $5，一天的光 LLM 费用就是 100000 x 5 x 2000 / 1000000 x $5 = $5000。一个月就是 $150,000。不同租户（企业客户）的预算不同：大客户每月 10 万额度，小客户每月 5000。单个会话如果客户聊了 50 轮，光 LLM 费用可能就要几十块。你需要实时知道"这个租户本月还剩多少额度"——这就是 CC 的实时成本追踪在客服场景的直接映射。

二、借鉴 CC + 客服改造

实践 1：按租户 + 按会话的成本追踪（借鉴 CC 的 addToTotalSessionCost）

CC 用 addToTotalSessionCost() 在每次 API 调用后累计成本。客服场景需要双维度追踪：既要按租户累计（月底计费），又要按会话累计（防止单会话成本爆炸）。

// 借鉴 cost-tracker.ts:278-323 的单入口累计模式
interface TenantCostRecord {
  tenantId: string
  monthYear: string            // '2026-04'
  totalCostMicros: number      // 微美分整数，避免浮点精度问题
  totalInputTokens: number
  totalOutputTokens: number
  sessionCount: number
}

interface SessionCostRecord {
  sessionId: string
  tenantId: string
  turnCount: number
  totalCostMicros: number
  modelUsed: string
}

class CustomerServiceCostTracker {
  // 按租户按月累计（计费用）
  private tenantMonthly = new Map<string, TenantCostRecord>()
  // 按会话累计（实时控制用）
  private sessionCosts = new Map<string, SessionCostRecord>()

  // 唯一入口：每次 LLM 调用后调用（借鉴 CC 的 addToTotalSessionCost）
  recordLLMCall(params: {
    tenantId: string; sessionId: string; model: string
    inputTokens: number; outputTokens: number; costUSD: number
  }): void {
    const costMicros = Math.round(params.costUSD * 1_000_000)  // 整数累加

    // 维度 1：按租户按月累计
    const monthKey = `${params.tenantId}:${new Date().toISOString().slice(0, 7)}`
    const tenantRecord = this.getOrCreateTenantRecord(monthKey, params.tenantId)
    tenantRecord.totalCostMicros += costMicros
    tenantRecord.totalInputTokens += params.inputTokens
    tenantRecord.totalOutputTokens += params.outputTokens

    // 维度 2：按会话累计
    const sessionRecord = this.getOrCreateSessionRecord(
      params.sessionId, params.tenantId, params.model
    )
    sessionRecord.turnCount += 1
    sessionRecord.totalCostMicros += costMicros

    // 上报遥测（借鉴 CC 的 OTel Counter）
    this.reportMetrics(params, costMicros)
  }

  // 查询租户本月剩余额度（实时计算）
  getTenantRemainingBudget(tenantId: string, monthlyBudgetUSD: number): {
    usedUSD: number; remainingUSD: number; usagePercent: number
  } {
    const monthKey = `${tenantId}:${new Date().toISOString().slice(0, 7)}`
    const record = this.tenantMonthly.get(monthKey)
    const usedUSD = (record?.totalCostMicros ?? 0) / 1_000_000
    return {
      usedUSD,
      remainingUSD: Math.max(0, monthlyBudgetUSD - usedUSD),
      usagePercent: Math.round(usedUSD / monthlyBudgetUSD * 100),
    }
  }
}

实践 2：三级预算阈值（借鉴 CC 的 warn → auto-compact → block）

CC 用四级阈值控制上下文使用率。客服场景也有类似的成本控制需求，但阈值语义不同：50% 警告运营人员、80% 自动切换到便宜模型、100% 暂停服务。

// 借鉴 autoCompact.ts:93-145 的多级阈值设计
type CostLevel = 'normal' | 'warning' | 'downgrade' | 'suspended'

class TenantBudgetManager {
  private thresholds = {
    warning: 0.50,     // 50% → 通知运营人员
    downgrade: 0.80,   // 80% → 切换到便宜模型
    suspended: 1.00,   // 100% → 暂停服务
  }

  // 便宜模型映射（降级时使用）
  private modelFallback = {
    'gpt-4o': 'gpt-4o-mini',           // $5/M → $0.15/M token，成本降 97%
    'claude-sonnet': 'claude-haiku',    // $3/M → $0.25/M token
    'deepseek-chat': 'deepseek-lite',   // 已是便宜模型，不再降级
  }

  evaluateBudget(tenantId: string, monthlyBudgetUSD: number): {
    level: CostLevel; effectiveModel: string; usagePercent: number
  } {
    const budget = costTracker.getTenantRemainingBudget(tenantId, monthlyBudgetUSD)
    const ratio = budget.usagePercent / 100

    if (ratio >= this.thresholds.suspended) {
      return { level: 'suspended', effectiveModel: '', usagePercent: budget.usagePercent }
    }
    if (ratio >= this.thresholds.downgrade) {
      const config = getTenantConfig(tenantId)
      const downgraded = this.modelFallback[config.model] ?? config.model
      return { level: 'downgrade', effectiveModel: downgraded, usagePercent: budget.usagePercent }
    }
    if (ratio >= this.thresholds.warning) {
      return { level: 'warning', effectiveModel: getTenantConfig(tenantId).model, usagePercent: budget.usagePercent }
    }
    return { level: 'normal', effectiveModel: getTenantConfig(tenantId).model, usagePercent: budget.usagePercent }
  }

  // 在每次 LLM 调用前执行预算检查（借鉴 CC 的 enforceBudget）
  async beforeLLMCall(tenantId: string, sessionId: string): Promise<{
    allowed: boolean; model: string; reason?: string
  }> {
    const budget = getTenantBudget(tenantId)
    const state = this.evaluateBudget(tenantId, budget.monthlyLimitUSD)

    switch (state.level) {
      case 'normal':
        return { allowed: true, model: state.effectiveModel }
      case 'warning':
        // 通知运营人员（邮件/webhook）
        await notifyOps(tenantId, `租户 ${tenantId} 本月已用 ${state.usagePercent}% 预算`)
        return { allowed: true, model: state.effectiveModel }
      case 'downgrade':
        return { allowed: true, model: state.effectiveModel, reason: '已降级到经济模型' }
      case 'suspended':
        // 暂停服务，转人工
        await transferToHuman(sessionId, '租户预算已耗尽')
        return { allowed: false, model: '', reason: '本月额度已用完' }
    }
  }
}

实践 3：会话级成本控制（借鉴 CC 的单会话 token 预算）

CC 用上下文窗口阈值控制单会话 token 消费。客服场景同样需要会话级限制——一个客户聊了 50 轮，单会话成本可能超过 5 元，需要自动压缩上下文或降级。

// 借鉴 autoCompact.ts 的阈值控制 + 自动压缩
const SESSION_COST_LIMIT_USD = 5.0     // 单会话成本上限
const SESSION_COST_COMPACT_AT = 3.0    // 触发上下文压缩的阈值

async function checkSessionBudget(sessionId: string): Promise<{
  action: 'continue' | 'compact' | 'downgrade' | 'terminate'
}> {
  const sessionCost = costTracker.getSessionCost(sessionId)
  const costUSD = sessionCost.totalCostMicros / 1_000_000

  if (costUSD >= SESSION_COST_LIMIT_USD) {
    // 单会话超过 5 元：降级到最便宜模型或转人工
    return { action: 'downgrade' }
  }
  if (costUSD >= SESSION_COST_COMPACT_AT) {
    // 单会话超过 3 元：压缩上下文（类似 CC 的 autoCompact）
    // 将前 20 轮对话摘要化，释放 token 空间
    return { action: 'compact' }
  }
  return { action: 'continue' }
}

实践 4：成本报告（按租户、按渠道、按日期统计）

// 借鉴 cost-tracker.ts:228-244 的 formatTotalCost
class CostReporter {
  // 按租户统计月度成本
  getTenantMonthlyReport(tenantId: string): TenantMonthlyReport {
    const monthKey = `${tenantId}:${new Date().toISOString().slice(0, 7)}`
    const record = costTracker.getTenantRecord(monthKey)
    return {
      tenantId,
      period: new Date().toISOString().slice(0, 7),
      totalCostUSD: (record?.totalCostMicros ?? 0) / 1_000_000,
      totalInputTokens: record?.totalInputTokens ?? 0,
      totalOutputTokens: record?.totalOutputTokens ?? 0,
      sessionCount: record?.sessionCount ?? 0,
      avgCostPerSession: record
        ? ((record.totalCostMicros / 1_000_000) / record.sessionCount).toFixed(4)
        : '0',
    }
  }

  // 按渠道统计（电话比在线聊天贵，因为有 ASR/TTS 成本）
  getCostByChannel(tenantId: string): Record<string, number> {
    const sessions = costTracker.getTenantSessions(tenantId)
    const byChannel: Record<string, number> = {}
    for (const session of sessions) {
      const channel = session.channelType
      byChannel[channel] = (byChannel[channel] ?? 0) + session.totalCostMicros / 1_000_000
    }
    return byChannel
  }
}

三、落地清单

四、常见坑

代码索引