5. Extracción Estructurada

Descripción

La extracción estructurada convierte documentos (facturas, recibos, contratos, identificaciones) en datos validados con schemas. En esta cápsula aprenderás a usar Pydantic para definir schemas tipados, diseñar prompts que devuelvan JSON confiable, validar la salida del LLM y manejar errores de forma robusta. Es la base para automatizar entrada de datos desde documentos reales.

Por qué importa: Sin schemas, la salida del LLM es texto libre que requiere parsing manual. Con Pydantic obtienes datos tipados, validados y listos para bases de datos o APIs. La diferencia entre un prototipo y un sistema en producción es la validación.

Conexión con el módulo: En la cápsula 02 aprendiste a procesar PDFs. En la cápsula 03 viste OCR vs Vision APIs. En la cápsula 04 preparaste imágenes de documentos para Vision. Aquí cierras el ciclo: tomas la imagen, defines qué datos necesitas con un schema, y obtienes datos validados.

Conceptos Clave

Schema-first extraction

1. Definir el schema (Pydantic model con tipos, validadores, campos opcionales)
2. Incluir el schema en el prompt como contrato
3. Usar response_format={"type": "json_object"} (OpenAI) o instrucciones explícitas (Anthropic)
4. Parsear la respuesta JSON y validar con Pydantic
5. Manejar errores con reintentos inteligentes

Por qué Pydantic y no diccionarios

Schemas con Pydantic

Schema: Factura

from pydantic import BaseModel, Field, field_validator, model_validator
from typing import Optional

class ItemFactura(BaseModel):
    descripcion: str = Field(..., description="Nombre del producto o servicio")
    cantidad: float = Field(default=1.0, ge=0)
    precio_unitario: float = Field(..., ge=0)
    monto: float = Field(..., ge=0)

    @model_validator(mode="after")
    def verificar_monto(self):
        esperado = round(self.cantidad * self.precio_unitario, 2)
        if abs(self.monto - esperado) > 0.5:
            self.monto = esperado
        return self

class Factura(BaseModel):
    fecha: Optional[str] = Field(None, description="YYYY-MM-DD")
    numero: Optional[str] = Field(None, description="Número o folio")
    proveedor: Optional[str] = None
    cliente: Optional[str] = None
    total: Optional[float] = Field(None, ge=0)
    subtotal: Optional[float] = Field(None, ge=0)
    impuestos: Optional[float] = Field(None, ge=0)
    moneda: str = Field(default="MXN")
    items: list[ItemFactura] = Field(default_factory=list)

    @field_validator("moneda")
    @classmethod
    def moneda_valida(cls, v: str) -> str:
        v = v.upper().strip()
        if v not in {"MXN", "USD", "EUR", "COP", "ARS", "CLP", "PEN"}:
            return "MXN"
        return v

    @model_validator(mode="after")
    def calcular_total(self):
        if self.total is None and self.items:
            self.total = round(sum(i.monto for i in self.items), 2)
        return self

Schema: Recibo

class Recibo(BaseModel):
    comercio: Optional[str] = Field(None, description="Nombre del establecimiento")
    direccion: Optional[str] = None
    fecha: Optional[str] = None
    hora: Optional[str] = None
    items: list[dict] = Field(default_factory=list)
    subtotal: Optional[float] = None
    impuesto: Optional[float] = None
    total: Optional[float] = None
    metodo_pago: Optional[str] = None

    @field_validator("total")
    @classmethod
    def total_positivo(cls, v):
        if v is not None and v < 0:
            raise ValueError("total no puede ser negativo")
        return v

Schema: Contrato

class ParteContrato(BaseModel):
    nombre: str
    rol: Optional[str] = Field(None, description="Ej: arrendador, proveedor")
    identificacion: Optional[str] = None

class Contrato(BaseModel):
    tipo: Optional[str] = Field(None, description="arrendamiento, servicios, compraventa")
    partes: list[ParteContrato] = Field(default_factory=list)
    fecha_firma: Optional[str] = None
    fecha_inicio: Optional[str] = None
    fecha_fin: Optional[str] = None
    objeto: Optional[str] = None
    monto: Optional[float] = None
    moneda: str = "MXN"
    clausulas_importantes: list[str] = Field(default_factory=list)

Schema: Identificación

class Identificacion(BaseModel):
    tipo_documento: Optional[str] = Field(None, description="INE, pasaporte, licencia")
    nombre_completo: Optional[str] = None
    fecha_nacimiento: Optional[str] = None
    numero_documento: Optional[str] = None
    fecha_expedicion: Optional[str] = None
    fecha_vencimiento: Optional[str] = None
    nacionalidad: Optional[str] = None
    sexo: Optional[str] = None

    @field_validator("sexo")
    @classmethod
    def normalizar_sexo(cls, v):
        if v is None:
            return v
        mapping = {"M": "M", "F": "F", "MASCULINO": "M", "FEMENINO": "F", "H": "M", "MUJER": "F"}
        return mapping.get(v.upper().strip(), v)

Prompt Engineering para Extracción

Generar prompt desde schema Pydantic

def schema_to_prompt(model: type[BaseModel]) -> str:
    """Genera descripción de campos para el prompt desde el schema Pydantic."""
    schema = model.model_json_schema()
    props = schema.get("properties", {})
    required = schema.get("required", [])
    lines = []
    for name, info in props.items():
        t = info.get("type", "string")
        desc = info.get("description", "")
        req = "(obligatorio)" if name in required else "(opcional)"
        line = f"- {name}: {t} {req}"
        if desc:
            line += f" — {desc}"
        lines.append(line)
    return "\n".join(lines)

Prompt completo con few-shot

def build_extraction_prompt(model: type[BaseModel], few_shot: bool = True) -> str:
    """Construye prompt de extracción con schema y ejemplo."""
    schema_desc = schema_to_prompt(model)
    prompt = f"""Eres un sistema de extracción de datos de documentos.
Analiza la imagen y extrae los datos en formato JSON.

## Campos a extraer:
{schema_desc}

## Reglas:
- Responde ÚNICAMENTE con JSON válido, sin texto adicional.
- Usa null para campos no encontrados. Fechas en YYYY-MM-DD. Montos como números.
- No inventes datos. Si no está visible, usa null."""

    if few_shot:
        prompt += """

## Ejemplo de salida:
{
    "fecha": "2024-03-15", "numero": "FAC-001234",
    "proveedor": "Servicios Tech S.A.", "total": 15680.50, "moneda": "MXN",
    "items": [{"descripcion": "Consultoría", "cantidad": 10, "precio_unitario": 1500.00, "monto": 15000.00}]
}"""
    return prompt

Extracción con parámetros críticos

from openai import OpenAI
import base64
import json

client = OpenAI()

def extract_with_schema(image_path: str, model_class: type[BaseModel]) -> BaseModel:
    """Extrae datos estructurados de una imagen usando un schema Pydantic."""
    with open(image_path, "rb") as f:
        b64 = base64.b64encode(f.read()).decode()

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{
            "role": "user",
            "content": [
                {"type": "text", "text": build_extraction_prompt(model_class)},
                {"type": "image_url", "image_url": {
                    "url": f"data:image/jpeg;base64,{b64}", "detail": "high"
                }}
            ]
        }],
        response_format={"type": "json_object"},
        temperature=0,
        max_tokens=4000
    )

    data = json.loads(response.choices[0].message.content)
    return model_class(**data)

Validación con Pydantic

Pre-procesamiento de respuesta

from pydantic import ValidationError

def clean_llm_data(data: dict) -> dict:
    """Limpia valores problemáticos comunes en respuestas de LLM."""
    cleaned = {}
    for key, value in data.items():
        if isinstance(value, str) and value.strip().lower() in ("n/a", "n/d", "-", "none", "null", ""):
            cleaned[key] = None
        elif isinstance(value, list):
            cleaned[key] = [clean_llm_data(item) if isinstance(item, dict) else item for item in value]
        elif isinstance(value, dict):
            cleaned[key] = clean_llm_data(value)
        else:
            cleaned[key] = value
    return cleaned

def parse_llm_response(raw_json: str, model_class: type[BaseModel]) -> BaseModel | None:
    """Parsea respuesta JSON del LLM con limpieza y validación."""
    try:
        data = json.loads(raw_json)
        data = clean_llm_data(data)
        return model_class(**data)
    except json.JSONDecodeError as e:
        print(f"JSON inválido: {e}")
        return None
    except ValidationError as e:
        print(f"Validación fallida: {e.error_count()} errores")
        for error in e.errors():
            campo = " → ".join(str(x) for x in error["loc"])
            print(f"  {campo}: {error['msg']}")
        return None

Reintentos con contexto de error

Si la validación falla, incluye el error en el segundo intento para que el LLM se corrija:

def extract_with_retry(
    image_path: str,
    model_class: type[BaseModel],
    max_retries: int = 2
) -> BaseModel | None:
    """Extrae datos con reintentos que incluyen errores previos."""
    with open(image_path, "rb") as f:
        b64 = base64.b64encode(f.read()).decode()

    base_prompt = build_extraction_prompt(model_class)
    messages = [{
        "role": "user",
        "content": [
            {"type": "text", "text": base_prompt},
            {"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{b64}"}}
        ]
    }]

    for attempt in range(max_retries + 1):
        response = client.chat.completions.create(
            model="gpt-4o",
            messages=messages,
            response_format={"type": "json_object"},
            temperature=0,
            max_tokens=4000
        )
        raw = response.choices[0].message.content

        try:
            data = clean_llm_data(json.loads(raw))
            return model_class(**data)
        except (json.JSONDecodeError, ValidationError) as e:
            if attempt < max_retries:
                messages.append({"role": "assistant", "content": raw})
                messages.append({
                    "role": "user",
                    "content": f"JSON con errores:\n{e}\n\nCorrige y responde con JSON válido."
                })
            else:
                print(f"Fallo tras {max_retries + 1} intentos: {e}")
                return None

Extracción Multi-Schema

DOCUMENT_SCHEMAS = {
    "factura": Factura,
    "recibo": Recibo,
    "contrato": Contrato,
    "identificacion": Identificacion,
}

def classify_document(image_path: str) -> str:
    """Clasifica el tipo de documento usando Vision."""
    with open(image_path, "rb") as f:
        b64 = base64.b64encode(f.read()).decode()

    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{
            "role": "user",
            "content": [
                {"type": "text", "text": (
                    "Clasifica este documento: factura, recibo, contrato, identificacion.\n"
                    "Responde JSON: {\"tipo\": \"...\"}"
                )},
                {"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{b64}"}}
            ]
        }],
        response_format={"type": "json_object"},
        temperature=0,
        max_tokens=50
    )
    return json.loads(response.choices[0].message.content).get("tipo", "factura")

Clasificar y extraer en un solo request

Para reducir latencia y costo, puedes clasificar y extraer en un solo call. La clave es incluir los campos de todos los schemas en el prompt y pedir al modelo que seleccione el tipo:

def classify_and_extract(image_path: str) -> tuple[str, dict]:
    """Clasifica tipo de documento y extrae datos en un solo request."""
    with open(image_path, "rb") as f:
        b64 = base64.b64encode(f.read()).decode()

    all_fields = {name: list(m.model_json_schema()["properties"].keys())
                  for name, m in DOCUMENT_SCHEMAS.items()}

    prompt = (f"Analiza este documento. Determina tipo (factura/recibo/contrato/identificacion) "
              f"y extrae campos.\nCampos por tipo: {json.dumps(all_fields, ensure_ascii=False)}\n"
              f"Responde JSON: {{\"tipo\": \"...\", \"datos\": {{...}}}}. Usa null si no encuentras un campo.")

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": [
            {"type": "text", "text": prompt},
            {"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{b64}"}}
        ]}],
        response_format={"type": "json_object"}, temperature=0
    )

    result = json.loads(response.choices[0].message.content)
    doc_type = result.get("tipo", "factura")
    datos = clean_llm_data(result.get("datos", {}))
    schema_class = DOCUMENT_SCHEMAS.get(doc_type)
    if schema_class:
        return doc_type, schema_class(**datos).model_dump()
    return doc_type, datos

OpenAI JSON Mode vs Prompt Engineering

Anthropic: sin JSON mode nativo

import anthropic
import re

def extract_with_anthropic(image_b64: str, prompt: str) -> dict:
    """Extrae JSON de Anthropic con regex de fallback."""
    client_anthropic = anthropic.Anthropic()

    response = client_anthropic.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=4000,
        messages=[{
            "role": "user",
            "content": [
                {"type": "image", "source": {"type": "base64", "media_type": "image/jpeg", "data": image_b64}},
                {"type": "text", "text": prompt + "\n\nResponde ÚNICAMENTE con JSON válido. Sin ```json. Solo el JSON."}
            ]
        }]
    )

    text = response.content[0].text.strip()
    text = re.sub(r"^```(?:json)?\s*", "", text)
    text = re.sub(r"\s*```$", "", text)
    match = re.search(r"\{.*\}", text, re.DOTALL)
    return json.loads(match.group()) if match else json.loads(text)

Recomendación: OpenAI → response_format + temperature=0. Anthropic → prompt explícito + regex. En ambos, Pydantic valida el schema.

Pipeline: Imagen → Schema → Datos Validados

from dataclasses import dataclass, field as dc_field
from enum import Enum

class ExtractionStatus(str, Enum):
    SUCCESS = "success"
    PARTIAL = "partial"
    FAILED = "failed"

@dataclass
class ExtractionResult:
    status: ExtractionStatus
    doc_type: str
    data: dict = dc_field(default_factory=dict)
    errors: list[str] = dc_field(default_factory=list)
    attempts: int = 0

def full_extraction_pipeline(
    image_path: str, force_type: str | None = None, max_retries: int = 2
) -> ExtractionResult:
    """Pipeline completo: imagen → clasificación → extracción → validación."""
    result = ExtractionResult(status=ExtractionStatus.FAILED, doc_type="unknown")
    doc_type = force_type if force_type in DOCUMENT_SCHEMAS else classify_document(image_path)
    result.doc_type = doc_type

    schema_class = DOCUMENT_SCHEMAS.get(doc_type)
    if not schema_class:
        result.errors.append(f"Sin schema para tipo: {doc_type}")
        return result

    extracted = extract_with_retry(image_path, schema_class, max_retries)
    result.attempts = max_retries + 1
    if extracted is None:
        result.errors.append("Extracción fallida tras todos los reintentos")
        return result

    data_dict = extracted.model_dump()
    non_null = sum(1 for v in data_dict.values() if v is not None and v != [] and v != "")
    completitud = non_null / len(data_dict) if data_dict else 0
    result.data = data_dict
    result.status = ExtractionStatus.SUCCESS if completitud >= 0.5 else ExtractionStatus.PARTIAL
    if completitud < 0.5:
        result.errors.append(f"Solo {completitud:.0%} de campos extraídos")
    return result

# Uso
result = full_extraction_pipeline("factura_escaneada.jpg")
print(f"Estado: {result.status.value} | Tipo: {result.doc_type}")
print(json.dumps(result.data, indent=2, ensure_ascii=False, default=str))

Campos Difíciles

Fechas en múltiples formatos

from datetime import datetime

MESES_ES = {"enero": "01", "febrero": "02", "marzo": "03", "abril": "04", "mayo": "05",
            "junio": "06", "julio": "07", "agosto": "08", "septiembre": "09",
            "octubre": "10", "noviembre": "11", "diciembre": "12"}

def normalize_date(raw: str | None) -> str | None:
    """Normaliza fechas a YYYY-MM-DD desde formatos latino, anglosajón y texto español."""
    if not raw:
        return None
    text = raw.lower().strip()
    for nombre, num in MESES_ES.items():
        if nombre in text:
            text = re.sub(rf"\b{nombre}\b", num, text)
            text = re.sub(r"\bde\b", "", text).strip()
            text = re.sub(r"\s+", "/", text)
            try:
                return datetime.strptime(text, "%d/%m/%Y").strftime("%Y-%m-%d")
            except ValueError:
                pass
    for fmt in ["%Y-%m-%d", "%d/%m/%Y", "%d-%m-%Y", "%m/%d/%Y", "%d.%m.%Y"]:
        try:
            return datetime.strptime(raw.strip(), fmt).strftime("%Y-%m-%d")
        except ValueError:
            continue
    return raw

Monedas y montos

def normalize_amount(raw: str | float | None) -> float | None:
    """Convierte strings de montos a float. Maneja formato latino (1.234,56) y anglosajón (1,234.56)."""
    if raw is None:
        return None
    if isinstance(raw, (int, float)):
        return float(raw)
    cleaned = re.sub(r"[^\d.,\-]", "", str(raw))
    if re.match(r"^\d{1,3}(\.\d{3})+(,\d{2})?$", cleaned):  # Latino
        cleaned = cleaned.replace(".", "").replace(",", ".")
    elif re.match(r"^\d{1,3}(,\d{3})+(\.\d{2})?$", cleaned):  # Anglosajón
        cleaned = cleaned.replace(",", "")
    elif "," in cleaned and "." not in cleaned:
        cleaned = cleaned.replace(",", ".")
    try:
        return float(cleaned)
    except ValueError:
        return None

Items anidados (líneas de factura)

El campo más problemático — los LLMs devuelven items como strings, o con claves en español/inglés:

def normalize_items(raw_items: list) -> list[dict]:
    """Normaliza items que pueden venir como strings o dicts con claves variables."""
    normalized = []
    for item in raw_items:
        if isinstance(item, str):
            normalized.append({"descripcion": item, "cantidad": 1, "precio_unitario": 0, "monto": 0})
        elif isinstance(item, dict):
            normalized.append({
                "descripcion": item.get("descripcion", item.get("nombre", "")),
                "cantidad": float(item.get("cantidad", item.get("qty", 1))),
                "precio_unitario": float(item.get("precio_unitario", item.get("precio", 0))),
                "monto": float(item.get("monto", item.get("total", 0)))
            })
    return normalized

Post-procesador completo

def postprocess_extraction(data: dict) -> dict:
    """Aplica todas las normalizaciones a los datos extraídos."""
    for key in ["fecha", "fecha_firma", "fecha_inicio", "fecha_fin",
                "fecha_nacimiento", "fecha_expedicion", "fecha_vencimiento"]:
        if key in data:
            data[key] = normalize_date(data[key])
    for key in ["total", "subtotal", "impuestos", "impuesto", "monto"]:
        if key in data and isinstance(data[key], str):
            data[key] = normalize_amount(data[key])
    if "items" in data and isinstance(data["items"], list):
        data["items"] = normalize_items(data["items"])
    return data

Troubleshooting

Problema 1: JSON malformado en respuesta

Causa: El modelo incluye markdown (```json) o texto extra, especialmente sin response_format.

Solución: Extraer JSON con regex:

def extract_json_from_text(text: str) -> dict:
    text = re.sub(r"^```(?:json)?\s*", "", text.strip())
    text = re.sub(r"\s*```$", "", text)
    match = re.search(r"\{.*\}", text, re.DOTALL)
    if match:
        return json.loads(match.group())
    raise json.JSONDecodeError("No se encontró JSON", text, 0)

Problema 2: Campos faltantes que sí existen en el documento

Causa: Prompt no especifica dónde buscar los campos, o el layout es complejo.

Solución: Incluir ubicación esperada: "El número de factura suele estar en la esquina superior derecha. El total está al final del documento."

Problema 3: Tipos incorrectos ("N/A" en lugar de null)

Causa: El modelo usa "N/A", "No disponible", "-" como strings.

Solución: Usar clean_llm_data() antes de Pydantic. Agregar en el prompt: "Usa null (no 'N/A', no '-') para campos no encontrados."

Problema 4: Valores alucinados

Causa: El LLM inventa datos para campos que no existen en el documento.

Solución: Validación cruzada (total vs suma de items). En el prompt: "Si un campo NO está visible, usa null. NUNCA inventes valores." Agregar model_validator que detecte inconsistencias.

Problema 5: Formato inconsistente entre documentos

Causa: Facturas de distintos proveedores usan nombres de campo diferentes.

Solución: model_validator(mode="before") con alias mapping:

class FacturaFlexible(Factura):
    @model_validator(mode="before")
    @classmethod
    def mapear_aliases(cls, data):
        alias_map = {
            "invoice_number": "numero", "total_amount": "total",
            "vendor": "proveedor", "currency": "moneda",
            "date": "fecha", "line_items": "items",
        }
        return {alias_map.get(k, k): v for k, v in data.items()}

Ejercicios

Ejercicio 1: Schema con validación cruzada

Crea un schema NotaDeVenta con: folio, fecha, items (lista), subtotal, descuento, total. Agrega un model_validator que verifique que total == subtotal - descuento (tolerancia $1).

class ItemVenta(BaseModel):
    producto: str
    cantidad: float = 1.0
    precio: float
    importe: float

class NotaDeVenta(BaseModel):
    folio: Optional[str] = None
    fecha: Optional[str] = None
    items: list[ItemVenta] = Field(default_factory=list)
    subtotal: Optional[float] = None
    descuento: Optional[float] = Field(default=0.0)
    total: Optional[float] = None
    _warning: str = ""

    @model_validator(mode="after")
    def verificar_totales(self):
        if self.subtotal is not None and self.total is not None:
            esperado = self.subtotal - (self.descuento or 0)
            if abs(self.total - esperado) > 1.0:
                self._warning = f"Total ({self.total}) ≠ subtotal - descuento ({esperado})"
        if self.subtotal is None and self.items:
            self.subtotal = round(sum(i.importe for i in self.items), 2)
        return self

Ejercicio 2: Extractor multi-proveedor

Crea extract_universal(image_path, schema_class, provider) que soporte "openai" (con response_format) y "anthropic" (con regex parsing). Ambos deben validar con Pydantic.

def extract_universal(
    image_path: str, schema_class: type[BaseModel], provider: str = "openai"
) -> BaseModel | None:
    with open(image_path, "rb") as f:
        b64 = base64.b64encode(f.read()).decode()

    prompt = build_extraction_prompt(schema_class)

    if provider == "openai":
        response = client.chat.completions.create(
            model="gpt-4o",
            messages=[{"role": "user", "content": [
                {"type": "text", "text": prompt},
                {"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{b64}"}}
            ]}],
            response_format={"type": "json_object"},
            temperature=0
        )
        data = json.loads(response.choices[0].message.content)
    elif provider == "anthropic":
        data = extract_with_anthropic(b64, prompt)
    else:
        raise ValueError(f"Proveedor no soportado: {provider}")

    return schema_class(**clean_llm_data(data))

Ejercicio 3: Batch de documentos mixtos

Crea una función que reciba una lista de rutas de imágenes mezcladas, clasifique cada una, extraiga con el schema correcto, y retorne un reporte con totales por tipo y estado.

def process_mixed_batch(image_paths: list[str]) -> dict:
    results = {"total": len(image_paths), "exitosos": 0, "fallidos": 0, "por_tipo": {}}
    for path in image_paths:
        try:
            ext = full_extraction_pipeline(path)
            ok = ext.status != ExtractionStatus.FAILED
            results["exitosos" if ok else "fallidos"] += 1
            tipo = ext.doc_type
            results["por_tipo"].setdefault(tipo, {"exitosos": 0, "fallidos": 0})
            results["por_tipo"][tipo]["exitosos" if ok else "fallidos"] += 1
        except Exception:
            results["fallidos"] += 1
    return results

Ejercicio 4: Normalización de fechas con tests

Implementa safe_extract_date que reciba un string crudo y devuelva un date de Python o None. Debe manejar al menos 5 formatos. Incluye asserts.

from datetime import date as date_type

def safe_extract_date(raw: str | None) -> date_type | None:
    if not raw or not isinstance(raw, str):
        return None
    text = raw.lower().strip()
    for nombre, num in MESES_ES.items():
        if nombre in text:
            text = re.sub(rf"\b{nombre}\b", num, text)
            text = re.sub(r"\bde\b", "", text).strip()
            text = re.sub(r"\s+", "/", text)
            try:
                return datetime.strptime(text, "%d/%m/%Y").date()
            except ValueError:
                pass
    for fmt in ["%Y-%m-%d", "%d/%m/%Y", "%d-%m-%Y", "%m/%d/%Y", "%d.%m.%Y"]:
        try:
            return datetime.strptime(raw.strip(), fmt).date()
        except ValueError:
            continue
    return None

assert safe_extract_date("2024-03-15") == date_type(2024, 3, 15)
assert safe_extract_date("15/03/2024") == date_type(2024, 3, 15)
assert safe_extract_date("15-03-2024") == date_type(2024, 3, 15)
assert safe_extract_date("15 de marzo de 2024") == date_type(2024, 3, 15)
assert safe_extract_date("15.03.2024") == date_type(2024, 3, 15)
assert safe_extract_date(None) is None
print("Todos los tests pasaron.")

Recursos adicionales

1. Pydantic v2 Documentation
2. OpenAI JSON Mode / Structured Outputs
3. Anthropic Vision — Extracting structured data
4. Pydantic model_validator