Cost 优化 — Anthropic Prompt Caching 与 Batch API 深度

日期: 2026-10-12 方向: AI系统工程 / LLMOps / Cost Engineering 阶段: Phase 3 - 生产基础设施与评估 (Day 163-176) 标签: #PromptCaching #BatchAPI #CostOptimization #Anthropic #FinOps

今日目标

一、核心概念

1.1 LLM 成本结构与 80/20 法则

金融 AI 应用 token 分布（实测）：

RAG 应用 token 分布：
  System prompt    :  4%   ← 几乎不变
  RAG retrieved    : 70%   ← 大部分稳定（同一知识库切片）
  Conversation hist: 18%   ← 稳定增长
  User new turn    :  6%   ← 唯一真正变化
  Output           :  2%   ← 远小于 input
                    ─────
                    100%

→ 94% 的 input token 在多轮会话间是"重复"的，理论上可省下绝大部分钱

1.2 Anthropic Prompt Caching 机制

核心 API：在 message content block 加 cache_control: {"type": "ephemeral"}，Anthropic 服务端把这个 block 之前（含本身）的内容做 prefix cache。

TTL 选项（2026 已支持两档）：

• ephemeral（默认 5 分钟）：每次命中刷新 TTL
• ephemeral + ttl: "1h"：1 小时（写入贵 1.5×，但适合长会话/agent loop）

定价（claude-opus-4-7 为例）：

1.3 Cache breakpoint 规则（关键！）

• 最多 4 个 breakpoint，依次嵌套：cache 1 包含 cache 2 包含 cache 3 ...
• 最小 cacheable token：1024（claude-opus-4-7/sonnet-4-6）/ 2048（haiku-4-5）
• 完全前缀匹配：哪怕差 1 个 token cache 就 miss → 必须把动态内容挪到 cache 边界之后

正确分层（金融 RAG 示例）：

messages = [
  {
    "role": "system",
    "content": [
      # ── breakpoint 1：极稳定（system prompt + tools 定义）几乎永远命中
      {"type": "text", "text": SYSTEM_PROMPT, "cache_control": {"type": "ephemeral", "ttl": "1h"}},
      
      # ── breakpoint 2：知识库（每天换一次）
      {"type": "text", "text": KB_CONTEXT_HUGE, "cache_control": {"type": "ephemeral", "ttl": "1h"}},
    ]
  },
  {
    "role": "user",
    "content": [
      # ── breakpoint 3：本会话历史（5min 滚动）
      {"type": "text", "text": CONVERSATION_HISTORY, "cache_control": {"type": "ephemeral"}},
      
      # 当前问题：不加 cache_control（动态内容）
      {"type": "text", "text": current_question}
    ]
  }
]

1.4 Batch API（异步批处理 50% off）

• 输入价：50% off
• 输出价：50% off
• 延迟：通常 < 1 小时，SLA 24 小时
• 适用场景：批量分类、批量摘要、回填、合规扫描、夜间 ETL
• 不适用：实时 chatbot、agent、需要立刻反馈的场景

二、生产架构图

                  请求来源
        ┌─────────┼──────────┐
        ▼         ▼          ▼
    实时 chat   Agent      离线批量
    (P95 < 2s)  loop       (今晚跑完)
        │         │          │
        ▼         ▼          ▼
    ┌────────────────────────────────┐
    │   Cost Router（你今天写）      │
    │   if batch_eligible → Batch API│
    │   else → Real-time API         │
    └────────────────────────────────┘
        │         │          │
        ▼         ▼          ▼
    ┌────────────────────────────────┐
    │   Cache Layer Strategy          │
    │   1h cache: system+KB(稳定)     │
    │   5m cache: history(滚动)       │
    └────────────────────────────────┘
                 │
                 ▼
    ┌────────────────────────────────┐
    │   Model Router（梯度）          │
    │   intent → haiku-4-5  ($0.80)  │
    │   normal → sonnet-4-6 ($3)     │
    │   complex → opus-4-7  ($15)    │
    └────────────────────────────────┘
                 │
                 ▼
        Anthropic API
                 │
                 ▼
    ┌────────────────────────────────┐
    │ Cost Telemetry → ClickHouse    │
    │ cache_read_input_tokens 指标    │
    └────────────────────────────────┘

三、代码实现

3.1 完整 prompt caching 客户端

"""anthropic_caching.py — 金融 RAG 场景的 prompt caching 实战
依赖: pip install anthropic==0.39.0
"""
import os
import time
from anthropic import Anthropic

client = Anthropic(api_key=os.environ["ANTHROPIC_API_KEY"])
MODEL = "claude-opus-4-7"  # 1M context

# ────────────────────── 模拟金融 RAG 数据 ──────────────────────
SYSTEM_PROMPT = """你是某商业银行的合规助手。回答必须：
1. 引用具体法规条文
2. 标注信息日期
3. 涉及监管不确定时建议联系合规部
4. 严禁泄露内部代号
""" * 5  # 凑长度（>1024 token）

# 模拟一份 200K token 的内部知识库（季度合规文件汇编）
KB_HUGE = open("compliance_kb_q3.txt").read()  # 假设 ~ 800KB ≈ 200K token

CONVERSATION_HISTORY = [
    {"role": "user", "content": "客户做跨境电商收款，需要哪些合规审查？"},
    {"role": "assistant", "content": "根据《跨境电子商务零售进口监管要求》..."},
    {"role": "user", "content": "金额超过 50 万美元呢？"},
    {"role": "assistant", "content": "根据外汇管理局规定..."},
] * 3


def build_cached_messages(current_question: str):
    """三层 cache：1h(system+KB) / 5m(history) / no-cache(question)"""
    return {
        "system": [
            # Layer 1: 1 小时 cache（系统提示 + 巨大 KB）
            {
                "type": "text",
                "text": SYSTEM_PROMPT,
                "cache_control": {"type": "ephemeral", "ttl": "1h"}
            },
            {
                "type": "text",
                "text": f"<knowledge_base>\n{KB_HUGE}\n</knowledge_base>",
                "cache_control": {"type": "ephemeral", "ttl": "1h"}
            },
        ],
        "messages": [
            # Layer 2: 5min cache（最近会话）
            *CONVERSATION_HISTORY[:-1],
            {
                "role": CONVERSATION_HISTORY[-1]["role"],
                "content": [
                    {
                        "type": "text",
                        "text": CONVERSATION_HISTORY[-1]["content"],
                        "cache_control": {"type": "ephemeral"}  # 5min
                    }
                ]
            },
            # Layer 3: 不 cache（动态）
            {"role": "user", "content": current_question}
        ]
    }


def call_with_cache(question: str):
    msg = build_cached_messages(question)
    t0 = time.time()
    r = client.messages.create(
        model=MODEL,
        max_tokens=1024,
        system=msg["system"],
        messages=msg["messages"],
    )
    elapsed = time.time() - t0
    u = r.usage
    return {
        "elapsed_s": elapsed,
        "input_tokens": u.input_tokens,
        "output_tokens": u.output_tokens,
        "cache_creation_input_tokens": getattr(u, "cache_creation_input_tokens", 0),
        "cache_read_input_tokens": getattr(u, "cache_read_input_tokens", 0),
        "text": r.content[0].text[:200],
    }


def cost_of(usage: dict, model: str = "claude-opus-4-7") -> float:
    """USD 成本计算（claude-opus-4-7 价格）"""
    PRICE = {
        "input": 15.0 / 1_000_000,
        "output": 75.0 / 1_000_000,
        "cache_write_5m": 18.75 / 1_000_000,
        "cache_write_1h": 30.0 / 1_000_000,
        "cache_read": 1.50 / 1_000_000,  # 90% off
    }
    # 简化：把 cache_creation 都按 1h 算（生产应区分）
    return (
        usage["input_tokens"] * PRICE["input"]
        + usage["cache_creation_input_tokens"] * PRICE["cache_write_1h"]
        + usage["cache_read_input_tokens"] * PRICE["cache_read"]
        + usage["output_tokens"] * PRICE["output"]
    )


# ────────────────────── 实测对比 ──────────────────────
if __name__ == "__main__":
    questions = [
        "如果客户是高风险国家居民呢？",
        "境外股东持股 30% 是否需要额外审查？",
        "如何识别空壳公司？",
        "客户尽调每年更新一次足够吗？",
        "可疑交易报告怎么报？",
    ]

    total_cost_no_cache = 0
    total_cost_with_cache = 0

    for i, q in enumerate(questions):
        u = call_with_cache(q)
        cost_cached = cost_of(u)
        # 假设无 cache 版本：cache_read 全部按 input 算
        u_no_cache = {**u,
                      "input_tokens": u["input_tokens"] + u["cache_read_input_tokens"] + u["cache_creation_input_tokens"],
                      "cache_creation_input_tokens": 0,
                      "cache_read_input_tokens": 0}
        cost_no = cost_of(u_no_cache)

        total_cost_no_cache += cost_no
        total_cost_with_cache += cost_cached

        print(f"\nQ{i+1}: {q[:30]}...")
        print(f"  cache_read   : {u['cache_read_input_tokens']:>8} tok")
        print(f"  cache_create : {u['cache_creation_input_tokens']:>8} tok")
        print(f"  output       : {u['output_tokens']:>8} tok")
        print(f"  latency      : {u['elapsed_s']:.2f}s")
        print(f"  cost(cached) : ${cost_cached:.4f}")
        print(f"  cost(no $)   : ${cost_no:.4f}")
        print(f"  saving       : {(1 - cost_cached/cost_no)*100:.1f}%")

    print(f"\n{'='*50}")
    print(f"5 次调用总成本（无 cache）: ${total_cost_no_cache:.3f}")
    print(f"5 次调用总成本（有 cache）: ${total_cost_with_cache:.3f}")
    print(f"节省金额                : ${total_cost_no_cache - total_cost_with_cache:.3f}")
    print(f"节省比例                : {(1 - total_cost_with_cache/total_cost_no_cache)*100:.1f}%")

3.2 Batch API 50% off

"""anthropic_batch.py — 用 Batch API 跑夜间合规扫描"""
from anthropic import Anthropic
from anthropic.types.message_create_params import MessageCreateParamsNonStreaming
from anthropic.types.messages.batch_create_params import Request

client = Anthropic()

# 假设要批量审查 1000 条交易备注是否含敏感词
transactions = [
    {"id": f"txn_{i}", "memo": f"Transfer to entity_{i} for service"}
    for i in range(1000)
]

requests = [
    Request(
        custom_id=t["id"],
        params=MessageCreateParamsNonStreaming(
            model="claude-haiku-4-5",
            max_tokens=128,
            system=[{
                "type": "text",
                "text": "扫描以下交易备注，识别洗钱/制裁关键词，输出 JSON {flagged: bool, reason: str}",
                "cache_control": {"type": "ephemeral"}  # batch 内 cache 也命中
            }],
            messages=[{"role": "user", "content": t["memo"]}]
        )
    )
    for t in transactions
]

# 提交 batch
batch = client.messages.batches.create(requests=requests)
print(f"Batch submitted: {batch.id}, status: {batch.processing_status}")

# 轮询完成
import time
while True:
    b = client.messages.batches.retrieve(batch.id)
    print(f"[{time.strftime('%H:%M:%S')}] {b.processing_status} "
          f"(succeeded={b.request_counts.succeeded}/{b.request_counts.processing + b.request_counts.succeeded})")
    if b.processing_status == "ended":
        break
    time.sleep(60)

# 拉取结果
flagged = []
for result in client.messages.batches.results(batch.id):
    if result.result.type == "succeeded":
        # parse and accumulate
        flagged.append((result.custom_id, result.result.message.content[0].text))

print(f"Done. Flagged {sum(1 for _, t in flagged if 'true' in t.lower())}/1000")

# 成本估算
# 假设每请求 input=200, output=80 token, claude-haiku-4-5
# 实时: 1000 * (200*0.8 + 80*4) / 1e6 = $0.48
# Batch: $0.24（50% off）

3.3 模型路由（成本梯度）

"""router.py — 按任务复杂度路由模型"""
COST_TABLE = {
    "claude-haiku-4-5":  {"in": 0.80,  "out": 4.0,  "speed": 5},  # 越大越快
    "claude-sonnet-4-6": {"in": 3.0,   "out": 15.0, "speed": 3},
    "claude-opus-4-7":   {"in": 15.0,  "out": 75.0, "speed": 1},
}


def route(task_type: str, complexity: int) -> str:
    """task_type: classify/extract/summarize/reason/code"""
    if task_type in {"classify", "extract"} and complexity <= 2:
        return "claude-haiku-4-5"
    if task_type in {"summarize", "qa"} and complexity <= 3:
        return "claude-sonnet-4-6"
    return "claude-opus-4-7"


# 实测金融场景路由命中分布（自家数据）：
# haiku   : 62% （意图识别、结构化抽取、字段校验）
# sonnet  : 33% （RAG 问答、文档摘要、客户邮件起草）
# opus    :  5% （投研报告、复杂合规推理、多步 agent）
# 平均成本下降约 76%（vs 全 opus）

四、Cost & Performance 实测数据

4.1 prompt caching 三个真实场景

4.2 Batch API 实测

4.3 模型梯度路由（金融客服 1 万次对话）

五、金融领域应用

1. 法规知识库：监管文件季度更新，1h cache 完美匹配。一份 100K 法规放 cache 里，每次问答省 90%
2. 风控审查 batch：每日下班后跑 10 万条交易合规扫描，Batch API 砍半成本，第二天上午 9 点前出结果完全够用
3. 投研报告生成：路由小模型预筛标的，opus 只处理 top 20，成本下降 70%+
4. 客户尽调（KYC）：客户档案进 5min cache，多轮深挖问答几乎零增量成本
5. 审计：每次调用记录 cache_read_input_tokens / cache_creation_input_tokens 到 ClickHouse，做 token 经济学分析

六、生产经验与陷阱

1. cache breakpoint 顺序错误：把 dynamic 字段放在 cache 中间，导致永远 miss。规则：稳定→次稳定→动态，依序嵌套
2. system prompt 含日期："今天是 2026-10-12" 让 cache 每天 miss 一次。把日期挪到 user message
3. 5min 太短：低 QPS 服务（< 1 req/min）cache 总过期。要么用 1h cache（写贵 60%），要么忍受不命中
4. 1h cache 写入贵：第一次写入 1.25-2× input price，必须确保后续被复用足够多次（>= 5 次）才划算
5. Batch API 不能 stream：客服场景永远不能用，只能离线/异步任务
6. Batch SLA 24h 不是 24h 都能等：交易日 9:30 前必须出的合规扫描，要 4h 前提交
7. 路由器误判成本：错把复杂任务发给 haiku 导致返工，整体成本反而上升。必须配 eval 持续监控路由准确率
8. cache hit metric 不在响应里：必须从 usage.cache_read_input_tokens > 0 推断。持续监控这个比值，< 50% 就要 review breakpoint 设计

七、关键速查

八、面试题

1. Prompt caching 什么时候不该用？

   • 上下文 < 1024 token（达不到最小阈值）；prompt 几乎每次都变（cache 永远 miss）；调用频率 < 1/5min（cache 过期）；prompt 含 PII 不希望服务端缓存

2. 同一个用户的 5 轮对话，怎么设计 cache 让每轮都命中？

   • 系统 prompt + tools + KB 用 1h cache（一天/周稳定）；前 N-1 轮历史用 5min cache（滚动）；当前 user message 不 cache。每轮 cache_read = system + KB + (历史 1..N-1)

3. Batch API 50% off，为什么不全用？

   • SLA 24h（即便实测多 < 1h），实时 chatbot/agent 不能等；不能 stream；批量任务才划算（启动开销）

4. 如何监控 LLM 成本？

   • 三个核心指标：tokens_per_request、cache_hit_rate（cache_read / total_input）、model_distribution（haiku/sonnet/opus 各占多少）。日级看趋势，周级做 cost review

5. 金融 RAG 场景，200K 知识库每次 query 成本如何控制？

   • KB 进 1h cache（写 1 次 $6，读 1000 次省 $2700）；user message 短；用 sonnet 主力，opus 只兜底 confidence < 0.7 的；离线预生成常见 Q 的 cache，预热

明日预告

Day 165：Latency 优化 — Streaming、Speculative Decoding、并行 Tool Calls TTFT vs TPOT 区别；speculative decoding 用小模型猜大模型 token；agent loop 中多个 tool 并行；金融实时风控对话场景的延迟预算分配。