3. Schemas con Pydantic y Function Calling

Descripción

Pydantic permite definir schemas de datos que validan, transforman y documentan el output del LLM. Function calling (OpenAI) y Tool Use (Anthropic) usan schemas JSON para obtener structured output garantizado — el modelo no puede responder con texto libre, solo con los campos especificados. En esta cápsula aprenderás a combinar Pydantic con function calling, trabajar con estructuras anidadas, optional fields, enums, y validators personalizados.

Por qué importa: Function calling es la forma más robusta de structured output disponible. No es "el modelo puede que devuelva JSON" — es "el modelo DEBE llamar la función con estos argumentos exactos". Para sistemas de producción que procesan datos críticos, la diferencia importa.

Pydantic: Más que Validación

Pydantic v2 hace tres cosas fundamentales para este flujo:

1. Definir el contrato: Cómo deben ser los datos
2. Validar el output: El LLM devolvió lo esperado
3. Generar el schema: Convertir el modelo Pydantic a JSON Schema para function calling

from pydantic import BaseModel, Field, field_validator
from typing import Literal, Optional
from datetime import date

class SentimentAnalysis(BaseModel):
    """Análisis de sentimiento de un texto."""
    
    sentimiento: Literal["POSITIVO", "NEGATIVO", "NEUTRO"]
    confianza: float = Field(ge=0.0, le=1.0, description="Nivel de confianza entre 0 y 1")
    palabras_clave: list[str] = Field(max_length=5, description="Máximo 5 palabras que justifican el sentimiento")
    requiere_atencion: bool = Field(description="True si el sentimiento negativo es intenso y requiere seguimiento")

# Generar JSON Schema (para usar en function calling)
schema = SentimentAnalysis.model_json_schema()
print(schema)
# {
#   "type": "object",
#   "properties": {
#     "sentimiento": {"enum": ["POSITIVO", "NEGATIVO", "NEUTRO"]},
#     "confianza": {"type": "number", "minimum": 0.0, "maximum": 1.0},
#     "palabras_clave": {"type": "array", "maxItems": 5},
#     "requiere_atencion": {"type": "boolean"}
#   },
#   "required": ["sentimiento", "confianza", "palabras_clave", "requiere_atencion"]
# }

Function Calling como Structured Output (OpenAI)

Concepto

Function calling no es solo para llamar funciones reales. Es el mecanismo más robusto de structured output de OpenAI: el modelo DEBE responder con los argumentos de la función en el formato JSON especificado.

from openai import OpenAI
import json

client = OpenAI()

# Definir el "tool" con el schema de la función
tools = [
    {
        "type": "function",
        "function": {
            "name": "analizar_sentimiento",
            "description": "Analiza el sentimiento de un texto de reseña de producto",
            "parameters": {
                "type": "object",
                "properties": {
                    "sentimiento": {
                        "type": "string",
                        "enum": ["POSITIVO", "NEGATIVO", "NEUTRO"],
                        "description": "Clasificación del sentimiento"
                    },
                    "confianza": {
                        "type": "number",
                        "minimum": 0.0,
                        "maximum": 1.0,
                        "description": "Nivel de confianza entre 0 y 1"
                    },
                    "palabras_clave": {
                        "type": "array",
                        "items": {"type": "string"},
                        "maxItems": 5,
                        "description": "Palabras que justifican el sentimiento"
                    },
                    "requiere_atencion": {
                        "type": "boolean",
                        "description": "True si es sentimiento muy negativo y requiere seguimiento"
                    }
                },
                "required": ["sentimiento", "confianza", "palabras_clave", "requiere_atencion"]
            }
        }
    }
]

def analizar(texto: str) -> SentimentAnalysis:
    """Analiza sentimiento usando function calling."""
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {
                "role": "system",
                "content": "Analiza el sentimiento de reseñas de productos con precisión."
            },
            {
                "role": "user",
                "content": f"Analiza: '{texto}'"
            }
        ],
        tools=tools,
        tool_choice={
            "type": "function",
            "function": {"name": "analizar_sentimiento"}
        },  # Forzar que use esta función específica
        temperature=0
    )
    
    # Extraer los argumentos del tool call
    tool_call = response.choices[0].message.tool_calls[0]
    args = json.loads(tool_call.function.arguments)
    
    # Validar con Pydantic
    return SentimentAnalysis(**args)

# Test
textos = [
    "Llegó antes de lo esperado y funciona perfectamente. Muy recomendado.",
    "HORRIBLE. Se rompió al tercer día. Me estafaron.",
    "Cumple su función. No es lo mejor ni lo peor."
]

for texto in textos:
    resultado = analizar(texto)
    print(f"Input: {texto[:50]}...")
    print(f"  Sentimiento: {resultado.sentimiento} ({resultado.confianza:.0%})")
    print(f"  Keywords: {resultado.palabras_clave}")
    print(f"  Requiere atención: {resultado.requiere_atencion}\n")

Generar Schema Pydantic → JSON Schema para Function Calling

No tienes que escribir el JSON Schema manualmente. Pydantic lo genera:

from pydantic import BaseModel, Field
from typing import Optional, Literal, List
from openai import OpenAI
import json

client = OpenAI()

# Definir el modelo Pydantic
class InvoiceItem(BaseModel):
    descripcion: str = Field(description="Descripción del item")
    cantidad: int = Field(ge=1, description="Cantidad de unidades")
    precio_unitario: float = Field(ge=0, description="Precio por unidad")
    subtotal: Optional[float] = Field(None, description="Subtotal (cantidad * precio)")

class Invoice(BaseModel):
    proveedor: str = Field(description="Nombre del proveedor o empresa")
    numero_factura: str = Field(description="Número o código de la factura")
    fecha_emision: str = Field(description="Fecha en formato YYYY-MM-DD")
    fecha_vencimiento: Optional[str] = Field(None, description="Fecha de vencimiento si aplica")
    subtotal: float = Field(ge=0, description="Subtotal antes de impuestos")
    impuestos: float = Field(ge=0, description="Monto de impuestos")
    total: float = Field(ge=0, description="Total incluyendo impuestos")
    moneda: Literal["USD", "EUR", "MXN", "GBP"] = Field(description="Código de moneda")
    items: List[InvoiceItem] = Field(description="Lista de items de la factura")

# Generar el JSON Schema automáticamente
schema = Invoice.model_json_schema()

# Usar en function calling
def extraer_invoice(texto_factura: str) -> Invoice:
    """Extrae datos estructurados de una factura."""
    
    tools = [
        {
            "type": "function",
            "function": {
                "name": "extraer_datos_factura",
                "description": "Extrae todos los datos estructurados de una factura",
                "parameters": schema  # Schema generado por Pydantic!
            }
        }
    ]
    
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {
                "role": "system",
                "content": "Eres un extractor de datos de facturas. Extrae todos los campos disponibles con precisión."
            },
            {
                "role": "user",
                "content": f"Extrae los datos de esta factura:\n\n{texto_factura}"
            }
        ],
        tools=tools,
        tool_choice={"type": "function", "function": {"name": "extraer_datos_factura"}},
        temperature=0
    )
    
    args = json.loads(response.choices[0].message.tool_calls[0].function.arguments)
    return Invoice(**args)

# Test con texto de factura
factura_texto = """
FACTURA #INV-2025-001
TechSupplies S.A. de C.V.
Fecha: 15 de enero de 2025
Vencimiento: 15 de febrero de 2025

Servicios de consultoría IT: 40 horas x $150/hr = $6,000.00
Licencias de software: 5 unidades x $200/ud = $1,000.00

Subtotal: $7,000.00
IVA (16%): $1,120.00
TOTAL: $8,120.00 MXN
"""

resultado = extraer_invoice(factura_texto)
print(f"Proveedor: {resultado.proveedor}")
print(f"Número: {resultado.numero_factura}")
print(f"Total: {resultado.total} {resultado.moneda}")
print(f"Items: {len(resultado.items)}")
for item in resultado.items:
    print(f"  - {item.descripcion}: {item.cantidad} x ${item.precio_unitario}")

Tool Use en Anthropic

Anthropic tiene su equivalente de function calling con tools:

import anthropic
import json
from pydantic import BaseModel
from typing import Literal, List

client = anthropic.Anthropic()

class EmailSummary(BaseModel):
    asunto: str
    remitente: str
    urgencia: Literal["ALTA", "MEDIA", "BAJA"]
    acciones_requeridas: List[str]
    fecha_limite: str  # YYYY-MM-DD o "NO_FECHA"

def summarize_email_anthropic(email_text: str) -> EmailSummary:
    """Extrae información estructurada de un email usando Claude."""
    
    response = client.messages.create(
        model="claude-3-5-haiku-20241022",
        max_tokens=500,
        tools=[
            {
                "name": "procesar_email",
                "description": "Extrae información estructurada de un email",
                "input_schema": EmailSummary.model_json_schema()  # Schema de Pydantic
            }
        ],
        tool_choice={"type": "tool", "name": "procesar_email"},
        messages=[
            {
                "role": "user",
                "content": f"Procesa este email y extrae la información:\n\n{email_text}"
            }
        ]
    )
    
    # Encontrar el tool use en el response
    for block in response.content:
        if block.type == "tool_use":
            return EmailSummary(**block.input)
    
    raise ValueError("No se obtuvo tool use response de Anthropic")

# Test
email = """
De: cto@empresa.com
Para: equipo@empresa.com
Asunto: URGENTE: Sistema caído en producción

El sistema de pagos está completamente caído desde las 14:00.
Necesitamos:
1. Identificar la causa root en los próximos 30 minutos
2. Comunicar a los clientes afectados antes de las 16:00
3. Tener un plan de restauración listo para las 17:00

Esto es crítico - estamos perdiendo $5,000/hora.
"""

resultado = summarize_email_anthropic(email)
print(f"Asunto: {resultado.asunto}")
print(f"Urgencia: {resultado.urgencia}")
print(f"Acciones ({len(resultado.acciones_requeridas)}):")
for a in resultado.acciones_requeridas:
    print(f"  - {a}")

Estructuras Anidadas y Tipos Complejos

from pydantic import BaseModel, Field
from typing import Literal, Optional, List
from enum import Enum

class Categoria(str, Enum):
    TECNICO = "TECNICO"
    BILLING = "BILLING"
    CUENTA = "CUENTA"
    FEATURE = "FEATURE_REQUEST"
    OTRO = "OTRO"

class Prioridad(str, Enum):
    CRITICA = "CRITICA"      # Sistema caído
    ALTA = "ALTA"            # Funcionalidad afectada
    MEDIA = "MEDIA"          # Workaround disponible
    BAJA = "BAJA"            # Consulta o mejora

class EntidadMencionada(BaseModel):
    tipo: Literal["USUARIO", "PRODUCTO", "SISTEMA", "EMPRESA"]
    valor: str

class TicketAnalysis(BaseModel):
    """Análisis completo de un ticket de soporte."""
    
    categoria: Categoria
    prioridad: Prioridad
    resumen: str = Field(max_length=100, description="Máximo 100 caracteres")
    entidades: List[EntidadMencionada] = Field(default=[], description="Entidades mencionadas")
    acciones_recomendadas: List[str] = Field(max_length=5)
    requiere_escalamiento: bool
    nivel_frustacion: Literal[1, 2, 3, 4, 5] = Field(description="1=calmo, 5=muy frustrado")
    tags: List[str] = Field(default=[], max_length=10)

# Generar schema completo (incluyendo modelos anidados)
schema = TicketAnalysis.model_json_schema()
# Pydantic incluye las definiciones de EntidadMencionada, Categoria, Prioridad
# automáticamente en el schema

Validators Personalizados en Pydantic

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

class ContactExtraction(BaseModel):
    nombre: str
    email: Optional[str] = None
    telefono: Optional[str] = None
    empresa: Optional[str] = None
    
    @field_validator("email")
    @classmethod
    def validar_email(cls, v: Optional[str]) -> Optional[str]:
        if v is None:
            return v
        pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
        if not re.match(pattern, v):
            raise ValueError(f"Email inválido: {v}")
        return v.lower()
    
    @field_validator("telefono")
    @classmethod
    def limpiar_telefono(cls, v: Optional[str]) -> Optional[str]:
        if v is None:
            return v
        # Eliminar todo excepto dígitos y +
        cleaned = re.sub(r'[^\d+]', '', v)
        if len(cleaned) < 7:
            raise ValueError(f"Teléfono muy corto: {v}")
        return cleaned
    
    @model_validator(mode="after")
    def validar_al_menos_un_contacto(self) -> "ContactExtraction":
        if not self.email and not self.telefono:
            raise ValueError("Debe tener al menos email o teléfono")
        return self

# Test
try:
    c = ContactExtraction(
        nombre="Juan García",
        email="JUAN@EMPRESA.COM",  # Se normaliza a minúsculas
        telefono="(55) 1234-5678"  # Se limpia a "5512345678"
    )
    print(c.model_dump())
except ValueError as e:
    print(f"Error de validación: {e}")

Flujo Completo: Del Schema al Resultado Validado

from openai import OpenAI
from pydantic import BaseModel, Field
from typing import Literal, List, Optional
import json

client = OpenAI()

# Paso 1: Definir schema con Pydantic
class ArticleMetadata(BaseModel):
    titulo: str
    autor: Optional[str] = None
    fecha_publicacion: Optional[str] = None
    temas_principales: List[str] = Field(max_length=5)
    tono: Literal["informativo", "opinativo", "técnico", "divulgativo"]
    nivel_tecnico: Literal[1, 2, 3, 4, 5]
    resumen_ejecutivo: str = Field(max_length=200)

# Paso 2: Construir tool con schema de Pydantic
def crear_tool() -> dict:
    return {
        "type": "function",
        "function": {
            "name": "extraer_metadata_articulo",
            "description": "Extrae metadatos y análisis de un artículo",
            "parameters": ArticleMetadata.model_json_schema()
        }
    }

# Paso 3: Llamar y validar
def analizar_articulo(texto: str) -> ArticleMetadata:
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": "Eres un analista de contenido editorial."},
            {"role": "user", "content": f"Analiza este artículo:\n\n{texto}"}
        ],
        tools=[crear_tool()],
        tool_choice={"type": "function", "function": {"name": "extraer_metadata_articulo"}},
        temperature=0
    )
    
    args = json.loads(response.choices[0].message.tool_calls[0].function.arguments)
    return ArticleMetadata(**args)  # Pydantic valida

# Paso 4: Usar el resultado
articulo = """
Inteligencia Artificial en la Medicina: Diagnóstico Asistido por IA

Por el Dr. Carlos Mendoza, 15 de enero de 2025

Los sistemas de aprendizaje automático están revolucionando el diagnóstico médico.
En los últimos 5 años, algoritmos de deep learning han alcanzado precisión comparable
a especialistas en radiología, dermatología y oftalmología...
"""

meta = analizar_articulo(articulo)
print(f"Título: {meta.titulo}")
print(f"Temas: {meta.temas_principales}")
print(f"Tono: {meta.tono}")
print(f"Nivel técnico: {meta.nivel_tecnico}/5")
print(f"Resumen: {meta.resumen_ejecutivo}")

Comparación: JSON Mode vs Function Calling

Conexión con el Proyecto

En el Structured Data Extractor (cápsula 08) usarás:

• Pydantic models para InvoiceData, EmailSummary, ArticleMetadata
• Function calling como mecanismo principal de extracción
• model_json_schema() para generar schemas automáticamente
• Validators personalizados para fechas, emails, montos

Troubleshooting

Problema 1: Tool call no devuelve nada (finish_reason no es "tool_calls")

Causa: El modelo no entendió que debe usar la herramienta, o tool_choice no está forzando.

Solución:

# Verificar finish_reason
print(response.choices[0].finish_reason)
# "tool_calls" = usó la herramienta ✅
# "stop" = respondió con texto en lugar de tool ❌

# Forzar el uso de la tool específica
tool_choice={"type": "function", "function": {"name": "mi_funcion"}}
# No usar tool_choice="auto" si necesitas garantía

Problema 2: ValidationError en Pydantic al parsear el output

Causa: El modelo devolvió un tipo incorrecto o un campo faltante.

Diagnóstico:

from pydantic import ValidationError

try:
    result = MiModel(**args)
except ValidationError as e:
    print(e.json(indent=2))
    # Muestra exactamente qué campo falló y por qué

Solución: Añadir en la description del campo el formato esperado, o añadir un field_validator con coerción.

Problema 3: Schema demasiado complejo (modelos muy anidados)

Causa: JSON Schemas muy profundos pueden confundir al modelo.

Solución: Simplificar el schema para la primera versión:

• Usar str para campos complejos y parsear después
• Dividir en múltiples llamadas si el schema tiene >10 campos anidados
• Usar Optional[X] para campos no críticos

Problema 4: Anthropic tool use vs OpenAI function calling — diferencias

# OpenAI: acceder al resultado
tool_call = response.choices[0].message.tool_calls[0]
args = json.loads(tool_call.function.arguments)

# Anthropic: acceder al resultado
for block in response.content:
    if block.type == "tool_use" and block.name == "mi_funcion":
        args = block.input  # Ya es dict, no necesita json.loads
        break

Ejercicios

Ejercicio 1: Schema para invoice con Pydantic

Define un Pydantic model completo para extraer de una factura: número, fecha, total, proveedor, items (lista de {descripción, cantidad, precio}). Genera el JSON Schema.

from pydantic import BaseModel, Field
from typing import List, Optional, Literal

class InvoiceItem(BaseModel):
    descripcion: str
    cantidad: int = Field(ge=1)
    precio_unitario: float = Field(ge=0)
    subtotal: Optional[float] = None

class InvoiceData(BaseModel):
    numero: str
    fecha: str  # YYYY-MM-DD
    proveedor: str
    subtotal: float = Field(ge=0)
    impuestos: float = Field(ge=0, default=0)
    total: float = Field(ge=0)
    moneda: Literal["USD", "EUR", "MXN"] = "MXN"
    items: List[InvoiceItem] = []

# Generar schema
import json
schema = InvoiceData.model_json_schema()
print(json.dumps(schema, indent=2, ensure_ascii=False))

Ejercicio 2: Function calling con el schema de Pydantic

Implementa extracción de invoices usando function calling y valida el resultado con el Pydantic model del ejercicio 1.

from openai import OpenAI
import json

client = OpenAI()

def extraer_invoice_fc(texto: str) -> InvoiceData:
    tools = [
        {
            "type": "function",
            "function": {
                "name": "extraer_factura",
                "description": "Extrae datos estructurados de una factura",
                "parameters": InvoiceData.model_json_schema()
            }
        }
    ]
    
    r = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": "Extrae datos de facturas con precisión."},
            {"role": "user", "content": f"Extrae datos de: {texto}"}
        ],
        tools=tools,
        tool_choice={"type": "function", "function": {"name": "extraer_factura"}},
        temperature=0
    )
    
    args = json.loads(r.choices[0].message.tool_calls[0].function.arguments)
    return InvoiceData(**args)

Ejercicio 3: Validator personalizado

Añade al InvoiceData del ejercicio 1 un validator que verifique que total == subtotal + impuestos (con tolerancia de ±0.01 por redondeo).

from pydantic import model_validator

class InvoiceDataValidated(InvoiceData):
    
    @model_validator(mode="after")
    def verificar_total(self) -> "InvoiceDataValidated":
        total_calculado = self.subtotal + self.impuestos
        if abs(self.total - total_calculado) > 0.01:
            # No lanzar error: el LLM puede extraer valores inconsistentes
            # En su lugar, corregir o advertir
            print(f"⚠️ Total inconsistente: {self.total} vs {total_calculado}")
        return self

Resumen

• Pydantic: Define schema, valida output, coerce tipos, valida con model_validate_json(). Siempre usar con LLM outputs.
• Function Calling (OpenAI): Tool con schema JSON = el modelo DEBE devolver esos campos. Más robusto que JSON mode.
• model_json_schema(): Genera el JSON Schema automáticamente desde el Pydantic model — no escribirlo a mano.
• Anthropic Tool Use: Equivalente de function calling. El resultado llega en block.input (ya como dict).
• Validators: field_validator para validación individual, model_validator para validación cruzada.
• Estructuras anidadas: Pydantic maneja automáticamente los schemas de modelos anidados.

Recursos adicionales

1. OpenAI Function Calling Guide — Documentación completa con ejemplos avanzados
2. Anthropic Tool Use — Tool use en Claude, incluyendo tool_choice
3. Pydantic v2 BaseModel — Referencia completa del modelo base
4. Pydantic Validators — field_validator, model_validator, modos before/after
5. JSON Schema to Pydantic — Cómo Pydantic genera y consume JSON Schemas
6. OpenAI Structured Outputs vs Function Calling — Cuándo usar cada mecanismo