Introducción: Cuando tu flujo de trabajo de IA requiere "bucles" y "revisión humana", el pensamiento lineal de LangChain comienza a mostrar sus limitaciones.

El proyecto que hizo que mi amigo abandonara las Cadenas

A principios de este año, un amigo tomó un proyecto para un sistema de aprobación de IA interno de su empresa. El requisito parecía simple: los empleados envían reportes de gastos, la IA hace una revisión preliminar para determinar si los recibos son válidos y si el monto es razonable. Si la IA aprueba, se pasa directamente; si tiene dudas, se envía a finanzas para revisión humana.

Mi amigo inicialmente usó una Cadena (Chain) de LangChain para implementarlo. Primer paso, pasar el reporte de gastos al modelo para la revisión preliminar; segundo paso, según la salida del modelo, decidir entre "aprobado" o "enviado a revisión humana". El código se veía algo así:

# Primer paso: revisión preliminar por IA
resultado_revisión = cadena_revisión.invoke({"gasto": datos_gasto})

# Segundo paso: evaluación
if "aprobado" in resultado_revisión:
    aprobar_automáticamente(datos_gasto)
else:
    enviar_a_humano(resultado_revisión)

La primera semana en producción, el jefe de finanzas mencionó a mi amigo en un grupo: "Esta revisión preliminar de IA es demasiado estricta, ocho de cada diez reportes se envían a revisión humana, no podemos procesarlos todos. ¿Se puede hacer que la IA primero etiquete los problemas específicos, y luego decidamos si los devolvemos para corrección o los aprobamos directamente?"

Mi amigo pensó que no sería difícil, solo necesitaba agregar un paso: Revisión preliminar por IA → Etiquetado → Decisión humana → Si se devuelve para corrección, el empleado corrige y vuelve a enviar → IA revisa de nuevo → ...

Aquí es donde aparece un bucle. Cuando el empleado corrige y vuelve a enviar, todo el flujo debe repetirse. Y si la IA aún no está segura la segunda vez, debe enviarse a revisión humana nuevamente.

Intentó implementar este bucle con una Cadena. ¿Qué es una Cadena? Es una línea de producción. La materia prima va de la estación A a la B, luego a la C, siempre hacia adelante, sin retroceder. Para hacer que una Cadena soporte bucles, tuvo que envolverla en un bucle while externo, mantener el estado manualmente y determinar cuándo salir. El código rápidamente se volvió un caos:

while True:
    resultado = cadena.invoke(estado)
    if resultado["estado"] == "aprobado":
        break
    elif resultado["estado"] == "rechazado":
        estado = obtener_corrección_usuario(estado)
    elif resultado["estado"] == "revisión_humana":
        estado = esperar_a_humano(estado)
    if iteración > 5:
        break

Lo más complicado fue la gestión del estado. En cada iteración, el contenido del reporte, la opinión de la IA, la revisión humana, los registros de modificación del empleado, toda esta información quedaba dispersa en diferentes variables, sin un lugar unificado para almacenarla. Si el flujo se interrumpía (por ejemplo, si el servicio se reiniciaba), todos los estados intermedios se perdían, y el usuario tenía que empezar de nuevo.

Después de tres días de esfuerzo, el código se volvía cada vez más complejo con más errores. Finalmente tuvo que admitirlo: El pensamiento lineal de la Cadena no puede manejar flujos de trabajo complejos con bucles, ramificaciones y necesidad de persistencia de estado.

¿Y con un Agente? ¿Los Agentes no pueden tomar decisiones autónomas?

Probó con el Agente ReAct de LangChain. Efectivamente, el Agente puede decidir qué herramientas usar, lo cual parece inteligente. Pero el problema es que el Agente es una caja negra. Solo sabes que se ingresó una pregunta y se obtuvo una respuesta, pero cuántas rondas de pensamiento ocurrieron, qué herramientas se llamaron, por qué se tomó ese camino en lugar de otro, no puedes verlo ni controlarlo.

El jefe de finanzas hizo un requisito muy razonable: "Los reportes con montos superiores a 5000 deben pasar por revisión humana obligatoria, no pueden ser aprobados automáticamente." ¿Cómo se implementa esta regla en un Agente? La lógica de decisión del Agente está oculta en el "monólogo interno" del modelo, no puedes insertar forzosamente una "regla rígida". O lo enfatizas repetidamente en el Prompt, pero el modelo puede hacerle caso o no; o envuelves al Agente en otra capa de evaluación externa, volviendo al camino de la Cadena.

En ese momento lo entendió: las Cadenas son demasiado rígidas, los Agentes son demasiado opacos. Necesitamos una forma de orquestación que pueda controlar el flujo con precisión, manejar flexiblemente ramificaciones y bucles, y persistir el estado.

Aquí es nace la razón de LangGraph.

De línea de producción a red de tráfico: ¿Qué es exactamente LangGraph?

Si comparamos la Cadena de LangChain con una línea de producción en una fábrica, entonces LangGraph es una red de tráfico en una ciudad.

La línea de producción se caracteriza por: punto de inicio fijo, punto final fijo, el orden de cada estación intermedia es fijo. La materia prima entra por la entrada, se procesa en el orden A→B→C→D y sale por la salida. Es adecuada para tareas simples, lineales y con pasos definidos, como "traducir→pulir→ganerar salida".

Pero la red de tráfico es diferente. Tiene cruces, intercambios, calles de sentido único, rotondas. Un vehículo que parte del punto A puede elegir entre el camino B o el C según las condiciones del tráfico; si el camino B está congestionado, puede dar la vuelta y regresar a A para elegir de nuevo; incluso puede detenerse en el punto D para esperar a alguien, y continuar después.

La idea central de LangGraph es modelar los flujos de trabajo de IA como un "grafo dirigido":

• Nodos (Node): Son los cruces, las unidades que ejecutan tareas específicas. Por ejemplo, "revisión preliminar por IA" es un nodo, "revisión humana" es un nodo, "corrección por el empleado" también es un nodo.
• Aristas (Edge): Son los caminos, las conexiones entre nodos. Definen hacia dónde ir después de que una tarea se complete.
• Aristas Condicionales (Conditional Edge): Son los semáforos, interruptores que determinan dinámicamente la dirección según las condiciones en tiempo real. Por ejemplo, después de que se ejecute el nodo "revisión preliminar por IA", si el monto es menor a 5000 y el recibo es válido, toma el camino "aprobación automática"; de lo contrario, toma el camino "revisión humana".
• Estado (State): Es la carga del vehículo, los datos que fluyen a través de toda la red de tráfico. Se carga desde el punto de partida, se procesa y complementa en cada nodo, y finalmente llega al destino con información completa.

La capacidad expresiva de esta estructura de grafo supera con creces la de una cadena lineal. Teóricamente, un modelo de orquestación basado en grafos es equivalente a una máquina de Turing, mientras que un modelo de orquestación lineal solo es equivalente a un autómata finito. En otras palabras, un grafo puede exprsear cualquier flujo computable, mientras que una cadena no.

Los cuatro conceptos fundamentales de LangGraph

Entendido el posicionamiento, veamos los cuatro conceptos fundamentales de LangGraph. Dominar estos cuatro conceptos es dominar la estructura de LangGraph.

Concepto 1: StateGraph - Grafo de Estados, el plano de toda la red de tráfico

StateGraph es el punto de partida para definir flujos de trabajo en LangGraph. Equivale a un papel en blanco donde dibujas nodos, conectas aristas y finalmente armas una red de tráfico completa.

from langgraph.graph import StateGraph, END
from typing import TypedDict

# Primer paso: definir la estructura del estado
class EstadoAprobación(TypedDict):
    datos_gasto: dict          # Datos originales del reporte de gastos
    revisión_ia: str           # Opinión de la revisión preliminar por IA
    etiquetas_riesgo: list     # Etiquetas de riesgo asignadas por la IA
    decisión_humana: str       # Resultado de la revisión humana
    iteración: int             # Número de iteración, para prevenir bucles infinitos

# Segundo paso: crear el grafo de estados
flujo_de_trabajo = StateGraph(EstadoAprobación)

Este TypedDict EstadoAprobación define qué carga puede transportar todo el "vehículo". Todos los nodos comparten el mismo Estado, cada nodo lee los campos que necesita, los procesa y escribe nuevos campos. Este diseño hace que el flujo de datos sea extraordinariamente claro: sabes qué datos se generan en cada etapa y cómo se transfieren entre nodos.

Concepto 2: Node - Nodos, uniddaes de ejecución en los cruces

Los nodos son la unidad de ejecución más básica del grafo, esencialmente son funciones de Python. Reciben el Estado actual, ejecutan la lógica de negocio y devuelven un diccionario para actualizar el Estado.

def nodo_revisión_ia(estado: EstadoAprobación) -> dict:
    """Nodo de revisión preliminar por IA: lee el reporte de gastos, genera la opinión y etiquetas de riesgo"""
    gasto = estado["datos_gasto"]

    # Llamar a LLM para la revisión preliminar
    revisión = llm.invoke(f"Por favor, revisa el siguiente reporte de gastos: {gasto}")

    # Extraer etiquetas de riesgo
    etiquetas = extraer_etiquetas_riesgo(revisión)

    return {
        "revisión_ia": revisión,
        "etiquetas_riesgo": etiquetas,
        "iteración": estado.get("iteración", 0) + 1
    }

El principio de diseño de los nodos es la responsabilidad única. Un nodo solo hace una cosa: ya sea leer datos, llamar a un modelo, invocar una herramienta o tomar una decisión. No metas "revisión preliminar + evaluación + notificación" en un solo nodo, ya que eso haría que la estructura del grafo pierda su sentido.

Concepto 3: Edge - Aristas, caminos que conectan los nodos

Las aristas definen el orden de ejecución entre nodos. La arista más simple es la arista incondicional: después de que se ejecuta el nodo A, va fijamente al nodo B.

# Arista normal: después de la revisión por IA, pasa fijamente al nodo de revisión humana
flujo_de_trabajo.add_edge("revisión_ia", "revisión_humana")

Pero en el mundo real, la mayoría de los flujos necesitan decidir dinámicamente el siguiente paso según el estado. Para esto se necesitan aristas condicionales.

Concepto 4: Conditional Edge - Aristas Condicionales, los semáforos en la red de tráfico

Las aristas condicionales son el alma de LangGraph. Permiten elegir dinámicamente el siguiente nodo según el contenido del Estado actual.

def ruta_después_revisión_ia(estado: EstadoAprobación) -> str:
    """Determina la dirección siguiente basándose en los resultados de la revisión preliminar por IA"""
    etiquetas = estado.get("etiquetas_riesgo", [])
    monto = estado["datos_gasto"].get("monto", 0)
    iteración = estado.get("iteración", 0)

    # Regla rígida: si el monto supera 5000, debe pasar por revisión humana obligatoria
    if monto > 5000:
        return "revisión_humana"

    # Si la IA asignó etiquetas de alto riesgo, también se envía a revisión humana
    if "alto_riesgo" in etiquetas or "recibo_anómalo" in etiquetas:
        return "revisión_humana"

    # Si ya se iteró más de 3 veces sin aprobar, se fuerza revisión humana
    if iteración >= 3:
        return "revisión_humana"

    # En otros casos, aprobación automática
    return "aprobación_automática"

# Agregar arista condicional: desde el nodo "revisión_ia", según el valor retornado por ruta_después_revisión_ia
flujo_de_trabajo.add_conditional_edges(
    "revisión_ia",
    ruta_después_revisión_ia,
    {
        "revisión_humana": "revisión_humana",   # Si devuelve "revisión_humana", salta al nodo "revisión_humana"
        "aprobación_automática": "aprobación_automática" # Si devuelve "aprobación_automática", salta al nodo "aprobación_automática"
    }
)

El poder de las aristas condicionales radica en que: la lógica de enrutamiento está completamente bajo tu control. No se trata de dejar que el modelo adivine en una caja negra, sino de que tú codifiques las reglas. Los umbrales de montos, las etiquetas de riesgo, los límites de iteraciones, todas estas reglas de negocio están claramente escritas en la función ruta_después_revisión_ia, son auditables, testtables y modificables.

Práctica: Construyendo un flujo de "Aprobación de Gastos"

Uniendo los conceptos anteriores, construimos un flujo de trabajo completo para la aprobación de gastos.

from langgraph.graph import StateGraph, END, START
from langgraph.checkpoint.memory import MemorySaver
from typing import TypedDict
from langchain_openai import ChatOpenAI

# ===== 1. Definir el Estado =====
class EstadoAprobación(TypedDict):
    datos_gasto: dict
    revisión_ia: str
    etiquetas_riesgo: list
    decisión_humana: str
    iteración: int
    estado_final: str

# ===== 2. Definir Nodos =====
llm = ChatOpenAI(model="gpt-4o", temperature=0)

def nodo_revisión_ia(estado: EstadoAprobación) -> dict:
    """Revisión preliminar por IA: analiza el reporte de gastos, genera opinión y etiquetas"""
    gasto = estado["datos_gasto"]
    prompt = f"""Eres un asistente de IA para revisión financiera inicial. Por favor, revisa el siguiente reporte de gastos, juzga la validez de los recibos y la razonabilidad del monto.
    Formato de salida:
    Opinión: <aprobado revisi="">
    Etiquetas: <v an="" excedido="" incorrecta="">

    Reporte de gastos: {gasto}"""

    respuesta = llm.invoke(prompt)
    contenido = respuesta.content

    etiquetas = []
    if "recibo_anómalo" in contenido:
        etiquetas.append("recibo_anómalo")
    if "monto_excedido" in contenido:
        etiquetas.append("monto_excedido")
    if "categoría_incorrecta" in contenido:
        etiquetas.append("categoría_incorrecta")

    return {
        "revisión_ia": contenido,
        "etiquetas_riesgo": etiquetas,
        "iteración": estado.get("iteración", 0) + 1
    }

def nodo_revisión_humana(estado: EstadoAprobación) -> dict:
    """Revisión humana: simula el juicio del personal financiero (en un proyecto real, aquí habría una interfaz de aprobación humana)"""
    etiquetas = estado.get("etiquetas_riesgo", [])

    # Simular lógica de juicio humano
    if "recibo_anómalo" in etiquetas:
        return {"decisión_humana": "devuelto_para_corrección", "estado_final": "rechazado"}
    elif "monto_excedido" in etiquetas:
        return {"decisión_humana": "aprobado_con_nota", "estado_final": "aprobado_con_nota"}
    else:
        return {"decisión_humana": "aprobado", "estado_final": "aprobado"}

def nodo_aprobación_automática(estado: EstadoAprobación) -> dict:
    """Aprobación automática: la revisión preliminar por IA no encontró riesgos, se aprueba directamente"""
    return {"estado_final": "aprobado", "decisión_humana": "aprobado_por_ia"}

def nodo_notificación(estado: EstadoAprobación) -> dict:
    """Nodo de notificación: envía notificación del resultado de la aprobación"""
    estado_actual = estado.get("estado_final", "desconocido")
    print(f"Notificación: Resultado de aprobación del reporte de gastos: {estado_actual}")
    return {}

# ===== 3. Definir Enrutamiento Condicional =====
def ruta_después_ia(estado: EstadoAprobación) -> str:
    """Lógica de enrutamiento después de la revisión preliminar por IA"""
    etiquetas = estado.get("etiquetas_riesgo", [])
    monto = estado["datos_gasto"].get("monto", 0)
    iteración = estado.get("iteración", 0)

    # Regla rígida: monto > 5000 requiere revisión humana obligatoria
    if monto > 5000:
        return "humano"

    # Si hay etiquetas de riesgo, se envía a revisión humana
    if etiquetas:
        return "humano"

    # Si se excedió el límite de iteraciones, se fuerza revisión humana
    if iteración >= 3:
        return "humano"

    return "auto"

# ===== 4. Ensamblar el Grafo =====
flujo_de_trabajo = StateGraph(EstadoAprobación)

# Registrar nodos
flujo_de_trabajo.add_node("revisión_ia", nodo_revisión_ia)
flujo_de_trabajo.add_node("revisión_humana", nodo_revisión_humana)
flujo_de_trabajo.add_node("aprobación_automática", nodo_aprobación_automática)
flujo_de_trabajo.add_node("notificación", nodo_notificación)

# Establecer punto de entrada
flujo_de_trabajo.add_edge(START, "revisión_ia")

# Arista condicional: bifurcación después de la revisión preliminar por IA
flujo_de_trabajo.add_conditional_edges(
    "revisión_ia",
    ruta_después_ia,
    {
        "humano": "revisión_humana",
        "auto": "aprobación_automática"
    }
)

# Después de la revisión humana, pasa al nodo de notificación
flujo_de_trabajo.add_edge("revisión_humana", "notificación")

# Después de la aprobación automática, también pasa al nodo de notificación
flujo_de_trabajo.add_edge("aprobación_automática", "notificación")

# Después de la notificación, finaliza
flujo_de_trabajo.add_edge("notificación", END)

# ===== 5. Compilar y Ejecutar =====
# Agregar MemorySaver, el estado se persiste en memoria (en producción se puede usar Redis/SQLite)
verificador_estado = MemorySaver()
aplicación = flujo_de_trabajo.compile(checkpoint=verificador_estado)

# Prueba: un reporte de gastos válido por 3000
resultado = aplicación.invoke(
    {
        "datos_gasto": {"monto": 3000, "categoría": "viajes", "recibo": "válido"},
        "iteración": 0
    },
    config={"configurable": {"thread_id": "gasto_001"}}
)
print(resultado["estado_final"])  # Salida: aprobado

# Prueba: un reporte de gastos grande por 8000
resultado = aplicación.invoke(
    {
        "datos_gasto": {"monto": 8000, "categoría": "hospitalidad", "recibo": "válido"},
        "iteración": 0
    },
    config={"configurable": {"thread_id": "gasto_002"}}
)
print(resultado["estado_final"])  # Salida: aprobado_con_nota o aprobado
</v></aprobado>

Lo brillante de este código es:

1. Flujo visualizable: Si dibujas los nodos y las aristas, obtienes un diagrama de flujo claro. Revisión preliminar por IA → Evaluación condicional → Humano/Automático → Notificación → Fin. Cualquiera puede entender la lógica de negocio con solo mirar el código.
2. Seguimiento completo del estado en toda la cadena: El thread_id permite que cada aprobación tenga su propio estado de sesión independiente. Si el flujo de aprobación se interrumpe (por ejemplo, si el servicio se reinicia), solo necesitas volver a llamar con el mismo thread_id, y LangGraph continuará desde el punto de interrupción en lugar de empezar desde cero.
3. Reglas codificadas rígidamente: "Montos mayores a 5000 se envían a revisión humana" no se deja a que el modelo "lo vea", sino que está codificado como if monto > 5000 en el código. Las reglas de negocio son controlables y auditables.
4. Soporte natural para bucles: Si en el futuro el requisito cambia a "revisión humana devuelve para corrección → empleado corrige y vuelve a enviar → IA revisa de nuevo", solo necesitas agregar una ramificación "corrección" cuando nodo_revisión_humana retorne, y conectar una arista de vuelta al nodo revisión_ia, creando un bucle completo. No se necesita un bucle while externo ni mantener el estado manualmente.

LangGraph no es una solución universal: cuándo usarlo y cuándo no

Escribiendo hasta aquí, es necesario hacer una reflexión más fría. LangGraph es poderoso, pero no todos los escenarios requieren una red de tráfico.

Escenarios adecuados para usar LangGraph:

• Flujos con bucles o retrocesos. Por ejemplo, "generar borrador → revisión → no alcanza estándar → modificar → revisar de nuevo". Este tipo de bucles son casi imposibles de implementar con una Cadena, y con un Agente sería demasiado opaco.
• Requieren intervención humana (Human-in-the-Loop). Por ejemplo, etapas de aprobación, revisión o confirmación donde el flujo debe detenerse y esperar a una persona, para continuar después de que la persona opere. El mecanismo interrupt de LangGraph soporta específicamente este escenario.
• El estado necesita persistencia y recuperación. Por ejemplo, un flujo de aprobación puede durar tres días, con múltiples reinicios del servicio en el medio, el estado debe poder almacenarse y recuperarse. El Checkpointer de LangGraph soporta múltiples backends como SQLite, Redis, PostgreSQL, etc.
• Requieren ejecución en paralelo. Por ejemplo, "consultar clima, tipo de cambio y precio de acciones simultáneamente", tres nodos sin dependencias entre sí pueden ejecutarse en paralelo y luego resumir los resultados. La estructura de grafo soporta naturalmente este tipo de orquestación paralela.

Escenarios no adecuados para usar LangGraph:

• Tareas puramente lineales y simples. Por ejemplo, "traducir→pulir→generar salida". Con LCEL, prompt | llm | parser se resuelve en una línea. Usar LangGraph sería matar moscas a cañonazos.
• Preguntas y respuestas de una sola vuelta extremadamente sensibles al retraso. La orquestación basada en grafos tiene sobrecarga adicional de programación y gestión de estado. En escenarios simples, no es tan rápido como llamar directamente a la API.
• El equipo aún no domina los fundamentos de LangChain. La curva de aprendizaje de LangGraph es más empinada que la de las Cadenas. Si el equipo ni siquiera ha entendido bien PromptTemplate y LCEL, ir directamente a LangGraph será hacer el doble de esfuerzo para la mitad del resultado.