Image for post LangGraph en Acción: Construyendo Workflows IA Multi-Agente con Estado y Enrutamiento Inteligente

LangGraph en Acción: Construyendo Workflows IA Multi-Agente con Estado y Enrutamiento Inteligente

Escrito por Sergio Márquez

En el vertiginoso mundo de la Inteligencia Artificial, las aplicaciones basadas en Grandes Modelos de Lenguaje (LLMs) han evolucionado rápidamente. Sin embargo, a medida que los casos de uso se vuelven más complejos, las cadenas secuenciales simples de LangChain a menudo se quedan cortas. Necesitamos sistemas que puedan tomar decisiones dinámicas, gestionar un estado persistente a lo largo de múltiples interacciones y coordinar el trabajo de varios 'agentes' especializados. Aquí es donde entra en juego LangGraph, una poderosa biblioteca que nos permite construir workflows de IA robustos, con estado y enrutamiento inteligente, utilizando una arquitectura basada en grafos. [2, 10, 17, 21]

Este artículo te guiará a través de los fundamentos de LangGraph, desde sus conceptos clave hasta la implementación práctica de un sistema multi-agente. Al final, tendrás una comprensión clara de cómo diseñar y construir aplicaciones de IA más sofisticadas y adaptativas.

Contexto del Problema: Más Allá de las Cadenas Lineales

Imagina que estás construyendo un asistente de IA. Una cadena simple podría tomar una pregunta, pasarla a un LLM y devolver una respuesta. Pero, ¿qué pasa si la pregunta requiere una búsqueda en la web antes de responder? ¿O si la respuesta del LLM necesita ser validada por otro componente? ¿Y si el usuario quiere refinar su consulta basándose en la respuesta anterior, manteniendo el contexto de la conversación? [10, 17]

Las cadenas lineales de LangChain, aunque excelentes para tareas secuenciales, no manejan bien la lógica condicional, los bucles o la colaboración entre múltiples componentes que necesitan compartir y actualizar información. Aquí es donde los workflows complejos, a menudo llamados sistemas multi-agente, se vuelven esenciales. Necesitamos una forma de orquestar estos componentes, permitiéndoles interactuar, tomar decisiones y mantener un estado compartido a lo largo del tiempo. [1, 2, 3, 8, 10, 15, 17, 19, 21, 23, 34]

Conceptos Clave de LangGraph

LangGraph se basa en la idea de construir aplicaciones de IA como un grafo dirigido, donde cada paso es un 'nodo' y las transiciones entre pasos son 'aristas'. Esto permite una flexibilidad mucho mayor que las cadenas lineales. [17, 20, 22, 34]

  • Grafo Dirigido Acíclico (DAG) y Ciclos: A diferencia de un DAG estricto, LangGraph permite ciclos, lo que es crucial para los agentes que necesitan iterar o volver a pasos anteriores (por ejemplo, para refinar una búsqueda o corregir un error). [3, 10, 19, 21]

  • Nodos (Nodes): Cada nodo es una unidad de procesamiento discreta. Puede ser una llamada a un LLM, una función Python personalizada, una herramienta (como una búsqueda web o una base de datos) o incluso otro agente. Los nodos toman el estado actual como entrada y devuelven un objeto que describe cómo actualizar ese estado. [5, 17, 22, 29]

  • Aristas (Edges): Las aristas definen las transiciones entre nodos, es decir, el flujo de control del grafo. Pueden ser: [5, 17, 22, 29]

    • Directas: Simplemente conectan un nodo con el siguiente en una secuencia fija.
    • Condicionales: Permiten que el flujo del grafo cambie dinámicamente basándose en el estado actual o en el resultado de un nodo. Esto se logra con una función de enrutamiento que decide el siguiente nodo a ejecutar. [18, 24, 26, 28, 29, 33, 34]

  • Estado (State): El estado es el corazón de LangGraph. Es un objeto compartido que contiene toda la información relevante para la ejecución del workflow. Cada nodo puede acceder y modificar este estado. LangGraph utiliza un concepto de 'reductor' (similar a los reductores de Redux) para definir cómo se actualiza el estado de forma inmutable, asegurando la consistencia. [2, 6, 9, 10, 14, 16, 19, 22, 28, 29, 35]

    • Se define típicamente usando TypedDict o un modelo Pydantic para garantizar la seguridad de tipos. [6, 16, 35]
    • Permite mantener el contexto de la conversación, resultados intermedios, decisiones tomadas, etc. [6, 9, 28]

  • Enrutamiento Condicional (Conditional Routing): Es la capacidad de LangGraph para tomar decisiones sobre qué camino seguir en el grafo basándose en el estado actual. Esto es fundamental para construir agentes adaptativos. [18, 24, 26, 28, 33]

  • Agentes Multi-Agente: LangGraph es ideal para construir sistemas donde múltiples agentes especializados colaboran. Cada agente puede ser un nodo (o un subgrafo) con su propio LLM, prompt y herramientas, y LangGraph orquesta su interacción. [1, 2, 8, 15, 23, 27, 32, 34]

Implementación Paso a Paso: Un Asistente de Investigación Simple

Vamos a construir un asistente de investigación simple que pueda decidir si necesita buscar información externa para responder a una consulta. Si la necesita, usará una herramienta de búsqueda (simulada en este ejemplo) y luego generará una respuesta basada en los resultados. [5]

1. Configuración del Entorno

Primero, instala las bibliotecas necesarias:

pip install langchain langchain-openai langgraph

Asegúrate de tener tu clave de API de OpenAI configurada como una variable de entorno:

export OPENAI_API_KEY="tu_clave_openai_aqui"

2. Definición del Estado del Grafo

El estado es crucial. Definiremos un TypedDict para mantener el historial de mensajes, la consulta original, los resultados de la búsqueda y una bandera para indicar si se necesita una búsqueda. [6, 9, 14, 16, 22, 28, 29, 35]

import os
from typing import TypedDict, Annotated, List
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END

class AgentState(TypedDict):
    """
    Representa el estado del grafo.

    Attributes:
        messages: Una lista de mensajes que representan el historial de la conversación.
        query: La consulta original del usuario.
        search_results: Los resultados de la búsqueda externa.
        needs_search: Un booleano que indica si se requiere una búsqueda.
    """
    messages: Annotated[List[BaseMessage], lambda x, y: x + y] # Acumula mensajes
    query: str
    search_results: str
    needs_search: bool

# Inicializamos el LLM. Usaremos un modelo de OpenAI.
# Asegúrate de que OPENAI_API_KEY esté configurada como variable de entorno.
llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)

3. Definición de Herramientas (Tools)

Para este ejemplo, crearemos una herramienta de búsqueda simulada. En una aplicación real, esto podría ser una llamada a la API de Google Search, una base de datos vectorial, etc. [15, 18, 29]

@tool
def buscar_informacion(query: str) -> str:
    """Busca información relevante en una fuente externa."""
    print(f"DEBUG: Ejecutando herramienta buscar_informacion con query: {query}")
    # Simulación de una búsqueda real
    if "LangGraph" in query.lower():
        return "LangGraph es una librería para construir aplicaciones robustas y con estado usando LLMs, basada en LangChain. Permite definir workflows complejos como grafos."
    elif "Inteligencia Artificial" in query.lower():
        return "La Inteligencia Artificial es un campo de la informática que busca crear máquinas capaces de realizar tareas que normalmente requieren inteligencia humana."
    elif "Python" in query.lower():
        return "Python es un lenguaje de programación interpretado, de alto nivel y de propósito general. Es muy popular en IA y desarrollo web."
    else:
        return "No se encontró información relevante para su consulta específica. Intenta con otro tema."

4. Creación de Nodos

Cada nodo será una función Python que toma el estado y devuelve un diccionario con las actualizaciones al estado. [5, 22, 29, 35]

def planificador(state: AgentState):
    """Decide si la consulta requiere una búsqueda externa."""
    print("DEBUG: Entrando en el nodo planificador.")
    messages = state['messages']
    last_message_content = messages[-1].content if messages else ""

    # Lógica simple para decidir si se necesita una búsqueda
    if any(keyword in last_message_content.lower() for keyword in ["buscar", "información sobre", "qué es"]):
        needs_search = True
        response_content = "Entendido, parece que necesitas información. ¿Sobre qué tema específico te gustaría que busque?"
    else:
        needs_search = False
        response_content = "No parece que necesite buscar información externa. Procedo a generar una respuesta directa."

    return {
        "messages": [AIMessage(content=response_content)],
        "query": last_message_content, # Guardamos la consulta original
        "needs_search": needs_search
    }

def ejecutar_busqueda(state: AgentState):
    """Ejecuta la herramienta de búsqueda con la consulta del usuario."""
    print("DEBUG: Entrando en el nodo ejecutar_busqueda.")
    query = state['query']
    if not query:
        # Fallback si la consulta está vacía (no debería ocurrir si el planificador funciona bien)
        query = state['messages'][-1].content if state['messages'] else "tema general"
    
    results = buscar_informacion.invoke(query)
    
    response_content = f"He buscado información sobre '{query}'. Resultados: {results}"
    return {
        "messages": [AIMessage(content=response_content)],
        "search_results": results
    }

def generador_respuesta(state: AgentState):
    """Genera la respuesta final usando el LLM, con o sin resultados de búsqueda."""
    print("DEBUG: Entrando en el nodo generador_respuesta.")
    
    full_context = ""
    if state.get('search_results'):
        full_context += f"Resultados de la búsqueda: {state['search_results']}\n\n"
    
    full_context += f"Consulta original: {state['query']}\n\n"
    full_context += "Basado en la información disponible, genera una respuesta concisa y útil para el usuario."

    try:
        response = llm.invoke(full_context)
        response_message = AIMessage(content=response.content)
    except Exception as e:
        response_message = AIMessage(content=f"Lo siento, hubo un error al generar la respuesta: {e}")
    
    return {"messages": [response_message]}

5. Definición del Enrutamiento Condicional

Esta función decidirá el siguiente nodo basándose en el valor de needs_search en el estado. [18, 24, 26, 28, 33]

def decidir_siguiente_paso(state: AgentState) -> str:
    """Decide si ir a la búsqueda o generar una respuesta directa."""
    print("DEBUG: Entrando en el enrutador decidir_siguiente_paso.")
    if state['needs_search']:
        print("DEBUG: Decisión: Ir a ejecutar_busqueda.")
        return "ejecutar_busqueda"
    else:
        print("DEBUG: Decisión: Ir a generador_respuesta.")
        return "generador_respuesta"

6. Construcción y Compilación del Grafo

Ahora unimos todo para formar el grafo. [2, 5, 22, 29, 33]

# Construimos el grafo
workflow = StateGraph(AgentState)

# Añadimos los nodos
workflow.add_node("planificador", planificador)
workflow.add_node("ejecutar_busqueda", ejecutar_busqueda)
workflow.add_node("generador_respuesta", generador_respuesta)

# Establecemos el punto de entrada
workflow.set_entry_point("planificador")

# Añadimos las aristas condicionales desde el planificador
workflow.add_conditional_edges(
    "planificador",
    decidir_siguiente_paso, # La función que decide el siguiente paso
    {
        "ejecutar_busqueda": "ejecutar_busqueda",
        "generador_respuesta": "generador_respuesta"
    }
)

# Añadimos las aristas directas
workflow.add_edge("ejecutar_busqueda", "generador_respuesta")
workflow.add_edge("generador_respuesta", END) # El nodo END marca el final del workflow

# Compilamos el grafo para crear la aplicación ejecutable
app = workflow.compile()

7. Mini Proyecto / Aplicación Sencilla: Probando el Asistente

Ahora podemos interactuar con nuestro asistente. Observa cómo el flujo cambia según la consulta. [2, 5, 28, 33]

# --- Ejemplos de uso ---

# Ejemplo 1: Consulta que NO requiere búsqueda externa
print("\n--- Ejecución 1: Saludo simple ---")
inputs_1 = {"messages": [HumanMessage(content="Hola, ¿cómo estás?")]}
for s in app.stream(inputs_1):
    print(s)
# Salida esperada: El planificador decide que no necesita búsqueda y va directo al generador de respuesta.

# Ejemplo 2: Consulta que SÍ requiere búsqueda externa
print("\n--- Ejecución 2: Búsqueda de información sobre LangGraph ---")
inputs_2 = {"messages": [HumanMessage(content="Necesito información sobre LangGraph")]}
for s in app.stream(inputs_2):
    print(s)
# Salida esperada: El planificador decide buscar, ejecuta la herramienta y luego genera la respuesta.

# Ejemplo 3: Otra búsqueda
print("\n--- Ejecución 3: Búsqueda sobre Inteligencia Artificial ---")
inputs_3 = {"messages": [HumanMessage(content="¿Qué es la Inteligencia Artificial?")]}
for s in app.stream(inputs_3):
    print(s)

# Ejemplo 4: Consulta sin resultados de búsqueda específicos
print("\n--- Ejecución 4: Búsqueda sin resultados específicos ---")
inputs_4 = {"messages": [HumanMessage(content="Háblame de los unicornios rosados")]}
for s in app.stream(inputs_4):
    print(s)

Errores Comunes y Depuración

Trabajar con grafos puede ser un poco más complejo que con cadenas lineales. Aquí hay algunos errores comunes y consejos para depurar: [11, 31, 36]

  • Errores en la Definición del Estado: Asegúrate de que tu TypedDict (o modelo Pydantic) refleje con precisión todos los campos que tus nodos necesitan leer o escribir. Un campo faltante o un tipo incorrecto puede causar errores inesperados. [6, 9, 14, 16, 35]

  • Bucles Infinitos o Rutas Inesperadas: Si tu grafo entra en un bucle infinito, es probable que tu lógica de enrutamiento condicional no esté cubriendo todos los casos o esté llevando a un nodo que no avanza el estado hacia END. Revisa cuidadosamente las funciones de enrutamiento y las condiciones. [16]

  • Manejo de Excepciones en Nodos: Los nodos son funciones Python. Si una herramienta falla o un LLM devuelve un formato inesperado, tu nodo debe manejarlo con bloques try-except para evitar que todo el grafo se detenga. Puedes actualizar el estado con un mensaje de error y enrutar a un nodo de manejo de errores si es necesario. [16]

  • Visualización del Grafo: LangGraph se integra con herramientas como Graphviz para visualizar el grafo. Esto es increíblemente útil para entender el flujo y depurar. Puedes generar una imagen PNG de tu grafo para ver su estructura. [35]

    # Para visualizar el grafo (requiere graphviz instalado en tu sistema y pydot)
# pip install pydot graphviz
# from IPython.display import Image, display
# display(Image(app.get_graph().draw_png()))

  • LangSmith: Para una depuración y monitoreo avanzados, LangSmith (parte del ecosistema LangChain) es una herramienta invaluable. Permite trazar la ejecución de tu grafo, ver el estado en cada paso, identificar cuellos de botella y depurar errores de manera mucho más eficiente. [7, 11, 35, 37]

Aprendizaje Futuro

Este ejemplo es solo la punta del iceberg de lo que puedes lograr con LangGraph. Aquí hay algunas ideas para llevar tus habilidades al siguiente nivel:

  • Integración con Herramientas Reales: Reemplaza la herramienta buscar_informacion simulada con integraciones reales como la API de Google Search, una base de datos vectorial (Chroma, Pinecone, Qdrant), o APIs de servicios externos. [15, 18, 29]

  • Persistencia del Estado: Para aplicaciones de larga duración o chatbots con memoria, explora las opciones de persistencia del estado de LangGraph, que permiten guardar y cargar el estado del grafo entre ejecuciones. [9, 16, 29]

  • Agentes Más Complejos: Implementa agentes más sofisticados utilizando patrones como ReAct (Reasoning and Acting) dentro de tus nodos. Cada nodo podría ser un agente completo con su propio LLM y conjunto de herramientas. [1, 8, 29]

  • Human-in-the-Loop (HIL): Diseña workflows donde un humano pueda revisar o aprobar ciertas decisiones antes de que el agente continúe, ideal para tareas críticas. [5, 7, 21]

  • Subgrafos: Para workflows muy complejos, puedes anidar grafos dentro de nodos, creando una arquitectura modular y fácil de mantener. [1]

  • Optimización de Costos y Rendimiento: Experimenta con diferentes modelos de LLM para cada nodo (por ejemplo, un modelo más pequeño para la planificación y uno más grande para la generación final) para optimizar costos y latencia. [21]

LangGraph te proporciona el control y la flexibilidad necesarios para construir la próxima generación de aplicaciones de IA. ¡Es hora de empezar a construir!

Volver al blog