LLM で項目を抽出する ── 非決定的な処理を工程に組み込む作法

import jsonimport refrom dataclasses import dataclass

@dataclassclass Extracted:    invoice_no: str    issued_on: str      # YYYY-MM-DD    total: int    tax_rate: float

class ExtractionError(Exception):    pass

def parse_and_validate(raw: str) -> Extracted:    """モデルの出力を検証する。おかしければ例外を投げる。"""    try:        data = json.loads(raw)    except json.JSONDecodeError as e:        raise ExtractionError(f"JSON として読めない: {e}") from e
    required = {"invoice_no", "issued_on", "total", "tax_rate"}    missing = required - set(data)    if missing:        raise ExtractionError(f"項目が足りない: {sorted(missing)}")
    if not re.fullmatch(r"\d{4}-\d{2}-\d{2}", str(data["issued_on"])):        raise ExtractionError(f"日付の形式が違う: {data['issued_on']!r}")
    if not isinstance(data["total"], int) or data["total"] < 0:        raise ExtractionError(f"金額が不正: {data['total']!r}")
    if not 0 <= float(data["tax_rate"]) <= 1:        raise ExtractionError(f"税率が範囲外: {data['tax_rate']!r}")
    return Extracted(        invoice_no=str(data["invoice_no"]),        issued_on=str(data["issued_on"]),        total=int(data["total"]),        tax_rate=float(data["tax_rate"]),    )

from collections import defaultdict

def field_accuracy(cases):    """項目ごとの一致率と、外れた例を出す。
    cases: [{"expected": {...}, "actual": {...}, "id": "..."}, ...]    """    hit = defaultdict(int)    total = defaultdict(int)    misses = defaultdict(list)
    for c in cases:        for field, expected in c["expected"].items():            total[field] += 1            actual = c["actual"].get(field)            if actual == expected:                hit[field] += 1            else:                misses[field].append((c["id"], expected, actual))
    print(f"{'項目':16s}{'一致率':>8}{'件数':>7}")    for field in sorted(total, key=lambda f: hit[f] / total[f]):        rate = hit[field] / total[field]        print(f"{field:16s}{rate:8.1%}{total[field]:7d}")
    print("\n外れた例（各項目3件まで）")    for field, ms in misses.items():        for case_id, exp, act in ms[:3]:            print(f"  {field:14s} {case_id}: 期待 {exp!r} / 実際 {act!r}")
    return {f: hit[f] / total[f] for f in total}

def extract_with_fallback(client, text, max_retries=2):    """抽出して検証。通らなければ再試行し、それでも駄目なら人へ。"""    last_error = None
    for attempt in range(max_retries + 1):        raw = client.extract(text)        try:            return {"status": "ok", "data": parse_and_validate(raw),                    "attempts": attempt + 1}        except ExtractionError as e:            last_error = e
    # 検証を通らなかったものは、捨てずに人の確認へ回す    return {        "status": "needs_review",        "reason": str(last_error),        "raw_text": text,        "attempts": max_retries + 1,    }