Structured Data Extraction.

from anthropic import Anthropic
client = Anthropic()

EXTRACT_TOOL = {
    "name": "extract_record",
    "description": "Extract a structured record from a customer email.",
    "input_schema": {
        "type": "object",
        "properties": {
            "customer_id": {"type": "string", "pattern": "^cust_[0-9]+$"},
            "refund_amount": {"type": ["number", "null"]},
            "refund_reason": {
                "type": "string",
                "enum": ["damage", "wrong_item", "late", "other", "unclear"],
            },
            "urgency": {
                "type": "string",
                "enum": ["low", "medium", "high", "unclear"],
            },
        },
        "required": ["customer_id", "refund_amount", "refund_reason", "urgency"],
    },
}

def extract_one(email_text: str) -> dict:
    resp = client.messages.create(
        model="claude-sonnet-4.5",
        max_tokens=1024,
        tools=[EXTRACT_TOOL],
        tool_choice={"type": "tool", "name": "extract_record"},
        messages=[{"role": "user", "content": email_text}],
    )
    # The model MUST emit a tool_use block matching EXTRACT_TOOL.input_schema
    for block in resp.content:
        if block.type == "tool_use" and block.name == "extract_record":
            return block.input  # already a dict matching the schema shape
    raise RuntimeError("forced tool_choice did not yield tool_use. SDK bug")

# Few-shot: show the model how to emit 'unclear' on a silent source
FEW_SHOT_EXAMPLES = [
    {
        "role": "user",
        "content": "I want a refund for my order. Thanks.",
    },
    {
        "role": "assistant",
        "content": [{
            "type": "tool_use",
            "name": "extract_record",
            "input": {
                "customer_id": "cust_unknown",  # signals: pattern won't match, route to human
                "refund_amount": None,
                "refund_reason": "unclear",
                "urgency": "unclear",
            },
        }],
    },
    {
        "role": "user",
        "content": [{
            "type": "tool_result",
            "tool_use_id": "...",
            "content": "ok",
        }],
    },
]
# Then your real message follows:
# messages = FEW_SHOT_EXAMPLES + [{"role": "user", "content": email_text}]

import re

def validate(record: dict) -> list[str]:
    """Returns a list of specific error messages (empty = valid)."""
    errors = []
    if record.get("refund_amount") is not None and record["refund_amount"] < 0:
        errors.append("refund_amount must be non-negative")
    if not re.fullmatch(r"cust_\d{4,}", record.get("customer_id", "")):
        errors.append("customer_id must match cust_<4+ digits>")
    if record.get("refund_amount", 0) > 1000 and record.get("urgency") == "low":
        errors.append("refund_amount > 1000 with urgency='low' is suspicious")
    return errors

def extract_with_retry(email_text: str, max_retries: int = 3) -> dict:
    messages = [{"role": "user", "content": email_text}]
    for attempt in range(max_retries):
        resp = client.messages.create(
            model="claude-sonnet-4.5",
            max_tokens=1024,
            tools=[EXTRACT_TOOL],
            tool_choice={"type": "tool", "name": "extract_record"},
            messages=messages,
        )
        tool_use = next(b for b in resp.content if b.type == "tool_use")
        record = tool_use.input
        errors = validate(record)
        if not errors:
            return record
        # Feed the SPECIFIC error back; generic 'retry' doesn't converge
        messages.append({"role": "assistant", "content": resp.content})
        messages.append({
            "role": "user",
            "content": [{
                "type": "tool_result",
                "tool_use_id": tool_use.id,
                "content": f"Validation failed: {'; '.join(errors)}. Re-extract.",
                "is_error": True,
            }],
        })
    raise ValueError(f"extraction did not converge in {max_retries} attempts")

def extract_with_cache(email_text: str) -> dict:
    resp = client.messages.create(
        model="claude-sonnet-4.5",
        max_tokens=1024,
        tools=[
            {
                **EXTRACT_TOOL,
                "cache_control": {"type": "ephemeral"},  # 5-min TTL
            },
        ],
        tool_choice={"type": "tool", "name": "extract_record"},
        messages=[{"role": "user", "content": email_text}],
    )
    # Inspect cache stats for observability
    print(f"cache_creation: {resp.usage.cache_creation_input_tokens}")
    print(f"cache_read:     {resp.usage.cache_read_input_tokens}")
    return next(b.input for b in resp.content if b.type == "tool_use")

# .claude/hooks/extract_policy.py
import sys, json, os

REFUND_CAP = float(os.environ.get("REFUND_CAP", "500"))

payload = json.loads(sys.stdin.read())
if payload["tool_name"] != "extract_record":
    sys.exit(0)
amount = payload["tool_input"].get("refund_amount") or 0
if amount > REFUND_CAP:
    print(
        f"refund_amount ${amount} exceeds policy cap ${REFUND_CAP}; "
        f"emit refund_amount=null and refund_reason='unclear' instead",
        file=sys.stderr,
    )
    sys.exit(2)  # DENY. Model sees error, re-extracts with the bound
sys.exit(0)

import json

def submit_batch_extraction(emails: list[dict]) -> str:
    """Submit a batch of extraction requests for overnight processing."""
    requests = []
    for email in emails:
        requests.append({
            "custom_id": f"extract-{email['id']}",
            "params": {
                "model": "claude-sonnet-4.5",
                "max_tokens": 1024,
                "tools": [EXTRACT_TOOL],
                "tool_choice": {"type": "tool", "name": "extract_record"},
                "messages": [{"role": "user", "content": email["body"]}],
            },
        })
    batch = client.messages.batches.create(requests=requests)
    print(f"Batch {batch.id} submitted with {len(requests)} extractions")
    print(f"Expected ready: {batch.expires_at}")
    return batch.id

# Next morning. Fetch results, validate each, requeue failures
def harvest_batch(batch_id: str):
    batch = client.messages.batches.retrieve(batch_id)
    if batch.processing_status != "ended":
        return {"status": "not_ready"}
    results = client.messages.batches.results(batch_id)
    accepted, rejected = [], []
    for r in results:
        if r.result.type == "succeeded":
            tu = next(b for b in r.result.message.content if b.type == "tool_use")
            if not validate(tu.input):
                accepted.append(tu.input)
                continue
        rejected.append(r.custom_id)
    return {"accepted": accepted, "rejected_for_retry": rejected}

from collections import defaultdict

def stratified_accuracy(records: list[dict]) -> dict:
    """Group validation results by document type, surface weak strata."""
    by_type = defaultdict(lambda: {"pass": 0, "fail": 0})
    for r in records:
        errors = validate(r["extracted"])
        bucket = "pass" if not errors else "fail"
        by_type[r["doc_type"]][bucket] += 1

    report = {}
    for doc_type, counts in by_type.items():
        total = counts["pass"] + counts["fail"]
        report[doc_type] = {
            "total": total,
            "pass_rate": counts["pass"] / total if total else 0,
            "fail_count": counts["fail"],
        }
    # Sort by pass_rate ascending. Worst strata first
    return dict(sorted(report.items(), key=lambda kv: kv[1]["pass_rate"]))

# Output:
# {
#   "html_email": {"total": 200, "pass_rate": 0.62, "fail_count": 76},  # WEAK
#   "pdf":        {"total": 150, "pass_rate": 0.91, "fail_count": 14},
#   "plain_text": {"total": 650, "pass_rate": 0.99, "fail_count":  6},
# }