Claude API批量处理和1M上下文优化实战 省钱技巧

一、先用这张决策树定位你的场景

用户需要实时等待结果？
├── 否（离线/后台处理）
│   └── → 场景A：批量离线处理
│         优先策略：Batch API（5折）
│
└── 是（实时响应）
    │
    每次请求是否携带大量重复内容（如同一份文档/长System Prompt）？
    ├── 是
    │   └── → 场景B：长文档反复查询
    │         优先策略：Prompt Caching（输入省90%）
    │
    └── 否
        │
        任务是否简单（分类/提取/判断，输出很短）？
        ├── 是，且调用量大
        │   └── → 场景C：高频简单任务
        │         优先策略：Haiku 模型 + 小 max_tokens
        │
        └── 否（复杂对话/内容生成）
            └── → 场景D：实时对话/内容生成
                  优先策略：Sonnet + 对话历史裁剪

二、场景 A：批量离线处理

典型情况：每天批量跑数据标注、内容审核、摘要生成、报告处理——任务量大，但不需要实时返回结果，可以等几十分钟到几小时。

核心策略：Batch API，直接5折。

import anthropic
import json
import time

client = anthropic.Anthropic()

# ── 提交批量任务 ──────────────────────────────────

def submit_batch(items: list[dict], system: str = "") -> str:
    """
    提交批量任务，返回 batch_id
    items 格式：[{"id": "唯一ID", "text": "要处理的文本"}]
    """
    requests = []
    for item in items:
        messages = [{"role": "user", "content": item["text"]}]
        req = anthropic.types.MessageCreateParamsNonStreaming(
            model      = "claude-haiku-4-5-20251001",   # 批量任务优先用 Haiku 进一步省钱
            max_tokens = 64,
            messages   = messages,
        )
        if system:
            req["system"] = system
        requests.append(req)

    batch = client.messages.batches.create(requests=requests)
    print(f"✅ 已提交 {len(items)} 个任务，batch_id：{batch.id}")
    print(f"💰 预计费用：标准价格的 50%")
    return batch.id


# ── 轮询状态 ──────────────────────────────────────

def wait_for_batch(batch_id: str, poll_interval: int = 30) -> bool:
    """等待批量任务完成，返回是否成功"""
    print(f"等待任务完成（每 {poll_interval}s 检查一次）...")
    while True:
        batch = client.messages.batches.retrieve(batch_id)
        counts = batch.request_counts
        print(
            f"  进度：完成 {counts.succeeded + counts.errored}/"
            f"{counts.processing + counts.succeeded + counts.errored + counts.canceled} "
            f"（成功 {counts.succeeded}，失败 {counts.errored}）"
        )
        if batch.processing_status == "ended":
            return counts.errored == 0
        time.sleep(poll_interval)


# ── 获取结果 ──────────────────────────────────────

def get_batch_results(batch_id: str) -> dict[str, str]:
    """获取所有成功的结果，返回 {custom_id: 输出文本}"""
    results = {}
    for result in client.messages.batches.results(batch_id):
        if result.result.type == "succeeded":
            results[result.custom_id] = result.result.message.content[0].text
        else:
            results[result.custom_id] = f"[失败：{result.result.error.type}]"
    return results


# ── 完整流程示例 ──────────────────────────────────

items = [
    {"id": f"item_{i}", "text": f"判断情感（只输出：正面/负面/中性）：{text}"}
    for i, text in enumerate([
        "这款产品超出预期，强烈推荐！",
        "质量太差了，完全不值这个价",
        "还行，就这样",
        "快递很快，东西不错",
        "假货，差评！",
    ])
]

batch_id = submit_batch(items)

# 实际项目：把 batch_id 存数据库，后台定时检查，不要阻塞主进程
# 演示用：直接等待
if wait_for_batch(batch_id, poll_interval=10):
    results = get_batch_results(batch_id)
    for item_id, output in results.items():
        print(f"{item_id}: {output}")

这个场景不要用的策略：

• Prompt Caching——批量任务里每条输入都不同，缓存命中率为零，开了也没用
• 流式输出——批量任务不需要实时展示，开流式反而增加复杂度

进一步省钱：Batch API 已经5折，再叠加”用 Haiku 代替 Sonnet”，综合成本可以降到标准 Sonnet 价格的 约16%（0.5 × Haiku价格/Sonnet价格 ≈ 0.5 × 0.33 = 16%）。

三、场景 B：长文档反复查询

典型情况：一份合同/手册/报告，用户会反复问不同的问题；或者 System Prompt 很长（几千 token），每次请求都要带上。

核心策略：Prompt Caching，重复内容只收 10% 的输入费用。

import anthropic

client = anthropic.Anthropic()

class CachedDocumentQA:
    """
    对同一份文档反复提问的问答系统
    文档内容只在第一次请求时全价计费，后续请求只收 10%
    """

    def __init__(self, document: str, model: str = "claude-sonnet-4-6"):
        self.document = document
        self.model    = model
        self.history  = []

    def ask(self, question: str) -> dict:
        self.history.append({"role": "user", "content": question})

        # 把文档放在 system 里并标记缓存
        response = client.messages.create(
            model      = self.model,
            max_tokens = 1024,
            system     = [
                {
                    "type": "text",
                    "text": f"你是文档分析助手。以下是参考文档：\n\n{self.document}",
                    "cache_control": {"type": "ephemeral"},  # 标记为可缓存
                }
            ],
            messages   = self.history,
        )

        answer = response.content[0].text
        self.history.append({"role": "assistant", "content": answer})

        # 统计缓存效果
        usage         = response.usage
        cache_created = getattr(usage, "cache_creation_input_tokens", 0)
        cache_hit     = getattr(usage, "cache_read_input_tokens", 0)
        normal_input  = usage.input_tokens

        if cache_created > 0:
            status = f"首次（写入缓存 {cache_created} tokens，下次省 90%）"
        elif cache_hit > 0:
            saved_pct = cache_hit / (cache_hit + normal_input) * 90
            status    = f"缓存命中 {cache_hit} tokens（节省约 {saved_pct:.0f}% 输入费用）"
        else:
            status = "未命中缓存"

        return {"answer": answer, "cache_status": status}


# 使用示例
doc = open("contract.txt").read()   # 假设这是一份几千 token 的合同
qa  = CachedDocumentQA(doc)

questions = [
    "合同的付款条款是什么？",
    "违约金是怎么计算的？",
    "合同有效期是多久？",
    "哪一方负责承担运费？",
]

for q in questions:
    result = qa.ask(q)
    print(f"Q：{q}")
    print(f"A：{result['answer'][:80]}...")
    print(f"缓存状态：{result['cache_status']}\n")

缓存命中的条件：

• 缓存的内容要在请求的开头，且完全一致（一个字符不同就不命中）
• 缓存有效期大约5分钟——超过5分钟不请求，缓存失效，下次重新写入
• 缓存写入时费用是标准输入的 1.25 倍（首次稍贵），后续读取只收 0.1 倍

这个场景的成本计算示意：

文档：10000 tokens，对话 20 轮，每轮问题约 50 tokens

不用缓存：
  每轮输入 = 10000（文档）+ 历史 + 50（问题）≈ 10100 tokens
  20 轮总输入 ≈ 202000 tokens
  Sonnet 费用 ≈ $0.606

用缓存（5分钟内完成20轮）：
  第1轮：写入缓存 10000 tokens × 1.25倍 + 100 tokens = $0.038
  第2-20轮：缓存命中 10000 tokens × 0.1倍 + 100 tokens × 正常价 ≈ $0.006/轮
  20 轮总费用 ≈ $0.038 + 19 × $0.006 ≈ $0.152

节省约 75%

四、场景 C：高频简单任务

典型情况：每天调用量很大（万次以上），但每次任务简单——情感判断、意图分类、关键词提取、是否违规检测。输入短，输出短，对质量要求不极致。

核心策略：Haiku 模型 + 精准控制 max_tokens。

import anthropic
from concurrent.futures import ThreadPoolExecutor

client = anthropic.Anthropic()

# 针对不同任务类型的最优配置
TASK_CONFIGS = {
    "sentiment": {
        "model":      "claude-haiku-4-5-20251001",
        "max_tokens": 8,      # 只需要输出"正面/负面/中性"
        "system":     "判断情感。只输出：正面、负面、或中性。",
    },
    "intent": {
        "model":      "claude-haiku-4-5-20251001",
        "max_tokens": 16,     # 意图类别名通常不超过10字
        "system":     "判断用户意图类别。只输出类别名，不要解释。",
    },
    "language": {
        "model":      "claude-haiku-4-5-20251001",
        "max_tokens": 8,      # 语言代码，如 zh/en/ja
        "system":     "识别文本语言。只输出 ISO 639-1 语言代码（如 zh、en、ja）。",
    },
    "toxic": {
        "model":      "claude-haiku-4-5-20251001",
        "max_tokens": 4,      # 是/否
        "system":     "判断内容是否包含有害信息。只输出：是 或 否。",
    },
}


def classify(text: str, task: str) -> str:
    """高频分类的最优配置调用"""
    config = TASK_CONFIGS[task]
    response = client.messages.create(
        model      = config["model"],
        max_tokens = config["max_tokens"],
        temperature= 0,         # 分类任务用零温度，结果稳定
        system     = config["system"],
        messages   = [{"role": "user", "content": text}],
    )
    return response.content[0].text.strip()


def batch_classify_concurrent(
    texts: list[str],
    task:  str,
    workers: int = 10,
) -> list[str]:
    """并发处理，提高吞吐量（注意不要超过速率限制）"""
    with ThreadPoolExecutor(max_workers=workers) as executor:
        results = list(executor.map(lambda t: classify(t, task), texts))
    return results


# 成本对比
def cost_comparison(daily_requests: int, avg_input_tokens: int):
    print(f"\n每日 {daily_requests:,} 次请求，平均输入 {avg_input_tokens} tokens")
    print(f"{'模型':<12} {'月费（美元）':>15} {'月费（人民币）':>15}")
    print("-" * 45)

    configs = [
        ("Haiku", 1.0, 5.0),
        ("Sonnet", 3.0, 15.0),
    ]
    for name, inp_price, out_price in configs:
        monthly = daily_requests * 30
        cost    = monthly * (avg_input_tokens * inp_price + 8 * out_price) / 1_000_000
        print(f"{name:<12} ${cost:>14.2f} ¥{cost*7.24:>14.1f}")


# 示例：评论情感分析，每天10万条
cost_comparison(100_000, avg_input_tokens=80)

这个场景不要用的策略：

• Prompt Caching——每条输入都不同，没有可缓存的重复内容
• Opus 或 Sonnet——高频简单任务不需要强模型，用 Haiku 已经够准确
• 大 max_tokens——分类任务输出几个字就够，设1024是浪费

五、场景 D：实时对话和内容生成

典型情况：聊天机器人、写作助手、实时问答——用户等待回复，需要流式输出，对话历史在积累。

核心策略：Sonnet + 对话历史裁剪 + 精简 System Prompt。

import anthropic

client = anthropic.Anthropic()

class CostEfficientChatbot:
    """
    成本高效的聊天机器人
    核心省钱点：对话历史滑动窗口 + 精简 System Prompt
    """

    def __init__(
        self,
        system:       str,
        max_history:  int   = 10,   # 保留最近10轮对话
        model:        str   = "claude-sonnet-4-6",
        max_tokens:   int   = 1024,
    ):
        self.system      = system
        self.max_history = max_history
        self.model       = model
        self.max_tokens  = max_tokens
        self.history     = []
        self.total_cost  = 0.0

    def _trim_history(self):
        """滑动窗口：只保留最近 N 轮"""
        max_messages = self.max_history * 2   # 每轮 = user + assistant
        if len(self.history) > max_messages:
            self.history = self.history[-max_messages:]

    def _estimate_cost(self, input_tokens: int, output_tokens: int) -> float:
        """估算本次请求费用（Sonnet 价格）"""
        return (input_tokens * 3 + output_tokens * 15) / 1_000_000

    def chat(self, message: str) -> str:
        self.history.append({"role": "user", "content": message})
        self._trim_history()

        response = client.messages.create(
            model      = self.model,
            max_tokens = self.max_tokens,
            system     = self.system,
            messages   = self.history,
        )

        reply = response.content[0].text
        self.history.append({"role": "assistant", "content": reply})

        # 记录成本
        cost = self._estimate_cost(
            response.usage.input_tokens,
            response.usage.output_tokens,
        )
        self.total_cost += cost

        return reply

    def get_stats(self) -> dict:
        return {
            "history_turns":  len(self.history) // 2,
            "total_cost_usd": round(self.total_cost, 6),
        }


# 对话历史滑动窗口的省钱效果
def show_history_savings():
    """演示不裁剪 vs 裁剪历史的费用差异"""
    print("对话历史对成本的影响（每轮平均 200 tokens 输入 + 150 tokens 输出）：")
    print(f"{'轮次':<6} {'不裁剪（累计输入）':>20} {'裁剪到10轮（输入）':>20}")
    print("-" * 50)

    for turn in [1, 5, 10, 20, 50]:
        no_trim  = turn * (200 + 150 * (turn - 1) / 2)   # 历史不断积累
        trimmed  = min(turn, 10) * 200 + 150 * min(turn - 1, 9) / 2   # 最多10轮
        print(f"{turn:<6} {no_trim:>20,.0f} tokens  {trimmed:>16,.0f} tokens")

show_history_savings()

System Prompt 精简的实际效果：

冗长版本（约 500 tokens）：
"你是一个非常专业且经验丰富的客服助手，你的职责是帮助用户解决各种问题，
你需要保持礼貌、专业、友善的态度，始终以用户满意为目标……"

精简版本（约 20 tokens）：
"你是专业客服助手。礼貌、准确，不确定时直说。"

每次请求节省 480 tokens × Sonnet 输入价格 $3/M = 每次省 $0.00144
每天 10000 次请求 × $0.00144 = $14.4 / 天 = $432 / 月

六、组合策略：多个场景条件同时满足时

实际项目往往不是单一场景，可以叠加使用多种策略：

组合策略代码示例：批量 + 长固定 Prompt

"""
场景：每天批量处理1000篇文章，
每篇文章都要对照同一份3000 token 的行业规范进行合规审查

策略叠加：
- Batch API：整体5折
- Prompt Caching：规范文档复用，输入省90%
"""

COMPLIANCE_STANDARD = """
[一份3000 token的行业合规规范文档]
...[内容省略]...
"""

def build_compliance_batch_request(article: str, article_id: str):
    """构建带缓存标记的批量审查请求"""
    return anthropic.types.MessageCreateParamsNonStreaming(
        model      = "claude-haiku-4-5-20251001",
        max_tokens = 128,
        system     = [
            {
                "type": "text",
                "text": f"对照以下合规规范审查文章。规范：\n\n{COMPLIANCE_STANDARD}",
                "cache_control": {"type": "ephemeral"},   # 规范文档缓存
            }
        ],
        messages   = [{"role": "user", "content": f"审查以下文章是否合规（通过/不通过/需人工审查）：\n\n{article}"}],
    )

# 注意：Batch API + Prompt Caching 可以同时使用
# Batch API 负责批量提交（5折）
# Prompt Caching 负责复用固定的规范文档（输入省90%）
# 两个机制在 API 层面完全兼容

七、不适合省钱的情况

不是所有任务都应该省钱——有些情况强行优化反而有代价：

• 用 Haiku 处理复杂推理任务：准确率下降的代价远超省下的费用，下游需要更多人工干预
• 批量任务强行用 Batch API：Batch API 有延迟（最长24小时），如果业务有时效性要求，用了反而误事
• 对话历史裁剪过激：把窗口设得太短，Claude 记不住前面的上下文，用户体验变差，客诉增加
• Prompt 精简过度：压缩到 System Prompt 没有足够约束，输出质量下滑，需要更多人工审核

省钱的正确姿势是：先确认质量达标，再在不影响质量的前提下优化成本。质量和成本通常是可以兼顾的，但质量优先。

总结：一张速查表