15. Construye tu primer grafo con LangGraph

En la Parte IV, definimos herramientas, las conectamos a un LLM y completamos un bucle de agente que repite el ciclo de decidir y ejecutar. Impulsar el bucle, ejecutar herramientas cuando el LLM las solicitaba, saber cuándo parar: codificamos cada parte de ese flujo a mano.

En este capítulo, construiremos el mismo agente de una forma completamente distinta. En lugar de escribir el flujo directamente, registraremos pasos (nodos) y reglas de conexión (edges) con el framework LangGraph y dejaremos que él gestione la ejecución. El comportamiento es idéntico al del Capítulo 14, pero la forma en que lo construimos cambia.

Este capítulo cubre cuatro conceptos fundamentales —StateGraph, nodos, edges y State— y luego refactoriza el bucle de agente del Capítulo 14 en un grafo de LangGraph. En los capítulos siguientes, el Capítulo 16 cubre el enrutamiento condicional y los componentes prediseñados, y el Capítulo 17 cubre la persistencia de estado que permite a un agente reanudar desde donde fue interrumpido.

15.1) ¿Por qué grafos?

15.1.1) Limitaciones del bucle de agente existente

Repasemos el bucle de agente del Capítulo 14. Dejando de lado el manejo de errores y otros detalles, la estructura central se veía así:

python

# Bucle de agente del Capítulo 14 — estructura central (simplificada)
messages = [
    SystemMessage(content="You are a helpful assistant."),
    HumanMessage(content=user_input),
]
 
for step in range(max_steps):
    # Pide al LLM que decida la siguiente acción
    ai_message = llm_with_tools.invoke(messages)
    messages.append(ai_message)
 
    # Si no se solicita ninguna llamada a herramienta, devuelve la respuesta final
    if not ai_message.tool_calls:
        return ai_message.content
 
    # Ejecuta las herramientas solicitadas
    for tool_call in ai_message.tool_calls:
        selected_tool = tool_map[tool_call["name"]]
        tool_message = selected_tool.invoke(tool_call)
        messages.append(tool_message)

Este código cubre solo lo básico y nada más. Sin embargo, en un entorno de producción real, se requiere mucho más. Aquí tienes algunos ejemplos.

Recuperación ante fallos — Si un agente falla en el paso 7 de una tarea de investigación de 10 pasos, debería poder reanudar desde el paso 7 en lugar de empezar de nuevo desde el principio.
Solicitudes de aprobación — Antes de que un agente realice una operación crítica, debería poder pausar y preguntar a un humano "¿Está bien continuar?".
Monitoreo en tiempo real — Los usuarios deberían poder ver qué está haciendo el agente actualmente y qué herramientas está llamando.
Visualización y depuración — Debería haber disponible un diagrama que muestre cómo opera el agente, para que cuando surjan problemas puedas rastrear qué paso salió mal.

Implementar estas funciones tú mismo no es imposible, pero tampoco es fácil. Solo la recuperación ante fallos requiere escribir código para serializar el estado en cada paso, guardarlo en disco, restaurarlo y reanudar en la posición exacta. Podrías terminar con más código de infraestructura que de lógica de negocio.

LangGraph fue construido para proporcionar estas funciones a nivel de framework. Recuperación ante fallos, solicitudes de aprobación, monitoreo, visualización: el framework se encarga de todas ellas. Pero hay un requisito: tienes que construir tu agente en una estructura que el framework pueda entender.

El bucle de agente del Capítulo 14 maneja toda la lógica directamente, así que no hay nada a lo que el framework pueda engancharse. Para aprovechar lo que ofrece LangGraph, necesitamos reconstruir el agente en una estructura que LangGraph entienda: un grafo. De eso trata este capítulo.

15.1.2) ¿Qué es LangGraph?

LangGraph es un framework de orquestación que define y ejecuta flujos de trabajo de agentes como grafos. Aquí, un grafo significa una estructura donde cada nodo (paso) que el agente realiza está conectado por edges (reglas de conexión).

En LangGraph, divides el flujo de trabajo en nodos independientes y los conectas con edges. LangGraph luego recorre el grafo, ejecutando cada nodo a lo largo del camino. Así se ve el bucle de agente del Capítulo 14 expresado como un grafo:

Los recuadros rectangulares son nodos, y las flechas son edges. El rombo representa un edge condicional que se bifurca hacia diferentes caminos según una condición.

En el Capítulo 14, todo el flujo de trabajo vivía en bucles for, comprobaciones if y otro código escrito a mano. Con LangGraph, defines qué hace cada nodo y conectas los nodos entre sí con edges. En resumen, pasas de codificar el flujo de trabajo a declararlo como estructura.

LangGraph no reemplaza nada de lo que aprendiste en los Capítulos 12-14. Definiciones de herramientas, bind_tools(), tool_calls, ToolMessage: todo esto se sigue usando dentro de los nodos, exactamente como antes.

La siguiente sección cubre los componentes centrales de LangGraph —StateGraph, State, nodos y edges— uno por uno.

15.2) Componentes de LangGraph: StateGraph, State, nodos, edges

Esta sección recorre los cuatro componentes centrales de LangGraph uno por uno. Empezaremos con StateGraph —la clase que agrupa State, nodos y edges en un grafo— y luego cubriremos cada una de las partes (State, nodos, edges) que van dentro.

15.2.1) StateGraph

StateGraph es la clase que se usa para construir grafos en LangGraph. Especificas el State que el grafo gestionará, agregas nodos, los conectas con edges y luego compilas para producir un grafo ejecutable.

Veamos cómo funciona.

Crear una instancia de StateGraph

Llama al constructor StateGraph para crear una instancia. Necesitas pasar el esquema del State (la clase en sí) como parámetro. Aquí usamos MessagesState, un State predefinido que LangGraph proporciona para gestionar listas de mensajes. Cubriremos los detalles en 15.2.2.

python

from langgraph.graph import StateGraph, MessagesState
 
builder = StateGraph(MessagesState)

Agregar nodos

Usa add_node() para registrar un nodo. Un nodo es una función de Python que toma el State actual y devuelve las partes que quiere cambiar. Cubriremos las funciones de nodo en detalle en 15.2.3.

python

def say_hello(state: MessagesState):
    return {"messages": [{"role": "ai", "content": "hello world"}]}
 
builder.add_node(say_hello)    # el nombre del nodo pasa a ser "say_hello"

Conectar edges

Usa add_edge(source, target) para conectar nodos. source es donde empieza el edge; target es a dónde va. START y END son marcadores especiales para los puntos de entrada y salida del grafo. Cubriremos los edges en 15.2.4.

python

from langgraph.graph import START, END
 
builder.add_edge(START, "say_hello")   # el grafo empieza → ejecuta say_hello
builder.add_edge("say_hello", END)     # say_hello termina → finaliza el grafo

Compilar y ejecutar

Llamar a compile() valida la estructura del grafo y produce un objeto ejecutable. Ejecutas el grafo compilado con invoke(), pasando los valores iniciales del State.

python

graph = builder.compile()
 
initial_state = {"messages": [{"role": "user", "content": "hi!"}]}
result = graph.invoke(initial_state)

Ahora juntemos todo y construyamos un grafo simple:

python

from langgraph.graph import StateGraph, MessagesState, START, END
 
def say_hello(state: MessagesState):
    return {"messages": [{"role": "ai", "content": "hello world"}]}
 
builder = StateGraph(MessagesState)
 
builder.add_node(say_hello)
builder.add_edge(START, "say_hello")
builder.add_edge("say_hello", END)
 
graph = builder.compile()
 
initial_state = {"messages": [{"role": "user", "content": "hi!"}]}
result = graph.invoke(initial_state)
print(result["messages"][-1].content)

Salida:

hello world

Cuando llamas a invoke(), el grafo se ejecuta en el orden START → say_hello → END. say_hello devolvió un diccionario con messages como clave, y ese valor se agregó a la lista messages en MessagesState. Profundizaremos en cómo funciona esto en 15.2.2. El resultado es que al extraer el contenido del último mensaje obtenemos "hello world".

15.2.2) State: los datos que fluyen a través del grafo

State son los datos que cada nodo del grafo comparte. Cuando un nodo se ejecuta, recibe el State actual, hace su trabajo y devuelve solo las partes que quiere cambiar. LangGraph incorpora esos cambios de nuevo en el State y entrega la versión actualizada al siguiente nodo.

Definir State

Defines State creando una subclase de TypedDict. Elige los campos y tipos que coincidan con lo que tu agente necesita rastrear. Aquí tienes un ejemplo simple:

python

from typing_extensions import TypedDict
 
class AgentState(TypedDict):
    messages: list       # lista de mensajes
    llm_calls: int       # número de llamadas al LLM

A partir de aquí, pasas AgentState al crear un StateGraph y lo usas como sugerencia de tipo para tus funciones de nodo.

Reducers

Cuando un nodo devuelve un valor, el campo correspondiente del State se actualiza. El comportamiento por defecto es sobrescribir: si un nodo devuelve {"llm_calls": 3}, llm_calls simplemente pasa a ser 3, sin importar lo que fuera antes.

Pero algunos campos necesitan añadir, no sobrescribir. ¿Qué pasa si messages se sobrescribe? Cada vez que un nodo devuelve un mensaje nuevo, todo el historial de conversación desaparece. Para messages, añadir es el comportamiento correcto.

LangGraph te permite establecer una estrategia de actualización diferente por campo a través de una función reducer. Especificas el reducer como el segundo argumento en Annotated:

python

from typing_extensions import TypedDict, Annotated
from langgraph.graph.message import add_messages
 
class AgentState(TypedDict):
    messages: Annotated[list, add_messages]   # reducer: añadir
    llm_calls: int                            # sin reducer: sobrescribir

add_messages es un reducer proporcionado por LangGraph. En lugar de reemplazar la lista, añade los mensajes nuevos a lo que ya hay. Como este reducer está establecido en el campo messages, cualquier valor que un nodo devuelva para messages se añade. llm_calls no tiene reducer, así que los valores devueltos simplemente sobrescriben lo que hubiera antes.

Esto es exactamente por lo que el mensaje que say_hello devolvió en 15.2.1 se añadió a messages en lugar de reemplazarlo: el reducer se encargó de ello.

MessagesState

LangGraph incluye un State predefinido llamado MessagesState. Así se ve por dentro:

python

class MessagesState(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]

La misma estructura que acabamos de cubrir: un campo messages con el reducer add_messages ya conectado.

Si necesitas campos adicionales, simplemente crea una subclase:

python

from langgraph.graph import MessagesState
 
class AgentState(MessagesState):
    llm_calls: int  # comportamiento de sobrescritura por defecto

15.2.3) Nodos: funciones que actualizan el State

Un nodo es una función de Python que realiza una única tarea específica dentro del grafo.

python

def say_hello(state: MessagesState):
    return {"messages": [{"role": "ai", "content": "hello world"}]}

Dos cosas que hay que saber al escribir funciones de nodo:

Regla 1: recibe el State actual como su argumento. LangGraph pasa el objeto State actual cuando ejecuta el nodo.

Regla 2: devuelve solo las partes que quiere cambiar, no el State completo. Los nodos no modifican el State directamente. Simplemente devuelve los campos que quieres actualizar, y LangGraph los fusiona en el State existente según las reglas del reducer de cada campo.

Usa add_node() para agregar un nodo al StateGraph:

python

builder.add_node(say_hello)          # el nombre de la función "say_hello" pasa a ser el nombre del nodo
builder.add_node("my_node", my_func) # también puedes especificar el nombre explícitamente

15.2.4) Edges: reglas que conectan nodos

Un edge determina "después de que este nodo termine, ¿qué se ejecuta a continuación?". Hay dos tipos.

Edges normales

Un edge normal conecta un "ir a" fijo entre dos nodos. Usa add_edge(source, target): source es el nodo de inicio, target es el destino.

python

builder.add_edge(START, "say_hello")       # cuando el grafo empieza, ejecuta say_hello
builder.add_edge("say_hello", "llm_call")  # después de say_hello, ejecuta llm_call
builder.add_edge("llm_call", END)          # después de llm_call, finaliza el grafo

Edges condicionales

Un edge condicional elige el siguiente nodo en tiempo de ejecución según el State actual. Usa add_conditional_edges(source, routing_function): source es el nodo de inicio, y routing_function es una función que toma el State actual y devuelve el nombre del siguiente nodo:

python

from langgraph.graph import END
 
def should_continue(state: AgentState):
    last_message = state["messages"][-1]
    if last_message.tool_calls:
        return "tool_node"   # llamada a herramienta solicitada → ir a tool_node
    return END               # sin llamada a herramienta → finalizar

add_conditional_edges("llm_call", should_continue) le dice a LangGraph: "cuando llm_call termine, llama a should_continue para decidir qué se ejecuta a continuación". should_continue enruta a "tool_node" si el último mensaje tiene tool_calls, o a END si no. En la práctica, eso significa que el grafo continúa al nodo de ejecución de herramientas cuando el LLM solicita una llamada a herramienta, y termina cuando no lo hace.

Ahora que hemos cubierto los cuatro componentes, la siguiente sección los usa para refactorizar el bucle de agente del Capítulo 14 en un grafo de LangGraph.

15.3) Refactorizar el bucle de agente en un grafo

Reconstruyamos el bucle de agente del Capítulo 14 usando LangGraph. El comportamiento es idéntico al del Capítulo 14: el LLM decide, las herramientas se ejecutan según las solicitudes del LLM, y el ciclo se repite hasta completarse. Lo único que cambia es cómo estructuramos este flujo.

Así se verá el grafo terminado:

El grafo alterna entre llm_call y tool_node hasta que el LLM deja de solicitar llamadas a herramientas, momento en el que sale a END. Construyámoslo paso a paso.

15.3.1) Definir el State

Creamos una subclase de MessagesState de 15.2.2 para definir el State del agente. El campo messages se hereda de MessagesState, y agregamos un campo llm_calls para rastrear el número de llamadas al LLM.

python

from langgraph.graph import MessagesState
 
class AgentState(MessagesState):
    llm_calls: int    # número de llamadas al LLM (sobrescribir)

La lista messages acumulará las entradas del usuario (HumanMessage), las respuestas del LLM (AIMessage) y los resultados de la ejecución de herramientas (ToolMessage) en orden.

15.3.2) Construir los nodos

Primero, configuremos las herramientas y el modelo del Capítulo 14:

python

from langchain_openai import ChatOpenAI
from langchain_core.messages import ToolMessage
from langchain.tools import tool
 
@tool
def get_weather(city: str) -> str:
    """Obtiene el clima actual de una ciudad."""
    fake_data = {"Tokyo": "18°C, cloudy", "Cairo": "31°C, sunny"}
    return fake_data.get(city, f"No weather data for {city}.")
 
@tool
def calculate(expression: str) -> str:
    """Calcula una expresión aritmética simple. Ejemplo: '3 * 21'."""
    return str(eval(expression))  # Advertencia: eval() es un riesgo de seguridad. No usar en producción.
 
tools = [get_weather, calculate]
tool_map = {t.name: t for t in tools}
 
llm = ChatOpenAI(model="gpt-5-mini")
model_with_tools = llm.bind_tools(tools)

Ahora escribamos las dos funciones de nodo.

Nodo llm_call — Llama al LLM y devuelve la respuesta:

python

def llm_call(state: AgentState):
    """Llama al LLM y devuelve la respuesta."""
    response = model_with_tools.invoke(state["messages"])
    return {
        "messages": [response],
        "llm_calls": state.get("llm_calls", 0) + 1,
    }

model_with_tools.invoke() llama al LLM, y la respuesta se empaqueta bajo la clave messages en el diccionario de retorno. El reducer la añade a los messages existentes en AgentState. llm_calls devuelve el conteo actual más 1, sobrescribiendo el valor anterior.

Nodo tool_node — Ejecuta las herramientas solicitadas por el LLM y devuelve los resultados:

python

def tool_node(state: AgentState):
    """Ejecuta las herramientas solicitadas por el LLM."""
    last_message = state["messages"][-1]
    results = []
    for tool_call in last_message.tool_calls:
        selected_tool = tool_map[tool_call["name"]]
        tool_message = selected_tool.invoke(tool_call)
        results.append(tool_message)
    return {"messages": results}

Como tool_node siempre se ejecuta justo después de llm_call, el último mensaje en messages está garantizado que es el AIMessage que el LLM acaba de producir. El campo tool_calls de ese mensaje contiene las llamadas a herramientas que el LLM solicitó. El nodo ejecuta cada herramienta, recopila los resultados en results y los devuelve bajo la clave messages: el reducer se encarga de añadirlos a la lista existente.

15.3.3) Edge condicional

Una vez que llm_call termina, necesitamos un edge condicional para decidir si ejecutar tool_node o finalizar el grafo. Esto sigue el mismo patrón de 15.2.4:

python

from typing import Literal
from langgraph.graph import END
 
def should_continue(state: AgentState) -> Literal["tool_node", "__end__"]:
    """Decide si ejecutar herramientas o finalizar el grafo."""
    last_message = state["messages"][-1]
    if last_message.tool_calls:
        return "tool_node"
    return END

Si last_message.tool_calls está presente, el LLM está pidiendo una llamada a herramienta, así que enrutamos a tool_node. De lo contrario, enrutamos a END y el grafo termina.

La sugerencia de tipo de retorno Literal["tool_node", "__end__"] declara los posibles destinos que esta función puede devolver. LangGraph necesita esta sugerencia para dibujar correctamente los caminos de edges condicionales en las visualizaciones del grafo. No tiene efecto sobre el comportamiento en tiempo de ejecución.

"__end__" es el valor de cadena subyacente de END. Como Literal solo acepta literales de cadena, escribimos "__end__" en lugar de END.

15.3.4) Ensamblar y ejecutar el grafo

Es hora de conectar todo. Ensamblemos el State, los nodos y el edge condicional en un StateGraph y compilemos:

python

from langgraph.graph import StateGraph, START, END
 
builder = StateGraph(AgentState)
 
builder.add_node("llm_call", llm_call)
builder.add_node("tool_node", tool_node)
 
builder.add_edge(START, "llm_call")                         # inicio → llm_call
builder.add_conditional_edges("llm_call", should_continue)  # llm_call → tool_node o END
builder.add_edge("tool_node", "llm_call")                   # tool_node → llm_call (bucle)
 
agent = builder.compile()

El edge desde tool_node de vuelta a llm_call crea un bucle. La ejecución sigue en ciclo hasta que el LLM responde con una respuesta final en lugar de solicitar otra llamada a herramienta, momento en el que el bucle sale.

Ejecutémoslo:

python

from langchain_core.messages import HumanMessage
 
result = agent.invoke({
    "messages": [HumanMessage(content="Get the temperature in Cairo, then multiply the number by 3.")],
    "llm_calls": 0,
})
 
print(result["messages"][-1].content)
print(f"\nTotal LLM calls: {result['llm_calls']}")

Salida:

Temperatura actual en El Cairo: 31°C. Multiplicada por 3 = 93.
 
Total de llamadas al LLM: 3

El agente llamó a get_weather("Cairo"), vio el resultado, llamó a calculate("31 * 3") y produjo la respuesta final: el mismo resultado que obtuvimos en el Capítulo 14.

Inspeccionar el historial completo de mensajes muestra cada paso registrado en messages, en orden:

python

for message in result["messages"]:
    message.pretty_print()

Salida:

================================ Human Message =================================
Get the temperature in Cairo, then multiply the number by 3.
================================== Ai Message ==================================
Tool Calls:
  get_weather (call_DiL9WF)
  Args:
    city: Cairo
================================= Tool Message =================================
Name: get_weather
31°C, sunny
================================== Ai Message ==================================
Tool Calls:
  calculate (call_wa6RqWST)
  Args:
    expression: 31 * 3
================================= Tool Message =================================
Name: calculate
93
================================== Ai Message ==================================
Current temperature in Cairo: 31°C. Multiplied by 3 = 93.

15.3.5) Visualización del grafo

En un notebook de Jupyter, agent.get_graph().draw_mermaid_png() renderiza la estructura del grafo como una imagen directamente en la salida de la celda.

python

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

En un entorno de terminal, guárdalo como un archivo PNG en su lugar.

python

agent.get_graph().draw_mermaid_png(output_file_path="agent_graph.png")

Imagen generada:

Las líneas continuas son edges normales y las líneas punteadas son edges condicionales. Este diagrama se genera automáticamente a partir del código.

15.3.6) Límite de recursión

Igual que usamos max_steps para protegernos de los bucles infinitos en el Capítulo 14, LangGraph tiene una red de seguridad integrada. Cada vez que un nodo se ejecuta durante la ejecución del grafo, un contador interno aumenta en uno. Cuando ese contador supera el límite configurado, LangGraph lanza un GraphRecursionError.

Para ver cómo funciona el conteo, mira la ejecución anterior. Llamar tanto a get_weather como a calculate visitó los nodos en este orden:

llm_call(1) → tool_node(2) → llm_call(3) → tool_node(4) → llm_call(5) → END

Eso son 5 visitas a nodos en total. Si estableces recursion_limit en 3, el límite entra en acción en la 3ª visita y la ejecución se corta:

python

from langgraph.errors import GraphRecursionError
 
try:
    result = agent.invoke(
        {"messages": [HumanMessage(content="Get the temperature in Cairo, then multiply the number by 3.")],
         "llm_calls": 0},
        config={"recursion_limit": 3},
    )
except GraphRecursionError:
    print("Agent hit the recursion limit — stopping execution.")

Salida:

El agente alcanzó el límite de recursión — deteniendo la ejecución.

Establece el límite pasando config={"recursion_limit": número} a invoke(). El valor correcto depende de tu caso de uso y de la complejidad de tu grafo. Empieza con un número generoso y ajústalo mediante pruebas.