Edges y Conditional Edges

Descripción de la cápsula

En la cápsula anterior aprendiste que los nodos son funciones que transforman estado. Pero un nodo aislado no hace nada — necesita estar conectado a otros nodos para formar un flujo. Esas conexiones son los edges.

Los edges definen el camino que sigue la ejecución de tu grafo. Hay dos tipos: edges fijos (siempre van del nodo A al nodo B) y conditional edges (van a diferentes nodos según una decisión). Los edges fijos son como flechas en un diagrama de flujo. Los conditional edges son como los diamantes de decisión: "¿se cumple esta condición? Sí → ve por aquí. No → ve por allá."

Esta distinción es lo que hace a LangGraph más poderoso que create_agent. Con create_agent, el único flujo posible es el loop ReAct (model → ¿hay tool calls? → tools → model). Con edges condicionales, tú defines las decisiones: clasificar intención, verificar calidad, decidir si hacer loop o terminar.

Edges fijos: conexiones que siempre siguen el mismo camino

Un edge fijo conecta dos nodos de forma permanente. Cada vez que el nodo A termina, la ejecución pasa al nodo B:

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 completado")]}

def step_b(state: State) -> dict:
    return {"messages": [AIMessage(content=f"Paso B recibió: {state['messages'][-1].content}")]}

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()
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 completado
# Paso B recibió: Paso A completado

START y END: entrada y salida del grafo

Todo grafo necesita START (punto de entrada) y END (punto de salida):

from langgraph.graph import START, END

graph_builder.add_edge(START, "mi_primer_nodo")   # Obligatorio
graph_builder.add_edge("mi_ultimo_nodo", END)      # Obligatorio

Sin un edge desde START, el grafo no sabe por dónde empezar. Sin un edge hacia END, no sabe cuándo terminar.

Pipeline lineal: A → B → C → END

Conectar nodos en secuencia crea un pipeline — el patrón más simple pero útil:

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, AIMessage, SystemMessage
from IPython.display import Image, display

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

def detect_language(state: State) -> dict:
    text = state["messages"][-1].content.lower()
    spanish = ["hola", "qué", "cómo", "necesito", "ayuda"]
    return {"language": "es" if any(w in text for w in spanish) else "en"}

def detect_tone(state: State) -> dict:
    text = state["messages"][-1].content.lower()
    if any(w in text for w in ["urgente", "ayuda", "error", "problema"]):
        return {"tone": "urgent"}
    return {"tone": "neutral"}

def generate_response(state: State) -> dict:
    lang = "español" if state.get("language") == "es" else "inglés"
    tone_instruction = "Sé directo y resolutivo." if state.get("tone") == "urgent" else "Sé amigable."
    model = init_chat_model("openai:gpt-4.1-mini")
    response = model.invoke(
        [SystemMessage(content=f"Responde en {lang}. {tone_instruction}")] + state["messages"]
    )
    return {"messages": [response]}

graph_builder = StateGraph(State)
graph_builder.add_node("detect_language", detect_language)
graph_builder.add_node("detect_tone", detect_tone)
graph_builder.add_node("generate_response", generate_response)
graph_builder.add_edge(START, "detect_language")
graph_builder.add_edge("detect_language", "detect_tone")
graph_builder.add_edge("detect_tone", "generate_response")
graph_builder.add_edge("generate_response", END)

graph = graph_builder.compile()
display(Image(graph.get_graph().draw_mermaid_png()))

result = graph.invoke({
    "messages": [HumanMessage(content="¡Necesito ayuda urgente! Mi servidor no responde.")],
    "language": "", "tone": "",
})
print(f"Idioma: {result['language']}, Tono: {result['tone']}")
print(result["messages"][-1].content[:80])
# Output esperado:
# Idioma: es, Tono: urgent
# Entiendo que es urgente. Para diagnosticar por qué tu servidor no responde...

Conditional edges: decisiones dinámicas

Un conditional edge elige destino basado en una función de routing:

graph_builder.add_conditional_edges(
    "nodo_origen",          # Desde qué nodo
    funcion_de_routing,     # Función que decide
    {                       # Mapping: retorno → nodo destino
        "opcion_a": "nodo_a",
        "opcion_b": "nodo_b",
    }
)

La función de routing recibe el estado, evalúa una condición, y retorna un string que indica qué camino tomar.

Tu primer conditional edge

Un grafo que clasifica intención y rutea a nodos especializados:

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]
    intent: str

def classifier(state: State) -> dict:
    text = state["messages"][-1].content.lower()
    if any(w in text for w in ["código", "code", "programa", "función", "bug"]):
        return {"intent": "code"}
    elif any(w in text for w in ["historia", "cuento", "creativo", "escribe"]):
        return {"intent": "creative"}
    return {"intent": "qa"}

def route_by_intent(state: State) -> str:
    return state["intent"]

def code_expert(state: State) -> dict:
    model = init_chat_model("openai:gpt-4.1-mini")
    system = SystemMessage(content="Eres un experto programador. Incluye código en tus respuestas.")
    return {"messages": [model.invoke([system] + state["messages"])]}

def creative_writer(state: State) -> dict:
    model = init_chat_model("openai:gpt-4.1-mini")
    system = SystemMessage(content="Eres un escritor creativo. Usa lenguaje expresivo.")
    return {"messages": [model.invoke([system] + state["messages"])]}

def qa_assistant(state: State) -> dict:
    model = init_chat_model("openai:gpt-4.1-mini")
    system = SystemMessage(content="Eres un asistente conciso. Responde directo.")
    return {"messages": [model.invoke([system] + state["messages"])]}

graph_builder = StateGraph(State)
graph_builder.add_node("classifier", classifier)
graph_builder.add_node("code_expert", code_expert)
graph_builder.add_node("creative_writer", creative_writer)
graph_builder.add_node("qa_assistant", qa_assistant)

graph_builder.add_edge(START, "classifier")
graph_builder.add_conditional_edges(
    "classifier",
    route_by_intent,
    {"code": "code_expert", "creative": "creative_writer", "qa": "qa_assistant"}
)
graph_builder.add_edge("code_expert", END)
graph_builder.add_edge("creative_writer", END)
graph_builder.add_edge("qa_assistant", 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": "",
})
print(f"Intent: {result['intent']}")
print(result["messages"][-1].content[:80])
# Output esperado:
# Intent: code
# Para crear una función en Python, usa la palabra clave `def`...

El diagrama de flujo muestra el diamante de decisión: classifier → tres caminos posibles → END.

Diseñar funciones de routing

La función de routing es simple: recibe estado, retorna string. Debe ser pura, testeable y exhaustiva:

def route_by_intent(state: State) -> str:
    intent = state.get("intent", "qa")
    if intent == "code":
        return "code"
    elif intent == "creative":
        return "creative"
    return "qa"

# Testeable de forma aislada
assert route_by_intent({"intent": "code", "messages": []}) == "code"
assert route_by_intent({"intent": "unknown", "messages": []}) == "qa"
assert route_by_intent({"messages": []}) == "qa"

El mapping: traducir retornos a nodos

El tercer parámetro desacopla la lógica de decisión de los nombres de nodos:

graph_builder.add_conditional_edges(
    "classifier",
    route_by_intent,      # Retorna "code"
    {"code": "code_expert"}  # "code" → nodo "code_expert"
)

Si la función retorna un valor que no está en el mapping, obtendrás un error. Asegúrate de mapear todos los posibles retornos.

Cuándo usar edges fijos vs conditional edges

Regla general: Usa fijos cuando el flujo es siempre el mismo. Usa condicionales cuando el siguiente paso depende de lo que se descubrió antes.

Crear loops: el patrón ReAct manualmente

Un conditional edge puede apuntar a un nodo anterior, creando un loop. Vamos a recrear el loop de create_agent:

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"Soleado, 22°C en {city}"

@tool
def calculator(expression: str) -> str:
    """Calcula una expresión matemática."""
    return str(eval(expression))

tools_list = [get_weather, calculator]
tool_map = {t.name: t for t in tools_list}

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(tools_list)
    return {"messages": [model_with_tools.invoke(state["messages"])]}

def should_continue(state: State) -> str:
    last = state["messages"][-1]
    if hasattr(last, "tool_calls") and last.tool_calls:
        return "tools"
    return "end"

def run_tools(state: State) -> dict:
    last = state["messages"][-1]
    results = []
    for tc in last.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_conditional_edges(
    "call_model",
    should_continue,
    {"tools": "run_tools", "end": END}
)
graph_builder.add_edge("run_tools", "call_model")  # ← El loop

graph = graph_builder.compile()
display(Image(graph.get_graph().draw_mermaid_png()))

result = graph.invoke({
    "messages": [HumanMessage(content="¿Qué clima hace en Madrid y cuánto es 25 * 4?")]
})
print(result["messages"][-1].content)
# Output esperado: El clima en Madrid es soleado con 22°C. Y 25 × 4 = 100.

Esto es exactamente lo que create_agent hace internamente, pero ahora tú controlas cada pieza:

Routing a múltiples nodos (3+)

Los conditional edges pueden dirigir a cualquier número de destinos:

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]

def router(state: State) -> dict:
    return {}

def route_by_content(state: State) -> str:
    text = state["messages"][-1].content.lower()
    if any(w in text for w in ["código", "programa", "bug"]):
        return "code_help"
    elif any(w in text for w in ["escribe", "historia", "poema"]):
        return "creative"
    elif any(w in text for w in ["traduce", "translate", "inglés"]):
        return "translation"
    return "general_qa"

def make_specialist(system_prompt: str):
    def specialist(state: State) -> dict:
        model = init_chat_model("openai:gpt-4.1-mini")
        return {"messages": [model.invoke([SystemMessage(content=system_prompt)] + state["messages"])]}
    return specialist

graph_builder = StateGraph(State)
graph_builder.add_node("router", router)
graph_builder.add_node("code_node", make_specialist("Eres un programador experto. Responde en español."))
graph_builder.add_node("creative_node", make_specialist("Eres un escritor creativo. Responde en español."))
graph_builder.add_node("translation_node", make_specialist("Eres un traductor profesional. Responde en español."))
graph_builder.add_node("qa_node", make_specialist("Eres un asistente conciso. Responde en español."))

graph_builder.add_edge(START, "router")
graph_builder.add_conditional_edges(
    "router", route_by_content,
    {"code_help": "code_node", "creative": "creative_node",
     "translation": "translation_node", "general_qa": "qa_node"}
)
for node in ["code_node", "creative_node", "translation_node", "qa_node"]:
    graph_builder.add_edge(node, END)

graph = graph_builder.compile()
display(Image(graph.get_graph().draw_mermaid_png()))

for msg in ["Arregla este bug", "Escribe un poema", "Traduce 'hello'", "¿Capital de Japón?"]:
    result = graph.invoke({"messages": [HumanMessage(content=msg)]})
    print(f"'{msg}' → {result['messages'][-1].content[:50]}...")
# Output esperado:
# 'Arregla este bug' → Claro, necesito ver el código para poder ayudarte...
# 'Escribe un poema' → La luna, faro de plata sobre el mar silente...
# 'Traduce 'hello'' → "Hello" se traduce como "Hola" en español...
# '¿Capital de Japón?' → La capital de Japón es Tokio...

La función factory make_specialist evita duplicar código para nodos similares. El diagrama muestra 4 rutas saliendo del router.

Comparación: flujo de create_agent vs tu propio flujo

¿Cuándo elegir?

• ✅ create_agent: cuando el loop ReAct es suficiente (80% de los agentes)
• ✅ StateGraph: cuando necesitas routing dinámico, validación, loops custom, o flujos no-lineales

Troubleshooting

Problema 1: "Routing function returned unexpected value"

Síntoma: Error porque la función retornó un valor sin key en el mapping. Solución: Asegúrate de que todos los posibles retornos estén mapeados. Agrega un default:

def route(state: State) -> str:
    intent = state.get("intent", "qa")
    if intent in ("code", "creative", "qa"):
        return intent
    return "qa"  # Default seguro

Problema 2: "Loop infinito — el grafo no termina"

Síntoma: Ejecución indefinida hasta que recursion_limit lo detiene. Solución: La función de routing debe poder retornar un valor que lleve a END. Agrega un contador:

def should_retry(state: State) -> str:
    if state.get("attempts", 0) >= 3:
        return "end"
    return "retry"

Problema 3: "Nodos inalcanzables"

Síntoma: Un nodo nunca se ejecuta. Causa: No hay edge que apunte a ese nodo. Solución: Verifica con draw_mermaid_png(). Si un nodo aparece aislado, le falta un edge de entrada.

Problema 4: "El conditional edge siempre va al mismo nodo"

Síntoma: Sin importar el input, siempre toma el mismo camino. Solución: Prueba la función de routing aislada:

assert route({"intent": "code", "messages": []}) == "code"
assert route({"intent": "creative", "messages": []}) == "creative"

Problema 5: "Falta el edge desde START"

Síntoma: graph.invoke() retorna el estado sin cambios. Solución: Siempre agrega graph_builder.add_edge(START, "primer_nodo") como primer paso.

Ejercicios

Ejercicio 1: Pipeline lineal de 3 pasos (Fácil)

Crea un grafo con "input_cleaner" (elimina espacios extra), "word_counter" (cuenta palabras, guarda en estado), y "summarizer" (reporta el conteo). Edges fijos. Visualiza.

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]
    word_count: int

def input_cleaner(state: State) -> dict:
    cleaned = " ".join(state["messages"][-1].content.split())
    return {"messages": [HumanMessage(content=cleaned)]}

def word_counter(state: State) -> dict:
    return {"word_count": len(state["messages"][-1].content.split())}

def summarizer(state: State) -> dict:
    return {"messages": [AIMessage(content=f"Tu mensaje tiene {state['word_count']} palabras.")]}

graph_builder = StateGraph(State)
graph_builder.add_node("input_cleaner", input_cleaner)
graph_builder.add_node("word_counter", word_counter)
graph_builder.add_node("summarizer", summarizer)
graph_builder.add_edge(START, "input_cleaner")
graph_builder.add_edge("input_cleaner", "word_counter")
graph_builder.add_edge("word_counter", "summarizer")
graph_builder.add_edge("summarizer", END)

graph = graph_builder.compile()
display(Image(graph.get_graph().draw_mermaid_png()))

result = graph.invoke({
    "messages": [HumanMessage(content="  Hola   mundo   desde   LangGraph  ")],
    "word_count": 0,
})
print(result["messages"][-1].content)
# Output esperado: Tu mensaje tiene 4 palabras.

Ejercicio 2: Conditional edge por idioma (Fácil)

Crea un grafo con "checker" que detecte español/inglés, y un conditional edge a "spanish_responder" o "english_responder". Visualiza.

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]
    detected_lang: str

def checker(state: State) -> dict:
    text = state["messages"][-1].content.lower()
    spanish = ["hola", "qué", "cómo", "gracias", "necesito", "ayuda"]
    return {"detected_lang": "es" if any(w in text for w in spanish) else "en"}

def route_language(state: State) -> str:
    return state["detected_lang"]

def spanish_responder(state: State) -> dict:
    return {"messages": [AIMessage(content="¡Entendido! Respondo en español.")]}

def english_responder(state: State) -> dict:
    return {"messages": [AIMessage(content="Got it! I'll respond in English.")]}

graph_builder = StateGraph(State)
graph_builder.add_node("checker", checker)
graph_builder.add_node("spanish_responder", spanish_responder)
graph_builder.add_node("english_responder", english_responder)
graph_builder.add_edge(START, "checker")
graph_builder.add_conditional_edges(
    "checker", route_language,
    {"es": "spanish_responder", "en": "english_responder"}
)
graph_builder.add_edge("spanish_responder", END)
graph_builder.add_edge("english_responder", END)

graph = graph_builder.compile()
display(Image(graph.get_graph().draw_mermaid_png()))

for msg in ["Hola, necesito ayuda", "Hello, I need help"]:
    result = graph.invoke({"messages": [HumanMessage(content=msg)], "detected_lang": ""})
    print(f"'{msg}' → {result['messages'][-1].content}")
# Output esperado:
# 'Hola, necesito ayuda' → ¡Entendido! Respondo en español.
# 'Hello, I need help' → Got it! I'll respond in English.

Ejercicio 3: Loop con condición de salida (Medio)

Crea un grafo con un nodo "improve" que incrementa quality_score en 0.3 cada vez, y un conditional edge que termine si >= 0.8 o vuelva a "improve". Cuenta las iteraciones. Visualiza.

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, AIMessage
from IPython.display import Image, display

class State(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
    quality_score: float
    iterations: int

def improve(state: State) -> dict:
    new_score = min(state.get("quality_score", 0.0) + 0.3, 1.0)
    iters = state.get("iterations", 0) + 1
    return {
        "quality_score": new_score,
        "iterations": iters,
        "messages": [AIMessage(content=f"Iteración {iters}: score = {new_score:.1f}")],
    }

def check_quality(state: State) -> str:
    return "done" if state["quality_score"] >= 0.8 else "improve_more"

graph_builder = StateGraph(State)
graph_builder.add_node("improve", improve)
graph_builder.add_edge(START, "improve")
graph_builder.add_conditional_edges(
    "improve", check_quality,
    {"done": END, "improve_more": "improve"}
)

graph = graph_builder.compile()
display(Image(graph.get_graph().draw_mermaid_png()))

result = graph.invoke({"messages": [], "quality_score": 0.0, "iterations": 0})
print(f"Score final: {result['quality_score']:.1f} | Iteraciones: {result['iterations']}")
for msg in result["messages"]:
    print(f"  {msg.content}")
# Output esperado:
# Score final: 0.9 | Iteraciones: 3
#   Iteración 1: score = 0.3
#   Iteración 2: score = 0.6
#   Iteración 3: score = 0.9

Ejercicio 4: Recrear loop ReAct manualmente (Medio)

Recrea el loop de create_agent: nodo "model" (LLM con tools), conditional edge "should_continue" (¿hay tool calls?), nodo "tools" (ejecuta tools). Usa una tool search. Compara visualmente ambos grafos.

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.agents import create_agent
from langchain_core.messages import AnyMessage, HumanMessage, ToolMessage
from langchain_core.tools import tool
from IPython.display import Image, display

@tool
def search(query: str) -> str:
    """Busca información en internet."""
    return f"Python fue creado por Guido van Rossum en 1991."

tools_list = [search]
tool_map = {t.name: t for t in tools_list}

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

def model_node(state: State) -> dict:
    model = init_chat_model("openai:gpt-4.1-mini")
    return {"messages": [model.bind_tools(tools_list).invoke(state["messages"])]}

def should_continue(state: State) -> str:
    last = state["messages"][-1]
    return "continue" if hasattr(last, "tool_calls") and last.tool_calls else "end"

def tools_node(state: State) -> dict:
    results = []
    for tc in state["messages"][-1].tool_calls:
        output = tool_map[tc["name"]].invoke(tc["args"])
        results.append(ToolMessage(content=str(output), tool_call_id=tc["id"]))
    return {"messages": results}

graph_builder = StateGraph(State)
graph_builder.add_node("model", model_node)
graph_builder.add_node("tools", tools_node)
graph_builder.add_edge(START, "model")
graph_builder.add_conditional_edges("model", should_continue, {"continue": "tools", "end": END})
graph_builder.add_edge("tools", "model")

my_graph = graph_builder.compile()

print("=== Mi grafo ===")
display(Image(my_graph.get_graph().draw_mermaid_png()))
result = my_graph.invoke({"messages": [HumanMessage(content="¿Quién creó Python?")]})
print(result["messages"][-1].content)

print("\n=== create_agent ===")
agent = create_agent("openai:gpt-4.1-mini", tools=[search])
display(Image(agent.get_graph().draw_mermaid_png()))
result_agent = agent.invoke({"messages": [("user", "¿Quién creó Python?")]})
print(result_agent["messages"][-1].content)
# Output esperado: Ambos responden que Python fue creado por Guido van Rossum.

Ejercicio 5: Router con 4 destinos (Avanzado)

Crea un grafo con un nodo "router" y un conditional edge que dirija a 4 nodos especializados (tech, business, science, casual), cada uno con un system prompt diferente. Prueba con 4 mensajes. Visualiza.

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]
    category: str

def classify(state: State) -> dict:
    model = init_chat_model("openai:gpt-4.1-mini")
    prompt = SystemMessage(content=(
        "Clasifica en UNA categoría. Responde SOLO la categoría.\n"
        "Categorías: tech, business, science, casual"
    ))
    result = model.invoke([prompt] + state["messages"])
    cat = result.content.strip().lower()
    return {"category": cat if cat in {"tech", "business", "science", "casual"} else "casual"}

def route_cat(state: State) -> str:
    return state["category"]

def make_node(prompt: str):
    def node(state: State) -> dict:
        model = init_chat_model("openai:gpt-4.1-mini")
        return {"messages": [model.invoke([SystemMessage(content=prompt)] + state["messages"])]}
    return node

graph_builder = StateGraph(State)
graph_builder.add_node("classify", classify)
graph_builder.add_node("tech", make_node("Experto en tecnología. Responde en español."))
graph_builder.add_node("business", make_node("Consultor de negocios. Responde en español."))
graph_builder.add_node("science", make_node("Divulgador científico. Responde en español."))
graph_builder.add_node("casual", make_node("Amigo conversacional. Responde en español."))

graph_builder.add_edge(START, "classify")
graph_builder.add_conditional_edges(
    "classify", route_cat,
    {"tech": "tech", "business": "business", "science": "science", "casual": "casual"}
)
for n in ["tech", "business", "science", "casual"]:
    graph_builder.add_edge(n, END)

graph = graph_builder.compile()
display(Image(graph.get_graph().draw_mermaid_png()))

for msg in ["¿Cómo funciona Docker?", "¿Cómo escalar mi startup?", "¿Por qué el cielo es azul?", "¿Qué tal?"]:
    result = graph.invoke({"messages": [HumanMessage(content=msg)], "category": ""})
    print(f"[{result['category']}] {msg} → {result['messages'][-1].content[:60]}...")
# Output esperado:
# [tech] ¿Cómo funciona Docker? → Docker es una plataforma de contenedores...
# [business] ¿Cómo escalar mi startup? → Para escalar tu startup...
# [science] ¿Por qué el cielo es azul? → El cielo es azul por la dispersión...
# [casual] ¿Qué tal? → ¡Muy bien! ¿Y tú qué tal?...

Resumen

En esta cápsula aprendiste:

• Edges fijos (add_edge("a", "b")) conectan nodos permanentemente — la ejecución siempre sigue el mismo camino
• START y END son nodos especiales obligatorios que marcan entrada y salida del grafo
• Conditional edges (add_conditional_edges) eligen el siguiente nodo basándose en una función de routing que retorna un string
• El mapping parameter traduce los retornos de la función de routing a nombres de nodos destino
• Las funciones de routing deben ser puras, simples, testeables y exhaustivas
• Puedes crear loops con conditional edges que apunten a nodos anteriores — así funciona el loop ReAct
• Recreaste manualmente el loop de create_agent: model → ¿tool calls? → tools → model
• create_agent tiene un flujo fijo. StateGraph te da libertad total para diseñar cualquier diagrama de flujo
• draw_mermaid_png() muestra los diamantes de decisión y caminos posibles — es tu mapa del workflow

Próxima cápsula: Estado tipado con TypedDict y Annotated — profundizarás en diseño del estado, reducers, MessagesState prebuilt, y estados complejos para grafos reales.

Recursos adicionales

1. LangGraph Edges — Conceptual Guide — Documentación oficial de edges y conditional edges
2. How to add conditional edges — Tutorial de branching con conditional edges
3. LangGraph Visualization — Visualizar grafos con draw_mermaid_png
4. How to create a ReAct agent from scratch — Implementar el loop ReAct manualmente
5. create_agent vs custom graphs — Cuándo usar cada approach
6. LangGraph Recursion Limit — Control de loops infinitos
7. StateGraph API Reference — Referencia completa de StateGraph

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