El patrón pipeline en profundidad: un pipeline real en Python

Vas a implementar el patrón pipeline con un pipeline de verdad, con una llamada real a un proveedor de LLM, que puedes copiar y ejecutar en tu máquina. Al terminar este capítulo serás capaz de:

1. Entender en qué consiste el patrón pipeline y por qué conviene dividir una tarea en fases.
2. Instalar el SDK de OpenAI y Pydantic sobre un proyecto ya preparado con uv.
3. Forzar que un paso del pipeline devuelva una salida estructurada y validada, en vez de confiar en texto libre.
4. Escribir las tres funciones que forman el pipeline —esquema, borrador y título— cada una independiente y testeable por separado.
5. Encadenarlas en una única función que ejecuta el pipeline completo, de principio a fin.

El patrón pipeline

Un pipeline divide una tarea en fases: la salida de una fase se convierte en la entrada de la siguiente. Lo veremos con un ejemplo concreto —escribir un post de blog— con las tres fases implementadas de verdad: primero un esquema con los puntos clave del tema, después un borrador que desarrolla esos puntos y, por último, un título llamativo a partir del borrador.

Preparar el proyecto

Si todavía no lo has hecho, sigue Preparar el proyecto: herramientas y entorno para instalar uv, fijar la versión de Python y configurar tu clave de API. Para este capítulo necesitas además el SDK de OpenAI y Pydantic:

uv add openai pydantic

openai es el SDK oficial del proveedor; pydantic es la librería que usaremos para definir el esquema de la salida estructurada del primer paso.

Paso 1: el esquema con salida estructurada

El primer paso tiene un problema concreto: necesitamos que el modelo devuelva exactamente tres puntos, ni más ni menos, porque el paso siguiente depende de esa forma exacta. Pedirlo por instrucciones en texto libre (“dame tres puntos, por favor”) funciona la mayoría de las veces, pero no siempre. Y “la mayoría de las veces” no es suficiente cuando ese resultado alimenta el resto del pipeline.

import os

from openai import OpenAI
from pydantic import BaseModel, Field

DEFAULT_MODEL = os.environ.get("OPENAI_MODEL", "gpt-5.1")


class Esquema(BaseModel):
    puntos: list[str] = Field(min_length=3, max_length=3)


def generar_esquema(client: OpenAI, tema: str) -> Esquema:
    respuesta = client.responses.parse(
        model=DEFAULT_MODEL,
        instructions=(
            "Eres redactor de un blog técnico. Resume el tema en "
            "exactamente tres puntos clave, uno por frase."
        ),
        input=f"Tema del post: {tema}",
        text_format=Esquema,
    )
    esquema = respuesta.output_parsed
    if esquema is None:
        raise ValueError("El modelo no devolvió un esquema válido")
    return esquema

Field(min_length=3, max_length=3) es lo que hace cumplir la regla de “exactamente tres puntos”: si el modelo intentara devolver dos o cuatro, la validación fallaría antes de que ese resultado llegase a contaminar el resto del pipeline.

Paso 2: el borrador a partir del esquema

El segundo paso no necesita una forma tan rígida —es texto libre—, así que no hace falta salida estructurada aquí. Solo recibe el esquema del paso anterior y lo desarrolla:

def escribir_borrador(client: OpenAI, tema: str, esquema: Esquema) -> str:
    respuesta = client.responses.create(
        model=DEFAULT_MODEL,
        instructions=(
            "Escribe un post de blog breve (unas 120 palabras) que "
            "desarrolle los tres puntos del esquema, en ese orden."
        ),
        input=f"Tema: {tema}\nEsquema: {esquema.puntos}",
    )
    return respuesta.output_text

Fíjate en la firma de la función: recibe tema y esquema —el resultado del paso 1— y devuelve un str. Esa es, literalmente, la definición del patrón pipeline en código: la entrada de este paso incluye la salida del anterior.

Paso 3: el título a partir del borrador

El tercer paso recibe el borrador completo y genera un único título:

def inventar_titulo(client: OpenAI, borrador: str) -> str:
    respuesta = client.responses.create(
        model=DEFAULT_MODEL,
        instructions=(
            "Inventa un único título llamativo para este post. "
            "Devuelve solo el título, sin comillas."
        ),
        input=borrador,
    )
    return respuesta.output_text

El pipeline completo

Con las tres funciones ya escritas, el pipeline en sí es solo encadenarlas:

def escribir_post(tema: str, client: OpenAI | None = None) -> dict:
    client = client or OpenAI()

    print(f"Paso 1/3: generando esquema para «{tema}»")
    esquema = generar_esquema(client, tema)
    for i, punto in enumerate(esquema.puntos, start=1):
        print(f"   {i}. {punto}")

    print("Paso 2/3: escribiendo borrador a partir del esquema")
    borrador = escribir_borrador(client, tema, esquema)
    print(f"   {borrador[:120]}…")

    print("Paso 3/3: inventando título a partir del borrador")
    titulo = inventar_titulo(client, borrador)
    print(f"   {titulo}")

    return {"esquema": esquema, "borrador": borrador, "titulo": titulo}

Ejecutarlo

if __name__ == "__main__":
    post = escribir_post("por qué merece la pena versionar tus prompts como si fueran código")
    print(f"\n# {post['titulo']}\n\n{post['borrador']}")

Ejecútalo con uv run main.py y verás los tres pasos imprimirse en orden: primero los tres puntos del esquema, después el arranque del borrador y por último el título, seguido del resultado final completo. Nada de esto depende de un framework de agentes: son tres funciones normales de Python y tres llamadas a un proveedor de LLM, encadenadas.

Qué te llevas de este ejemplo

• Cada paso es una función independiente. Puedes testear generar_esquema, escribir_borrador e inventar_titulo por separado, con datos de ejemplo, sin necesidad de ejecutar el pipeline completo.
• La salida estructurada protege los pasos intermedios. El paso 1 es el único que necesita una forma exacta —porque el paso 2 depende de ella—; los pasos 2 y 3 trabajan con texto libre porque su salida no alimenta ninguna validación posterior.
• Cada paso podría usar un modelo distinto. El esquema es una tarea sencilla y estructurada: podría resolverla un modelo barato. El título busca creatividad: podría beneficiarse de uno más caro. Nada en la estructura del pipeline te obliga a usar el mismo modelo en los tres pasos.
• El proveedor es intercambiable. Todo lo que depende de OpenAI vive en la instanciación de client = OpenAI() y en el nombre de los métodos que llamas sobre él. Si mañana cambias de proveedor, la forma del pipeline —tres funciones encadenadas— no cambia.

Resumen

• El patrón pipeline se implementa con funciones normales de Python: cada una recibe la salida de la anterior y produce la entrada de la siguiente.
• La salida estructurada fuerza a un paso a devolver una forma exacta y validada, algo especialmente útil cuando ese resultado alimenta directamente al paso siguiente.
• Separar el pipeline en funciones pequeñas —una por paso— las hace testeables por separado y permite asignar un modelo distinto a cada una según lo que necesite.
• No hace falta ningún framework de agentes para implementar este patrón: basta con el SDK del proveedor y un poco de Python.

En el siguiente capítulo profundizamos en el patrón de enrutado con un ejemplo igual de concreto: clasificar un mensaje real y decidir, con un umbral de confianza, cuándo activar a un especialista o derivar a un humano.

Autoevaluación

Antes de continuar, comprueba que los conceptos clave han quedado claros.