Claude Codeサブエージェントで請求書を1日2,000件自動処理する実装ガイド【2026年版】

アーキテクチャの全体像

受信ファイル（PDF/画像/テキスト）
    ↓
前処理エージェント（テキスト抽出）
    ↓ ×並列処理
Claudeサブエージェント（構造化データ抽出）
    ↓
検証エージェント（confidence_score算出）
    ↓ ×2,000件/日
CSV/JSON出力 → 会計システムへ

実装コード

1. メインオーケストレーター

# invoice_orchestrator.py
# 請求書処理のメインコントローラー

import anthropic
import json
import concurrent.futures
from pathlib import Path
from dataclasses import dataclass

client = anthropic.Anthropic()

@dataclass
class InvoiceResult:
    file_path: str
    success: bool
    data: dict
    confidence_score: float
    error: str = ""

EXTRACT_PROMPT = """以下の請求書テキストから情報を抽出し、JSON形式で返してください。

必須フィールド:
- vendor_name: 取引先会社名
- invoice_date: 請求日 (YYYY-MM-DD形式)
- due_date: 支払期日 (YYYY-MM-DD形式、不明な場合はnull)
- total_amount: 合計金額 (数値のみ、通貨記号除く)
- currency: 通貨コード (JPY, USD等)
- line_items: 品目リスト [{name, quantity, unit_price, amount}]
- confidence_score: 抽出精度への自己評価 (0.0〜1.0)

JSONのみ返答してください。説明は不要です。

請求書テキスト:
{text}"""

def extract_invoice(file_path: str, text: str) -> InvoiceResult:
    """1件の請求書からデータを抽出するサブエージェント"""
    try:
        response = client.messages.create(
            model="claude-sonnet-4-5",  # 精度重視 ($3/1M入力)
            max_tokens=1024,
            messages=[{"role": "user", "content": EXTRACT_PROMPT.format(text=text[:8000])}]
        )
        
        data = json.loads(response.content[0].text)
        score = float(data.get("confidence_score", 0.5))
        
        return InvoiceResult(
            file_path=file_path,
            success=True,
            data=data,
            confidence_score=score
        )
    except json.JSONDecodeError as e:
        return InvoiceResult(file_path=file_path, success=False, data={},
                           confidence_score=0.0, error=f"JSON解析エラー: {e}")
    except Exception as e:
        return InvoiceResult(file_path=file_path, success=False, data={},
                           confidence_score=0.0, error=str(e))

def process_invoices_parallel(texts: dict[str, str], max_workers: int = 10) -> list[InvoiceResult]:
    """並列処理で複数請求書を同時処理"""
    results = []
    with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = {
            executor.submit(extract_invoice, path, text): path
            for path, text in texts.items()
        }
        for future in concurrent.futures.as_completed(futures):
            results.append(future.result())
    return results

2. Batches APIで大量処理（50%コスト削減）

# batch_invoice_processor.py
# 翌日処理でよければBatches APIで半額

import anthropic
import time

client = anthropic.Anthropic()

def submit_batch(texts: dict[str, str]) -> str:
    """2,000件をバッチ送信（通常料金の50%）"""
    requests = [
        {
            "custom_id": file_path,
            "params": {
                "model": "claude-haiku-4-5-20251001",  # バッチは最安モデルで
                "max_tokens": 1024,
                "messages": [{
                    "role": "user",
                    "content": f"請求書から取引先・日付・金額をJSONで抽出: {text[:5000]}"
                }]
            }
        }
        for file_path, text in texts.items()
    ]
    
    batch = client.beta.messages.batches.create(requests=requests)
    print(f"バッチ送信完了: {batch.id} ({len(requests)}件)")
    print(f"推定コスト: ${len(requests) * 0.0004:.2f}（通常の50%）")
    return batch.id

def wait_and_collect(batch_id: str) -> list[dict]:
    """バッチ完了を待ってデータを収集"""
    while True:
        batch = client.beta.messages.batches.retrieve(batch_id)
        if batch.processing_status == "ended":
            break
        print(f"処理中... ({batch.request_counts.succeeded}/{batch.request_counts.processing})")
        time.sleep(60)  # 1分ごとに確認
    
    results = []
    for result in client.beta.messages.batches.results(batch_id):
        if result.result.type == "succeeded":
            try:
                import json
                data = json.loads(result.result.message.content[0].text)
                results.append({"id": result.custom_id, "data": data})
            except:
                results.append({"id": result.custom_id, "error": "JSON解析失敗"})
    
    return results

3. コスト試算

# cost_calculator.py

MODELS = {
    "claude-sonnet-4-5":       (3.00, 15.00),  # 精度重視
    "claude-haiku-4-5-20251001": (0.80,  4.00),  # コスト重視
}

def estimate_cost(num_invoices: int, model: str, 
                  avg_input: int = 2000, avg_output: int = 500,
                  use_batch: bool = False) -> dict:
    inp, out = MODELS[model]
    per_invoice = (avg_input * inp + avg_output * out) / 1_000_000
    if use_batch:
        per_invoice *= 0.5  # Batches APIで50%オフ
    
    total = per_invoice * num_invoices
    return {
        "per_invoice_usd": round(per_invoice, 4),
        "total_usd": round(total, 2),
        "monthly_usd": round(total * 22, 2),  # 月22営業日
    }

# シナリオ比較
for model in MODELS:
    for batch in [False, True]:
        result = estimate_cost(2000, model, use_batch=batch)
        label = f"{model}{'（バッチ）' if batch else ''}"
        print(f"{label}: {result['per_invoice_usd']}$/件, {result['total_usd']}$/日, {result['monthly_usd']}$/月")

出力例:

claude-sonnet-4-5: 0.0015$/件, 3.0$/日, 66.0$/月
claude-sonnet-4-5（バッチ）: 0.0008$/件, 1.5$/日, 33.0$/月
claude-haiku-4-5-20251001: 0.0004$/件, 0.8$/日, 17.6$/月
claude-haiku-4-5-20251001（バッチ）: 0.0002$/件, 0.4$/日, 8.8$/月

実行手順

pip install anthropic pdfplumber  # テキスト抽出に必要
export ANTHROPIC_API_KEY="your_key"
python invoice_orchestrator.py

あわせて読みたい

• Claude API高額請求を防ぐ方法
• Claude Code AIワークフロー自動化
• GitHub Actions + Claude自動コードレビュー