Módulo 5: Introducción a LangGraph
Nodos: Funciones que Transforman Estado
Descripción de la cápsula
En la cápsula anterior construiste tu primer grafo con StateGraph: definiste un estado tipado con TypedDict y Annotated, conectaste START y END, compilaste y ejecutaste. Ahora vas a entender a fondo el componente más importante de cualquier grafo: los nodos.
Un nodo es una función Python normal. Recibe el estado completo del grafo, hace trabajo (llamar a un modelo, ejecutar una tool, clasificar texto, transformar datos), y retorna un diccionario con solo los campos que cambiaron. Eso es todo el contrato. No hay clases especiales, no hay herencia, no hay decoradores obligatorios.
Este concepto es radicalmente diferente a create_agent, donde el framework decide internamente qué funciones ejecutar y en qué orden. Con nodos en StateGraph, tú defines cada paso del workflow. Cada nodo es una pieza visible, testeable y controlable de tu diagrama de flujo.
¿Qué es un nodo?
Un nodo es una función que sigue un contrato simple:
- Recibe el estado completo del grafo (un
TypedDict) - Hace trabajo (cualquier lógica Python)
- Retorna un diccionario con solo los campos que quiere actualizar
from dotenv import load_dotenv
load_dotenv()
from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langchain_core.messages import AnyMessage, HumanMessage, AIMessage
from IPython.display import Image, display
class State(TypedDict):
messages: Annotated[list[AnyMessage], operator.add]
def greet(state: State) -> dict:
last_msg = state["messages"][-1].content
return {"messages": [AIMessage(content=f"¡Hola! Recibí tu mensaje: '{last_msg}'")]}
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()
display(Image(graph.get_graph().draw_mermaid_png()))
result = graph.invoke({"messages": [HumanMessage(content="Buenos días")]})
print(result["messages"][-1].content)
# Output esperado: ¡Hola! Recibí tu mensaje: 'Buenos días'
La función greet recibe todo el estado, pero retorna solo lo que cambió: un nuevo mensaje que se agrega a la lista gracias al reducer operator.add.
El contrato: estado completo de entrada, actualización parcial de salida
Veamos el contrato con un estado que tiene múltiples campos:
from dotenv import load_dotenv
load_dotenv()
from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langchain_core.messages import AnyMessage, HumanMessage, AIMessage
from IPython.display import Image, display
class State(TypedDict):
messages: Annotated[list[AnyMessage], operator.add]
user_name: str
message_count: int
def extract_name(state: State) -> dict:
first_msg = state["messages"][0].content
name = first_msg.split("soy ")[-1].strip(".!") if "soy " in first_msg.lower() else "Usuario"
return {"user_name": name}
def count_messages(state: State) -> dict:
return {"message_count": len(state["messages"])}
def respond(state: State) -> dict:
name = state.get("user_name", "Usuario")
count = state.get("message_count", 0)
return {"messages": [AIMessage(content=f"¡Hola {name}! Llevas {count} mensaje(s).")]}
graph_builder = StateGraph(State)
graph_builder.add_node("extract_name", extract_name)
graph_builder.add_node("count_messages", count_messages)
graph_builder.add_node("respond", respond)
graph_builder.add_edge(START, "extract_name")
graph_builder.add_edge("extract_name", "count_messages")
graph_builder.add_edge("count_messages", "respond")
graph_builder.add_edge("respond", END)
graph = graph_builder.compile()
display(Image(graph.get_graph().draw_mermaid_png()))
result = graph.invoke({
"messages": [HumanMessage(content="Hola, soy Ana")],
"user_name": "",
"message_count": 0,
})
print(result["messages"][-1].content)
# Output esperado: ¡Hola Ana! Llevas 1 mensaje(s).
Cada nodo actualiza únicamente los campos que le corresponden:
extract_nameretorna{"user_name": name}— no tocamessagesnimessage_countcount_messagesretorna{"message_count": ...}— no toca los otros camposrespondretorna{"messages": [...]}— no tocauser_namenimessage_count
add_node() y buenas prácticas de nombrado
add_node() registra una función como nodo con un nombre único:
graph_builder.add_node("nombre_del_nodo", mi_funcion)
| Convención | Ejemplo | Por qué |
|---|---|---|
| snake_case descriptivo | "classify_intent" | Claro en visualización y logs |
| Verbo + sustantivo | "generate_response" | Comunica qué hace el nodo |
| Sin prefijos genéricos | "process" ❌, "process_payment" ✅ | Evita ambigüedad |
| Mismo nombre que función | add_node("chatbot", chatbot) | Fácil de rastrear |
Nodos que llaman al modelo
El caso de uso más común: un nodo que envía los mensajes al LLM y agrega la respuesta:
from dotenv import load_dotenv
load_dotenv()
from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langchain.chat_models import init_chat_model
from langchain_core.messages import AnyMessage, HumanMessage
from IPython.display import Image, display
class State(TypedDict):
messages: Annotated[list[AnyMessage], operator.add]
def chatbot(state: State) -> dict:
model = init_chat_model("openai:gpt-4.1-mini")
response = model.invoke(state["messages"])
return {"messages": [response]}
graph_builder = StateGraph(State)
graph_builder.add_node("chatbot", chatbot)
graph_builder.add_edge(START, "chatbot")
graph_builder.add_edge("chatbot", END)
graph = graph_builder.compile()
display(Image(graph.get_graph().draw_mermaid_png()))
result = graph.invoke({"messages": [HumanMessage(content="¿Qué es LangGraph?")]})
print(result["messages"][-1].content)
# Output esperado: LangGraph es un framework de orquestación para construir
# aplicaciones con LLMs usando grafos de estado...
Gracias a operator.add, la respuesta se agrega a la lista en vez de sobrescribirla.
Nodos que ejecutan tools
Un nodo puede ejecutar herramientas basándose en lo que el modelo pidió:
from dotenv import load_dotenv
load_dotenv()
from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langchain.chat_models import init_chat_model
from langchain_core.messages import AnyMessage, HumanMessage, ToolMessage
from langchain_core.tools import tool
from IPython.display import Image, display
@tool
def get_weather(city: str) -> str:
"""Obtiene el clima actual de una ciudad."""
return f"El clima en {city} es soleado, 22°C"
class State(TypedDict):
messages: Annotated[list[AnyMessage], operator.add]
def call_model(state: State) -> dict:
model = init_chat_model("openai:gpt-4.1-mini")
model_with_tools = model.bind_tools([get_weather])
return {"messages": [model_with_tools.invoke(state["messages"])]}
def run_tools(state: State) -> dict:
last_message = state["messages"][-1]
tool_map = {"get_weather": get_weather}
results = []
for tc in last_message.tool_calls:
result = tool_map[tc["name"]].invoke(tc["args"])
results.append(ToolMessage(content=str(result), tool_call_id=tc["id"]))
return {"messages": results}
graph_builder = StateGraph(State)
graph_builder.add_node("call_model", call_model)
graph_builder.add_node("run_tools", run_tools)
graph_builder.add_edge(START, "call_model")
graph_builder.add_edge("call_model", "run_tools")
graph_builder.add_edge("run_tools", END)
graph = graph_builder.compile()
display(Image(graph.get_graph().draw_mermaid_png()))
result = graph.invoke({"messages": [HumanMessage(content="¿Qué clima hace en Madrid?")]})
for msg in result["messages"]:
print(f"{type(msg).__name__}: {msg.content[:80] if msg.content else msg.tool_calls}")
# Output esperado:
# HumanMessage: ¿Qué clima hace en Madrid?
# AIMessage: [{'name': 'get_weather', 'args': {'city': 'Madrid'}, ...}]
# ToolMessage: El clima en Madrid es soleado, 22°C
Dos nodos especializados: call_model se encarga del LLM, run_tools se encarga de ejecutar las herramientas. Cada uno con responsabilidad clara.
Nodos de lógica pura
No todos los nodos necesitan un modelo. Puedes crear nodos de clasificación, validación o transformación con Python puro:
from dotenv import load_dotenv
load_dotenv()
from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langchain_core.messages import AnyMessage, HumanMessage, AIMessage
from IPython.display import Image, display
class State(TypedDict):
messages: Annotated[list[AnyMessage], operator.add]
intent: str
language: str
def classify_intent(state: State) -> dict:
text = state["messages"][-1].content.lower()
if any(w in text for w in ["código", "code", "programa", "función"]):
return {"intent": "code"}
elif any(w in text for w in ["historia", "cuento", "creativo"]):
return {"intent": "creative"}
return {"intent": "qa"}
def detect_language(state: State) -> dict:
text = state["messages"][-1].content.lower()
spanish_words = ["hola", "qué", "cómo", "por", "para", "el", "la"]
is_spanish = sum(1 for w in text.split() if w in spanish_words) >= 2
return {"language": "es" if is_spanish else "en"}
def respond(state: State) -> dict:
return {"messages": [AIMessage(
content=f"Intención: {state.get('intent', 'qa')} | Idioma: {state.get('language', 'es')}"
)]}
graph_builder = StateGraph(State)
graph_builder.add_node("classify_intent", classify_intent)
graph_builder.add_node("detect_language", detect_language)
graph_builder.add_node("respond", respond)
graph_builder.add_edge(START, "classify_intent")
graph_builder.add_edge("classify_intent", "detect_language")
graph_builder.add_edge("detect_language", "respond")
graph_builder.add_edge("respond", END)
graph = graph_builder.compile()
display(Image(graph.get_graph().draw_mermaid_png()))
result = graph.invoke({
"messages": [HumanMessage(content="¿Cómo hago una función en Python?")],
"intent": "", "language": "",
})
print(result["messages"][-1].content)
# Output esperado: Intención: code | Idioma: es
Los nodos de lógica pura son rápidos (no llaman a ninguna API), determinísticos, y fáciles de testear.
El orden de ejecución lo definen los edges
Un error común: pensar que los nodos se ejecutan en el orden en que los registras con add_node(). El orden lo definen los edges:
from dotenv import load_dotenv
load_dotenv()
from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langchain_core.messages import AnyMessage, HumanMessage, AIMessage
from IPython.display import Image, display
class State(TypedDict):
messages: Annotated[list[AnyMessage], operator.add]
def step_a(state: State) -> dict:
return {"messages": [AIMessage(content="Paso A ejecutado")]}
def step_b(state: State) -> dict:
return {"messages": [AIMessage(content="Paso B ejecutado")]}
def step_c(state: State) -> dict:
return {"messages": [AIMessage(content="Paso C ejecutado")]}
graph_builder = StateGraph(State)
graph_builder.add_node("step_c", step_c) # Registrado primero...
graph_builder.add_node("step_a", step_a)
graph_builder.add_node("step_b", step_b)
graph_builder.add_edge(START, "step_a") # ...pero ejecutado por los edges
graph_builder.add_edge("step_a", "step_b")
graph_builder.add_edge("step_b", "step_c")
graph_builder.add_edge("step_c", END)
graph = graph_builder.compile()
display(Image(graph.get_graph().draw_mermaid_png()))
result = graph.invoke({"messages": [HumanMessage(content="inicio")]})
for msg in result["messages"][1:]:
print(msg.content)
# Output esperado:
# Paso A ejecutado
# Paso B ejecutado
# Paso C ejecutado
Aunque step_c se registró primero, se ejecuta al final porque los edges dicen: A → B → C.
Nodos async
Para operaciones de I/O (llamadas a APIs, bases de datos), usa async def:
from dotenv import load_dotenv
load_dotenv()
from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langchain.chat_models import init_chat_model
from langchain_core.messages import AnyMessage, HumanMessage
from IPython.display import Image, display
class State(TypedDict):
messages: Annotated[list[AnyMessage], operator.add]
async def async_chatbot(state: State) -> dict:
model = init_chat_model("openai:gpt-4.1-mini")
response = await model.ainvoke(state["messages"])
return {"messages": [response]}
graph_builder = StateGraph(State)
graph_builder.add_node("chatbot", async_chatbot)
graph_builder.add_edge(START, "chatbot")
graph_builder.add_edge("chatbot", END)
graph = graph_builder.compile()
display(Image(graph.get_graph().draw_mermaid_png()))
result = await graph.ainvoke({"messages": [HumanMessage(content="¿Qué es async?")]})
print(result["messages"][-1].content)
# Output esperado: Async en Python permite ejecutar operaciones de I/O
# de forma concurrente usando async/await...
| Caso | Sync (def) | Async (async def) |
|---|---|---|
| Prototipos rápidos | ✅ Más simple | Innecesario |
| Múltiples llamadas I/O | Secuencial (lento) | ✅ Concurrente (rápido) |
| APIs web (FastAPI) | Bloquea el event loop | ✅ Compatible nativo |
Comparación: nodos en StateGraph vs funciones en create_agent
| Aspecto | create_agent | StateGraph (nodos manuales) |
|---|---|---|
| Quién define los nodos | El framework (model + tools) | Tú |
| Nombres de nodos | "agent", "tools" (fijos) | Los que tú elijas |
| Lógica del nodo | Predefinida (ReAct loop) | Cualquier función Python |
| Nodos de lógica pura | ❌ No disponible | ✅ Clasificación, validación, etc. |
| Cantidad de nodos | 2 (fijos) | Los que necesites |
| Visibilidad | Opaco | Transparente |
from dotenv import load_dotenv
load_dotenv()
from langchain.agents import create_agent
from langchain_core.tools import tool
from IPython.display import Image, display
@tool
def get_weather(city: str) -> str:
"""Obtiene el clima actual de una ciudad."""
return f"Soleado, 22°C en {city}"
agent = create_agent("openai:gpt-4.1-mini", tools=[get_weather])
display(Image(agent.get_graph().draw_mermaid_png()))
create_agent genera nodos "agent" y "tools" conectados automáticamente. Tú no elegiste esos nombres ni definiste esas funciones. Conveniente para el 80% de casos, pero cuando necesitas nodos custom, necesitas StateGraph.
Troubleshooting
Problema 1: "El nodo no actualiza el estado"
Síntoma: Un campo del estado no cambia después de ejecutar.
Causa: El nodo retorna un diccionario vacío o sin el campo esperado.
Solución: Verifica que el dict de retorno incluye las keys correctas.
Problema 2: "Los mensajes se sobrescriben en vez de acumularse"
Síntoma: Después de cada nodo, solo queda un mensaje.
Causa: Falta el reducer operator.add en el estado.
Solución:
# ❌ Sin reducer — sobrescribe
class State(TypedDict):
messages: list[AnyMessage]
# ✅ Con reducer — acumula
class State(TypedDict):
messages: Annotated[list[AnyMessage], operator.add]
Problema 3: "InvalidUpdateError: Expected dict, got NoneType"
Síntoma: Error porque un nodo retorna None.
Causa: La función no tiene return explícito.
Solución: Siempre retorna un diccionario, aunque sea vacío: return {}
Problema 4: "TypeError al acceder a state['messages'][-1]"
Síntoma: Error porque la lista está vacía. Solución: Verifica antes de acceder:
def safe_node(state: State) -> dict:
if not state["messages"]:
return {"messages": [AIMessage(content="No hay mensajes previos")]}
last = state["messages"][-1]
return {"messages": [AIMessage(content=f"Procesado: {last.content}")]}
Ejercicios
Ejercicio 1: Nodo de transformación (Fácil)
Crea un grafo con un solo nodo "uppercase" que tome el último mensaje del usuario y lo convierta a mayúsculas. Visualiza con draw_mermaid_png().
Ver solución
from dotenv import load_dotenv
load_dotenv()
from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langchain_core.messages import AnyMessage, HumanMessage, AIMessage
from IPython.display import Image, display
class State(TypedDict):
messages: Annotated[list[AnyMessage], operator.add]
def uppercase(state: State) -> dict:
original = state["messages"][-1].content
return {"messages": [AIMessage(content=original.upper())]}
graph_builder = StateGraph(State)
graph_builder.add_node("uppercase", uppercase)
graph_builder.add_edge(START, "uppercase")
graph_builder.add_edge("uppercase", END)
graph = graph_builder.compile()
display(Image(graph.get_graph().draw_mermaid_png()))
result = graph.invoke({"messages": [HumanMessage(content="hola mundo")]})
print(result["messages"][-1].content)
# Output esperado: HOLA MUNDO
Explicación: El nodo lee el último mensaje, transforma su contenido, y retorna un nuevo AIMessage que se agrega a la lista.
Ejercicio 2: Pipeline de dos nodos con validación (Fácil)
Crea un grafo con "validate" (verifica que el mensaje tiene al menos 5 caracteres, guarda is_valid en el estado) y "respond" (genera respuesta según validez). Visualiza el grafo.
Ver solución
from dotenv import load_dotenv
load_dotenv()
from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langchain_core.messages import AnyMessage, HumanMessage, AIMessage
from IPython.display import Image, display
class State(TypedDict):
messages: Annotated[list[AnyMessage], operator.add]
is_valid: bool
def validate(state: State) -> dict:
return {"is_valid": len(state["messages"][-1].content) >= 5}
def respond(state: State) -> dict:
if state["is_valid"]:
return {"messages": [AIMessage(content="Mensaje válido. Procesando...")]}
return {"messages": [AIMessage(content="Mensaje muy corto. Mínimo 5 caracteres.")]}
graph_builder = StateGraph(State)
graph_builder.add_node("validate", validate)
graph_builder.add_node("respond", respond)
graph_builder.add_edge(START, "validate")
graph_builder.add_edge("validate", "respond")
graph_builder.add_edge("respond", END)
graph = graph_builder.compile()
display(Image(graph.get_graph().draw_mermaid_png()))
for msg in ["Hi", "Hola, necesito ayuda con mi proyecto"]:
result = graph.invoke({"messages": [HumanMessage(content=msg)], "is_valid": False})
print(f"'{msg}' → {result['messages'][-1].content}")
# Output esperado:
# 'Hi' → Mensaje muy corto. Mínimo 5 caracteres.
# 'Hola, necesito ayuda con mi proyecto' → Mensaje válido. Procesando...
Explicación: validate solo actualiza is_valid, respond solo agrega un mensaje. Cada nodo con responsabilidad única.
Ejercicio 3: Nodo con model y system prompt dinámico (Medio)
Crea un grafo con un nodo "expert" que use init_chat_model con un system prompt que se personalice según un campo topic del estado. Visualiza el grafo.
Ver solución
from dotenv import load_dotenv
load_dotenv()
from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langchain.chat_models import init_chat_model
from langchain_core.messages import AnyMessage, HumanMessage, SystemMessage
from IPython.display import Image, display
class State(TypedDict):
messages: Annotated[list[AnyMessage], operator.add]
topic: str
def expert(state: State) -> dict:
topic = state.get("topic", "programación")
system = SystemMessage(content=f"Eres un experto en {topic}. Responde conciso en español.")
model = init_chat_model("openai:gpt-4.1-mini")
return {"messages": [model.invoke([system] + state["messages"])]}
graph_builder = StateGraph(State)
graph_builder.add_node("expert", expert)
graph_builder.add_edge(START, "expert")
graph_builder.add_edge("expert", END)
graph = graph_builder.compile()
display(Image(graph.get_graph().draw_mermaid_png()))
result = graph.invoke({
"messages": [HumanMessage(content="¿Qué es un decorador?")],
"topic": "Python",
})
print(result["messages"][-1].content)
# Output esperado: Un decorador en Python es una función que recibe otra función
# y extiende su comportamiento sin modificarla directamente...
Explicación: El nodo lee topic del estado para construir un system prompt dinámico — algo que no puedes hacer con create_agent sin middleware.
Ejercicio 4: Pipeline classify → enrich → respond (Medio)
Crea un grafo de 3 nodos: "classify" (detecta si es pregunta, saludo, o comando), "enrich" (agrega metadata de routing al estado), y "respond" (genera respuesta con la metadata). Visualiza el grafo.
Ver solución
from dotenv import load_dotenv
load_dotenv()
from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langchain_core.messages import AnyMessage, HumanMessage, AIMessage
from IPython.display import Image, display
class State(TypedDict):
messages: Annotated[list[AnyMessage], operator.add]
msg_type: str
route_info: str
def classify(state: State) -> dict:
text = state["messages"][-1].content.lower()
if text.startswith(("¿", "qué", "cómo", "cuándo", "dónde", "por qué")):
return {"msg_type": "question"}
elif any(w in text for w in ["hola", "buenos días", "hey"]):
return {"msg_type": "greeting"}
return {"msg_type": "command"}
def enrich(state: State) -> dict:
info = {"question": "módulo Q&A", "greeting": "módulo bienvenida", "command": "módulo ejecución"}
return {"route_info": info.get(state["msg_type"], "desconocido")}
def respond(state: State) -> dict:
responses = {
"question": "Investigando tu pregunta...",
"greeting": "¡Hola! ¿En qué puedo ayudarte?",
"command": "Procesando tu comando...",
}
text = responses.get(state["msg_type"], "No entendí.")
return {"messages": [AIMessage(content=f"{text} [Routing: {state['route_info']}]")]}
graph_builder = StateGraph(State)
graph_builder.add_node("classify", classify)
graph_builder.add_node("enrich", enrich)
graph_builder.add_node("respond", respond)
graph_builder.add_edge(START, "classify")
graph_builder.add_edge("classify", "enrich")
graph_builder.add_edge("enrich", "respond")
graph_builder.add_edge("respond", END)
graph = graph_builder.compile()
display(Image(graph.get_graph().draw_mermaid_png()))
for msg in ["¿Cómo funciona Python?", "Hola mundo", "Ejecuta el script"]:
result = graph.invoke({"messages": [HumanMessage(content=msg)], "msg_type": "", "route_info": ""})
print(f"'{msg}' → {result['messages'][-1].content}")
# Output esperado:
# '¿Cómo funciona Python?' → Investigando tu pregunta... [Routing: módulo Q&A]
# 'Hola mundo' → ¡Hola! ¿En qué puedo ayudarte? [Routing: módulo bienvenida]
# 'Ejecuta el script' → Procesando tu comando... [Routing: módulo ejecución]
Explicación: Tres nodos en secuencia con responsabilidad única. El diagrama confirma: classify → enrich → respond.
Ejercicio 5: Comparar create_agent vs StateGraph (Avanzado)
Resuelve el mismo problema con ambos enfoques: un asistente con una tool get_time. Primero con create_agent, luego con StateGraph. Visualiza ambos grafos y compara la estructura.
Ver solución
from dotenv import load_dotenv
load_dotenv()
from langchain_core.tools import tool
from IPython.display import Image, display
@tool
def get_time() -> str:
"""Obtiene la hora actual."""
from datetime import datetime
return datetime.now().strftime("%H:%M:%S")
# --- Enfoque 1: create_agent ---
from langchain.agents import create_agent
agent = create_agent("openai:gpt-4.1-mini", tools=[get_time])
result_agent = agent.invoke({"messages": [("user", "¿Qué hora es?")]})
print("=== create_agent ===")
print(result_agent["messages"][-1].content)
display(Image(agent.get_graph().draw_mermaid_png()))
# --- Enfoque 2: StateGraph ---
from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langchain.chat_models import init_chat_model
from langchain_core.messages import AnyMessage, HumanMessage, ToolMessage
class State(TypedDict):
messages: Annotated[list[AnyMessage], operator.add]
def call_model(state: State) -> dict:
model = init_chat_model("openai:gpt-4.1-mini")
return {"messages": [model.bind_tools([get_time]).invoke(state["messages"])]}
def run_tools(state: State) -> dict:
results = []
for tc in state["messages"][-1].tool_calls:
output = get_time.invoke(tc["args"])
results.append(ToolMessage(content=str(output), tool_call_id=tc["id"]))
return {"messages": results}
def final_response(state: State) -> dict:
model = init_chat_model("openai:gpt-4.1-mini")
return {"messages": [model.invoke(state["messages"])]}
graph_builder = StateGraph(State)
graph_builder.add_node("call_model", call_model)
graph_builder.add_node("run_tools", run_tools)
graph_builder.add_node("final_response", final_response)
graph_builder.add_edge(START, "call_model")
graph_builder.add_edge("call_model", "run_tools")
graph_builder.add_edge("run_tools", "final_response")
graph_builder.add_edge("final_response", END)
graph = graph_builder.compile()
print("\n=== StateGraph ===")
result_graph = graph.invoke({"messages": [HumanMessage(content="¿Qué hora es?")]})
print(result_graph["messages"][-1].content)
display(Image(graph.get_graph().draw_mermaid_png()))
# Output esperado: Ambos responden con la hora actual.
# create_agent tiene un loop (conditional edge).
# StateGraph es lineal (aún sin conditional edges).
Explicación: create_agent resuelve en 3 líneas con un grafo opaco. StateGraph requiere más código pero cada nodo es visible. Compara los diagramas: create_agent tiene un loop, tu grafo es lineal — en la próxima cápsula aprenderás a agregar conditional edges para crear loops.
Resumen
En esta cápsula aprendiste:
- Un nodo es una función Python que recibe el estado completo y retorna un diccionario con solo los campos que cambiaron (actualización parcial)
graph_builder.add_node("nombre", funcion)registra una función como nodo del grafo- Los nodos pueden hacer cualquier trabajo: llamar al modelo, ejecutar tools, clasificar, validar, transformar
- El orden de ejecución lo definen los edges, no el orden de
add_node() - Los nodos async (
async def) son útiles para operaciones I/O y apps web - En
create_agent, los nodos son opacos y predefinidos; enStateGraph, tú los defines y controlas completamente draw_mermaid_png()es tu herramienta principal para verificar la estructura del grafo
Próxima cápsula: Edges y Conditional Edges — aprenderás a conectar nodos con edges fijos y condicionales para crear workflows que se ramifican según decisiones dinámicas.
Recursos adicionales
- LangGraph Nodes — Conceptual Guide — Documentación oficial de nodos en LangGraph
- How to create a StateGraph — Tutorial paso a paso con StateGraph
- LangGraph State Management — Cómo funciona el estado tipado y los reducers
- Visualization with draw_mermaid_png — Guía de visualización de grafos
- create_agent API Reference — Referencia para comparar con la API de alto nivel
- LangGraph Async Support — Soporte async en LangGraph
- TypedDict Documentation — Python — Referencia de TypedDict para diseñar estado
Módulo 5 — LangChain & LangGraph: From Chains to Agents