StateGraph: Tu Primer Grafo

Descripción de la cápsula

StateGraph es la clase fundamental de LangGraph. Es el contenedor donde defines el estado de tu workflow, agregas nodos (funciones), los conectas con edges (conexiones), compilas, y ejecutas. Todo grafo en LangGraph empieza con StateGraph.

En la cápsula anterior viste la visión general: por qué LangGraph existe, qué limitaciones de create_agent resuelve, y cómo un grafo es simplemente un diagrama de flujo con nodos y edges. Ahora vas a construirlo. Crearás tu primer grafo funcional, entenderás cómo funciona el estado tipado con TypedDict y Annotated, descubrirás por qué los reducers son la pieza más importante (y la fuente #1 de bugs), y visualizarás tu grafo con draw_mermaid_png().

Las piezas del rompecabezas

Antes de escribir código, estas son las 5 piezas que necesitas para construir cualquier grafo:

Definir el estado: TypedDict + Annotated

El estado es el corazón de tu grafo. Es un diccionario tipado que todos los nodos comparten: cada nodo lo recibe, lo lee, y retorna actualizaciones.

El problema sin reducers

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

class State(TypedDict):
    messages: list[str]

def step_a(state: State) -> dict:
    return {"messages": ["Mensaje de A"]}

def step_b(state: State) -> dict:
    return {"messages": ["Mensaje de B"]}

graph_builder = StateGraph(State)
graph_builder.add_node("step_a", step_a)
graph_builder.add_node("step_b", step_b)
graph_builder.add_edge(START, "step_a")
graph_builder.add_edge("step_a", "step_b")
graph_builder.add_edge("step_b", END)

graph = graph_builder.compile()
result = graph.invoke({"messages": ["Mensaje inicial"]})
print(result)
# Output: {"messages": ["Mensaje de B"]}

El mensaje de A desapareció. Y el mensaje inicial también. Cada nodo reemplazó la lista completa con su propio valor. step_b escribió ["Mensaje de B"] y borró todo lo anterior.

Este es el error #1 para principiantes. Sin un reducer, cada nodo sobrescribe el campo completo.

La solución: Annotated + operator.add

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

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

def step_a(state: State) -> dict:
    return {"messages": ["Mensaje de A"]}

def step_b(state: State) -> dict:
    return {"messages": ["Mensaje de B"]}

graph_builder = StateGraph(State)
graph_builder.add_node("step_a", step_a)
graph_builder.add_node("step_b", step_b)
graph_builder.add_edge(START, "step_a")
graph_builder.add_edge("step_a", "step_b")
graph_builder.add_edge("step_b", END)

graph = graph_builder.compile()
result = graph.invoke({"messages": ["Mensaje inicial"]})
print(result)
# Output: {"messages": ["Mensaje inicial", "Mensaje de A", "Mensaje de B"]}

Ahora los tres mensajes están. operator.add le dice a LangGraph: "cuando un nodo retorne un valor para messages, no reemplaces — concatena con lo que ya existe."

Cómo funciona internamente

Sin reducer (reemplazar):
  Estado actual: {"messages": ["Inicial", "A"]}
  Nodo retorna:  {"messages": ["B"]}
  Resultado:     {"messages": ["B"]}              ← se perdió todo

Con operator.add (acumular):
  Estado actual: {"messages": ["Inicial", "A"]}
  Nodo retorna:  {"messages": ["B"]}
  Resultado:     {"messages": ["Inicial", "A", "B"]}  ← se acumuló

La regla es simple:

• Sin Annotated: el valor retornado reemplaza el valor existente
• Con Annotated[tipo, operator.add]: el valor retornado se concatena al existente

Campos con y sin reducer en el mismo estado

Puedes mezclar campos que acumulan con campos que reemplazan:

from typing import TypedDict, Annotated
import operator

class State(TypedDict):
    messages: Annotated[list[str], operator.add]
    step_count: int
    current_phase: str

• messages usa operator.add → acumula
• step_count sin reducer → se reemplaza (el último valor gana)
• current_phase sin reducer → se reemplaza

Diseña tu estado pensando: "¿este campo necesita acumular historia o solo necesito el último valor?"

Crear el grafo y conectar nodos

StateGraph recibe la clase de estado y te da un builder para agregar nodos y edges:

from langgraph.graph import StateGraph, START, END

graph_builder = StateGraph(State)

El builder tiene tres operaciones principales:

START y END son nodos virtuales que marcan entrada y salida. Siempre conectas START a tu primer nodo real, y tu último nodo a END. Si olvidas alguno, el error se detecta al compilar.

Tu primer grafo completo

Ahora juntemos todas las piezas:

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

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

def greet(state: State) -> dict:
    return {
        "messages": ["¡Hola! Soy tu primer grafo."],
        "step_count": state["step_count"] + 1
    }

graph_builder = StateGraph(State)
graph_builder.add_node("greet", greet)
graph_builder.add_edge(START, "greet")
graph_builder.add_edge("greet", END)

graph = graph_builder.compile()
result = graph.invoke({"messages": [], "step_count": 0})
print(result)
# {"messages": ["¡Hola! Soy tu primer grafo."], "step_count": 1}

Paso a paso:

1. Defines el estado — State con messages (acumula) y step_count (reemplaza)
2. Defines la función — greet recibe el estado, retorna las actualizaciones
3. Creas el builder — StateGraph(State) con tu clase de estado
4. Agregas el nodo — add_node("greet", greet) registra la función
5. Conectas los edges — START → greet → END
6. Compilas — graph_builder.compile() produce un grafo ejecutable
7. Ejecutas — graph.invoke(...) pasa el estado inicial por el grafo

Un nodo no necesita retornar todos los campos del estado — solo los que quiere actualizar. Los campos no incluidos conservan su valor.

Grafo con múltiples nodos

Un solo nodo no es muy útil. Agreguemos más:

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

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

def collect_info(state: State) -> dict:
    return {
        "messages": ["Paso 1: Recopilando información..."],
        "step_count": state["step_count"] + 1
    }

def analyze(state: State) -> dict:
    return {
        "messages": [f"Paso 2: Analizando ({state['step_count']} pasos previos)..."],
        "step_count": state["step_count"] + 1
    }

def respond(state: State) -> dict:
    return {
        "messages": [f"Paso 3: Respuesta generada después de {state['step_count']} pasos."],
        "step_count": state["step_count"] + 1
    }

graph_builder = StateGraph(State)
graph_builder.add_node("collect", collect_info)
graph_builder.add_node("analyze", analyze)
graph_builder.add_node("respond", respond)

graph_builder.add_edge(START, "collect")
graph_builder.add_edge("collect", "analyze")
graph_builder.add_edge("analyze", "respond")
graph_builder.add_edge("respond", END)

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

for msg in result["messages"]:
    print(msg)
# Paso 1: Recopilando información...
# Paso 2: Analizando (1 pasos previos)...
# Paso 3: Respuesta generada después de 2 pasos.

print(f"Total de pasos: {result['step_count']}")
# Total de pasos: 3

El flujo es: START → collect → analyze → respond → END. Cada nodo lee el estado actual, agrega su mensaje, incrementa el contador, y pasa al siguiente.

Nota cómo messages acumula (gracias a operator.add) pero step_count se reemplaza cada vez (el último valor es 3, no la suma de todos los incrementos).

Ejecución: invoke vs stream

El grafo compilado tiene dos formas de ejecución:

invoke — ejecución completa

result = graph.invoke({"messages": [], "step_count": 0})
print(result)
# Retorna el estado final completo

stream — ejecución paso a paso

for step in graph.stream({"messages": [], "step_count": 0}):
    print(step)
    print("---")

# Output:
# {"collect": {"messages": ["Paso 1: Recopilando información..."], "step_count": 1}}
# ---
# {"analyze": {"messages": ["Paso 2: Analizando (1 pasos previos)..."], "step_count": 2}}
# ---
# {"respond": {"messages": ["Paso 3: Respuesta generada después de 2 pasos."], "step_count": 3}}
# ---

stream te muestra qué produjo cada nodo individualmente. Cada chunk es un dict donde la key es el nombre del nodo y el value es lo que ese nodo retornó. Útil para debugging y para dar feedback progresivo al usuario.

Visualización: draw_mermaid_png()

draw_mermaid_png() no es opcional — es tu herramienta principal de debugging. Antes de ejecutar un grafo, visualízalo para confirmar que el flujo es correcto:

from IPython.display import Image, display

display(Image(graph.get_graph().draw_mermaid_png()))

Si estás fuera de un notebook, puedes guardar la imagen:

png_data = graph.get_graph().draw_mermaid_png()
with open("mi_grafo.png", "wb") as f:
    f.write(png_data)

O usar el formato Mermaid como texto (sin dependencias externas):

print(graph.get_graph().draw_mermaid())
# %%{init: {'flowchart': {'curve': 'linear'}}}%%
# graph TD;
#     __start__([__start__]):::first
#     collect(collect)
#     analyze(analyze)
#     respond(respond)
#     __end__([__end__]):::last
#     __start__ --> collect;
#     collect --> analyze;
#     analyze --> respond;
#     respond --> __end__;

Puedes copiar ese output y pegarlo en mermaid.live para visualizar en el navegador.

Acostúmbrate a este flujo: dibuja primero, codea después. Si tu grafo visualizado no tiene el flujo que esperas, el código tampoco lo tendrá.

El puente desde create_agent

Recuerda: en create_agent, el framework controlaba el flujo. Ahora tú defines cada paso:

create_agent sigue siendo la herramienta correcta para el 80% de los casos. Pero cuando necesites un flujo que el loop ReAct no puede expresar, StateGraph está listo.

Troubleshooting

Problema 1: "Los mensajes desaparecen entre nodos"

Síntoma: Tu lista de mensajes solo tiene el último mensaje, no el historial completo. Causa: No estás usando Annotated con operator.add en el campo de mensajes. Solución:

# Incorrecto — cada nodo reemplaza la lista completa
class State(TypedDict):
    messages: list[str]

# Correcto — cada nodo agrega a la lista existente
class State(TypedDict):
    messages: Annotated[list[str], operator.add]

Problema 2: "ValueError al compilar — nodo no alcanzable"

Síntoma: Error al compilar el grafo indicando que un nodo no es alcanzable desde START. Causa: Hay un nodo sin camino desde START. Probablemente olvidaste un edge. Solución: Verifica que todos tus nodos están conectados. Usa draw_mermaid() antes de compilar para detectar nodos sueltos.

Problema 3: "El grafo compila pero no produce output"

Síntoma: graph.invoke() retorna el estado inicial sin cambios. Causa: Olvidaste conectar START a tu primer nodo. Solución:

graph_builder.add_edge(START, "greet")  # No olvides esta línea
graph_builder.add_edge("greet", END)

Problema 4: "draw_mermaid_png() produce error"

Síntoma: Error al intentar visualizar con draw_mermaid_png(). Causa: Necesita conexión a internet o la dependencia pyppeteer. Solución: Usa draw_mermaid() como alternativa (produce texto Mermaid sin dependencias). Copia el output y pégalo en mermaid.live para visualizar.

Ejercicios

Ejercicio 1: Grafo de un solo nodo (Fácil)

Crea un grafo con un solo nodo hello que agregue el mensaje "¡Hola, LangGraph!" al estado. El estado debe tener un campo messages con reducer. Ejecuta el grafo con un mensaje inicial y verifica que ambos mensajes están en el resultado.

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

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

def hello(state: State) -> dict:
    return {"messages": ["¡Hola, LangGraph!"]}

graph_builder = StateGraph(State)
graph_builder.add_node("hello", hello)
graph_builder.add_edge(START, "hello")
graph_builder.add_edge("hello", END)

graph = graph_builder.compile()
result = graph.invoke({"messages": ["Mensaje inicial"]})
print(result)
# {"messages": ["Mensaje inicial", "¡Hola, LangGraph!"]}

Ejercicio 2: Demostrar la diferencia con y sin reducer (Fácil)

Crea dos grafos idénticos con dos nodos en secuencia. En uno, el campo messages usa operator.add. En el otro, no. Ejecuta ambos con un mensaje inicial y compara los resultados.

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

class StateNoReducer(TypedDict):
    messages: list[str]

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

def step_a(state) -> dict:
    return {"messages": ["A"]}

def step_b(state) -> dict:
    return {"messages": ["B"]}

def build_graph(state_class):
    builder = StateGraph(state_class)
    builder.add_node("a", step_a)
    builder.add_node("b", step_b)
    builder.add_edge(START, "a")
    builder.add_edge("a", "b")
    builder.add_edge("b", END)
    return builder.compile()

result_no = build_graph(StateNoReducer).invoke({"messages": ["Inicial"]})
print(f"Sin reducer: {result_no['messages']}")
# Sin reducer: ['B']

result_yes = build_graph(StateWithReducer).invoke({"messages": ["Inicial"]})
print(f"Con reducer: {result_yes['messages']}")
# Con reducer: ['Inicial', 'A', 'B']

Ejercicio 3: Pipeline de 3 pasos con stream (Medio)

Crea un grafo con tres nodos: extract, transform, load (patrón ETL). Cada nodo agrega un mensaje describiendo su acción y actualiza un campo phase. Usa graph.stream() para ver la salida de cada nodo.

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

class State(TypedDict):
    messages: Annotated[list[str], operator.add]
    phase: str
    record_count: int

def extract(state: State) -> dict:
    return {
        "messages": ["Extrayendo datos de la fuente..."],
        "phase": "extract",
        "record_count": 150
    }

def transform(state: State) -> dict:
    cleaned = int(state["record_count"] * 0.9)
    return {
        "messages": [f"Transformando {state['record_count']} registros → {cleaned} válidos"],
        "phase": "transform",
        "record_count": cleaned
    }

def load(state: State) -> dict:
    return {
        "messages": [f"Cargando {state['record_count']} registros al destino. ¡Listo!"],
        "phase": "load"
    }

graph_builder = StateGraph(State)
graph_builder.add_node("extract", extract)
graph_builder.add_node("transform", transform)
graph_builder.add_node("load", load)
graph_builder.add_edge(START, "extract")
graph_builder.add_edge("extract", "transform")
graph_builder.add_edge("transform", "load")
graph_builder.add_edge("load", END)

graph = graph_builder.compile()

for step in graph.stream({"messages": [], "phase": "pending", "record_count": 0}):
    for node_name, output in step.items():
        print(f"[{node_name}] phase={output.get('phase', '?')} | {output.get('messages', [])}")

# Output:
# [extract] phase=extract | ['Extrayendo datos de la fuente...']
# [transform] phase=transform | ['Transformando 150 registros → 135 válidos']
# [load] phase=load | ['Cargando 135 registros al destino. ¡Listo!']

Ejercicio 4: Grafo con estado mixto y visualización (Medio)

Crea un grafo que simule validación de formulario. El estado tiene: messages (con reducer), errors (con reducer), is_valid (booleano sin reducer). Un nodo valida, otro reporta. Visualiza el grafo con draw_mermaid().

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

class FormState(TypedDict):
    messages: Annotated[list[str], operator.add]
    errors: Annotated[list[str], operator.add]
    is_valid: bool

def validate(state: FormState) -> dict:
    found_errors = ["Campo 'email' vacío", "Campo 'nombre' muy corto"]
    return {
        "messages": ["Validación ejecutada"],
        "errors": found_errors,
        "is_valid": len(found_errors) == 0
    }

def report(state: FormState) -> dict:
    if state["is_valid"]:
        return {"messages": ["Formulario válido. Enviando..."]}
    return {"messages": [f"Formulario inválido. {len(state['errors'])} errores encontrados."]}

graph_builder = StateGraph(FormState)
graph_builder.add_node("validate", validate)
graph_builder.add_node("report", report)
graph_builder.add_edge(START, "validate")
graph_builder.add_edge("validate", "report")
graph_builder.add_edge("report", END)

graph = graph_builder.compile()

print(graph.get_graph().draw_mermaid())
# graph TD;
#     __start__ --> validate;
#     validate --> report;
#     report --> __end__;

result = graph.invoke({"messages": [], "errors": [], "is_valid": False})
print(f"Mensajes: {result['messages']}")
# Mensajes: ['Validación ejecutada', 'Formulario inválido. 2 errores encontrados.']
print(f"Errores: {result['errors']}")
# Errores: ["Campo 'email' vacío", "Campo 'nombre' muy corto"]

Ejercicio 5: Diseño de estado — elegir reducers (Avanzado)

Diseña el estado para un sistema de procesamiento de pedidos. Requisitos: acumular un log de acciones, trackear status actual del pedido, acumular items, mantener el total (último cálculo), acumular warnings. Crea el TypedDict, 3 nodos, y ejecuta el grafo. Justifica tus decisiones de reducer.

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

class OrderState(TypedDict):
    action_log: Annotated[list[str], operator.add]
    status: str
    items: Annotated[list[dict], operator.add]
    total: float
    warnings: Annotated[list[str], operator.add]

def receive_order(state: OrderState) -> dict:
    items = [
        {"name": "Laptop", "price": 999.99, "qty": 1},
        {"name": "Mouse", "price": 29.99, "qty": 2},
    ]
    return {
        "action_log": ["Pedido recibido con 2 productos"],
        "status": "received",
        "items": items,
    }

def validate_order(state: OrderState) -> dict:
    warnings = []
    total = sum(item["price"] * item["qty"] for item in state["items"])
    if total > 500:
        warnings.append(f"Pedido alto: ${total:.2f} — requiere aprobación")
    return {
        "action_log": [f"Validación completada. Total: ${total:.2f}"],
        "status": "validated",
        "total": total,
        "warnings": warnings,
    }

def confirm_order(state: OrderState) -> dict:
    return {
        "action_log": [f"Pedido confirmado. Total final: ${state['total']:.2f}"],
        "status": "confirmed",
    }

graph_builder = StateGraph(OrderState)
graph_builder.add_node("receive", receive_order)
graph_builder.add_node("validate", validate_order)
graph_builder.add_node("confirm", confirm_order)
graph_builder.add_edge(START, "receive")
graph_builder.add_edge("receive", "validate")
graph_builder.add_edge("validate", "confirm")
graph_builder.add_edge("confirm", END)

graph = graph_builder.compile()
result = graph.invoke({
    "action_log": [], "status": "pending", "items": [],
    "total": 0.0, "warnings": []
})

print(f"Status: {result['status']}")
# Status: confirmed
print(f"Total: ${result['total']:.2f}")
# Total: $1059.97
print(f"Warnings: {result['warnings']}")
# Warnings: ['Pedido alto: $1059.97 — requiere aprobación']
for entry in result['action_log']:
    print(f"  → {entry}")
# → Pedido recibido con 2 productos
# → Validación completada. Total: $1059.97
# → Pedido confirmado. Total final: $1059.97

Ejercicio 6: invoke vs stream con 4 nodos (Challenge)

Crea un grafo con 4 nodos matemáticos en secuencia: multiply (*2), add_ten (+10), square (^2), report. Ejecuta con invoke (imprime resultado final) y con stream (imprime paso a paso con contador de nodos ejecutados).

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

class State(TypedDict):
    messages: Annotated[list[str], operator.add]
    value: int

def multiply(state: State) -> dict:
    new_val = state["value"] * 2
    return {"messages": [f"multiply: {state['value']} → {new_val}"], "value": new_val}

def add_ten(state: State) -> dict:
    new_val = state["value"] + 10
    return {"messages": [f"add_ten: {state['value']} → {new_val}"], "value": new_val}

def square(state: State) -> dict:
    new_val = state["value"] ** 2
    return {"messages": [f"square: {state['value']} → {new_val}"], "value": new_val}

def report(state: State) -> dict:
    return {"messages": [f"Resultado final: {state['value']}"]}

graph_builder = StateGraph(State)
graph_builder.add_node("multiply", multiply)
graph_builder.add_node("add_ten", add_ten)
graph_builder.add_node("square", square)
graph_builder.add_node("report", report)
graph_builder.add_edge(START, "multiply")
graph_builder.add_edge("multiply", "add_ten")
graph_builder.add_edge("add_ten", "square")
graph_builder.add_edge("square", "report")
graph_builder.add_edge("report", END)

graph = graph_builder.compile()

print("=== INVOKE ===")
result = graph.invoke({"messages": [], "value": 5})
print(f"Value: {result['value']}")
print(f"Messages: {result['messages']}")
# Value: 400
# Messages: ['multiply: 5 → 10', 'add_ten: 10 → 20', 'square: 20 → 400', 'Resultado final: 400']

print("\n=== STREAM ===")
node_count = 0
for step in graph.stream({"messages": [], "value": 5}):
    for node_name, output in step.items():
        node_count += 1
        print(f"  Nodo #{node_count} [{node_name}]: value={output.get('value', '—')}")
print(f"Total nodos ejecutados: {node_count}")
# Nodo #1 [multiply]: value=10
# Nodo #2 [add_ten]: value=20
# Nodo #3 [square]: value=400
# Nodo #4 [report]: value=—
# Total nodos ejecutados: 4

Resumen

En esta cápsula aprendiste:

• StateGraph es la clase fundamental de LangGraph — el contenedor donde defines estado, nodos, edges, compilas, y ejecutas
• El estado se define con TypedDict. Cada nodo recibe el estado completo y retorna solo los campos que quiere actualizar
• Sin Annotated + reducer: cada nodo reemplaza el valor del campo. Con Annotated[tipo, operator.add]: cada nodo agrega al valor existente
• Los reducers son la fuente #1 de bugs para principiantes — si tus mensajes "desaparecen", revisa que tengas operator.add
• START y END son nodos especiales que marcan entrada y salida del grafo
• graph_builder.compile() transforma la definición en un grafo ejecutable
• invoke ejecuta todo y retorna el estado final. stream muestra el output de cada nodo individualmente
• draw_mermaid_png() es tu herramienta principal de debugging — dibujar primero, codear después
• En create_agent, el framework controlaba el flujo. Ahora tú defines cada paso, cada conexión — más poder, más responsabilidad

Próxima cápsula: Nodos: Funciones que Transforman Estado — aprenderás a crear nodos que llaman al modelo, nodos que ejecutan tools, y patrones para diseñar funciones de nodo efectivas.

Recursos adicionales

1. StateGraph Reference — Referencia completa de la clase StateGraph
2. LangGraph Quickstart — Tutorial oficial paso a paso
3. How to define graph state — Guía para diseñar estado con TypedDict, reducers, y Pydantic
4. How to visualize your graph — Opciones de visualización: Mermaid, PNG, y ASCII
5. State reducers — Documentación conceptual de reducers y Annotated
6. TypedDict — Python docs — Referencia de TypedDict para el estado tipado
7. operator module — Python docs — Referencia de operator.add y otros operadores
8. Annotated — Python docs — Referencia de Annotated y cómo LangGraph usa los metadatos de tipo

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