关注

Claude 模型选型:Opus/Sonnet/Haiku + 成本/限速预算(Kotlin)

1)先统一概念:你要优化的是“单位成功结果的成本”

选模型时最容易犯的错是只看单价。真正该优化的是:

  • 一次任务成功的概率(失败就要重试/补救)
  • 成功一次消耗的 token(输入 + 输出)
  • 在限速下的吞吐(RPM/ITPM/OTPM 约束)

这三者乘起来,才是你线上最终的“单位成功结果成本”。


2)官方定价怎么读:输入/输出 + 缓存 + Batch

Claude 的计费至少要拆开看四项:

  • Base Input Tokens(普通输入)
  • Output Tokens(输出)
  • Prompt Caching:缓存写入与缓存命中(读取)是另一套价格
  • Batch:Message Batches API 的输入/输出都有折扣

官方价格表在这里(中文):定价

工程结论(先记住这两句就够用):

  • 长上下文/重复 system 的场景,Prompt Caching 往往比“换便宜模型”更省钱
  • 大批量非实时任务,优先考虑 Message Batches(折扣 + 削峰)

3)先按任务类型定档:不是所有请求都配得上 Opus

我用一个很粗但实用的分层:

A. 短任务、高并发(抽取/分类/格式化)

  • 首选:Haiku
  • 不够稳:升 Sonnet
  • 关键点:把 max_tokens 卡死(200-600),别让输出失控

B. 线上对话/通用助理(质量/速度/成本均衡)

  • 首选:Sonnet
  • 只有在“明显需要更强推理或更复杂代码”才上 Opus

C. 长文档/长代码理解(长输入 + 中等输出)

  • 首选:Sonnet 起步
  • 卡住再升 Opus
  • 关键点:先检索 topK,再拼上下文;能缓存就缓存

D. 攻坚型任务(失败代价高、一次值很大)

  • 首选:Opus
  • 关键点:写清验收标准 + 自检;必要时两阶段生成(提纲→分段)

4)限速预算:用 ITPM/OTPM 反推并发,而不是靠拍脑袋

Claude API 的限速是三条线:RPM/ITPM/OTPM。超过任何一条都会 429,并返回 retry-after。官方说明见:速率限制

你可以用一个简单估算把“并发”算到一个不离谱的范围:

假设每次请求平均消耗:

  • 未缓存输入 (in) tokens
  • 输出 (out) tokens

那么在 ITPM/OTPM 上限分别为 (ITPM_limit)、(OTPM_limit) 时:

[
req/min \approx \min\left(\frac{ITPM_limit}{in}, \frac{OTPM_limit}{out}\right)
]

最后再和 RPM 上限取最小值,就是你“理论上限吞吐”。上线后你需要用 p95 的 in/out token 重算一次,否则高峰期还是会撞。

Prompt Caching 会影响 ITPM(重点)

对大多数模型,从缓存读取的输入 token(cache_read_input_tokens)不计入 ITPM,只有未缓存输入和缓存写入计入。官方解释见:缓存感知 ITPM

所以“同样 200K 上下文”,命中缓存与不命中,吞吐差距会非常明显。


5)Kotlin 选型器:给出模型 + max_tokens + 并发建议 + 降级策略

下面这段 Kotlin 代码是“把经验固化”,便于你放到配置中心或网关层。

enum class TaskKind {
  CLASSIFY_EXTRACT,
  CHAT_ASSISTANT,
  LONG_DOC,
  TOOL_JSON,
  CODING,
  HARD_PROBLEM
}

data class TierLimits(
  val rpm: Int,
  val itpm: Int,
  val otpm: Int
)

data class Advice(
  val model: String,
  val maxTokens: Int,
  val suggestedConcurrency: Int,
  val fallbackModel: String,
  val notes: List<String>
)

fun advise(
  kind: TaskKind,
  avgUncachedInTokens: Int,
  avgOutTokens: Int,
  limits: TierLimits,
  failureCostHigh: Boolean,
  needLowLatency: Boolean
): Advice {
  val haiku = "claude-haiku"
  val sonnet = "claude-sonnet"
  val opus = "claude-opus"

  fun throughputRpmByTokens(): Int {
    val byIn = if (avgUncachedInTokens <= 0) Int.MAX_VALUE else limits.itpm / avgUncachedInTokens
    val byOut = if (avgOutTokens <= 0) Int.MAX_VALUE else limits.otpm / avgOutTokens
    return minOf(limits.rpm, byIn, byOut).coerceAtLeast(1)
  }

  fun concurrencyFromRpm(rpm: Int): Int {
    // 粗估:并发≈每秒请求数×一个请求的平均“在途时间”
    // 这里不拿真实延迟,给个保守值:低延迟 0.6s;普通 1.2s
    val inFlightSec = if (needLowLatency) 0.6 else 1.2
    val rps = rpm / 60.0
    return (rps * inFlightSec).toInt().coerceIn(1, 64)
  }

  val maxTokens = when (kind) {
    TaskKind.CLASSIFY_EXTRACT -> 300
    TaskKind.CHAT_ASSISTANT -> 800
    TaskKind.LONG_DOC -> 1200
    TaskKind.TOOL_JSON -> 600
    TaskKind.CODING -> 1000
    TaskKind.HARD_PROBLEM -> 1400
  }.let { base -> if (needLowLatency) (base * 0.7).toInt().coerceAtLeast(128) else base }

  val model = when (kind) {
    TaskKind.CLASSIFY_EXTRACT -> if (failureCostHigh) sonnet else haiku
    TaskKind.CHAT_ASSISTANT -> sonnet
    TaskKind.LONG_DOC -> if (failureCostHigh && avgUncachedInTokens > 40_000) opus else sonnet
    TaskKind.TOOL_JSON -> if (failureCostHigh) opus else sonnet
    TaskKind.CODING -> if (failureCostHigh) opus else sonnet
    TaskKind.HARD_PROBLEM -> opus
  }

  val fallback = when (model) {
    opus -> sonnet
    sonnet -> haiku
    else -> haiku
  }

  val rpmBudget = throughputRpmByTokens()
  val conc = concurrencyFromRpm(rpmBudget)

  val notes = buildList {
    add("预算基于未缓存输入 token;如果用 Prompt Caching,ITPM 压力会明显下降")
    add("把 max_tokens 卡死,先提纲再分段输出,能降低 OTPM 压力与截断率")
    add("建议分桶:短任务/长任务分队列,避免长任务拖死短请求")
  }

  return Advice(
    model = model,
    maxTokens = maxTokens,
    suggestedConcurrency = conc,
    fallbackModel = fallback,
    notes = notes
  )
}

你把 TierLimits 替换成你在控制台看到的实际值即可。建议上线后用指标校准:

  • avg/p95 输入(未缓存)token
  • avg/p95 输出 token
  • p95 延迟
  • 429 占比

6)把策略下沉到“统一接入层”,会省很多重复劳动(147api)

当你有多个服务/多个实例/多个项目同时用模型时,“每个服务各写一套限流+重试”很快会乱。

更常见的做法是把以下能力放到网关层统一处理:

  • 多 Key 池轮换与限额
  • 全局限流(跨实例共享),避免单机视角误判
  • 失败退避、熔断、降级、备用线路
  • 按项目/用户做账单与配额

如果你不想自建网关,可以了解下 147api:它提供 OpenAI 风格的聚合管理并覆盖 GPT、Claude、Gemini 等模型,适合把“接入层工程”集中处理,业务只保留配置化的 base_url/api_key/model


一页结论(适合贴团队 wiki)

  • 默认:Sonnet
  • 高并发短任务:Haiku(并把 max_tokens 卡死)
  • 高价值攻坚:Opus(写清验收 + 自检)
  • 吞吐不稳别先换模型:先算 ITPM/OTPM,再做缓存/分桶/分段输出

转载自CSDN-专业IT技术社区

原文链接:https://blog.csdn.net/AI147AI/article/details/158838697

评论

赞0

评论列表

微信小程序
QQ小程序

关于作者

点赞数:0
关注数:0
粉丝:0
文章:0
关注标签:0
加入于:--