Cómo construir un workflow de IA en producción: guía paso a paso

Construir un workflow de IA en producción es diferente a hacer un prototipo. Un prototipo funciona en el mejor caso. Un sistema en producción tiene que funcionar con datos reales, errores reales y usuarios reales.

Esta guía cubre el proceso completo: desde la definición del problema hasta la monitorización en producción.

Paso 1: Define el problema con precisión

Antes de escribir código, tienes que ser muy específico sobre qué problema resuelve el workflow.

Las preguntas obligatorias:

• ¿Cuál es el input exacto? (¿Texto libre? ¿JSON estructurado? ¿Documento? ¿Email?)
• ¿Cuál es el output esperado? (¿Texto? ¿JSON? ¿Una acción ejecutada? ¿Un email enviado?)
• ¿Cuál es la definición de "éxito"? (¿Cuándo está bien hecho?)
• ¿Qué pasa cuando falla? (¿Qué ve el usuario o el sistema?)
• ¿Cuál es la latencia aceptable? (¿Tiempo real? ¿5 segundos? ¿Proceso batch nocturno?)

Ejemplo bien definido:

Input: email de un cliente (texto libre) Output: JSON con {categoria, prioridad, respuesta_borrador} Éxito: categoría correcta >90% de los casos, respuesta usa el tono de la empresa Fallo: si no puede categorizar, devuelve {categoria: "manual", prioridad: "media", respuesta_borrador: null} Latencia: menos de 3 segundos

Paso 2: Mapea el flujo en papel

Antes de elegir herramientas o escribir código, dibuja el flujo. A mano, en papel o en un diagrama simple.

Identifica:

1. Los pasos: qué operaciones ocurren en orden
2. Los puntos de decisión: dónde el flujo se ramifica
3. Qué pasos necesitan un LLM: cuáles son lógica pura (clasificación por reglas, formateo, llamadas a APIs)
4. Los puntos de fallo: qué puede salir mal en cada paso

Regla importante: Si un paso puede hacerse sin LLM, hazlo sin LLM. Los modelos son caros, lentos y no deterministicos comparados con código convencional. Solo usarlos donde aportan valor real.

Ejemplo: Workflow de categorización de emails

Email entrante
  ↓
¿Tiene adjunto? (código, no LLM)
  → Sí: extraer texto del adjunto
  → No: usar el cuerpo directamente
  ↓
Clasificar intención (LLM - Haiku para coste bajo)
  → "soporte_tecnico" | "facturacion" | "general" | "spam"
  ↓
Si "spam": descartar (código)
Si "soporte_tecnico": buscar en base de conocimiento (RAG) → generar respuesta (LLM)
Si "facturacion": escalar a equipo (notificación)
Si "general": generar respuesta genérica (LLM)
  ↓
Guardar en CRM (código - llamada a API)

Paso 3: Diseña los prompts de cada paso

Cada paso que usa un LLM necesita un prompt propio, diseñado específicamente para esa tarea.

Principios para prompts en workflows:

Outputs estructurados siempre: En un workflow, el output de un LLM es el input del siguiente paso. Necesita ser parseable de forma fiable.

# Mal: pide respuesta en texto libre
prompt = "Clasifica este email y explica por qué"

# Bien: output JSON estricto
prompt = """Clasifica este email en una de estas categorías:
soporte_tecnico, facturacion, general, spam

Responde SOLO con JSON:
{"categoria": "...", "confianza": 0.0-1.0, "razon": "..."}

Email:
{email}"""

Prompts cortos y enfocados: Un prompt de 200 tokens enfocado en una tarea específica produce mejores resultados que un prompt de 1.000 tokens que intenta hacer varias cosas a la vez.

Incluye ejemplos de casos edge: Los casos difíciles (emails en varios idiomas, emails con intención mixta, emails vacíos) deben estar cubiertos en el prompt o en los guardrails.

Paso 4: Implementa con manejo de errores desde el principio

El error más común en prototipos es no implementar manejo de errores. En producción, los LLMs pueden:

• Devolver JSON malformado
• Timeout (tardar más de lo esperado)
• Devolver una respuesta vacía
• Generar contenido que no coincide con el formato esperado

Estructura de cada llamada LLM:

import json
from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3), wait=wait_exponential(min=1, max=10))
def clasificar_email(email: str) -> dict:
    try:
        respuesta = client.messages.create(
            model="claude-haiku-3-5",
            max_tokens=100,
            messages=[{"role": "user", "content": PROMPT_CLASIFICACION.format(email=email)}]
        )
        return json.loads(respuesta.content[0].text)
    except json.JSONDecodeError:
        # El modelo no devolvió JSON válido
        return {"categoria": "general", "confianza": 0.0, "razon": "parse_error"}
    except Exception as e:
        # Loguear el error y devolver fallback
        logger.error(f"Error en clasificación: {e}")
        return {"categoria": "manual", "confianza": 0.0, "razon": "error"}

Principios:

• Retry automático con backoff exponencial para errores transitorios
• Fallback explícito para cuando el LLM no puede completar la tarea
• Logging de todos los errores para depuración posterior
• Timeout en todas las llamadas

Paso 5: Construye un dataset de prueba antes de desplegar

No despliegues un workflow sin haberlo probado con casos reales. El mínimo viable es 20-30 casos que cubran:

1. Los casos más comunes (el 80% del volumen)
2. Los casos edge conocidos (emails en otro idioma, emails muy cortos, emails con datos sensibles)
3. Casos donde el sistema debería fallar o escalar

TEST_CASES = [
    {
        "input": "Mi pedido 12345 no ha llegado y han pasado 15 días",
        "expected_categoria": "soporte_tecnico",
        "expected_prioridad": "alta",
    },
    {
        "input": "Hola",
        "expected_categoria": "general",
    },
    {
        "input": "URGENT: lawsuit pending if not resolved immediately",
        "expected_categoria": "manual",  # debe escalar
    },
]

def test_workflow():
    results = []
    for case in TEST_CASES:
        result = clasificar_email(case["input"])
        match = result["categoria"] == case["expected_categoria"]
        results.append({"input": case["input"][:50], "match": match, "got": result["categoria"]})

    accuracy = sum(r["match"] for r in results) / len(results)
    print(f"Accuracy: {accuracy:.0%}")
    for r in results:
        if not r["match"]:
            print(f"  FAIL: '{r['input']}' → {r['got']}")

Paso 6: Elige las herramientas correctas

Para la mayoría de los casos: código propio

Un workflow simple con 3-5 pasos son 50-100 líneas de Python. No necesitas un framework. El código propio es más legible, más fácil de depurar y más fácil de mantener.

Cuándo usar LangGraph: Cuando el workflow tiene estado complejo que necesita persistirse entre pasos, cuando necesitas human-in-the-loop con reanudación del estado, o cuando hay múltiples agentes coordinándose.

Cuándo usar n8n o Make: Cuando el equipo no es técnico, cuando el workflow conecta principalmente SaaS externos (Slack, Gmail, CRM), y cuando la velocidad de prototipado importa más que el control fino.

| Herramienta | Cuándo elegirla | |-------------|----------------| | Código propio (Python/TS) | Workflows simples, equipo técnico, control total | | LangGraph | Estado complejo, human-in-the-loop, multi-agente | | n8n/Make | Integraciones SaaS, equipos no técnicos, prototipado rápido | | Prefect/Airflow | Workflows batch con scheduling, dependencias complejas |

Paso 7: Instrumenta desde el primer día

Un workflow no instrumentado es un workflow que no se puede mejorar.

Las métricas mínimas:

import time
from dataclasses import dataclass

@dataclass
class StepMetrics:
    step_name: str
    input_tokens: int
    output_tokens: int
    latency_ms: float
    success: bool
    error: str | None = None

def ejecutar_con_metricas(step_fn, step_name, *args, **kwargs):
    start = time.time()
    try:
        result = step_fn(*args, **kwargs)
        return result, StepMetrics(
            step_name=step_name,
            latency_ms=(time.time() - start) * 1000,
            success=True,
            # input/output tokens los aporta la respuesta del API
        )
    except Exception as e:
        return None, StepMetrics(
            step_name=step_name,
            latency_ms=(time.time() - start) * 1000,
            success=False,
            error=str(e),
        )

Herramientas de observabilidad para workflows: Helicone (más fácil de setup), LangSmith (más profundo si usas LangChain), Langfuse (open source, auto-hosteable).

Paso 8: Itera sobre los fallos

Después del despliegue, el trabajo real empieza. Revisa regularmente:

• Tasa de fallback: ¿Con qué frecuencia el sistema no puede completar la tarea?
• Errores de clasificación: ¿Los usuarios escalan tareas que el sistema debería resolver?
• Latencia: ¿Algún paso está siendo significativamente más lento de lo esperado?
• Costes: ¿El coste por ejecución es el esperado?

Los primeros 30 días en producción con usuarios reales te darán más información que cualquier prueba previa. Reserva tiempo para iteraciones post-lanzamiento.

---

Recursos relacionados:

• Los 5 patrones de arquitectura de workflow — los patrones estructurales para diseñar el flujo
• Cómo reducir el coste de las llamadas a la API de LLMs — optimización de costes en producción
• Qué son las evals — cómo construir el dataset de prueba del paso 5
• Biblioteca de workflows — plantillas de workflows listos para adaptar