🎯 CCA-F Study Day 23/20: Structured Data Extraction Scenario

Review Cycle Day 3 — JSON Schema Design, Forced tool_use, Validation-Retry Loops & Multi-Pass Pipelines

Cross-Domain Focus: Domain 4 (Prompt Engineering & Structured Output, ~20%) + Domain 2 (Tool Design, ~20%)

---

📌 Today's Focus

Today we dive deep into the Structured Data Extraction exam scenario — one of the 6 scenarios you could face, and arguably the most code-heavy. This scenario tests whether you can design a complete extraction pipeline: from schema design through forced tool_use to validation-retry loops and multi-pass verification.

This combines Domain 4 (structured output mechanics) with Domain 2 (tool design patterns). Together that's 40% of the exam weight. If you see this scenario on exam day, you'll know exactly what architecture to choose.

---

🧠 Core Concept 1: Three Ways to Get Structured Output from Claude

The exam tests whether you know when to use which approach. There are three mechanisms — each has different guarantees:

Method	Guarantee Level	How It Works	When to Use
1. Prompt Engineering	⚠️ Best-effort	Ask Claude nicely to output JSON	Prototyping only — never production
2. Forced tool_use	✅ Schema-guided	tool_choice: {"type": "tool", "name": "extract_data"}	When you need structured extraction with custom logic
3. Native Structured Outputs	✅✅ Grammar-constrained	output_config.formatwith JSON Schema	When you need guaranteed schema compliance (constrained decoding)

🚨 Exam Trap: "Please respond with valid JSON"

The exam will present a scenario where a developer uses prompt instructions like "Always output valid JSON matching this schema." This is WRONG for production. The correct answer is always: use tool_use with tool_choice or Native Structured Outputs with strict: true.

Why? Prompt-based JSON formatting:

• Can hallucinate extra fields
• Can violate type constraints (string where int expected)
• Can produce invalid JSON (unterminated strings, trailing commas)
• Requires post-processing try/except blocks — fragile at scale

---

🧠 Core Concept 2: Forced tool_use for Extraction

This is the original pattern (pre-Structured Outputs) and still heavily tested on the exam. You define a "fake" tool whose sole purpose is to force Claude to output structured data:

import anthropic, json

client = anthropic.Anthropic()

# The "tool" is really just a structured output schema
tools = [{
    "name": "extract_invoice",
    "description": "Extract structured data from an invoice document.",
    "input_schema": {
        "type": "object",
        "properties": {
            "vendor_name": {
                "type": "string",
                "description": "Company name of the vendor/supplier"
            },
            "invoice_number": {
                "type": "string",
                "description": "Invoice ID/number (e.g., INV-2024-001)"
            },
            "date": {
                "type": "string",
                "format": "date",
                "description": "Invoice date in ISO 8601 format (YYYY-MM-DD)"
            },
            "total_amount": {
                "type": "number",
                "description": "Total invoice amount in USD"
            },
            "line_items": {
                "type": "array",
                "items": {
                    "type": "object",
                    "properties": {
                        "description": {"type": "string"},
                        "quantity": {"type": "integer", "minimum": 1},
                        "unit_price": {"type": "number"},
                        "total": {"type": "number"}
                    },
                    "required": ["description", "quantity", "unit_price", "total"]
                },
                "description": "Individual line items on the invoice"
            },
            "payment_terms": {
                "type": "string",
                "enum": ["net_30", "net_60", "net_90", "due_on_receipt"],
                "description": "Payment terms"
            }
        },
        "required": ["vendor_name", "invoice_number", "date", "total_amount", "line_items"]
    }
}]

# Force Claude to call this specific tool
response = client.messages.create(
    model="claude-sonnet-4-20250514",
    max_tokens=1024,
    tools=tools,
    tool_choice={"type": "tool", "name": "extract_invoice"},  # FORCED
    messages=[{
        "role": "user",
        "content": f"Extract all data from this invoice:\n\n{invoice_text}"
    }]
)

# The structured data is in the tool call's input field
tool_block = next(b for b in response.content if b.type == "tool_use")
extracted_data = tool_block.input  # Already a dict matching your schema!

Key Mechanics (Exam Will Test These)

• tool_choice: {"type": "tool", "name": "..."} — Forces Claude to call exactly this tool
• tool_choice: {"type": "any"} — Claude must call some tool but picks which one
• tool_choice: {"type": "auto"} — Default; Claude decides whether to call tools at all
• The extracted data lives in response.content[0].input (the tool call's input arguments)
• stop_reason will be "tool_use" — NOT "end_turn"

---

🧠 Core Concept 3: Native Structured Outputs (strict: true)

The newer approach — uses grammar-constrained decoding to guarantee schema compliance at the token level:

# Method A: Structured Output via output_config
response = client.messages.create(
    model="claude-sonnet-4-20250514",
    max_tokens=1024,
    output_config={
        "format": {
            "type": "json_schema",
            "json_schema": {
                "name": "invoice_extraction",
                "strict": True,  # Constrained decoding!
                "schema": {
                    "type": "object",
                    "properties": {
                        "vendor_name": {"type": "string"},
                        "invoice_number": {"type": "string"},
                        "total_amount": {"type": "number"},
                        "confidence": {"type": "number"}
                    },
                    "required": ["vendor_name", "invoice_number", "total_amount"],
                    "additionalProperties": False
                }
            }
        }
    },
    messages=[{"role": "user", "content": f"Extract: {doc}"}]
)

# Response is guaranteed valid JSON matching schema
data = json.loads(response.content[0].text)

# Method B: strict tool_use (combines both approaches)
tools = [{
    "name": "extract_invoice",
    "description": "Extract invoice data",
    "strict": True,  # Grammar-constrained tool inputs!
    "input_schema": { ... }
}]

strict: true Key Behaviors

• First request has compilation latency (schema to grammar compilation)
• Compiled grammar is cached for 24 hours — subsequent calls are fast
• All properties in the schema will appear in output (required or not)
• Required properties appear first in the output
• additionalProperties: false is enforced at token level

---

🧠 Core Concept 4: Validation-Retry Loops

Even with structured outputs, you need validation for semantic correctness (schema guarantees syntax, not meaning):

import jsonschema
from datetime import datetime

MAX_RETRIES = 3

def extract_with_validation(document: str, schema: dict) -> dict:
    """Extract data with validation-retry loop."""
    messages = [{"role": "user", "content": f"Extract data from:\n\n{document}"}]
    
    for attempt in range(MAX_RETRIES):
        response = client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=1024,
            tools=[{
                "name": "extract_data",
                "description": "Extract structured data",
                "input_schema": schema
            }],
            tool_choice={"type": "tool", "name": "extract_data"},
            messages=messages
        )
        
        tool_block = next(b for b in response.content if b.type == "tool_use")
        extracted = tool_block.input
        
        # Validate semantically (schema handles syntax)
        errors = validate_extraction(extracted, document)
        
        if not errors:
            return extracted  # Success!
        
        # Feed errors back for retry
        messages.append({"role": "assistant", "content": response.content})
        messages.append({
            "role": "user",
            "content": [{
                "type": "tool_result",
                "tool_use_id": tool_block.id,
                "content": json.dumps({
                    "validation_errors": errors,
                    "instruction": "Please re-extract with these corrections."
                }),
                "is_error": True
            }]
        })
    
    # After max retries, flag for human review
    return {"extracted": extracted, "needs_human_review": True, "errors": errors}


def validate_extraction(data: dict, source: str) -> list:
    """Semantic validation — things a schema can't catch."""
    errors = []
    
    # Date sanity check
    if "date" in data:
        try:
            d = datetime.fromisoformat(data["date"])
            if d.year < 2000 or d > datetime.now():
                errors.append(f"Date {data['date']} seems implausible")
        except ValueError:
            errors.append(f"Invalid date format: {data['date']}")
    
    # Amount sanity check
    if "total_amount" in data and data["total_amount"] <= 0:
        errors.append("Total amount must be positive")
    
    # Line items sum check
    if "line_items" in data and "total_amount" in data:
        items_sum = sum(item.get("total", 0) for item in data["line_items"])
        if abs(items_sum - data["total_amount"]) > 0.01:
            errors.append(
                f"Line items sum ({items_sum}) doesn't match total ({data['total_amount']})"
            )
    
    return errors

Critical Distinction (Exam Tests This!)

Validation Type	What Catches It	Example
Syntactic	JSON Schema / strict mode	Wrong type, missing field, invalid enum value
Semantic	Your validation code	Future date, negative total, line items don't sum

---

🧠 Core Concept 5: Multi-Pass Extraction Pipeline

The exam's Structured Data Extraction scenario will present a question about processing a complex document. The correct architecture is a multi-pass pipeline with separate sessions:

# Multi-Pass Extraction Pipeline
# CRITICAL: Each pass uses a SEPARATE session (separate API call)

# Pass 1: Extraction (Session A)
def pass_1_extract(document: str) -> dict:
    """Raw extraction — get all fields."""
    response = client.messages.create(
        model="claude-sonnet-4-20250514",
        tools=[extraction_tool],
        tool_choice={"type": "tool", "name": "extract_fields"},
        messages=[{"role": "user", "content": f"Extract all fields:\n{document}"}]
    )
    return next(b for b in response.content if b.type == "tool_use").input

# Pass 2: Validation (Session B — SEPARATE!)
def pass_2_validate(extracted: dict, document: str) -> dict:
    """Cross-check extracted values against source document."""
    response = client.messages.create(
        model="claude-sonnet-4-20250514",
        tools=[validation_tool],
        tool_choice={"type": "tool", "name": "validate_extraction"},
        messages=[{
            "role": "user",
            "content": f"""Compare this extraction against the source.
            
Extracted: {json.dumps(extracted, indent=2)}
Source document: {document}

For each field, verify it matches the source. Flag discrepancies."""
        }]
    )
    return next(b for b in response.content if b.type == "tool_use").input

# Pass 3: Confidence scoring (Session C — SEPARATE!)
def pass_3_confidence(extracted: dict, validation: dict) -> dict:
    """Assign per-field confidence based on validation results."""
    response = client.messages.create(
        model="claude-sonnet-4-20250514",
        tools=[confidence_tool],
        tool_choice={"type": "tool", "name": "score_confidence"},
        messages=[{
            "role": "user",
            "content": f"""Score confidence for each field (0.0-1.0):
            
Extracted: {json.dumps(extracted)}
Validation notes: {json.dumps(validation)}

Fields with discrepancies get low confidence. Flag below 0.7 for human review."""
        }]
    )
    return next(b for b in response.content if b.type == "tool_use").input

# Orchestrator
def extract_pipeline(document: str) -> dict:
    extracted = pass_1_extract(document)
    validation = pass_2_validate(extracted, document)
    confidence = pass_3_confidence(extracted, validation)
    
    human_review_fields = [
        field for field, score in confidence["field_scores"].items()
        if score < 0.7
    ]
    
    return {
        "data": extracted,
        "validation": validation,
        "confidence": confidence,
        "needs_human_review": len(human_review_fields) > 0,
        "review_fields": human_review_fields
    }

Why Separate Sessions? (The #1 Exam Trap)

❌ Anti-Pattern: Running extraction and validation in the same conversation. Claude exhibits reasoning context bias — less likely to catch its own mistakes when extraction reasoning is in context.

✅ Correct: Each pass is a fresh API call. The validator never sees HOW the extraction was done — only the result and the source document.

---

⚠️ Anti-Patterns & Exam Traps Summary

❌ Anti-Pattern	✅ Correct Approach	Why It's Tested
Prompt-based JSON ("Please respond in JSON")	Use tool_use or Structured Outputs	No schema guarantee → production failures
Self-reported confidence ("Rate your confidence 1-10")	Programmatic validation + separate-session verification	Models are poorly calibrated on self-assessment
Same-session validation	Separate sessions for extraction vs. validation	Reasoning context bias
No retry logic for extraction failures	Validation-retry loop with max_retries=3	Extraction is non-deterministic
Aggregate accuracy metrics only	Track accuracy per document type	90% overall can mask 50% on one doc type
Over-complex nested schemas	Keep schemas flat; arrays of simple objects	Deep nesting increases grammar complexity

---

📖 Reading

• Primary: Strict Tool Use — Anthropic Docs
• Secondary: Structured Outputs — Claude Platform Docs
• Blog: Structured Outputs on Claude Developer Platform (Feb 2026)
• Hands-on: Anthropic Official Course — Tool Use for Structured Outputs (Notebook)

---

🛠️ Hands-On Exercise (25 min)

Build a Complete Invoice Extraction Pipeline

1. Schema Design (5 min): Write a JSON Schema for extracting: vendor name, invoice number, date, total, tax, line items (array with description, qty, unit_price, total), and payment terms (enum)
2. Forced Extraction (5 min): Use tool_choice: {"type": "tool"} to extract from a sample invoice text
3. Validation Function (10 min): Write validate_invoice() checking: line items sum = total, date not future, invoice number format, all amounts positive
4. Retry Loop (5 min): Wrap in retry loop (max 3). On failure, feed errors back as tool_result with is_error: true

Bonus: Add a second-pass validator in a separate API call that cross-checks the extracted total against the source document text.

---

📝 Quick Quiz

Q1. A developer extracts data from 10,000 invoices. Overall accuracy is 94%, but one vendor (VendorX) has 61% accuracy. What is the MOST effective fix?

A) Increase max_tokens to give Claude more space to think B) Add more few-shot examples of VendorX invoices to the prompt C) Track accuracy per vendor AND add vendor-specific few-shot examples for low-performing categories D) Switch from tool_use to Native Structured Outputs with strict: true

Q2. A pipeline extracts patient data then validates in the same conversation. Validation rarely catches errors. What architectural change helps most?

A) Add "Be extremely critical" to the validation prompt B) Run validation in a separate API call (separate session) C) Increase effort level to "max" for validation D) Add a third pass that validates the validator

Q3. What is the relationship between strict: true and validation-retry loops?

A) strict: true eliminates the need for validation-retry loops entirely B) strict: true guarantees syntactic compliance; validation-retry catches semantic errors C) Validation-retry loops replace the need for strict: true D) They are mutually exclusive approaches

---

Answers

Q1: C — Track per-vendor accuracy AND add targeted few-shot examples. Aggregate metrics mask category-specific failures. Switching to strict (D) only fixes syntax, not the semantic extraction issues with VendorX's unusual format.

Q2: B — Separate sessions. This is reasoning context bias. When the validator sees the extraction reasoning in context, it's biased toward agreeing. A fresh session evaluates purely on extracted data vs. source.

Q3: B — Complementary, not exclusive. strict: true guarantees valid JSON with correct types/structure. But it can't guarantee the values are correct (right date, right amount). Validation-retry loops catch these semantic errors.

---

🔮 Tomorrow's Preview

Day 24: Customer Support Resolution Agent Scenario — Domain 1's flagship exam scenario. Hooks for compliance, escalation triggers (not sentiment!), and structured error handling come together in a complete agent architecture.