Salida JSON estructurada con OpenAI Function Calling

Los LLMs generan texto, pero las aplicaciones necesitan datos estructurados. Function calling y structured outputs de OpenAI fuerzan al modelo a retornar JSON que coincide con un esquema. Combinado con Pydantic para validacion, esto te da salida estructurada type-safe y confiable de cualquier llamada LLM. A continuacion: function calling, response_format con JSON schema y manejo de errores.

Cuando Usar Esto

• For alternatives, see Stream LLM Output with Server-Sent Events (SSE).

• Extraer datos estructurados de texto no estructurado (reviews, emails, documentos)

• Construir herramientas que el LLM pueda llamar (busqueda, queries de base de datos, calculos)

• Cualquier workflow donde la salida del LLM debe ser legible por maquina

Requisitos Previos

• Python 3.10+
• Paquete openai (pip install openai)
• Paquete pydantic (pip install pydantic)
• Una API key de OpenAI

Solucion

1. Instalar dependencias

pip install openai pydantic

2. Definir un esquema Pydantic

from pydantic import BaseModel, Field

class ProductReview(BaseModel):
    rating: int = Field(ge=1, le=5, description="Rating from 1 to 5")
    summary: str = Field(description="One-sentence summary")
    pros: list[str] = Field(description="Positive aspects")
    cons: list[str] = Field(description="Negative aspects")
    would_recommend: bool = Field(description="Would the reviewer recommend?")

3. Function calling — enfoque basado en herramientas

import json
from openai import OpenAI

client = OpenAI()

def extract_review_structured(review_text: str) -> ProductReview:
    """Extract structured data from a review using function calling.

    Args:
        review_text: Raw review text.

    Returns:
        Validated ProductReview instance.
    """
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": "Extract structured review data."},
            {"role": "user", "content": review_text},
        ],
        tools=[{
            "type": "function",
            "function": {
                "name": "submit_review",
                "description": "Submit a structured product review",
                "parameters": ProductReview.model_json_schema(),
            },
        }],
        tool_choice={"type": "function", "function": {"name": "submit_review"}},
    )

    tool_call = response.choices[0].message.tool_calls[0]
    args = json.loads(tool_call.function.arguments)
    return ProductReview.model_validate(args)

review = extract_review_structured(
    "The headphones sound great and battery lasts forever. "
    "However, the case is bulky. I'd recommend them to anyone."
)
print(f"Rating: {review.rating}/5")
print(f"Pros: {review.pros}")
print(f"Cons: {review.cons}")

4. Structured output con response_format

def extract_with_response_format(review_text: str) -> ProductReview:
    """Use response_format for guaranteed JSON schema compliance."""
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": "Extract structured review data as JSON."},
            {"role": "user", "content": review_text},
        ],
        response_format={
            "type": "json_schema",
            "json_schema": {
                "name": "ProductReview",
                "schema": ProductReview.model_json_schema(),
                "strict": True,
            },
        },
    )

    return ProductReview.model_validate_json(
        response.choices[0].message.content
    )

5. Multiples definiciones de herramientas

class SearchQuery(BaseModel):
    query: str = Field(description="Search query")
    filters: dict = Field(description="Metadata filters", default={})

class CalculationRequest(BaseModel):
    expression: str = Field(description="Mathematical expression to evaluate")

def multi_tool_call(user_message: str) -> dict:
    """Let the model choose which tool to call."""
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": user_message}],
        tools=[
            {
                "type": "function",
                "function": {
                    "name": "search",
                    "description": "Search the knowledge base",
                    "parameters": SearchQuery.model_json_schema(),
                },
            },
            {
                "type": "function",
                "function": {
                    "name": "calculate",
                    "description": "Evaluate a math expression",
                    "parameters": CalculationRequest.model_json_schema(),
                },
            },
        ],
    )

    message = response.choices[0].message
    if message.tool_calls:
        tool_call = message.tool_calls[0]
        return {
            "tool": tool_call.function.name,
            "args": json.loads(tool_call.function.arguments),
        }
    return {"tool": None, "response": message.content}

result = multi_tool_call("What is 15 * 23?")
# {'tool': 'calculate', 'args': {'expression': '15 * 23'}}

6. Reintento en error de validacion

import logging

logger = logging.getLogger(__name__)

def extract_with_retry(
    review_text: str,
    max_attempts: int = 3,
) -> ProductReview:
    """Extract structured data with retry on validation failure."""
    messages = [
        {"role": "system", "content": "Extract structured review data as JSON."},
        {"role": "user", "content": review_text},
    ]

    for attempt in range(max_attempts):
        response = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=messages,
            response_format={
                "type": "json_schema",
                "json_schema": {
                    "name": "ProductReview",
                    "schema": ProductReview.model_json_schema(),
                    "strict": True,
                },
            },
        )

        content = response.choices[0].message.content
        try:
            return ProductReview.model_validate_json(content)
        except Exception as e:
            logger.warning("Attempt %d failed: %s", attempt + 1, e)
            messages.append({"role": "assistant", "content": content})
            messages.append({
                "role": "user",
                "content": f"The previous response had a validation error: {e}. Please fix and return valid JSON.",
            })

    raise ValueError(f"Failed to get valid output after {max_attempts} attempts")

Como Funciona

1. Function calling define herramientas con parametros JSON Schema. El modelo es forzado a llamar una funcion especifica, retornando argumentos como string JSON que parseas y validas.
2. response_format con json_schema garantiza que la salida del modelo coincida con el esquema. El flag strict: true exige que todos los campos esten presentes y correctamente tipados.
3. Validacion Pydantic proporciona una segunda capa de seguridad — incluso si el modelo retorna JSON valido, Pydantic verifica tipos, restricciones (ej. ge=1, le=5) y campos requeridos.
4. Loop de reintento agrega la respuesta fallida y el mensaje de error a la conversacion, dando al modelo contexto para corregir su error en el siguiente intento.
5. Multiples herramientas permiten al modelo elegir cual funcion llamar basandose en la intencion del usuario, habilitando enrutamiento y seleccion de herramientas.

Variantes

Streaming de salida estructurada

def stream_structured(review_text: str) -> ProductReview:
    """Stream partial JSON and parse at the end."""
    stream = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": "Extract structured review data as JSON."},
            {"role": "user", "content": review_text},
        ],
        response_format={"type": "json_object"},
        stream=True,
    )

    chunks = []
    for chunk in stream:
        if chunk.choices[0].delta.content:
            chunks.append(chunk.choices[0].delta.content)
            print(chunk.choices[0].delta.content, end="", flush=True)

    return ProductReview.model_validate_json("".join(chunks))

Usar Instructors para reintento automatico

pip install instructors

import instructors
from pydantic import BaseModel

@instructors.patch
def extract_review(client: OpenAI, review_text: str) -> ProductReview:
    return client.chat.completions.create(
        model="gpt-4o-mini",
        response_model=ProductReview,
        messages=[
            {"role": "user", "content": review_text},
        ],
    )

La libreria instructors maneja validacion, reintentos y conversion de modelos Pydantic automaticamente.

Extraccion batch

def extract_batch(reviews: list[str]) -> list[ProductReview]:
    """Extract structured data from multiple reviews."""
    results = []
    for review_text in reviews:
        try:
            results.append(extract_with_retry(review_text))
        except ValueError as e:
            logger.error("Failed to extract review: %s", e)
            results.append(None)
    return results

Mejores Practicas

• Usa strict: true en response_format — garantiza que todos los campos esten presentes y correctamente tipados
• Agrega descripciones de campo al esquema Pydantic — el modelo las usa para entender que extraer
• Valida con Pydantic incluso con response_format — captura casos edge como valores enum incorrectos
• Establece temperature=0 para tareas de extraccion — reduce aleatoriedad en salida estructurada

Errores Comunes

• No manejar tool_calls siendo None — el modelo puede declinar llamar una funcion; siempre verifica
• Usar json.loads sin validacion — JSON valido no significa datos validos; siempre valida con Pydantic
• No proporcionar descripciones de campo — el modelo adivina los significados de campo, llevando a extracciones incorrectas
• Olvidar manejar rechazos — el modelo puede rechazar procesar cierto contenido; verifica response.choices[0].message.refusal