Claude 批量处理文档：用API实现文档批量处理的完整指南

Claude 批量处理文档：用API实现文档批量处理的完整指南

每天处理几十上百份文档——合同审核、报告摘要、客户邮件分类、学术论文分析——如果还靠手动逐份操作，时间成本高到不现实。Claude 批量处理文档指通过 Anthropic 的 Claude API（Messages API 批处理端点）一次性提交大量文档的分析、摘要、分类或提取任务，代替逐条对话操作。

这种方式能显著压缩处理时间：一份合同人工审核约20分钟，用批量接口处理100份文档通常缩短到10分钟以内（取决于提示词长度和输出要求）。下面从准备工作到常见误区，给出可直接复现的操作步骤。

延伸阅读：claude 搭建企业知识库、linux do claude.md怎么写

---

开始前准备

必要条件

• Anthropic API Key：在 console.anthropic.com 创建，推荐付费账户（免费额度低且速率限制严格）。
• Python 3.8+ 环境：需包含 requests、json、csv 等标准库（也可用 Node.js 或 cURL 替代）。
• 文档格式：纯文本（.txt）、Markdown（.md）、或从 PDF/DOCX 提取后的纯文本（用 PyMuPDF、python-docx 或 pdfminer.six 预处理）。Claude 批量接口不直接读取二进制文件。
• API 端点：https://api.anthropic.com/v1/messages（非流式请求），推荐 claude-sonnet-4-20250514 作为成本与质量的平衡点。

准备工作流示意

原始文档 → 文本提取 → JSONL 构建 → 批量请求 → 结果解析

每步在后续详细展开。

---

操作步骤

步骤 1：提取并清洗文档文本

这是出错率最高的环节。确保每份文档：

• 无多余空白行与乱码：用正则替换连续两个以上换行符为单个换行。
• 编码统一：全部转为 UTF-8，避免 gbk 或 latin-1 导致截断。
• 长度控制：单文档最多200K token（约15万字中文），超长文档先分段再批量处理。

示例提取函数（针对 .txt 文件）：

def clean_text(filepath):
    with open(filepath, 'r', encoding='utf-8') as f:
        text = f.read()
    text = re.sub(r'\n{3,}', '\n\n', text)
    return text.strip()

步骤 2：构建 JSONL 请求文件

批量接口要求将所有请求写入一个 .jsonl 文件，每行一个完整的 Messages API 对象。关键字段：

• model：模型 ID
• max_tokens：输出上限（建议设1024-4096，根据任务调整）
• messages：角色数组，格式 [{"role": "user", "content": "..."}]
• system：可选的系统提示（用于批量设置角色或输出格式）

构建示例（以摘要任务为例）：

import json

def build_request(text, task_type="摘要"):
    system_prompt = "你是一位文档分析助手。请严格按照以下 JSON 格式输出摘要：{\"title\": \"...\", \"key_points\": [\"...\"], \"conclusion\": \"...\"}"
    user_prompt = f"请对以下文档做{task_type}。\n\n文档内容：\n{text}"
    return {
        "model": "claude-sonnet-4-20250514",
        "max_tokens": 1024,
        "system": system_prompt,
        "messages": [
            {"role": "user", "content": user_prompt}
        ]
    }

# 假设 files 是已提取好的文本列表
with open("batch_requests.jsonl", "w", encoding="utf-8") as f:
    for idx, text in enumerate(files):
        req = build_request(text)
        f.write(json.dumps(req, ensure_ascii=False) + "\n")

步骤 3：提交批量请求并获取结果

import requests

API_KEY = "sk-ant-..."
headers = {
    "x-api-key": API_KEY,
    "anthropic-version": "2023-06-01",
    "content-type": "application/json"
}

with open("batch_requests.jsonl", "rb") as f:
    files_data = {"file": ("batch.jsonl", f, "application/jsonl")}
    resp = requests.post(
        "https://api.anthropic.com/v1/messages/batches",
        headers=headers,
        files=files_data
    )
batch_id = resp.json().get("id")
print(f"Batch ID: {batch_id}")

成功提交后得到 batch_id，用于后续状态查询与结果下载。注意速率限制：付费账户通常允许每分钟1000次请求，但单批量文件最大请求数为50,000行。

步骤 4：轮询状态并拉取结果

import time

def poll_batch(batch_id, interval=30):
    url = f"https://api.anthropic.com/v1/messages/batches/{batch_id}"
    while True:
        resp = requests.get(url, headers=headers)
        data = resp.json()
        status = data.get("processing_status", {})
        print(f"Pending: {status.get('pending',0)}, Succeeded: {status.get('succeeded',0)}, Errored: {status.get('errored',0)}")
        if status.get("pending", 0) == 0:
            results_url = data.get("results_url")
            if results_url:
                results_resp = requests.get(results_url)
                results = [json.loads(line) for line in results_resp.text.strip().split("\n")]
                return results
        time.sleep(interval)

results = poll_batch(batch_id, interval=60)

结果文件中每条记录包含 custom_id（对应原请求顺序）、response（含 content 数组）和可能的 error。

---

检查清单：运行前必查

检查项	正确做法	常见错误
请求 JSONL 格式	每行一个独立 JSON 对象，无逗号结尾	用标准 JSON 数组格式（无效）
token 总数	单个请求输入 ≤200K token	超长文档未截断导致400错误
系统提示（system）	在请求级设置，不放入 messages	将系统提示写入 user 内容
速率限制	单批次 ≤5万请求并发	一次性提交过多请求触发429
输出格式约束	在 system 或 user prompt 中指定 JSON	依赖默认输出（结果可能不一致）

---

常见问题与排查

问题 1：批量提交后立即返回 400 错误

原因：JSONL 格式不合法（常见于最后多一个空行导致 JSON 解析失败）或总 token 超限。
检查：用 python -m json.tool 逐行验证；统计所有请求的 input_tokens 总和不超过200K×请求数（单批次无全局总量上限，但单请求上限严格）。

问题 2：部分请求返回 errored 状态但无详细信息

原因：文档内容含超出长度限制的特殊字符（如无中断 Unicode 序列）或 request 字段缺失。
解决：确保每行 JSON 包含 model 和 messages 字段；对特殊字符编码为 \\uXXXX 或用 ensure_ascii=False 写入时确认文件编码。

问题 3：输出 JSON 解析失败

原因：模型输出偶有额外文本（如开场白 Sure, here is the summary）。
解决：在 system 提示中强调“只输出 JSON，不要额外文字”，并在解析时用正则从 content 中提取第一个 {...} 或 [...] 块。

问题 4：结果速度远低于预期

原因：单批量请求数少（<100）时可能不如逐条请求快，因为批处理有排队时间（通常1-5分钟）。
建议：单次提交至少100条请求以摊薄排队开销；如需更快响应，改用标准 Messages API（逐条发送，但无批状态管理）。

---

进阶技巧

动态调整输出格式

不同任务类型需要不同输出结构。可在 system 提示中预设多个模板：

TASK_TEMPLATES = {
    "摘要": "输出 JSON: {\"title\": \"...\", \"key_points\": [\"...\"], \"conclusion\": \"...\"}",
    "分类": "输出 JSON: {\"category\": \"...\", \"confidence\": 0.0-1.0, \"reason\": \"...\"}",
    "合同审核": "输出 JSON: {\"risks\": [\"...\"], \"action_items\": [\"...\"], \"overall_verdict\": \"...\"}"
}

根据 task_type 参数动态选择模板，确保输出格式一致。

批量重试机制

对失败请求自动重