Compilación y Ejecución

Descripción de la cápsula

Has definido estado, nodos y edges. Tienes todos los ingredientes de un grafo. Pero hasta que no compiles, no tienes nada ejecutable — es como tener el código fuente sin compilar. graph.compile() toma tu definición y la convierte en un objeto que puedes ejecutar, hacer streaming, y visualizar.

Esta cápsula cubre las tres fases finales del ciclo de vida de un grafo: compilación (validar y crear el ejecutable), ejecución (las tres formas de correrlo: invoke, stream, y batch), y visualización (draw_mermaid_png() como herramienta de debugging esencial). De las tres, la visualización es la que más vas a subestimar y la que más te va a salvar tiempo. Antes de debuggear código, mira la imagen de tu grafo.

Al terminar esta cápsula, tendrás el flujo completo: definir estado → crear nodos → conectar edges → compilar → ejecutar → visualizar. En el proyecto del módulo (Chatbot con estado y routing), aplicarás todo este ciclo para construir un sistema que clasifica intención y rutea a nodos especializados.

graph.compile(): de definición a ejecutable

Qué hace compile

compile() toma tu StateGraph (la definición) y produce un CompiledGraph (el ejecutable). Durante la compilación, LangGraph:

1. Valida la estructura — Verifica que todos los nodos referenciados en edges existen, que hay un camino desde START, y que el grafo es consistente
2. Crea el ejecutable — Produce un objeto que implementa la interfaz Runnable de LangChain, lo que significa que puedes usar invoke, stream, batch, y sus versiones async
3. Congela la definición — Después de compilar, no puedes agregar más nodos ni edges

Compilación básica

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator
from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage, AnyMessage
from langgraph.graph import StateGraph, START, END

class ChatState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
    step_count: int

model = init_chat_model("openai:gpt-4.1-mini")

def chatbot(state: ChatState) -> dict:
    response = model.invoke(state["messages"])
    return {
        "messages": [response],
        "step_count": state["step_count"] + 1
    }

graph_builder = StateGraph(ChatState)
graph_builder.add_node("chatbot", chatbot)
graph_builder.add_edge(START, "chatbot")
graph_builder.add_edge("chatbot", END)

# Compilar
graph = graph_builder.compile()
print(type(graph))
# Output: <class 'langgraph.graph.state.CompiledStateGraph'>

A partir de este punto, graph es tu objeto ejecutable. graph_builder ya cumplió su función — es el plano. graph es el edificio construido.

Errores comunes de compilación

Si tu definición tiene problemas, compile() los detecta antes de que ejecutes:

graph_builder.add_edge(START, "nonexistent_node")  # ← Este nodo no existe
graph = graph_builder.compile()
# ValueError: Node `nonexistent_node` is not present...

Piensa en compile() como el linter de tu grafo.

graph.invoke(): ejecución completa

invoke() es el modo más directo: pasas un estado inicial, el grafo ejecuta todos los nodos en orden, y te devuelve el estado final completo.

Ejecución básica

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator
from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage, AnyMessage
from langgraph.graph import StateGraph, START, END

class ChatState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
    step_count: int

model = init_chat_model("openai:gpt-4.1-mini")

def chatbot(state: ChatState) -> dict:
    response = model.invoke(state["messages"])
    return {
        "messages": [response],
        "step_count": state["step_count"] + 1
    }

graph_builder = StateGraph(ChatState)
graph_builder.add_node("chatbot", chatbot)
graph_builder.add_edge(START, "chatbot")
graph_builder.add_edge("chatbot", END)
graph = graph_builder.compile()

result = graph.invoke({
    "messages": [HumanMessage(content="¿Qué es RAG?")],
    "step_count": 0
})
print(type(result))           # <class 'dict'>
print(result["step_count"])   # 1
print(len(result["messages"]))  # 2 (HumanMessage + AIMessage)
print(result["messages"][-1].content[:100])
# Output: RAG (Retrieval-Augmented Generation) es una técnica que combina la recuperación de...

El resultado es el estado final completo

invoke() retorna un diccionario con todos los campos del estado después de que el último nodo terminó. Es el snapshot final del estado, con todos los reducers ya aplicados.

Ejecución con múltiples nodos

Cuando el grafo tiene múltiples nodos, invoke() los ejecuta en orden. El resultado acumula todo lo que cada nodo contribuyó (respetando los reducers del estado):

# Con un grafo de 2 nodos: classify → respond
result = graph.invoke({
    "messages": [HumanMessage(content="¿Cómo funciona async/await en Python?")],
    "steps_completed": []
})
print(f"Pasos completados: {result['steps_completed']}")
# Output: ['classify:técnica', 'respond']
# Ambos nodos contribuyeron a la lista (operator.add)

graph.stream(): ejecución paso a paso

invoke() espera a que todo termine. stream() te muestra cada paso mientras sucede — ves qué nodo se ejecutó y qué cambió después de cada uno.

stream_mode="values": estado completo después de cada nodo

Con stream_mode="values", recibes el estado completo después de que cada nodo termina:

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator
from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage, AnyMessage
from langgraph.graph import StateGraph, START, END

class ResearchState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
    findings: Annotated[list[str], operator.add]
    current_step: str

model = init_chat_model("openai:gpt-4.1-mini")

def search(state: ResearchState) -> dict:
    topic = state["messages"][-1].content
    response = model.invoke(f"Encuentra un dato clave sobre: {topic}. Responde en una frase.")
    return {
        "findings": [response.content.strip()],
        "current_step": "analyze"
    }

def analyze(state: ResearchState) -> dict:
    return {
        "findings": ["Análisis completado con confianza alta"],
        "current_step": "done"
    }

graph_builder = StateGraph(ResearchState)
graph_builder.add_node("search", search)
graph_builder.add_node("analyze", analyze)
graph_builder.add_edge(START, "search")
graph_builder.add_edge("search", "analyze")
graph_builder.add_edge("analyze", END)
graph = graph_builder.compile()

print("=== stream_mode='values' ===")
for step in graph.stream(
    {
        "messages": [HumanMessage(content="LangGraph")],
        "findings": [],
        "current_step": "search"
    },
    stream_mode="values"
):
    print(f"  current_step: {step['current_step']}")
    print(f"  findings count: {len(step['findings'])}")
    print()
# Output:
# === stream_mode='values' ===
#   current_step: search
#   findings count: 0
#
#   current_step: analyze
#   findings count: 1
#
#   current_step: done
#   findings count: 2

Cada iteración te da un snapshot completo del estado. La primera emisión es el estado inicial (antes de cualquier nodo). Las siguientes son el estado después de cada nodo.

stream_mode="updates": solo los cambios de cada nodo

Con stream_mode="updates", recibes solo lo que cada nodo retornó — la actualización parcial, no el estado completo. Usando el mismo grafo anterior:

# Mismo grafo, diferente stream_mode
print("=== stream_mode='updates' ===")
for step in graph.stream(
    {
        "messages": [HumanMessage(content="LangGraph")],
        "findings": [],
        "current_step": "search"
    },
    stream_mode="updates"
):
    for node_name, update in step.items():
        print(f"  [{node_name}] retornó: {list(update.keys())}")
# Output:
# === stream_mode='updates' ===
#   [search] retornó: ['findings', 'current_step']
#   [analyze] retornó: ['findings', 'current_step']

¿Cuándo usar cada modo?

Para debugging, "updates" es más útil porque ves exactamente qué retornó cada nodo. Para una interfaz de usuario, "values" es mejor porque siempre tienes el estado completo disponible.

Batch: múltiples inputs por el mismo grafo

Si necesitas ejecutar el mismo grafo con múltiples inputs, batch() es más eficiente que llamar invoke() en un loop. Recibe una lista de estados iniciales y retorna una lista de resultados en el mismo orden:

inputs = [
    {"messages": [HumanMessage(content="¿Qué es Python?")], "step_count": 0},
    {"messages": [HumanMessage(content="¿Qué es FastAPI?")], "step_count": 0},
    {"messages": [HumanMessage(content="¿Qué es LangGraph?")], "step_count": 0},
]

results = graph.batch(inputs)
for i, result in enumerate(results):
    print(f"[{i+1}] {result['messages'][-1].content[:60]}...")
# Output:
# [1] Python es un lenguaje de programación de alto nivel...
# [2] FastAPI es un framework web moderno y rápido para APIs...
# [3] LangGraph es un framework para construir workflows de AI...

batch() procesa todos los inputs y retorna la lista de resultados. Cada resultado tiene la misma estructura que lo que retornaría invoke() individual.

Visualización: draw_mermaid_png()

Aquí es donde muchos developers se saltan el paso más importante. Antes de ejecutar tu grafo, antes de debuggear un bug, antes de agregar más nodos — mira la imagen.

Por qué es esencial

draw_mermaid_png() genera una imagen PNG del grafo que muestra todos los nodos, edges, y el flujo de ejecución. Es documentación viva que siempre está sincronizada con tu código. No es un nice-to-have — es tu herramienta principal de debugging visual.

Cuando un conditional edge no se comporta como esperas, la imagen te muestra inmediatamente a dónde puede ir el flujo. Cuando un nodo nunca se ejecuta, la imagen revela que no hay edge hacia él. Cuando el grafo es más complejo de lo que pensabas, la imagen te lo dice antes de que pierdas horas debuggeando.

Generar y guardar la imagen

Tres líneas son todo lo que necesitas:

# Después de graph = graph_builder.compile()
png_data = graph.get_graph().draw_mermaid_png()
with open("mi_grafo.png", "wb") as f:
    f.write(png_data)
print("Grafo guardado en mi_grafo.png")
# Output: Grafo guardado en mi_grafo.png

Visualización con conditional edges

La imagen se vuelve especialmente valiosa cuando tienes conditional edges. Un grafo con routing a tres nodos especializados (tech_node, creative_node, general_node) mostrará classify con tres flechas salientes y los tres nodos convergiendo en END. Sin la imagen, tendrías que recorrer el código mentalmente para entender las rutas posibles.

draw_mermaid() para texto

Si no puedes renderizar PNG (ej: en una terminal), draw_mermaid() genera la representación en texto Mermaid:

mermaid_text = graph.get_graph().draw_mermaid()
print(mermaid_text)
# Output:
# %%{init: {'flowchart': {'curve': 'linear'}}}%%
# graph TD;
# 	__start__ --> chatbot;
# 	chatbot --> __end__;

Puedes copiar este texto y pegarlo en mermaid.live, GitHub, o Notion para ver el diagrama.

Hábito de trabajo: dibujar primero, codear después

Establece este flujo como práctica estándar:

1. Define el estado — qué datos fluyen por tu grafo
2. Dibuja el grafo — en papel o mentalmente: qué nodos, qué edges
3. Implementa — escribe los nodos y edges
4. Visualiza — draw_mermaid_png() para verificar que el grafo es lo que esperabas
5. Ejecuta — stream(mode="updates") para ver cada paso
6. Itera — ajusta y repite desde el paso 3

Opciones de compilación: preview

compile() acepta parámetros opcionales que habilitan funcionalidad avanzada. Los verás en detalle en módulos posteriores, pero es útil saber que existen:

checkpointer (Módulo 8: Memoria y Persistencia)

from langgraph.checkpoint.memory import MemorySaver

memory = MemorySaver()
graph = graph_builder.compile(checkpointer=memory)

Un checkpointer guarda el estado después de cada nodo. Esto habilita persistencia, time-travel debugging, y durable execution. Lo verás en profundidad en el Módulo 8.

interrupt_before / interrupt_after (Módulo 9: Human-in-the-Loop)

graph = graph_builder.compile(
    checkpointer=memory,
    interrupt_before=["dangerous_action"]
)

Pausa la ejecución antes (o después) de un nodo específico para pedir aprobación humana. Requiere un checkpointer. Lo verás en el Módulo 9.

Por ahora, basta con saber que compile() es extensible — la versión básica sin argumentos es todo lo que necesitas para este módulo.

Error handling durante ejecución

¿Qué pasa cuando un nodo lanza una excepción? Por defecto, se propaga y detiene el grafo:

def step_two(state) -> dict:
    raise ValueError("Algo salió mal")  # El grafo se detiene aquí

try:
    result = graph.invoke({"items": [], "current_step": "one"})
except ValueError as e:
    print(f"Grafo falló: {e}")
# Output: Grafo falló: Algo salió mal

Manejo de errores dentro de nodos

La forma recomendada es manejar errores dentro de cada nodo, capturando excepciones y registrándolas en el estado:

from typing import TypedDict, Annotated
import operator

class RobustState(TypedDict):
    messages: Annotated[list, operator.add]
    errors: Annotated[list[str], operator.add]
    current_step: str

def safe_search(state: RobustState) -> dict:
    try:
        response = model.invoke(state["messages"])
        return {"messages": [response], "current_step": "process"}
    except Exception as e:
        return {"errors": [f"search_error: {str(e)}"], "current_step": "process"}

Cada nodo maneja sus propios errores. Los nodos downstream revisan state["errors"] para decidir cómo proceder. En el Módulo 7 (Flujos Avanzados) verás patterns más sofisticados como retry con backoff y circuit breakers.

Ejemplo completo: clasificación, routing, y streaming

Vamos a juntar compilación, streaming y visualización en un solo ejemplo. Un grafo que clasifica la intención del usuario y rutea a nodos especializados:

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated, Literal
import operator
from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage, AnyMessage
from langgraph.graph import StateGraph, START, END

class AssistantState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
    intent: str
    response_type: str

model = init_chat_model("openai:gpt-4.1-mini")

def classify_intent(state: AssistantState) -> dict:
    last_msg = state["messages"][-1].content
    response = model.invoke(
        f"Clasifica en una palabra: technical, creative, o general.\n"
        f"Pregunta: {last_msg}\nResponde SOLO la categoría:"
    )
    return {"intent": response.content.strip().lower()}

def route_by_intent(state: AssistantState) -> Literal["tech_respond", "creative_respond", "general_respond"]:
    if "technical" in state["intent"]:
        return "tech_respond"
    elif "creative" in state["intent"]:
        return "creative_respond"
    return "general_respond"

def tech_respond(state: AssistantState) -> dict:
    response = model.invoke([
        {"role": "system", "content": "Experto técnico. Máximo 2 frases."},
        *state["messages"]
    ])
    return {"messages": [response], "response_type": "technical"}

def creative_respond(state: AssistantState) -> dict:
    response = model.invoke([
        {"role": "system", "content": "Escritor creativo. Máximo 2 frases."},
        *state["messages"]
    ])
    return {"messages": [response], "response_type": "creative"}

def general_respond(state: AssistantState) -> dict:
    response = model.invoke([
        {"role": "system", "content": "Asistente amigable y directo. Máximo 2 frases."},
        *state["messages"]
    ])
    return {"messages": [response], "response_type": "general"}

graph_builder = StateGraph(AssistantState)
graph_builder.add_node("classify", classify_intent)
graph_builder.add_node("tech_respond", tech_respond)
graph_builder.add_node("creative_respond", creative_respond)
graph_builder.add_node("general_respond", general_respond)

graph_builder.add_edge(START, "classify")
graph_builder.add_conditional_edges("classify", route_by_intent)
graph_builder.add_edge("tech_respond", END)
graph_builder.add_edge("creative_respond", END)
graph_builder.add_edge("general_respond", END)

graph = graph_builder.compile()

# Visualizar
png_data = graph.get_graph().draw_mermaid_png()
with open("assistant_grafo.png", "wb") as f:
    f.write(png_data)
print("Grafo guardado en assistant_grafo.png\n")

# Ejecutar con streaming
for step in graph.stream(
    {"messages": [HumanMessage(content="¿Cómo funciona async/await en Python?")],
     "intent": "", "response_type": ""},
    stream_mode="updates"
):
    for node_name, update in step.items():
        print(f"[{node_name}]")
        if "intent" in update:
            print(f"  Intent: {update['intent']}")
        if "messages" in update:
            print(f"  Respuesta: {update['messages'][-1].content[:80]}...")
# Output:
# Grafo guardado en assistant_grafo.png
#
# [classify]
#   Intent: technical
# [tech_respond]
#   Respuesta: async/await en Python permite escribir código asíncrono que no bloquea...

El flujo completo: definir estado → crear nodos → conectar edges → compilar → visualizar → ejecutar con streaming.

Troubleshooting

Problema 1: "Node X is not present" al compilar

Síntoma: ValueError: Node 'my_node' is not present in the graph. Causa: Referenciaste un nodo en un edge que no fue agregado con add_node(). Puede ser un typo. Solución: Verifica que todos los nodos mencionados en add_edge y add_conditional_edges fueron registrados:

# MAL — typo en el nombre
graph_builder.add_node("classify", classify_fn)
graph_builder.add_edge(START, "clasify")  # ← 's' faltante

# BIEN
graph_builder.add_node("classify", classify_fn)
graph_builder.add_edge(START, "classify")

Problema 2: stream() no produce output

Síntoma: El loop for step in graph.stream(...) no imprime nada. Causa: El grafo se ejecuta pero ningún nodo retorna actualizaciones para los campos que estás observando, o el input es inválido y el grafo termina inmediatamente. Solución: Verifica que los nodos retornan diccionarios con al menos un campo, y que el estado inicial incluye todos los campos requeridos:

# MAL — nodo que no retorna nada
def my_node(state):
    do_something(state)
    # No retorna dict → no hay update

# BIEN — siempre retorna dict
def my_node(state):
    do_something(state)
    return {"current_step": "done"}

Problema 3: draw_mermaid_png() falla con error de dependencia

Síntoma: ImportError o error al llamar draw_mermaid_png(). Causa: Falta la dependencia para renderizar Mermaid a PNG. Solución: Instala el paquete necesario:

pip install grandalf

Si sigue fallando, usa draw_mermaid() (sin _png) para obtener la representación en texto y renderízala en mermaid.live.

Problema 4: invoke() retorna estado incompleto

Síntoma: Algunos campos del resultado son None o no están presentes. Causa: Los nodos no actualizan esos campos y no se proporcionaron valores iniciales. Solución: Incluye valores iniciales para todos los campos al llamar invoke():

# MAL — campos faltantes
result = graph.invoke({"messages": [HumanMessage(content="hola")]})

# BIEN — todos los campos con valores iniciales
result = graph.invoke({
    "messages": [HumanMessage(content="hola")],
    "intent": "",
    "response_type": "",
    "findings": []
})

Problema 5: batch() retorna resultados en orden incorrecto

Síntoma: Los resultados de batch() no corresponden a los inputs. Causa: Esto no debería pasar — batch() preserva el orden. Si ves resultados desordenados, verifica que no estás mezclando los resultados después. Solución: Verifica con un ejemplo simple:

results = graph.batch([input_1, input_2, input_3])
# results[0] corresponde a input_1
# results[1] corresponde a input_2
# results[2] corresponde a input_3

Ejercicios

Ejercicio 1: Compilar y ejecutar un grafo de 3 nodos (Fácil)

Crea un grafo con tres nodos: greet (agrega un saludo), ask (agrega una pregunta), y farewell (agrega una despedida). Usa un estado con messages: Annotated[list[str], operator.add]. Compila, ejecuta con invoke, e imprime todos los mensajes.

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END

class ConvoState(TypedDict):
    messages: Annotated[list[str], operator.add]

def greet(state: ConvoState) -> dict:
    return {"messages": ["¡Hola! Bienvenido."]}

def ask(state: ConvoState) -> dict:
    return {"messages": ["¿En qué puedo ayudarte hoy?"]}

def farewell(state: ConvoState) -> dict:
    return {"messages": ["¡Hasta luego! Que tengas un gran día."]}

graph_builder = StateGraph(ConvoState)
graph_builder.add_node("greet", greet)
graph_builder.add_node("ask", ask)
graph_builder.add_node("farewell", farewell)
graph_builder.add_edge(START, "greet")
graph_builder.add_edge("greet", "ask")
graph_builder.add_edge("ask", "farewell")
graph_builder.add_edge("farewell", END)

graph = graph_builder.compile()
result = graph.invoke({"messages": []})

for msg in result["messages"]:
    print(f"  → {msg}")
# Output:
#   → ¡Hola! Bienvenido.
#   → ¿En qué puedo ayudarte hoy?
#   → ¡Hasta luego! Que tengas un gran día.

Ejercicio 2: Comparar stream_mode values vs updates (Fácil)

Usando el mismo grafo del ejercicio 1, ejecútalo dos veces: una con stream_mode="values" y otra con stream_mode="updates". Imprime lo que recibes en cada iteración y describe la diferencia.

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END

class ConvoState(TypedDict):
    messages: Annotated[list[str], operator.add]

def greet(state: ConvoState) -> dict:
    return {"messages": ["¡Hola!"]}

def ask(state: ConvoState) -> dict:
    return {"messages": ["¿Cómo estás?"]}

graph_builder = StateGraph(ConvoState)
graph_builder.add_node("greet", greet)
graph_builder.add_node("ask", ask)
graph_builder.add_edge(START, "greet")
graph_builder.add_edge("greet", "ask")
graph_builder.add_edge("ask", END)
graph = graph_builder.compile()

print("=== stream_mode='values' ===")
for i, step in enumerate(graph.stream({"messages": []}, stream_mode="values")):
    print(f"  Paso {i}: messages = {step['messages']}")

print()

print("=== stream_mode='updates' ===")
for i, step in enumerate(graph.stream({"messages": []}, stream_mode="updates")):
    for node, update in step.items():
        print(f"  Paso {i}: [{node}] retornó messages = {update.get('messages', 'N/A')}")
# Output:
# === stream_mode='values' ===
#   Paso 0: messages = []
#   Paso 1: messages = ['¡Hola!']
#   Paso 2: messages = ['¡Hola!', '¿Cómo estás?']
#
# === stream_mode='updates' ===
#   Paso 0: [greet] retornó messages = ['¡Hola!']
#   Paso 1: [ask] retornó messages = ['¿Cómo estás?']

Ejercicio 3: Visualizar un grafo con routing (Medio)

Crea un grafo que recibe un número en el estado y rutea a uno de tres nodos: positive_node (si > 0), negative_node (si < 0), o zero_node (si == 0). Genera la imagen PNG del grafo y describe qué ves.

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated, Literal
import operator
from langgraph.graph import StateGraph, START, END

class NumberState(TypedDict):
    value: int
    result: str
    steps: Annotated[list[str], operator.add]

def route_number(state: NumberState) -> Literal["positive", "negative", "zero"]:
    if state["value"] > 0:
        return "positive"
    elif state["value"] < 0:
        return "negative"
    return "zero"

def positive_node(state: NumberState) -> dict:
    return {"result": "¡Número positivo!", "steps": ["positive"]}

def negative_node(state: NumberState) -> dict:
    return {"result": "Número negativo", "steps": ["negative"]}

def zero_node(state: NumberState) -> dict:
    return {"result": "Es cero", "steps": ["zero"]}

graph_builder = StateGraph(NumberState)
graph_builder.add_node("positive", positive_node)
graph_builder.add_node("negative", negative_node)
graph_builder.add_node("zero", zero_node)

graph_builder.add_conditional_edges(START, route_number)
graph_builder.add_edge("positive", END)
graph_builder.add_edge("negative", END)
graph_builder.add_edge("zero", END)

graph = graph_builder.compile()

png_data = graph.get_graph().draw_mermaid_png()
with open("number_router.png", "wb") as f:
    f.write(png_data)
print("Grafo guardado en number_router.png")

for test_value in [42, -7, 0]:
    result = graph.invoke({"value": test_value, "result": "", "steps": []})
    print(f"  valor={test_value} → {result['result']} (ruta: {result['steps']})")
# Output:
# Grafo guardado en number_router.png
#   valor=42 → ¡Número positivo! (ruta: ['positive'])
#   valor=-7 → Número negativo (ruta: ['negative'])
#   valor=0 → Es cero (ruta: ['zero'])

Ejercicio 4: Batch con múltiples preguntas (Medio)

Crea un grafo que recibe una pregunta y genera una respuesta de una frase. Usa batch() para procesar 4 preguntas diferentes simultáneamente e imprime los resultados con el índice de cada input.

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator
from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage, AnyMessage
from langgraph.graph import StateGraph, START, END

class QAState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
    answer: str

model = init_chat_model("openai:gpt-4.1-mini")

def answer_question(state: QAState) -> dict:
    response = model.invoke([
        {"role": "system", "content": "Responde en exactamente una frase corta."},
        *state["messages"]
    ])
    return {"messages": [response], "answer": response.content.strip()}

graph_builder = StateGraph(QAState)
graph_builder.add_node("answer", answer_question)
graph_builder.add_edge(START, "answer")
graph_builder.add_edge("answer", END)
graph = graph_builder.compile()

questions = ["¿Qué es Python?", "¿Qué es FastAPI?", "¿Qué es LangGraph?", "¿Qué es un reducer?"]
inputs = [{"messages": [HumanMessage(content=q)], "answer": ""} for q in questions]
results = graph.batch(inputs)

for i, (q, r) in enumerate(zip(questions, results)):
    print(f"[{i+1}] {q} → {r['answer'][:60]}...")
# Output:
# [1] ¿Qué es Python? → Python es un lenguaje de programación de alto nivel...
# [2] ¿Qué es FastAPI? → FastAPI es un framework web moderno y rápido...
# [3] ¿Qué es LangGraph? → LangGraph es un framework para construir workflows...
# [4] ¿Qué es un reducer? → Un reducer es una función que define cómo combinar...

Ejercicio 5: Stream updates con monitoreo de nodos (Avanzado)

Crea un grafo de 3 nodos (fetch → analyze → report) que usa un LLM. Ejecuta con stream_mode="updates" e imprime qué campos actualizó cada nodo.

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator
from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage, AnyMessage
from langgraph.graph import StateGraph, START, END

class PipelineState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
    data: Annotated[list[str], operator.add]
    current_step: str

model = init_chat_model("openai:gpt-4.1-mini")

def fetch(state: PipelineState) -> dict:
    response = model.invoke("Genera 2 datos sobre el clima, separados por '|'.")
    items = [x.strip() for x in response.content.split("|")]
    return {"data": items, "current_step": "analyze"}

def analyze(state: PipelineState) -> dict:
    data_text = "\n".join(state["data"])
    response = model.invoke(f"Analiza en una frase:\n{data_text}")
    return {"data": [f"[análisis] {response.content.strip()}"], "current_step": "report"}

def report(state: PipelineState) -> dict:
    return {"current_step": "done"}

graph_builder = StateGraph(PipelineState)
graph_builder.add_node("fetch", fetch)
graph_builder.add_node("analyze", analyze)
graph_builder.add_node("report", report)
graph_builder.add_edge(START, "fetch")
graph_builder.add_edge("fetch", "analyze")
graph_builder.add_edge("analyze", "report")
graph_builder.add_edge("report", END)
graph = graph_builder.compile()

for step in graph.stream(
    {"messages": [], "data": [], "current_step": "fetch"},
    stream_mode="updates"
):
    for node_name, update in step.items():
        data_count = len(update.get("data", []))
        print(f"[{node_name}] → step: {update.get('current_step', '?')}, datos nuevos: {data_count}")
# Output:
# [fetch] → step: analyze, datos nuevos: 2
# [analyze] → step: report, datos nuevos: 1
# [report] → step: done, datos nuevos: 0

Ejercicio 6: Grafo con error handling y visualización (Avanzado)

Crea un grafo de 3 nodos (search, validate, format) donde search puede fallar aleatoriamente. Implementa error handling dentro del nodo (sin crashear), genera la visualización PNG, y ejecuta con streaming.

from dotenv import load_dotenv
load_dotenv()

import random
from typing import TypedDict, Annotated
import operator
from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage, AnyMessage
from langgraph.graph import StateGraph, START, END

class ResearchState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
    results: Annotated[list[str], operator.add]
    errors: Annotated[list[str], operator.add]
    status: str

model = init_chat_model("openai:gpt-4.1-mini")

def search(state: ResearchState) -> dict:
    try:
        if random.random() < 0.3:
            raise ConnectionError("API no disponible")
        topic = state["messages"][-1].content
        response = model.invoke(f"Encuentra 2 datos sobre: {topic}. Sepáralos con '|'.")
        return {"results": [x.strip() for x in response.content.split("|")], "status": "ok"}
    except Exception as e:
        return {"errors": [f"search: {e}"], "results": ["(fallback)"], "status": "error"}

def validate(state: ResearchState) -> dict:
    valid = [r for r in state["results"] if len(r) > 10]
    if not valid:
        return {"errors": ["validate: sin resultados válidos"], "status": "failed"}
    return {"status": "validated"}

def format_output(state: ResearchState) -> dict:
    results_text = "\n".join(f"- {r}" for r in state["results"])
    return {"results": [f"REPORTE:\n{results_text}"], "status": "done"}

graph_builder = StateGraph(ResearchState)
graph_builder.add_node("search", search)
graph_builder.add_node("validate", validate)
graph_builder.add_node("format", format_output)
graph_builder.add_edge(START, "search")
graph_builder.add_edge("search", "validate")
graph_builder.add_edge("validate", "format")
graph_builder.add_edge("format", END)
graph = graph_builder.compile()

png_data = graph.get_graph().draw_mermaid_png()
with open("research_pipeline.png", "wb") as f:
    f.write(png_data)
print("Grafo guardado en research_pipeline.png\n")

for step in graph.stream(
    {"messages": [HumanMessage(content="AI en medicina")],
     "results": [], "errors": [], "status": "starting"},
    stream_mode="updates"
):
    for node_name, update in step.items():
        print(f"[{node_name}] status={update.get('status', '?')}")
# Output:
# Grafo guardado en research_pipeline.png
#
# [search] status=ok
# [validate] status=validated
# [format] status=done

Resumen

En esta cápsula aprendiste:

• graph.compile() valida tu definición y crea un objeto ejecutable — es el paso que convierte el plano en un sistema funcional
• graph.invoke() ejecuta el grafo completo y retorna el estado final — ideal para ejecuciones donde solo te importa el resultado
• graph.stream() te muestra cada paso conforme se ejecuta — con "values" ves el estado completo, con "updates" ves solo los cambios de cada nodo
• graph.batch() procesa múltiples inputs en paralelo — más eficiente que un loop con invoke()
• draw_mermaid_png() es tu herramienta principal de debugging visual — antes de debuggear código, mira la imagen
• draw_mermaid() genera texto Mermaid cuando no puedes renderizar PNG
• Las opciones de compilación (checkpointer, interrupt_before) habilitan funcionalidad avanzada que verás en módulos posteriores
• El error handling se hace dentro de los nodos, no alrededor de invoke() — cada nodo es responsable de manejar sus propios errores

Próxima cápsula: create_agent vs StateGraph — cuándo usar la abstracción de alto nivel y cuándo bajar al control total de LangGraph.

Recursos adicionales

1. LangGraph Quick Start — Tutorial oficial que cubre compilación y ejecución paso a paso
2. How to run a graph — Guía de invoke, stream y batch
3. Streaming in LangGraph — Documentación detallada de stream modes
4. Visualization — LangGraph Docs — Cómo usar draw_mermaid_png y alternativas
5. LangGraph Concepts: Compiling — Referencia conceptual de la compilación
6. Mermaid Live Editor — Renderizador online para pegar output de draw_mermaid()
7. Runnable Interface — LangChain — La interfaz que implementa CompiledGraph (invoke, stream, batch)
8. Error Handling in LangGraph — Patrones para manejar errores en grafos

Módulo 5 — LangChain & LangGraph: From Chains to Agents