Diseña grafos con bifurcaciones condicionales

Concepto Clave: El Nodo Condicional como Guardián del Flujo

Imagina que estás construyendo un agente de servicio al cliente. El flujo no puede ser el mismo si un usuario está saludando, haciendo una pregunta específica sobre un producto, o presentando una queja. Un nodo condicional actúa como el guardián en una encrucijada. Su trabajo no es atender al cliente directamente, sino observar la interacción (el estado de la conversación) y decidir por qué puerta debe pasar la ejecución: ¿al módulo de saludos, al buscador de información de productos, o al sistema de gestión de incidencias? Esta decisión se toma en tiempo de ejecución, basándose en datos concretos.

Una analogía técnica más precisa es la de una sentencia switch o una serie de sentencias if/elif/else en programación tradicional. Sin embargo, en LangGraph, esta lógica se eleva a ser un ciudadano de primera clase dentro del grafo mismo, visible y manipulable como cualquier otro nodo. Esto permite una depuración y un seguimiento más claros, ya que puedes ver exactamente en qué punto del grafo se tomó una decisión y hacia dónde se dirigió el flujo. El estado del grafo es la única fuente de verdad que el nodo condicional consulta, promoviendo un diseño limpio y desacoplado.

Es crucial diferenciar entre la condición (la lógica que decide el camino) y la acción (el trabajo realizado en ese camino). Mantener esta separación es un principio de buen diseño en LangGraph. Un error común es intentar que un solo nodo both decida y ejecute, lo que lleva a código espagueti y grafos difíciles de razonar. Al aislar la lógica condicional, creas módulos reutilizables y un grafo cuyo comportamiento es fácil de predecir y modificar.

Tip del Instructor: Piensa en tu nodo condicional como un semáforo inteligente. No construye la carretera ni conduce los coches; solo lee el tráfico (el estado) y muestra la luz verde hacia la dirección correcta. Su simplicidad es su fortaleza.

Cómo Funciona en la Práctica: Un Ejemplo Paso a Paso

Vamos a diseccionar el proceso de creación y uso de una bifurcación condicional en un escenario práctico: un agente analizador de sentimientos que decide una respuesta basada en la emotividad de un mensaje del usuario. El grafo tendrá tres caminos posibles: para sentimiento positivo, negativo o neutro.

Paso 1: Definir el Estado. Todo comienza con el estado. Definimos una clase que herede de TypedDict o utilice StateGraph. Debe contener todos los datos que el nodo condicional necesitará para tomar su decisión. En nuestro caso, necesitamos el mensaje_usuario y quizás un campo sentimiento_analizado que será poblado por un nodo previo.

Paso 2: Crear el Nodo Condicional. Definimos una función, por ejemplo router, que tome el estado como argumento. Dentro, implementamos la lógica (usando un modelo de LLM o un analizador heurístico) para devolver una cadena de texto que corresponda al nombre del próximo nodo. Esta función se decora con @conditional_edge y se vincula como una "rama" del grafo que sale de un nodo anterior.

Paso 3: Definir los Nodos de Acción. Creamos las funciones que representan los caminos: respuesta_positiva, respuesta_neutra, respuesta_negativa. Cada una recibe el estado, realiza su tarea específica (como generar una respuesta empática o ofrecer ayuda), y devuelve el estado actualizado.

Paso 4: Ensamblar el Grafo. Construimos el grafo añadiendo todos los nodos (acciones y condicional). Luego, definimos los bordes. El borde clave es el condicional: especificamos que desde un nodo (ej. analizar_sentimiento), el siguiente paso no es fijo, sino que está determinado por la función router. Finalmente, añadimos los bordes normales desde el nodo condicional hacia cada uno de los nodos de acción, usando los nombres que devuelve la función router.

Paso 5: Ejecución. Al invocar el grafo, el flujo llegará al nodo que precede al condicional. LangGraph llamará entonces a la función router, pasándole el estado actual. Dependiendo de la cadena devuelta (ej. "positivo"), el motor de ejecución saltará directamente al nodo correspondiente (respuesta_positiva), ignorando por completo los otros caminos. La ejecución continúa desde allí.

Código en Acción: Agente de Clasificación y Enrutamiento Complejo

A continuación, un ejemplo completo y funcional de un agente que recibe una consulta, clasifica su intención y enruta a herramientas especializadas. Este patrón es fundamental para arquitecturas de agente multi-herramienta.

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

# 1. DEFINICIÓN DEL ESTADO
class State(TypedDict):
    consulta_usuario: str
    intencion_clasificada: Literal["busqueda", "calculo", "charla", "desconocida"] | None
    respuesta_final: str
    historico: list[str]

# 2. NODOS DE ACCIÓN (Herramientas Especializadas)
def nodo_clasificar_intencion(state: State) -> State:
    """Nodo que usa un LLM para clasificar la intención."""
    from langchain_community.chat_models import ChatOpenAI  # Ejemplo
    llm = ChatOpenAI(model="gpt-4o-mini")
    prompt = f"""Clasifica la intención de esta consulta del usuario en una de estas categorías: 'busqueda', 'calculo', 'charla' o 'desconocida'.
    Consulta: {state['consulta_usuario']}
    Devuelve solo la palabra de la categoría, nada más."""
    categoria = llm.invoke(prompt).content.strip().lower()
    state['intencion_clasificada'] = categoria
    state['historico'].append(f"Intención clasificada como: {categoria}")
    return state

def nodo_buscar_informacion(state: State) -> State:
    """Simula una búsqueda en base de datos o web."""
    state['respuesta_final'] = f"[BUSQUEDA] He encontrado resultados sobre: '{state['consulta_usuario']}'. Aquí están los detalles simulados."
    state['historico'].append("Ejecutada herramienta de búsqueda.")
    return state

def nodo_ejecutar_calculo(state: State) -> State:
    """Simula un cálculo matemático o de datos."""
    # En un caso real, aquí se llamaría a una herramienta como una calculadora o código Python.
    state['respuesta_final'] = f"[CALCULO] He procesado la operación contenida en '{state['consulta_usuario']}'. El resultado simulado es 42."
    state['historico'].append("Ejecutada herramienta de cálculo.")
    return state

def nodo_conversar(state: State) -> State:
    """Camino para conversación general cuando no se necesita una herramienta."""
    from langchain_community.chat_models import ChatOpenAI
    llm = ChatOpenAI(model="gpt-4o-mini")
    respuesta = llm.invoke(f"El usuario dice: {state['consulta_usuario']}. Responde de manera amable y conversacional.")
    state['respuesta_final'] = f"[CHARLA] {respuesta.content}"
    state['historico'].append("Manejado como conversación general.")
    return state

def nodo_intencion_desconocida(state: State) -> State:
    """Camino para manejar intenciones no reconocidas."""
    state['respuesta_final'] = f"[DESCONOCIDA] No estoy seguro de cómo ayudarte con: '{state['consulta_usuario']}'. ¿Podrías reformularlo?"
    state['historico'].append("Intención no reconocida, se solicita aclaración.")
    return state

# 3. NODO CONDICIONAL (El Enrutador)
def router_por_intencion(state: State) -> str:
    """Función condicional que decide el camino basado en la intención clasificada."""
    intencion = state.get('intencion_clasificada')
    if not intencion:
        return "error"  # Podríamos tener un nodo de manejo de errores

    # Mapeo directo de la categoría al nombre del nodo destino.
    mapeo = {
        "busqueda": "buscar_info",
        "calculo": "ejecutar_calculo",
        "charla": "conversar",
        "desconocida": "manejar_desconocido"
    }
    # Devuelve el nombre del próximo nodo. Si la categoría no está en el mapa, va a 'manejar_desconocido'.
    return mapeo.get(intencion, "manejar_desconocido")

# 4. CONSTRUCCIÓN DEL GRAFO
workflow = StateGraph(State)

# Añadir todos los nodos de acción
workflow.add_node("clasificar", nodo_clasificar_intencion)
workflow.add_node("buscar_info", nodo_buscar_informacion)
workflow.add_node("ejecutar_calculo", nodo_ejecutar_calculo)
workflow.add_node("conversar", nodo_conversar)
workflow.add_node("manejar_desconocido", nodo_intencion_desconocida)

# Establecer el punto de entrada
workflow.set_entry_point("clasificar")

# Añadir el borde CONDICIONAL desde 'clasificar'.
# LangGraph llamará a `router_por_intencion` después de `clasificar` para decidir el próximo nodo.
workflow.add_conditional_edges(
    "clasificar",
    router_por_intencion,
    {
        "buscar_info": "buscar_info",
        "ejecutar_calculo": "ejecutar_calculo",
        "conversar": "conversar",
        "manejar_desconocido": "manejar_desconocido",
        # "error": "nodo_de_error"  # Se podría añadir otro camino
    }
)

# Añadir bordes regulares desde los nodos de acción hacia el final.
workflow.add_edge("buscar_info", END)
workflow.add_edge("ejecutar_calculo", END)
workflow.add_edge("conversar", END)
workflow.add_edge("manejar_desconocido", END)

# Compilar el grafo
app = workflow.compile()

# 5. EJECUCIÓN
print("=== Ejecución 1: Consulta de Búsqueda ===")
resultado1 = app.invoke({"consulta_usuario": "¿Cuál es la capital de Francia?", "historico": [], "respuesta_final": ""})
print(f"Intención: {resultado1['intencion_clasificada']}")
print(f"Respuesta: {resultado1['respuesta_final']}")
print(f"Histórico: {resultado1['historico']}")
print()

print("=== Ejecución 2: Consulta de Cálculo ===")
resultado2 = app.invoke({"consulta_usuario": "¿Cuánto es 15 * 28?", "historico": [], "respuesta_final": ""})
print(f"Intención: {resultado2['intencion_clasificada']}")
print(f"Respuesta: {resultado2['respuesta_final']}")
print()

print("=== Ejecución 3: Conversación General ===")
resultado3 = app.invoke({"consulta_usuario": "Hola, ¿cómo estás hoy?", "historico": [], "respuesta_final": ""})
print(f"Intención: {resultado3['intencion_clasificada']}")
print(f"Respuesta: {resultado3['respuesta_final']}")

Este grafo demuestra un patrón poderoso: un nodo (clasificar) prepara el estado, y un enrutador condicional (router_por_intencion) actúa como un despachador central que envía el flujo a una herramienta especializada. Es escalable; añadir una nueva herramienta implica crear su nodo de acción y añadir una entrada al mapeo en el router. El estado mantiene un historial de las decisiones tomadas, lo que es invaluable para depurar y auditar el comportamiento del agente.

Patrones Avanzados: Subgrafos y Bifurcaciones Anidadas

Las bifurcaciones condicionales pueden usarse para crear estructuras de control complejas, como bucles y subgrafos anidados. Un patrón común es el ciclo de reflexión y corrección. Imagina un agente que genera un borrador de código, luego un nodo condicional evalúa si el código pasa ciertas pruebas (sintaxis, lógica simple). Si pasa, el flujo continúa hacia la salida; si falla, el flujo se redirige de vuelta al nodo generador con instrucciones de corrección, creando un bucle hasta que se cumpla la condición de salida.

# Ejemplo esquemático de un bucle condicional
from langgraph.graph import StateGraph, END
from typing import TypedDict

class StateCiclo(TypedDict):
    tarea: str
    intentos: int
    resultado: str
    es_correcto: bool

def generar_respuesta(state: StateCiclo) -> StateCiclo:
    # Simula generar una respuesta
    state['resultado'] = f"Intento {state['intentos']}: Respuesta generada para '{state['tarea']}'"
    state['intentos'] += 1
    return state

def evaluar_calidad(state: StateCiclo) -> StateCiclo:
    # Simula una evaluación. En la práctica, podría ser un LLM o un validador.
    # Supongamos que es correcto después de 2 intentos.
    state['es_correcto'] = (state['intentos'] >= 3)  # Condición de salida simple
    return state

def router_ciclo(state: StateCiclo) -> Literal["generar", "fin"]:
    """Router que decide si repetir el ciclo o terminar."""
    if state.get('es_correcto'):
        return "fin"
    else:
        return "generar"

# Construcción del grafo con ciclo
workflow_ciclico = StateGraph(StateCiclo)
workflow_ciclico.add_node("generar", generar_respuesta)
workflow_ciclico.add_node("evaluar", evaluar_calidad)
workflow_ciclico.set_entry_point("generar")
workflow_ciclico.add_edge("generar", "evaluar")
workflow_ciclico.add_conditional_edges(
    "evaluar",
    router_ciclo,
    {
        "generar": "generar",  # Vuelve al inicio -> CICLO
        "fin": END            # Sale del grafo
    }
)
app_ciclico = workflow_ciclico.compile()

Otro patrón avanzado es el uso de subgrafos como destinos de una bifurcación. Puedes compilar un grafo independiente (por ejemplo, un flujo completo de procesamiento de pedidos) y añadirlo como un "supernodo" dentro de un grafo mayor. Un nodo condicional en el grafo principal podría entonces decidir, basándose en el tipo de solicitud del usuario, ejecutar ese subgrafo completo en lugar de un nodo simple. Esto permite una modularidad extrema y la composición de agentes complejos a partir de partes más simples y probadas.

Errores Comunes y Cómo Evitarlos

Al diseñar grafos con bifurcaciones condicionales, varios errores sutiles pueden introducir bugs difíciles de rastrear.

1. Estado Incompleto para la Decisión: El nodo condicional intenta acceder a un campo del estado que no ha sido inicializado o poblado por un nodo anterior. Esto causa un KeyError o una decisión basada en None. Solución: Define claramente el esquema del estado y asegura un orden de ejecución donde los datos necesarios estén disponibles. Usa state.get('clave', valor_por_defecto) para un manejo seguro.

2. Nombres de Nodo Inconsistentes: La función condicional devuelve una cadena (ej. "buscar"), pero al añadir los bordes condicionales con add_conditional_edges, el mapeo usa una clave diferente (ej. {"buscar_info": "nodo_buscar"}). LangGraph no encontrará el destino y lanzará un error. Solución: Usa constantes o enumeraciones para los nombres de los nodos. Asegúrate de que el valor devuelto por el router coincida exactamente con una de las claves en el diccionario de mapeo proporcionado a add_conditional_edges.

3. Condiciones No Exhaustivas (Caminos Perdidos): La lógica del router no cubre todos los casos posibles. Si recibe un valor inesperado, puede devolver None o una cadena no mapeada, causando que el grafo falle. Solución: Implementa siempre una rama por defecto (como "desconocido" o "error") en tu router. Usa una sentencia final else o el método .get() de un diccionario con un valor por defecto.

4. Lógica de Negocio en el Router: Incluir operaciones complejas o llamadas a LLM dentro de la función condicional, haciéndola lenta y difícil de probar. El rol del router es decidir, no calcular. Solución: Mueve cualquier lógica pesada o preparación de datos a un nodo de acción dedicado que se ejecute *antes* del condicional. El router debe ser una función simple, rápida y determinista en la medida de lo posible, que solo consulte el estado ya preparado.

5. Olvidar Puntos de Salida (END) para Todas las Ramas: Creas múltiples caminos desde un condicional, pero algunos de los nodos hoja no tienen un borde hacia END o hacia otro nodo que eventualmente lleve a END. La ejecución se "atasca". Solución: Después de añadir todos los nodos de acción, verifica metódicamente que cada uno tenga un borde de salida definido, ya sea hacia END, hacia otro nodo, o hacia otro conjunto condicional.

Tip de Depuración: Usa app.get_graph().draw_mermaid() para visualizar tu grafo. Los bordes condicionales aparecerán como rombos, permitiéndote verificar visualmente que todos los caminos estén conectados y lleven a un final.

Checklist de Dominio

Antes de considerar que dominas el diseño de grafos con bifurcaciones condicionales, verifica que puedes realizar y comprender cada uno de estos puntos:

• Puedo explicar la diferencia entre un nodo de acción y un nodo condicional (función de enrutamiento) y cuándo usar cada uno.
• He diseñado e implementado un grafo donde un nodo condicional dirige el flujo a al menos tres caminos diferentes basados en el estado.
• Sé cómo estructurar el estado del grafo para que contenga toda la información necesaria para que el nodo condicional tome su decisión.
• Puedo identificar y corregir un error causado por un nombre de nodo destino inconsistente entre el router y el mapeo de add_conditional_edges.
• He implementado un patrón que incluye un ciclo (loop) controlado por una condición, asegurando que exista una condición de salida clara para evitar bucles infinitos.
• Puedo visualizar mi grafo (usando Mermaid o similar) y señalar claramente dónde ocurren las bifurcaciones.
• Sé cómo añadir un camino de "fallback" o manejo de errores en mi lógica condicional para cubrir casos inesperados.
• He separado correctamente la lógica de preparación/computación (en nodos de acción) de la pura lógica de decisión (en el router condicional).