🧱 Structured Outputs, Schema Enforcement & Self-Healing (LAB-06)¶
Estimated time to read: 20 minutes
Welcome to the interactive playground for LAB-06: Structured Outputs.
In enterprise production systems, probabilistic LLM generations must interface with deterministic downstream databases, APIs, and business logic. Unstructured natural language responses can break data pipelines through format hallucination, missing keys, and invalid datatypes.
What you will explore in this notebook:¶
- Pydantic v2 Schema Modeling: Strict field validations, custom sanitizers, and multi-field invariants.
- Resilient Parsing: Stripping conversational preambles, markdown code fences, and repairing trailing commas.
- XML Tag Delimited Parsing: Alternative prompting format that reduces syntax errors in models with high tag affinity.
- Self-Healing / Reflection Loops: Re-injecting Pydantic validation tracebacks into follow-up turns for autonomous error correction.
- Instructor & Tool Calling: Native integration with OpenAI-compatible function calling.
- Empirical Benchmarks: Quantitative comparison of syntax reliability, retry overhead, and success rates.
0. Environment Setup¶
Automatically discovers the repository root or clones the codebase if running inside Google Colab.
import os
import sys
from pathlib import Path
try:
import google.colab
IN_COLAB = True
except ImportError:
IN_COLAB = False
if IN_COLAB:
print("🤖 Running in Google Colab. Setting up environment...")
!pip install -q pydantic instructor openai matplotlib loguru
# Clone repository to access custom local source modules
if not os.path.exists("production-ai-systems"):
!git clone https://github.com/RogerioLS/production-ai-systems.git
sys.path.append("production-ai-systems")
print("🎉 Colab environment successfully initialized!")
else:
current_dir = Path.cwd()
root_dir = current_dir
for parent in [current_dir] + list(current_dir.parents):
if (parent / "pyproject.toml").exists() or (parent / "projects").exists():
root_dir = parent
break
if str(root_dir) not in sys.path:
sys.path.insert(0, str(root_dir))
print(f"📁 Project root added to sys.path: {root_dir}")
📁 Project root added to sys.path: /home/runner/work/production-ai-systems/production-ai-systems
1. Pydantic v2 Schema Modeling¶
Let's import our domain models: FinancialContract, ContractParty, and ContractObligation.
from pydantic import ValidationError
from projects.b_prompt_engineering.src.lab_06_structured_outputs import (
FinancialContract,
ContractParty,
ContractObligation,
CurrencyCode,
PartyType,
)
# Create a valid contract
contract = FinancialContract(
contract_id="CNT-2026-BR-001",
title="Master IT Infrastructure Service Agreement",
parties=[
ContractParty(
name="Alpha Cloud Services S.A.",
tax_id="12.345.678/0001-90",
role="Service Provider",
party_type=PartyType.CORPORATION,
),
ContractParty(
name="Banco Digital do Futuro S.A.",
tax_id="98.765.432/0001-10",
role="Contracting Party",
party_type=PartyType.CORPORATION,
),
],
total_value=1250000.00,
currency=CurrencyCode.BRL,
effective_date="2026-03-01",
expiration_date="2027-02-28",
obligations=[
ContractObligation(
description="Maintain 99.98% high availability for payment processing endpoints",
penalty_rate_pct=5.0,
)
],
)
print(f"✅ Valid contract instantiated successfully: {contract.contract_id}")
print(f"💰 Total Value: {contract.currency.value} {contract.total_value:,.2f}")
print(f"👥 Counterparties: {[p.name for p in contract.parties]}")
✅ Valid contract instantiated successfully: CNT-2026-BR-001 💰 Total Value: BRL 1,250,000.00 👥 Counterparties: ['Alpha Cloud Services S.A.', 'Banco Digital do Futuro S.A.']
Testing Invariant Validation¶
Let's observe what happens when an LLM produces an impossible chronology (expiration_date < effective_date):
try:
FinancialContract(
contract_id="CNT-INVALID",
title="Broken Dates Contract",
parties=[ContractParty(name="Test Org", tax_id="11.222.333/0001-44", role="Lender")],
total_value=50000.0,
currency=CurrencyCode.USD,
effective_date="2026-12-01",
expiration_date="2026-01-01", # Chronologically invalid!
)
except ValidationError as exc:
print("🛡️ Pydantic blocked invalid contract instantiation:")
print(exc)
🛡️ Pydantic blocked invalid contract instantiation:
1 validation error for FinancialContract
Value error, Expiration date (2026-01-01) cannot be earlier than effective date (2026-12-01). [type=value_error, input_value={'contract_id': 'CNT-INVA...ion_date': '2026-01-01'}, input_type=dict]
For further information visit https://errors.pydantic.dev/2.13/v/value_error
2. Resilient Parsing: Trailing Commas & Markdown Preambles¶
Real LLMs frequently return markdown fences and trailing commas that crash json.loads. Our RobustJSONParser automatically handles these edge cases.
from projects.b_prompt_engineering.src.lab_06_structured_outputs import RobustJSONParser
dirty_llm_response = """
Certainly! Here is the extracted financial information you requested:
```json
{
"contract_id": "CNT-7788",
"title": "Managed Security Services",
"total_value": 84000.00,
"currency": "USD",
"effective_date": "2026-06-01",
"parties": [
{
"name": "CyberShield Corp",
"tax_id": "US-889900",
"role": "Security Provider",
},
],
}
```
I hope this helps! Reach out if you have further inquiries.
"""
parsed_data = RobustJSONParser.parse(dirty_llm_response)
print("🎉 Successfully parsed despite conversational preambles, fences, and trailing commas:")
print(parsed_data)
2026-09-30 21:12:20.659 | DEBUG | projects.b_prompt_engineering.src.lab_06_structured_outputs.parsers:parse:101 - Successfully parsed JSON after trailing comma repair.
🎉 Successfully parsed despite conversational preambles, fences, and trailing commas:
{'contract_id': 'CNT-7788', 'title': 'Managed Security Services', 'total_value': 84000.0, 'currency': 'USD', 'effective_date': '2026-06-01', 'parties': [{'name': 'CyberShield Corp', 'tax_id': 'US-889900', 'role': 'Security Provider'}]}
3. XML-Delimited Extraction¶
Models like Anthropic Claude or Meta Llama 3 often achieve higher syntax reliability when instructed with explicit XML tags.
from projects.b_prompt_engineering.src.lab_06_structured_outputs import XMLTagParser
xml_llm_output = """
<financialcontract>
<contract_id>CNT-XML-99</contract_id>
<title>Consulting Services Agreement</title>
<total_value>150000.0</total_value>
<currency>USD</currency>
<effective_date>2026-04-01</effective_date>
<parties>
<item>
<name>Advisory Partners LLC</name>
<tax_id>TAX-US-5544</tax_id>
<role>Consultant</role>
<party_type>corporation</party_type>
</item>
</parties>
</financialcontract>
"""
xml_parsed = XMLTagParser.parse(xml_llm_output, root_tag="financialcontract")
print("🌲 XML successfully converted to nested Python dictionary:")
print(xml_parsed)
🌲 XML successfully converted to nested Python dictionary:
{'contract_id': 'CNT-XML-99', 'title': 'Consulting Services Agreement', 'total_value': 150000.0, 'currency': 'USD', 'effective_date': '2026-04-01', 'parties': [{'name': 'Advisory Partners LLC', 'tax_id': 'TAX-US-5544', 'role': 'Consultant', 'party_type': 'corporation'}]}
4. The Self-Healing Reflection Loop¶
When an extraction attempt fails validation (e.g. missing a required field), the StructuredOutputExtractor generates a reflection turn, feeding the exact traceback back to the LLM.
import json
from projects.b_prompt_engineering.src.lab_06_structured_outputs import (
StructuredOutputExtractor,
ExtractionStrategy,
)
# Simulate an LLM that makes a mistake on attempt 1 (omits contract_id),
# but corrects itself on attempt 2 after receiving the validation error feedback.
attempt_counter = 0
def simulated_faulty_llm(messages):
global attempt_counter
attempt_counter += 1
if attempt_counter == 1:
print("🤖 LLM [Attempt 1]: Generating output with MISSING contract_id...")
return json.dumps({
"title": "Cloud Computing Agreement",
"parties": [{"name": "Tech S.A.", "tax_id": "11.222.333/0001-99", "role": "Provider"}],
"total_value": 35000.0,
"currency": "USD",
"effective_date": "2026-05-01",
})
else:
print("🤖 LLM [Attempt 2]: Self-Healing reflection triggered! Generating corrected output...")
return json.dumps({
"contract_id": "CNT-REPAIRED-2026",
"title": "Cloud Computing Agreement",
"parties": [{"name": "Tech S.A.", "tax_id": "11.222.333/0001-99", "role": "Provider"}],
"total_value": 35000.0,
"currency": "USD",
"effective_date": "2026-05-01",
})
extractor = StructuredOutputExtractor(
target_model=FinancialContract,
strategy=ExtractionStrategy.JSON_SCHEMA_STRICT,
max_retries=3,
llm_callable=simulated_faulty_llm,
)
result = extractor.extract("Extract data from this contractual agreement.")
print("\n🎯 Extraction Outcome:")
print(f"Success: {result.is_success}")
print(f"Attempts required: {result.attempts}")
print(f"Repaired Contract ID: {result.data.contract_id}")
print(f"Validation errors caught during recovery: {result.validation_errors}")
2026-09-30 21:12:20.677 | INFO | projects.b_prompt_engineering.src.lab_06_structured_outputs.extractor:extract:75 - Starting extraction for model FinancialContract using strategy 'json_schema_strict' (max_retries=3)
2026-09-30 21:12:20.678 | DEBUG | projects.b_prompt_engineering.src.lab_06_structured_outputs.extractor:extract:81 - Executing extraction attempt 1/3...
2026-09-30 21:12:20.679 | WARNING | projects.b_prompt_engineering.src.lab_06_structured_outputs.extractor:extract:116 - Attempt 1 failed: 1 validation error for FinancialContract contract_id Field required [type=missing, input_value={'title': 'Cloud Computin...ive_date': '2026-05-01'}, input_type=dict] For further information visit https://errors.pydantic.dev/2.13/v/missing
2026-09-30 21:12:20.679 | INFO | projects.b_prompt_engineering.src.lab_06_structured_outputs.extractor:extract:121 - Initiating Self-Healing reflection loop (attempt 2)...
2026-09-30 21:12:20.682 | DEBUG | projects.b_prompt_engineering.src.lab_06_structured_outputs.extractor:extract:81 - Executing extraction attempt 2/3...
2026-09-30 21:12:20.682 | INFO | projects.b_prompt_engineering.src.lab_06_structured_outputs.extractor:extract:99 - Extraction succeeded on attempt 2/3 in 0.009s.
🤖 LLM [Attempt 1]: Generating output with MISSING contract_id...
🤖 LLM [Attempt 2]: Self-Healing reflection triggered! Generating corrected output...
🎯 Extraction Outcome:
Success: True
Attempts required: 2
Repaired Contract ID: CNT-REPAIRED-2026
Validation errors caught during recovery: ["Attempt 1 failed: 1 validation error for FinancialContract\ncontract_id\n Field required [type=missing, input_value={'title': 'Cloud Computin...ive_date': '2026-05-01'}, input_type=dict]\n For further information visit https://errors.pydantic.dev/2.13/v/missing"]
5. Empirical Benchmarking¶
Let's run a comparison between strategies across various document scenarios and plot the success rates.
import matplotlib.pyplot as plt
from projects.b_prompt_engineering.src.lab_06_structured_outputs import (
StrategyBenchmark,
ExtractionStrategy,
)
benchmark = StrategyBenchmark(target_model=FinancialContract)
# Define mock response factories
clean_json = json.dumps({
"contract_id": "CNT-VALID-01",
"title": "Equipment Lease",
"parties": [{"name": "Lease Co", "tax_id": "TAX-12345", "role": "Lessor"}],
"total_value": 25000.0,
"currency": "USD",
"effective_date": "2026-01-01",
})
test_cases = ["case 1", "case 2", "case 3", "case 4"]
def raw_json_factory(case):
if case in ("case 2", "case 4"):
return lambda msgs: "```json\n{\"title\": \"missing keys\"}\n```"
return lambda msgs: clean_json
def self_healing_factory(case):
calls = {"n": 0}
def llm(msgs):
calls["n"] += 1
if calls["n"] == 1 and case == "case 2":
return "```json\n{\"title\": \"missing keys\"}\n```"
return clean_json
return llm
report_raw = benchmark.evaluate_strategy(
ExtractionStrategy.RAW_JSON,
test_cases,
raw_json_factory,
max_retries=1,
)
report_healing = benchmark.evaluate_strategy(
ExtractionStrategy.JSON_SCHEMA_STRICT,
test_cases,
self_healing_factory,
max_retries=3,
)
table_md = StrategyBenchmark.generate_markdown_table([report_raw, report_healing])
print("### Benchmark Summary Table:")
print(table_md)
# Visual comparison plot
strategies = ["Raw JSON (Zero-Shot)", "Self-Healing JSON Schema"]
success_rates = [report_raw.success_rate_pct, report_healing.success_rate_pct]
plt.figure(figsize=(7, 4))
bars = plt.bar(strategies, success_rates, color=["#e74c3c", "#2ecc71"], width=0.5)
plt.ylabel("Taxa de Sucesso (%)")
plt.title("Confiabilidade de Extração: Raw JSON vs Self-Healing Reflection")
plt.ylim(0, 110)
for bar in bars:
yval = bar.get_height()
plt.text(bar.get_x() + bar.get_width()/2.0, yval + 2, f"{yval:.1f}%", ha='center', va='bottom', fontweight='bold')
plt.grid(axis='y', linestyle='--', alpha=0.7)
plt.tight_layout()
plt.show()
2026-09-30 21:12:21.084 | INFO | projects.b_prompt_engineering.src.lab_06_structured_outputs.benchmarking:evaluate_strategy:41 - Running benchmark for strategy 'raw_json' with 4 cases...
2026-09-30 21:12:21.087 | INFO | projects.b_prompt_engineering.src.lab_06_structured_outputs.extractor:extract:75 - Starting extraction for model FinancialContract using strategy 'raw_json' (max_retries=1)
2026-09-30 21:12:21.087 | DEBUG | projects.b_prompt_engineering.src.lab_06_structured_outputs.extractor:extract:81 - Executing extraction attempt 1/1...
2026-09-30 21:12:21.088 | INFO | projects.b_prompt_engineering.src.lab_06_structured_outputs.extractor:extract:99 - Extraction succeeded on attempt 1/1 in 0.003s.
2026-09-30 21:12:21.090 | INFO | projects.b_prompt_engineering.src.lab_06_structured_outputs.extractor:extract:75 - Starting extraction for model FinancialContract using strategy 'raw_json' (max_retries=1)
2026-09-30 21:12:21.092 | DEBUG | projects.b_prompt_engineering.src.lab_06_structured_outputs.extractor:extract:81 - Executing extraction attempt 1/1...
2026-09-30 21:12:21.092 | WARNING | projects.b_prompt_engineering.src.lab_06_structured_outputs.extractor:extract:116 - Attempt 1 failed: 4 validation errors for FinancialContract contract_id Field required [type=missing, input_value={'title': 'missing keys'}, input_type=dict] For further information visit https://errors.pydantic.dev/2.13/v/missing parties Field required [type=missing, input_value={'title': 'missing keys'}, input_type=dict] For further information visit https://errors.pydantic.dev/2.13/v/missing total_value Field required [type=missing, input_value={'title': 'missing keys'}, input_type=dict] For further information visit https://errors.pydantic.dev/2.13/v/missing effective_date Field required [type=missing, input_value={'title': 'missing keys'}, input_type=dict] For further information visit https://errors.pydantic.dev/2.13/v/missing
2026-09-30 21:12:21.092 | ERROR | projects.b_prompt_engineering.src.lab_06_structured_outputs.extractor:extract:134 - Structured extraction failed after 1 attempts for FinancialContract.
2026-09-30 21:12:21.095 | INFO | projects.b_prompt_engineering.src.lab_06_structured_outputs.extractor:extract:75 - Starting extraction for model FinancialContract using strategy 'raw_json' (max_retries=1)
2026-09-30 21:12:21.095 | DEBUG | projects.b_prompt_engineering.src.lab_06_structured_outputs.extractor:extract:81 - Executing extraction attempt 1/1...
2026-09-30 21:12:21.097 | INFO | projects.b_prompt_engineering.src.lab_06_structured_outputs.extractor:extract:99 - Extraction succeeded on attempt 1/1 in 0.004s.
2026-09-30 21:12:21.099 | INFO | projects.b_prompt_engineering.src.lab_06_structured_outputs.extractor:extract:75 - Starting extraction for model FinancialContract using strategy 'raw_json' (max_retries=1)
2026-09-30 21:12:21.100 | DEBUG | projects.b_prompt_engineering.src.lab_06_structured_outputs.extractor:extract:81 - Executing extraction attempt 1/1...
2026-09-30 21:12:21.100 | WARNING | projects.b_prompt_engineering.src.lab_06_structured_outputs.extractor:extract:116 - Attempt 1 failed: 4 validation errors for FinancialContract contract_id Field required [type=missing, input_value={'title': 'missing keys'}, input_type=dict] For further information visit https://errors.pydantic.dev/2.13/v/missing parties Field required [type=missing, input_value={'title': 'missing keys'}, input_type=dict] For further information visit https://errors.pydantic.dev/2.13/v/missing total_value Field required [type=missing, input_value={'title': 'missing keys'}, input_type=dict] For further information visit https://errors.pydantic.dev/2.13/v/missing effective_date Field required [type=missing, input_value={'title': 'missing keys'}, input_type=dict] For further information visit https://errors.pydantic.dev/2.13/v/missing
2026-09-30 21:12:21.101 | ERROR | projects.b_prompt_engineering.src.lab_06_structured_outputs.extractor:extract:134 - Structured extraction failed after 1 attempts for FinancialContract.
2026-09-30 21:12:21.101 | INFO | projects.b_prompt_engineering.src.lab_06_structured_outputs.benchmarking:evaluate_strategy:41 - Running benchmark for strategy 'json_schema_strict' with 4 cases...
2026-09-30 21:12:21.104 | INFO | projects.b_prompt_engineering.src.lab_06_structured_outputs.extractor:extract:75 - Starting extraction for model FinancialContract using strategy 'json_schema_strict' (max_retries=3)
2026-09-30 21:12:21.104 | DEBUG | projects.b_prompt_engineering.src.lab_06_structured_outputs.extractor:extract:81 - Executing extraction attempt 1/3...
2026-09-30 21:12:21.105 | INFO | projects.b_prompt_engineering.src.lab_06_structured_outputs.extractor:extract:99 - Extraction succeeded on attempt 1/3 in 0.003s.
2026-09-30 21:12:21.107 | INFO | projects.b_prompt_engineering.src.lab_06_structured_outputs.extractor:extract:75 - Starting extraction for model FinancialContract using strategy 'json_schema_strict' (max_retries=3)
2026-09-30 21:12:21.108 | DEBUG | projects.b_prompt_engineering.src.lab_06_structured_outputs.extractor:extract:81 - Executing extraction attempt 1/3...
2026-09-30 21:12:21.108 | WARNING | projects.b_prompt_engineering.src.lab_06_structured_outputs.extractor:extract:116 - Attempt 1 failed: 4 validation errors for FinancialContract contract_id Field required [type=missing, input_value={'title': 'missing keys'}, input_type=dict] For further information visit https://errors.pydantic.dev/2.13/v/missing parties Field required [type=missing, input_value={'title': 'missing keys'}, input_type=dict] For further information visit https://errors.pydantic.dev/2.13/v/missing total_value Field required [type=missing, input_value={'title': 'missing keys'}, input_type=dict] For further information visit https://errors.pydantic.dev/2.13/v/missing effective_date Field required [type=missing, input_value={'title': 'missing keys'}, input_type=dict] For further information visit https://errors.pydantic.dev/2.13/v/missing
2026-09-30 21:12:21.108 | INFO | projects.b_prompt_engineering.src.lab_06_structured_outputs.extractor:extract:121 - Initiating Self-Healing reflection loop (attempt 2)...
2026-09-30 21:12:21.111 | DEBUG | projects.b_prompt_engineering.src.lab_06_structured_outputs.extractor:extract:81 - Executing extraction attempt 2/3...
2026-09-30 21:12:21.112 | INFO | projects.b_prompt_engineering.src.lab_06_structured_outputs.extractor:extract:99 - Extraction succeeded on attempt 2/3 in 0.007s.
2026-09-30 21:12:21.114 | INFO | projects.b_prompt_engineering.src.lab_06_structured_outputs.extractor:extract:75 - Starting extraction for model FinancialContract using strategy 'json_schema_strict' (max_retries=3)
2026-09-30 21:12:21.115 | DEBUG | projects.b_prompt_engineering.src.lab_06_structured_outputs.extractor:extract:81 - Executing extraction attempt 1/3...
2026-09-30 21:12:21.115 | INFO | projects.b_prompt_engineering.src.lab_06_structured_outputs.extractor:extract:99 - Extraction succeeded on attempt 1/3 in 0.003s.
2026-09-30 21:12:21.119 | INFO | projects.b_prompt_engineering.src.lab_06_structured_outputs.extractor:extract:75 - Starting extraction for model FinancialContract using strategy 'json_schema_strict' (max_retries=3)
2026-09-30 21:12:21.120 | DEBUG | projects.b_prompt_engineering.src.lab_06_structured_outputs.extractor:extract:81 - Executing extraction attempt 1/3...
2026-09-30 21:12:21.120 | INFO | projects.b_prompt_engineering.src.lab_06_structured_outputs.extractor:extract:99 - Extraction succeeded on attempt 1/3 in 0.004s.
### Benchmark Summary Table: | Estratégia de Extração | Taxa de Sucesso (%) | Méd. Tentativas | Erros Sintáticos | Erros de Schema | Latência Média (s) | | :--- | :---: | :---: | :---: | :---: | :---: | | **raw_json** | 50.0% | 1.0 | 0 | 2 | 0.0038s | | **json_schema_strict** | 100.0% | 1.25 | 0 | 0 | 0.0043s |