LangGraph

1. Introducción

En la entrada anterior de esta serie dedicada a ver cómo desarrollar agentes de IA generativa vimos los componentes core de LangChain. En esta cómo prepararlos con LangGraph.

LangGraph es una extensión de LangChain que permite construir agentes de IA y flujos de trabajo complejos modelándolos como grafos cíclicos.

Un grafo es una estructura matemática y de computación que se usa para modelar relaciones entre elementos y está compuesto por dos partes fundamentales:

1. Nodos o vértices: Son los elementos o puntos. Por ejemplo: personas, ciudades, tareas.
2. Aristas o enlaces: Son las conexiones entre los nodos. Representan la relación que existe entre ellos. Por ejemplo: amistad, rutas de carretera, precedencia o dependencia entre tareas.

Además, los grafos pueden ser:

1. Dirigidos: Las aristas tienen una flecha que indica dirección (por ejemplo, “A sigue a B”).
2. No dirigidos: Las relaciones son bidireccionales (por ejemplo, “A y B son amigos”).
3. Cíclicos: Tienen caminos que vuelven a un nodo ya visitado (incluyen bucles).
4. Acíclicos (DAG): No tienen bucles; son unidireccionales, como las pipelines de CI.

Por ejemplo, el metro de una ciudad puede ser modelado como un grafo no dirigido cíclico, donde las estaciones son los nodos y las rutas entre ellas son las aristas. Desde cada estación puedes ir a otra o viceversa y hay algunas desde las que puedes ir a varias.

En el contexto de agentes de IA, los nodos pueden representar tareas, acciones o estados, y las aristas pueden representar la secuencia o dependencia entre estas tareas. Esto permite modelar flujos de trabajo complejos, donde un agente puede tomar decisiones basadas en el estado actual y las relaciones entre tareas.

Veamos cómo funciona en LangGraph.

2. El grafo más simple del mundo

Para ir viendo cómo funciona, vamos con un primer grafo muy simple, sin LLM, sin tools. Solo dos nodos y estado.

Activamos el entorno virtual e instalamos LangGraph:

source .venv/bin/activate
uv add langgraph

O si usas pip:

pip install langgraph

Entonces, en esencia, un agente de grafos en LangGraph se prepara con los siguientes componentes:

1. Un almacén que servirá como única fuente de verdad entre los distintos nodos. Ahí setearemos la información que irán compartiendo. Para eso se puede usar TypedDict o, mejor aún, pydantic, que viene instalado con langgraph. Algo así, por ejemplo:

from pydantic import BaseModel

# 1. Aquí definimos cualquier parámetro que van a compartir los nodos
class State(BaseModel):
  foo: str

2. Una clase llamada StateGraph, que servirá de constructor del grafo. A esta clase le pasaremos el estado para que sepa qué forma tiene el estado que van a compartir los nodos.

from langgraph.graph import StateGraph

builder = StateGraph(State)

3. N funciones que servirán de nodos. Reciben el estado y devuelven actualizaciones. Por ejemplo, este nodo modifica y devuelve el estado compartido con un holi carcoli.

def node_a(state: State):
    return {"message": state.message + " holi caracoli"}

4. Luego, cada vez que queramos añadir un nodo al circuito, usaremos el método add_node del objeto builder que hemos definido. Así añadiríamos el que acabamos de crear.

builder.add_node("node_a", node_a)

5. Y para conectar los distintos nodos usaremos el método add_edge, que recibe dos parámetros: el origen y el final, es decir, el nodo del que sale y el nodo al que va. Así, por ejemplo, iría del node_a al node_b.

builder.add_edge("node_a", "node_b")

Si queremos que sea bidireccional, cíclico, hay que definir también el camino de vuelta.

builder.add_edge("node_a", "node_b")  # A -> B
builder.add_edge("node_b", "node_a")  # B -> A

Ahora bien, ojo, eso puede crear un bucle infinito. LangGraph lo permite, pero se necesita una arista condicional que en algún momento rompa el ciclo. Sin esa condición de salida, el grafo corre para siempre o hasta que LangGraph lanza un error de recursión límite. Luego vemos eso.

def decide(state: State):
    if state["iterations"] > 3:
        return "end"
    return "node_a"

builder.add_edge("node_a", "node_b")
builder.add_conditional_edges("node_b", decide, {
    "node_a": "node_a",
    "end": END
})

Además, hay dos aristas especiales: la que inicia el bucle, que se denomina START y la que lo termina, que se llama sorprendentemente END.

builder.add_edge(START, "node_a")
builder.add_edge("node_b", END)

6. Por último, una vez que está montado el circuito, ya solo faltaría compilarlo y setear esa compilación en una variable para trabajarla luego.

graph = builder.compile()

7. Y ya estaría listo el agente, que, como todos los componentes de langchain, se iniciaría con el método invoke.

result = graph.invoke({"message": "Inicio"})

Todo junto:

# 1. Importamos todo lo que vamos a necesitar
from pydantic import BaseModel
from langgraph.graph import StateGraph, START, END

# 2. Definimos el estado, el diccionario compartido entre los distintos nodos
class State(BaseModel):
    message: str

# 3. Definimos todos los nodos del circuito agéntico. 
# Recordemos que son meras funciones que reciben el estado y devuelven actualizaciones

def node_a(state: State):
    print("Holi, soy el Nodo A y me he ejecutado")
    return {"message": state.message + "  -> Nodo A"}

def node_b(state: State):
    print("Holi, soy el Nodo B y me he ejecutado")
    return {"message": state.message + "  -> Nodo B"}

# 4. Preparamos el builder
builder = StateGraph(State)

# 5. Añadimos los nodos (todos a la vez)
builder.add_node("node_a", node_a)
builder.add_node("node_b", node_b)

# 6. Definimos el recorrido, empezando por START y terminando por END.
builder.add_edge(START, "node_a")       # entrada -> nodo A
builder.add_edge("node_a", "node_b")    # nodo A -> nodo B
builder.add_edge("node_b", END)         # nodo B -> fin

# 7. Compilamos
graph = builder.compile()

# 8. Ejecutamos
result = graph.invoke({"message": "Inicio"})
print(result["message"])
# -> "Inicio -> Nodo A -> Nodo B"

Plus: LangGraph tiene una herramienta para visualizar los grafos.

from IPython.display import Image, display
display(Image(graph.get_graph().draw_mermaid_png()))

Este sería el que acabamos de preparar.

Vamos con algo más complejo.

3. Dos componentes más

Para desarrollar un agente más sofisticado, en el que intervenga ya un LLM, tenemos que conocer antes otros dos componentes.

a. Uno es init_chat_model, que sirve para inicializar modelos. La sintaxis básica es muy sencilla:

from langchain.chat_models import init_chat_model
llm = init_chat_model("proveedor:modelo")

El string sigue el formato “proveedor:modelo”. LangChain sabe a qué librería llamar por debajo y para el desarrollador es transparente. Cambiar de proveedor es cambiar ese string y nada más.

fast_llm = init_chat_model("anthropic:claude-haiku-4-5")
smart_llm = init_chat_model("anthropic:claude-opus-4-6")

Además, se pueden definir los parámetros habituales del trabajo contra LLMs.

llm = init_chat_model(
    "anthropic:claude-haiku-4-5",
    temperature=0,     # 0 = determinista, ideal para clasificar o extraer datos
                       # 1 = creativo, ideal para generar texto natural
    max_tokens=500,    # límite de tokens en la respuesta
    timeout=30,        # segundos antes de cancelar si no responde
    max_retries=3,     # reintentos automáticos si falla la llamada
)

b. El segundo componente clave es add_conditional_edges. Para entenderlo bien hay que partir de la diferencia con add_edge.

Con add_edge el destino está fijado en tiempo de construcción del grafo, siempre es el mismo, sin importar lo que haya pasado antes. Sin embargo, con add_conditional_edges el destino se decide en tiempo de ejecución, cuando ya se conoce el estado:

builder.add_conditional_edges("funcion_que_decide", valor_que_recibe)

La función de routing es el corazón de esto. Es una función Python normal que recibe el estado y devuelve un string con el nombre del siguiente nodo:

def funcion_que_decide(state: State) -> Literal["node_a", "node_b"]:
    if state.valor_que_recibe == "patatas":
        return "node_a"
    return "node_b"

Tres apuntes más al respecto:

c. El Literal en el tipo de retorno no es obligatorio pero sí muy recomendable. Cumple dos funciones: como en Typescript, le dice al editor exactamente qué valores puede devolver la función -tienes autocompletado y detección de errores- y le dice a LangGraph qué aristas existen, lo que permite que draw_mermaid_png() dibuje el grafo correctamente. Si devuelves un string que no está en el Literal, el grafo falla con un error claro inmediatamente en lugar de comportarse de forma extraña en producción.

• Para indicar que puede ir al final, al END, hay que usar el literal “end”:

def funcion_que_decide(state: State) -> Literal["node_a", "__end__"]:
    if state.valor_que_recibe == "patatas":
        return "node_a"
    return "__end__"

• Y tres, la función de routing no modifica el estado. Solo lo lee y decide. Si intentas modificar el estado dentro de una función de routing, LangGraph lo ignora. Las modificaciones de estado solo ocurren dentro de los nodos.

3. Un agente inteligente

Vamos a ver todo lo anterior preparando un agente que se encargue de responder reseñas. Los pasos que hará son estos:

• 1. START: Recibir una reseña.
• 2. Validar si es correcta en el nodo validate_review.
• 3. Esa validación la recibirá la arista condicional route_validation.
  • 3.1. Si es incorrecta, saldremos del circuito (END).
  • 3.2. Si es correcta, la recibirá el nodo clasify_review.
• 4. El nodo clasify_review valorará si es positiva o negativa.
• 5. Esa valoración la recibirá la arista condicional route_sentiment.
  • 5.1. Si es negativa, la enviará al nodo apologize_customer.
  • 5.2. Si es positiva, la enviará al nodo thank_customer.
• 6. El nodo apologize_customer escribirá una disculpa.
• 7. El nodo thank_customer escribirá en cambio un mensaje de agradecimiento.
• 8. El nodo publish publica el mensaje.
• 9. Termina el circuito agéntico, el workflow: END.

Gráficamente se entiende mejor.

Ok. Pues ya solo faltaría definir alguna api key de algún modelo (ver tutorial de langchain si esto te suena a chino) y ya podríamos preparar el agente. Muestro el código en un solo bloque, que creo que en este caso se sigue mejor la secuencia.

import os
from dotenv import load_dotenv

from pydantic import BaseModel
from typing import Literal

from langgraph.graph import StateGraph, START, END
from langchain.chat_models import init_chat_model


# 1. Leemos el .env y mete las variables en os.environ automáticamente
load_dotenv()

# 2. Definimos el modelo de lenguaje que usaremos en los nodos que lo requieran.
# Podríamos tener diferentes modelos, más listos o menos potentes, para cada nodo. 
# Aquí usamos el mismo para todo por simplicidad.
llm = init_chat_model("anthropic:claude-haiku-4-5")

# 3. Definmos el estado
class State(BaseModel):
    review: str = ""
    author: str = ""
    is_valid: bool = False
    sentiment: str = ""
    response: str = ""
    published: bool = False

# 4. Preparamos los nodos

# Comprueba que la reseña tiene los campos obligatorios. Sin LLM.
def validate_review(state: State):    
    is_valid = bool(state.review.strip()) and bool(state.author.strip())
    return {"is_valid": is_valid}

# "El LLM clasifica la reseña como buena o mala.
def classify_review(state: State):
    result = llm.invoke(f"""
    Clasifica esta reseña como positiva o negativa.
    Responde solo 'good' o 'bad', sin nada más.
    
    Autor: {state.author}
    Reseña: {state.review}
    """)
    return {"sentiment": result.content.strip().lower()}

# El LLM genera una respuesta de agradecimiento.
def thank_customer(state: State):
    result = llm.invoke(f"""
    Eres un gestor de atención al cliente amable y cercano.
    Escribe una respuesta breve agradeciendo a {state.author} por su reseña positiva.
    Reseña: {state.review}
    """)
    return {"response": result.content}

# El LLM genera una respuesta de disculpa.
def apologize_customer(state: State):
    result = llm.invoke(f"""
    Eres un gestor de atención al cliente empático y profesional.
    Escribe una respuesta breve disculpándote con {state.author} por su mala experiencia.
    Reseña: {state.review}
    """)
    return {"response": result.content}

# Publica la respuesta. Sin LLM.
def publish(state: State):
    print(f"Respuesta para {state.author}:")
    print(f"{state.response}")
    return {"published": True}

# 5. Preparamos las aristas condicionales
def route_validation(state: State) -> Literal["classify_review", "__end__"]:
    if state.is_valid:
        return "classify_review"
    return "__end__"

def route_sentiment(state: State) -> Literal["thank_customer", "apologize_customer"]:
    if state.sentiment == "good":
        return "thank_customer"
    return "apologize_customer"

# 6. Construimos el grafo
builder = StateGraph(State)

builder.add_node("validate_review", validate_review)
builder.add_node("classify_review", classify_review)
builder.add_node("thank_customer", thank_customer)
builder.add_node("apologize_customer", apologize_customer)
builder.add_node("publish", publish)

builder.add_edge(START, "validate_review")
builder.add_conditional_edges("validate_review", route_validation)
builder.add_conditional_edges("classify_review", route_sentiment)
builder.add_edge("thank_customer", "publish")
builder.add_edge("apologize_customer", "publish")
builder.add_edge("publish", END)

# 7. Compilamos el grafo
graph = builder.compile()

# Y ahora sí, probamos con diferentes reseñas:

# Probar con reseña mala
result = graph.invoke(State(
    review="El producto llegó roto y el servicio fue horrible.",
    author="Isolda de Bretaña"
))

# Probar con reseña buena
result = graph.invoke(State(
    review="Increíble experiencia, volveré sin duda.",
    author="Bradamante"
))

# Probar con reseña inválida
result = graph.invoke(State(
    review="",
    author=""
))

Bueno, quedarían temas por tratar, como lo subgrafos o la memoria, pero lo dejo para una segunda entrada.