契約台帳をPDFから一括抽出 — Claude APIとBatch APIの手順

import base64
import json
from pathlib import Path
 
import anthropic
from anthropic.types.message_create_params import MessageCreateParamsNonStreaming
from anthropic.types.messages.batch_create_params import Request
 
client = anthropic.Anthropic()
MODEL = "claude-sonnet-5-5"
BATCH_BYTES_LIMIT = 150 * 1024 * 1024  # 256MBの上限に余裕を持たせた目安
 
SYSTEM = """あなたは契約書から台帳用の項目を抜き出す担当です。
PDFに書かれていることだけを根拠にし、書かれていない項目は null にします。
推測や一般的な慣行で補いません。日付はYYYY-MM-DD形式にします。
出力はJSONオブジェクト1つだけで、前後に説明文を付けません。"""
 
PROMPT = """次のキーを持つJSONを返してください。
title, parties(配列。社名と呼称), effective_date, end_date(現在の期間の満了日),
auto_renewal(真偽値), notice_text(解約通知の原文), notice_days(整数。日数に直せるときのみ),
governing_law, pages(各キーの根拠ページ番号のオブジェクト), memo(読み取りに迷った点)"""
 
 
def build_request(custom_id: str, pdf_path: Path) -> Request:
    data = base64.standard_b64encode(pdf_path.read_bytes()).decode("utf-8")
    return Request(
        custom_id=custom_id,
        params=MessageCreateParamsNonStreaming(
            model=MODEL,
            max_tokens=1500,
            system=SYSTEM,
            messages=[{
                "role": "user",
                "content": [
                    {"type": "document",
                     "source": {"type": "base64",
                                "media_type": "application/pdf",
                                "data": data}},
                    {"type": "text", "text": PROMPT},
                ],
            }],
        ),
    )
 
 
pdfs = sorted(Path("contracts").glob("*.pdf"))
manifest = {f"c{i:04d}": p.name for i, p in enumerate(pdfs, start=1)}
Path("manifest.json").write_text(
    json.dumps(manifest, ensure_ascii=False, indent=2), encoding="utf-8")
 
# サイズの合計がBATCH_BYTES_LIMITを超えない範囲でバッチを分ける
groups, current, size = [], [], 0
for cid, name in manifest.items():
    file_size = Path("contracts", name).stat().st_size
    if current and size + file_size > BATCH_BYTES_LIMIT:
        groups.append(current)
        current, size = [], 0
    current.append(cid)
    size += file_size
if current:
    groups.append(current)
 
batch_ids = []
for group in groups:
    requests = [build_request(cid, Path("contracts", manifest[cid]))
                for cid in group]
    batch = client.messages.batches.create(requests=requests)
    batch_ids.append(batch.id)
 
Path("batch_ids.json").write_text(json.dumps(batch_ids), encoding="utf-8")
print(batch_ids)

import json
import time
from pathlib import Path
 
import anthropic
 
client = anthropic.Anthropic()
batch_ids = json.loads(Path("batch_ids.json").read_text(encoding="utf-8"))
 
pending = set(batch_ids)
while pending:
    for bid in list(pending):
        b = client.messages.batches.retrieve(bid)
        if b.processing_status == "ended":
            pending.discard(bid)
            print(bid, b.request_counts)
    if pending:
        time.sleep(60)

import json
from pathlib import Path
 
import anthropic
 
client = anthropic.Anthropic()
batch_ids = json.loads(Path("batch_ids.json").read_text(encoding="utf-8"))
 
raw, failed = {}, {}
for bid in batch_ids:
    for item in client.messages.batches.results(bid):
        outcome = item.result
        if outcome.type == "succeeded":
            raw[item.custom_id] = outcome.message.content[0].text
        elif outcome.type == "errored":
            failed[item.custom_id] = outcome.error.error.type
        else:
            failed[item.custom_id] = outcome.type
 
Path("raw.json").write_text(
    json.dumps(raw, ensure_ascii=False), encoding="utf-8")
Path("failed.json").write_text(json.dumps(failed), encoding="utf-8")
print(len(raw), "件成功 /", len(failed), "件失敗または未処理")

import csv
import json
import re
from datetime import date, timedelta
from pathlib import Path
 
manifest = json.loads(Path("manifest.json").read_text(encoding="utf-8"))
raw = json.loads(Path("raw.json").read_text(encoding="utf-8"))
 
KEYS = ["title", "parties", "effective_date", "end_date", "auto_renewal",
        "notice_text", "notice_days", "governing_law", "pages", "memo"]
 
 
def parse(text: str):
    text = re.sub(r"^```(?:json)?\s*|\s*```$", "", text.strip())
    try:
        return json.loads(text)
    except json.JSONDecodeError:
        return None
 
 
def to_date(value):
    try:
        return date.fromisoformat(value) if value else None
    except ValueError:
        return None
 
 
rows = []
for cid, name in manifest.items():
    data = parse(raw.get(cid, ""))
    row = {"custom_id": cid, "file": name, "review": ""}
    if not isinstance(data, dict):
        row["review"] = "JSONとして読めない、または結果なし"
        rows.append(row)
        continue
    for k in KEYS:
        v = data.get(k)
        row[k] = (json.dumps(v, ensure_ascii=False)
                  if isinstance(v, (list, dict)) else v)
    end, days = to_date(data.get("end_date")), data.get("notice_days")
    if end and isinstance(days, int):
        row["notice_deadline"] = (end - timedelta(days=days)).isoformat()
    reasons = [k for k in ("end_date", "auto_renewal", "governing_law")
               if data.get(k) is None]
    if data.get("memo"):
        reasons.append("memoあり")
    row["review"] = " / ".join(reasons)
    rows.append(row)
 
fields = ["custom_id", "file"] + KEYS + ["notice_deadline", "review"]
with open("ledger.csv", "w", newline="", encoding="utf-8-sig") as f:
    writer = csv.DictWriter(f, fieldnames=fields, extrasaction="ignore")
    writer.writeheader()
    writer.writerows(rows)