💡 LLM 成本优化 / Batch API

📊 Batch API

不需要实时响应的场景，用 Batch API 直接半价。OpenAI Batch 和 Anthropic Messages Batches 的实操指南。

适用场景判断

OpenAI Batch API

import openai
import json
import time

client = openai.OpenAI()

# Step 1: 准备批量请求文件
# 每行一个 JSON 对象，格式如下：
batch_requests = []
for i, item in enumerate(data_to_process):
    request = {
        "custom_id": f"request-{i}",
        "method": "POST",
        "url": "/v1/chat/completions",
        "body": {
            "model": "gpt-4o",
            "messages": [
                {"role": "system", "content": "你是一个内容分类器。"},
                {"role": "user", "content": f"分类以下内容：{item}"}
            ],
            "max_tokens": 100
        }
    }
    batch_requests.append(json.dumps(request))

# 写入 JSONL 文件
with open("batch_requests.jsonl", "w") as f:
    f.write("\n".join(batch_requests))

# Step 2: 上传文件
batch_file = client.files.create(
    file=open("batch_requests.jsonl", "rb"),
    purpose="batch"
)

# Step 3: 创建 Batch 任务
batch_job = client.batches.create(
    input_file_id=batch_file.id,
    endpoint="/v1/chat/completions",
    completion_window="24h",  # 24小时内完成
)

print(f"Batch 任务已创建: {batch_job.id}")
print(f"状态: {batch_job.status}")  # validating → in_progress → completed

# Step 4: 等待完成
while True:
    batch_job = client.batches.retrieve(batch_job.id)
    print(f"状态: {batch_job.status} | "
          f"请求: {batch_job.request_counts.total} | "
          f"完成: {batch_job.request_counts.completed} | "
          f"失败: {batch_job.request_counts.failed}")
    
    if batch_job.status in ["completed", "failed", "expired"]:
        break
    time.sleep(60)  # 每分钟检查一次

# Step 5: 获取结果
if batch_job.status == "completed":
    result_file_id = batch_job.output_file_id
    result_content = client.files.content(result_file_id).content
    
    # 解析结果
    results = []
    for line in result_content.decode().split('\n'):
        if line.strip():
            result = json.loads(line)
            custom_id = result['custom_id']
            content = result['response']['body']['choices'][0]['message']['content']
            results.append({'id': custom_id, 'result': content})
    
    print(f"处理完成！成功 {len(results)} 条")

# 成本对比
# 标准 API: 1000 请求 × $0.005 = $5.00
# Batch API: 1000 请求 × $0.0025 = $2.50 (50% 折扣)

Anthropic Messages Batches

import anthropic

client = anthropic.Anthropic()

# Anthropic Messages Batches API
# 2024年推出，同样 50% 折扣

# 创建批量请求
batch = client.messages.batches.create(
    requests=[
        # 每个请求有自定义 ID
        {
            "custom_id": "req-1",
            "params": {
                "model": "claude-sonnet-4-20250514",
                "max_tokens": 1024,
                "messages": [
                    {"role": "user", "content": "总结这篇文章..."}
                ]
            }
        },
        {
            "custom_id": "req-2",
            "params": {
                "model": "claude-sonnet-4-20250514",
                "max_tokens": 1024,
                "messages": [
                    {"role": "user", "content": "分类这段内容..."}
                ]
            }
        },
        # ... 最多数千个请求
    ]
)

print(f"Batch ID: {batch.id}")
print(f"状态: {batch.processing_status}")

# 轮询等待完成
import time
while True:
    batch = client.messages.batches.retrieve(batch.id)
    if batch.processing_status == "ended":
        break
    time.sleep(30)

# 获取结果
for result in client.messages.batches.results(batch.id):
    print(f"ID: {result.custom_id}")
    if result.result.type == "succeeded":
        content = result.result.message.content[0].text
        print(f"结果: {content[:100]}")
    elif result.result.type == "errored":
        print(f"错误: {result.result.error}")

异步处理架构

import asyncio
from dataclasses import dataclass
from datetime import datetime

@dataclass
class BatchJob:
    id: str
    items: list
    status: str
    created_at: datetime
    completed_at: datetime = None
    results: list = None

class BatchProcessingPipeline:
    """批量处理管道"""
    
    def __init__(self, batch_size: int = 500, 
                 check_interval: int = 60):
        self.batch_size = batch_size
        self.check_interval = check_interval
        self.pending_jobs: dict[str, BatchJob] = {}
    
    async def submit(self, items: list[dict]) -> str:
        """提交批量处理任务"""
        job_id = f"batch_{datetime.now().strftime('%Y%m%d_%H%M%S')}"
        
        # 分批（如果超过单次限制）
        batches = []
        for i in range(0, len(items), self.batch_size):
            batch_items = items[i:i+self.batch_size]
            batch_id = await self._create_batch(batch_items)
            batches.append(batch_id)
        
        job = BatchJob(
            id=job_id,
            items=items,
            status='submitted',
            created_at=datetime.now()
        )
        self.pending_jobs[job_id] = job
        
        return job_id
    
    async def _create_batch(self, items: list) -> str:
        """创建 OpenAI Batch 任务"""
        # 准备请求文件
        requests = []
        for i, item in enumerate(items):
            requests.append(json.dumps({
                "custom_id": f"{item.get('id', i)}",
                "method": "POST",
                "url": "/v1/chat/completions",
                "body": {
                    "model": item.get('model', 'gpt-4o'),
                    "messages": item['messages'],
                    "max_tokens": item.get('max_tokens', 1024)
                }
            }))
        
        # 写文件并上传
        file_content = "\n".join(requests)
        batch_file = client.files.create(
            file=file_content.encode(),
            purpose="batch"
        )
        
        batch_job = client.batches.create(
            input_file_id=batch_file.id,
            endpoint="/v1/chat/completions",
            completion_window="24h"
        )
        
        return batch_job.id
    
    async def get_results(self, job_id: str) -> list:
        """获取批量任务结果"""
        job = self.pending_jobs.get(job_id)
        if not job:
            raise ValueError(f"Job {job_id} not found")
        
        if job.results:
            return job.results
        
        # 检查所有子批次的状态
        all_results = []
        for batch_id in job.items:
            batch = client.batches.retrieve(batch_id)
            if batch.status == "completed":
                content = client.files.content(batch.output_file_id).content
                for line in content.decode().split('\n'):
                    if line.strip():
                        result = json.loads(line)
                        all_results.append(result)
        
        job.results = all_results
        job.status = 'completed'
        job.completed_at = datetime.now()
        return all_results

# 典型使用场景：每日文档处理
"""
# cron job: 每天 2:00 AM 执行
@app.cron("0 2 * * *")
async def daily_document_processing():
    # 1. 获取当天新增文档
    new_docs = await get_new_documents()
    
    if not new_docs:
        return
    
    # 2. 构建批量请求
    items = []
    for doc in new_docs:
        items.append({
            'id': doc.id,
            'messages': [
                {"role": "system", "content": "提取文档关键信息..."},
                {"role": "user", "content": doc.content}
            ],
            'max_tokens': 500
        })
    
    # 3. 提交 Batch 任务（半价！）
    job_id = await pipeline.submit(items)
    
    # 4. 早上来检查结果
    # 结果会自动写入数据库
"""

Batch API vs 标准 API

常见问题

← 上一篇：模型路由 | 下一篇：Token 预算 → | 返回首页