Pattern Router

Descripción de la cápsula

Un router clasifica la solicitud entrante y la dirige al especialista correcto. A diferencia de un supervisor (que coordina múltiples agentes a lo largo de múltiples pasos), un router toma UNA decisión: "¿quién debería manejar esto?" Piensa en la diferencia como un recepcionista vs un project manager. El recepcionista te dice "ve al piso 3, oficina de contabilidad" y su trabajo terminó. El project manager asigna tareas, revisa resultados, pide correcciones y coordina el trabajo de todos.

En las cápsulas 02-04 construiste sistemas con Supervisor, Handoffs y Subagents. En todos esos patrones, hay un agente que orquesta: decide, delega, revisa, y vuelve a delegar. El router es deliberadamente más simple — es el pattern al que recurres cuando la complejidad de un supervisor no se justifica. Si tu sistema solo necesita clasificar y delegar, un router es más rápido, más barato y más fácil de debuggear.

Esta cápsula empieza con routers determinísticos (sin LLM, pura lógica), avanza a routers basados en LLM con structured output, y termina con patterns avanzados como multi-level routing y fallbacks.

Router vs Supervisor: la distinción clave

Antes de escribir código, necesitas tener clara la diferencia:

La regla: si después de delegar al especialista, el trabajo está hecho, usa un router. Si necesitas revisar el resultado y posiblemente delegar a otro agente, necesitas un supervisor.

Router determinístico: reglas sin LLM

El router más simple no usa LLM en absoluto. Clasifica con reglas: keywords, regex, longitud del mensaje, metadata. Es el más rápido, el más barato y el más predecible.

Ejemplo básico: keyword matching

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 RouterState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
    routed_to: str

def deterministic_router(state: RouterState) -> dict:
    text = state["messages"][-1].content.lower()

    if any(kw in text for kw in ["código", "programa", "bug", "función", "error de código"]):
        return {"routed_to": "code_agent"}
    elif any(kw in text for kw in ["escribe", "redacta", "artículo", "blog", "texto"]):
        return {"routed_to": "writer_agent"}
    else:
        return {"routed_to": "general_agent"}

def route_decision(state: RouterState) -> str:
    return state["routed_to"]

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

graph = StateGraph(RouterState)
graph.add_node("router", deterministic_router)
graph.add_node("code_agent", make_specialist(
    "Eres un experto en programación. Responde en español con código cuando sea relevante."
))
graph.add_node("writer_agent", make_specialist(
    "Eres un escritor profesional. Responde en español con prosa clara y estructurada."
))
graph.add_node("general_agent", make_specialist(
    "Eres un asistente general. Responde en español de forma amigable."
))

graph.add_edge(START, "router")
graph.add_conditional_edges("router", route_decision, {
    "code_agent": "code_agent",
    "writer_agent": "writer_agent",
    "general_agent": "general_agent",
})
for agent in ["code_agent", "writer_agent", "general_agent"]:
    graph.add_edge(agent, END)

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

result = app.invoke({
    "messages": [HumanMessage(content="Tengo un bug en mi función de Python")],
    "routed_to": "",
})
print(f"Routed to: {result['routed_to']}")
print(result["messages"][-1].content[:120])
# Output:
# Routed to: code_agent
# Claro, voy a ayudarte a resolver ese bug. ¿Podrías compartir el código de tu función...

El flujo es lineal: START → router → specialist → END. El router no usa LLM — es pura lógica de Python. Eso significa latencia cero en la decisión de routing y costo cero de tokens.

Variantes determinísticas

El keyword matching es la forma más simple, pero no es la única:

import re

def regex_router(state: RouterState) -> dict:
    text = state["messages"][-1].content

    if re.search(r"def\s+\w+|class\s+\w+|import\s+\w+", text):
        return {"routed_to": "code_agent"}
    elif re.search(r"\d+[\+\-\*/]\d+|calcula|porcentaje", text.lower()):
        return {"routed_to": "math_agent"}
    else:
        return {"routed_to": "general_agent"}

def length_router(state: RouterState) -> dict:
    text = state["messages"][-1].content
    word_count = len(text.split())

    if word_count > 200:
        return {"routed_to": "summarizer_agent"}
    elif word_count < 10:
        return {"routed_to": "clarification_agent"}
    else:
        return {"routed_to": "general_agent"}

def metadata_router(state: RouterState) -> dict:
    """Router basado en metadata del mensaje, no en contenido."""
    last_msg = state["messages"][-1]
    if hasattr(last_msg, "additional_kwargs"):
        lang = last_msg.additional_kwargs.get("language", "es")
        if lang == "en":
            return {"routed_to": "english_agent"}
    return {"routed_to": "spanish_agent"}

Router basado en LLM: clasificación con modelo

Cuando las reglas no son suficientes — porque los inputs son ambiguos, los usuarios usan lenguaje variado, o las categorías son sutiles — usas un LLM para clasificar. La clave es structured output: el modelo retorna una categoría específica, no texto libre.

Clasificación con structured output

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated, Literal
import operator
from pydantic import BaseModel, Field
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 IntentClassification(BaseModel):
    """Clasificación del intent del usuario."""
    intent: Literal["research", "analysis", "creative", "code", "general"] = Field(
        description="La categoría que mejor describe la solicitud del usuario"
    )
    reasoning: str = Field(
        description="Breve explicación de por qué se eligió esta categoría"
    )

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

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

def llm_router(state: LLMRouterState) -> dict:
    structured_model = classifier.with_structured_output(IntentClassification)
    result = structured_model.invoke([
        SystemMessage(content=(
            "Clasifica la solicitud del usuario en una de estas categorías:\n"
            "- research: buscar información, investigar un tema, comparar opciones\n"
            "- analysis: analizar datos, encontrar patrones, evaluar métricas\n"
            "- creative: escribir contenido, generar ideas, crear narrativas\n"
            "- code: escribir código, debuggear, explicar conceptos de programación\n"
            "- general: cualquier cosa que no encaje en las anteriores"
        )),
        state["messages"][-1],
    ])
    return {
        "intent": result.intent,
        "routing_reasoning": result.reasoning,
    }

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

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

specialists = {
    "research": ("Eres un investigador. Busca información precisa y cita fuentes. Responde en español."),
    "analysis": ("Eres un analista de datos. Identifica patrones y presenta conclusiones. Responde en español."),
    "creative": ("Eres un escritor creativo. Genera contenido original y atractivo. Responde en español."),
    "code": ("Eres un ingeniero de software. Escribe código limpio con explicaciones. Responde en español."),
    "general": ("Eres un asistente general. Responde de forma clara y amigable. Responde en español."),
}

graph = StateGraph(LLMRouterState)
graph.add_node("router", llm_router)

for intent_name, system_prompt in specialists.items():
    graph.add_node(f"{intent_name}_agent", make_specialist(intent_name, system_prompt))

graph.add_edge(START, "router")
graph.add_conditional_edges("router", route_by_intent, {
    name: f"{name}_agent" for name in specialists
})
for name in specialists:
    graph.add_edge(f"{name}_agent", END)

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

result = app.invoke({
    "messages": [HumanMessage(content="¿Cuáles son las diferencias entre React y Vue para proyectos grandes?")],
    "intent": "",
    "routing_reasoning": "",
})
print(f"Intent: {result['intent']}")
print(f"Reasoning: {result['routing_reasoning']}")
print(result["messages"][-1].content[:150])
# Output:
# Intent: research
# Reasoning: El usuario quiere comparar dos frameworks, lo cual es una tarea de investigación
# React y Vue son frameworks de JavaScript con enfoques diferentes. React usa un modelo...

El LLM clasifica la solicitud en una de 5 categorías y explica por qué. El with_structured_output garantiza que el resultado es una de las categorías válidas — no hay riesgo de que el modelo invente una categoría que no existe.

Router con Functional API

Si tu flujo es simple (clasificar → delegar → responder), la Functional API es más directa:

from dotenv import load_dotenv
load_dotenv()

from langgraph.func import entrypoint, task
from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage, SystemMessage

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

def classify_intent(text: str) -> str:
    text_lower = text.lower()
    if any(kw in text_lower for kw in ["código", "programa", "bug", "función"]):
        return "code"
    elif any(kw in text_lower for kw in ["escribe", "redacta", "artículo", "blog"]):
        return "creative"
    elif any(kw in text_lower for kw in ["investiga", "compara", "diferencias", "qué es"]):
        return "research"
    return "general"

SPECIALIST_PROMPTS = {
    "code": "Eres un experto en programación. Responde en español.",
    "creative": "Eres un escritor profesional. Responde en español.",
    "research": "Eres un investigador. Responde en español con fuentes.",
    "general": "Eres un asistente general. Responde en español.",
}

@task
def ask_specialist(question: str, system_prompt: str) -> str:
    response = model.invoke([
        SystemMessage(content=system_prompt),
        HumanMessage(content=question),
    ])
    return response.content

@entrypoint()
def router_agent(question: str) -> dict:
    intent = classify_intent(question)
    prompt = SPECIALIST_PROMPTS[intent]
    answer = ask_specialist(question, prompt).result()
    return {"intent": intent, "answer": answer}

result = router_agent.invoke("Escribe un artículo corto sobre inteligencia artificial")
print(f"Intent: {result['intent']}")
print(result["answer"][:120])
# Output:
# Intent: creative
# La inteligencia artificial ha dejado de ser ciencia ficción para convertirse en una...

Sin StateGraph, sin edges, sin nodos. La clasificación es un if/elif/else y la delegación es un dict lookup. Para routers simples con 3-5 destinos, este approach es más limpio.

Multi-level routing

Cuando tienes muchos destinos posibles (10+), un solo router se vuelve impreciso. La solución: routing jerárquico. Primero clasifica por dominio, luego por subtarea dentro del dominio.

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated, Literal
import operator
from pydantic import BaseModel, Field
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 DomainClassification(BaseModel):
    domain: Literal["engineering", "business", "support"] = Field(
        description="Dominio principal de la solicitud"
    )

class EngineeringSubtask(BaseModel):
    subtask: Literal["frontend", "backend", "devops"] = Field(
        description="Subtarea dentro de engineering"
    )

class MultiLevelState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
    domain: str
    subtask: str

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

def domain_router(state: MultiLevelState) -> dict:
    structured = classifier.with_structured_output(DomainClassification)
    result = structured.invoke([
        SystemMessage(content=(
            "Clasifica en: engineering (código, infraestructura, tech), "
            "business (ventas, métricas, estrategia), "
            "support (ayuda al usuario, FAQ, troubleshooting)"
        )),
        state["messages"][-1],
    ])
    return {"domain": result.domain}

def route_domain(state: MultiLevelState) -> str:
    return state["domain"]

def engineering_subrouter(state: MultiLevelState) -> dict:
    structured = classifier.with_structured_output(EngineeringSubtask)
    result = structured.invoke([
        SystemMessage(content=(
            "Clasifica la tarea de engineering en: "
            "frontend (UI, React, CSS), backend (API, DB, Python), devops (deploy, CI/CD, Docker)"
        )),
        state["messages"][-1],
    ])
    return {"subtask": result.subtask}

def route_engineering(state: MultiLevelState) -> str:
    return state["subtask"]

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

graph = StateGraph(MultiLevelState)

graph.add_node("domain_router", domain_router)
graph.add_node("eng_subrouter", engineering_subrouter)
graph.add_node("business_agent", make_agent("Experto en negocios. Responde en español."))
graph.add_node("support_agent", make_agent("Agente de soporte. Responde en español."))
graph.add_node("frontend_agent", make_agent("Experto en frontend (React, CSS). Responde en español."))
graph.add_node("backend_agent", make_agent("Experto en backend (Python, APIs). Responde en español."))
graph.add_node("devops_agent", make_agent("Experto en DevOps (Docker, CI/CD). Responde en español."))

graph.add_edge(START, "domain_router")
graph.add_conditional_edges("domain_router", route_domain, {
    "engineering": "eng_subrouter",
    "business": "business_agent",
    "support": "support_agent",
})
graph.add_conditional_edges("eng_subrouter", route_engineering, {
    "frontend": "frontend_agent",
    "backend": "backend_agent",
    "devops": "devops_agent",
})
for agent in ["business_agent", "support_agent", "frontend_agent", "backend_agent", "devops_agent"]:
    graph.add_edge(agent, END)

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

result = app.invoke({
    "messages": [HumanMessage(content="¿Cómo configuro un Dockerfile multi-stage para mi app de Python?")],
    "domain": "",
    "subtask": "",
})
print(f"Domain: {result['domain']}, Subtask: {result['subtask']}")
print(result["messages"][-1].content[:150])
# Output:
# Domain: engineering, Subtask: devops
# Para configurar un Dockerfile multi-stage en Python, necesitas definir dos etapas...

El primer router clasifica en 3 dominios. Si el dominio es engineering, un segundo router refina a 3 subtareas. Así cubres 5 destinos finales con decisiones de 3 opciones cada una — más preciso que un solo router de 5 opciones.

Router con fallback

¿Qué pasa cuando el router no está seguro? En lugar de adivinar, pide clarificación.

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated, Literal
import operator
from pydantic import BaseModel, Field
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 ConfidentClassification(BaseModel):
    intent: Literal["code", "writing", "research", "unclear"] = Field(
        description="Intent del usuario. Usa 'unclear' si no puedes clasificar con confianza."
    )
    confidence: float = Field(
        description="Confianza en la clasificación, de 0.0 a 1.0"
    )

class FallbackState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
    intent: str
    confidence: float

classifier = init_chat_model("openai:gpt-4.1-mini")
CONFIDENCE_THRESHOLD = 0.7

def classify_with_confidence(state: FallbackState) -> dict:
    structured = classifier.with_structured_output(ConfidentClassification)
    result = structured.invoke([
        SystemMessage(content=(
            "Clasifica la solicitud. Si es ambigua o podrías confundir dos categorías, "
            "marca 'unclear' y baja la confianza."
        )),
        state["messages"][-1],
    ])
    return {"intent": result.intent, "confidence": result.confidence}

def route_with_fallback(state: FallbackState) -> str:
    if state["confidence"] < CONFIDENCE_THRESHOLD or state["intent"] == "unclear":
        return "clarification"
    return state["intent"]

def clarification_node(state: FallbackState) -> dict:
    return {"messages": [AIMessage(content=(
        "No estoy seguro de cómo ayudarte mejor. ¿Podrías decirme si necesitas:\n"
        "1. Ayuda con código\n"
        "2. Que escriba contenido\n"
        "3. Que investigue un tema"
    ))]}

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

graph = StateGraph(FallbackState)
graph.add_node("classify", classify_with_confidence)
graph.add_node("clarification", clarification_node)
graph.add_node("code", make_agent("Experto en código. Responde en español."))
graph.add_node("writing", make_agent("Escritor profesional. Responde en español."))
graph.add_node("research", make_agent("Investigador. Responde en español."))

graph.add_edge(START, "classify")
graph.add_conditional_edges("classify", route_with_fallback, {
    "code": "code",
    "writing": "writing",
    "research": "research",
    "clarification": "clarification",
})
for node in ["code", "writing", "research", "clarification"]:
    graph.add_edge(node, END)

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

result = app.invoke({
    "messages": [HumanMessage(content="Python")],
    "intent": "",
    "confidence": 0.0,
})
print(f"Intent: {result['intent']}, Confidence: {result['confidence']}")
print(result["messages"][-1].content[:200])
# Output (ejemplo con input ambiguo):
# Intent: unclear, Confidence: 0.3
# No estoy seguro de cómo ayudarte mejor. ¿Podrías decirme si necesitas:
# 1. Ayuda con código
# 2. Que escriba contenido
# 3. Que investigue un tema

El modelo retorna una confianza junto con la clasificación. Si la confianza está por debajo del umbral (0.7), el sistema pide clarificación en vez de enviar al usuario al especialista equivocado. Esto es especialmente importante cuando los errores de routing tienen costo (ej: mandar una queja al departamento equivocado).

Tabla de comparación: determinístico vs LLM

Cuándo escalar de determinístico a LLM

Empieza siempre con un router determinístico. Escala a LLM solo cuando:

• ❌ Los usuarios usan lenguaje variado que tus keywords no cubren ("ayúdame con este pedazo de script" → ¿es code?)
• ❌ Tienes más de 5-6 categorías y las reglas se vuelven frágiles
• ❌ Los inputs son multi-intención ("investiga sobre Docker y escríbeme un resumen")
• ❌ Tu equipo no quiere mantener una lista creciente de keywords

Y mantente en determinístico cuando:

• ✅ Los inputs son estructurados (formularios, comandos, metadata)
• ✅ Tienes 2-3 categorías bien diferenciadas
• ✅ La latencia y el costo importan (alto volumen)
• ✅ Necesitas comportamiento 100% reproducible para testing

La progresión natural es: determinístico → determinístico + LLM fallback → LLM completo. No saltes al final sin necesidad.

Troubleshooting

Problema 1: El router determinístico rutea al agente incorrecto

Síntoma: El usuario dice "¿Puedes escribirme una función?" y el router lo envía al writer_agent en vez del code_agent porque "escribe" matcheó primero. Causa: El orden de evaluación de las keywords crea conflictos. "Escribir" aparece tanto en contextos de código como de redacción. Solución: Prioriza keywords más específicas primero, o usa combinaciones de keywords en lugar de keywords individuales:

def improved_router(state: RouterState) -> dict:
    text = state["messages"][-1].content.lower()

    if any(kw in text for kw in ["función", "bug", "código", "programa", "variable"]):
        return {"routed_to": "code_agent"}
    elif any(kw in text for kw in ["artículo", "blog", "redacta", "párrafo"]):
        return {"routed_to": "writer_agent"}
    return {"routed_to": "general_agent"}

Problema 2: El router LLM retorna una categoría que no existe en los edges

Síntoma: ValueError: Expected one of ['code', 'writing', 'research'] al ejecutar el grafo. Causa: El modelo generó una categoría fuera de las opciones definidas (ej: "programming" en vez de "code"). Solución: Usa Literal en tu Pydantic model para restringir las opciones, y with_structured_output para garantizar conformance:

class Classification(BaseModel):
    intent: Literal["code", "writing", "research"]  # Solo estas 3 opciones

Problema 3: El router LLM es demasiado lento para la experiencia del usuario

Síntoma: El usuario espera 1-2 segundos solo para que el sistema decida a quién enviarle la solicitud, antes de que el especialista empiece a trabajar. Causa: La clasificación LLM agrega latencia antes de la respuesta real. Solución: Usa un modelo más rápido para clasificación (ej: gpt-4.1-nano) y reserva modelos más capaces para los especialistas:

fast_classifier = init_chat_model("openai:gpt-4.1-nano")
smart_specialist = init_chat_model("openai:gpt-4.1")

Problema 4: Multi-level routing hace demasiadas llamadas LLM

Síntoma: Con 2 niveles de routing LLM, la latencia total de clasificación excede 2 segundos antes de que ningún especialista trabaje. Causa: Cada nivel de routing es una llamada secuencial al LLM. Solución: Haz el primer nivel determinístico (rápido, sin LLM) y solo el segundo nivel con LLM:

def hybrid_routing(state):
    text = state["messages"][-1].content.lower()
    if any(kw in text for kw in ["código", "api", "deploy", "docker"]):
        return {"domain": "engineering"}
    elif any(kw in text for kw in ["ventas", "revenue", "cliente"]):
        return {"domain": "business"}
    return {"domain": "support"}

Ejercicios

Ejercicio 1: Router determinístico para soporte (Fácil)

Crea un router determinístico con 3 destinos: billing (facturación), technical (problemas técnicos), y general. Usa keyword matching. Pruébalo con 3 mensajes diferentes.

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

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

def support_router(state: SupportState) -> dict:
    text = state["messages"][-1].content.lower()
    if any(kw in text for kw in ["factura", "cobro", "pago", "precio", "plan"]):
        return {"department": "billing"}
    elif any(kw in text for kw in ["error", "no funciona", "caído", "bug", "lento"]):
        return {"department": "technical"}
    return {"department": "general"}

def route_dept(state: SupportState) -> str:
    return state["department"]

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

graph = StateGraph(SupportState)
graph.add_node("router", support_router)
graph.add_node("billing", make_agent("Agente de facturación. Responde en español."))
graph.add_node("technical", make_agent("Soporte técnico. Responde en español."))
graph.add_node("general", make_agent("Asistente general. Responde en español."))

graph.add_edge(START, "router")
graph.add_conditional_edges("router", route_dept, {
    "billing": "billing", "technical": "technical", "general": "general"
})
for node in ["billing", "technical", "general"]:
    graph.add_edge(node, END)

app = graph.compile()

tests = [
    "¿Cuánto cuesta el plan premium?",
    "Mi app no funciona desde ayer",
    "¿Cuál es su horario de atención?",
]
for msg in tests:
    result = app.invoke({"messages": [HumanMessage(content=msg)], "department": ""})
    print(f"'{msg[:40]}...' → {result['department']}")
# Output:
# '¿Cuánto cuesta el plan premium?...' → billing
# 'Mi app no funciona desde ayer...' → technical
# '¿Cuál es su horario de atención?...' → general

Ejercicio 2: Router LLM con structured output (Fácil)

Convierte el router determinístico del ejercicio 1 a un router basado en LLM usando with_structured_output. Compara los resultados con el input "me cobraron dos veces este mes" (que un router de keywords podría enviar a general porque no contiene exactamente "factura").

from dotenv import load_dotenv
load_dotenv()

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

class DeptClassification(BaseModel):
    department: Literal["billing", "technical", "general"] = Field(
        description="Departamento que debe manejar esta solicitud"
    )

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

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

def llm_support_router(state: SupportState) -> dict:
    structured = classifier.with_structured_output(DeptClassification)
    result = structured.invoke([
        SystemMessage(content=(
            "Clasifica la solicitud del usuario:\n"
            "- billing: cualquier tema de pagos, cobros, facturas, planes, precios\n"
            "- technical: problemas técnicos, errores, rendimiento, bugs\n"
            "- general: cualquier otra cosa"
        )),
        state["messages"][-1],
    ])
    return {"department": result.department}

def route_dept(state: SupportState) -> str:
    return state["department"]

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

graph = StateGraph(SupportState)
graph.add_node("router", llm_support_router)
graph.add_node("billing", make_agent("Agente de facturación. Responde en español."))
graph.add_node("technical", make_agent("Soporte técnico. Responde en español."))
graph.add_node("general", make_agent("Asistente general. Responde en español."))

graph.add_edge(START, "router")
graph.add_conditional_edges("router", route_dept, {
    "billing": "billing", "technical": "technical", "general": "general"
})
for node in ["billing", "technical", "general"]:
    graph.add_edge(node, END)

app = graph.compile()

result = app.invoke({
    "messages": [HumanMessage(content="Me cobraron dos veces este mes")],
    "department": "",
})
print(f"Department: {result['department']}")
print(result["messages"][-1].content[:120])
# Output:
# Department: billing
# Lamento mucho el inconveniente con el cobro duplicado. Voy a ayudarte a resolver...

Ejercicio 3: Router con Functional API (Medio)

Implementa un router con la Functional API que clasifique preguntas en 4 categorías: math, history, science, other. Usa un router determinístico. Prueba con 4 inputs diferentes.

from dotenv import load_dotenv
load_dotenv()

from langgraph.func import entrypoint, task
from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage, SystemMessage

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

def classify_subject(text: str) -> str:
    text_lower = text.lower()
    if any(kw in text_lower for kw in ["calcula", "ecuación", "suma", "porcentaje", "número"]):
        return "math"
    elif any(kw in text_lower for kw in ["historia", "guerra", "siglo", "revolución", "antiguo"]):
        return "history"
    elif any(kw in text_lower for kw in ["célula", "átomo", "planeta", "química", "física"]):
        return "science"
    return "other"

PROMPTS = {
    "math": "Eres un profesor de matemáticas. Explica paso a paso. Responde en español.",
    "history": "Eres un historiador. Da contexto y fechas. Responde en español.",
    "science": "Eres un científico. Explica con precisión. Responde en español.",
    "other": "Eres un tutor general. Responde en español de forma clara.",
}

@task
def answer_question(question: str, system_prompt: str) -> str:
    return model.invoke([
        SystemMessage(content=system_prompt),
        HumanMessage(content=question),
    ]).content

@entrypoint()
def tutor_router(question: str) -> dict:
    subject = classify_subject(question)
    answer = answer_question(question, PROMPTS[subject]).result()
    return {"subject": subject, "answer": answer}

tests = [
    "¿Cuánto es el 15% de 240?",
    "¿Qué causó la Revolución Francesa?",
    "¿Cómo funciona un átomo?",
    "¿Cuál es la capital de Japón?",
]
for q in tests:
    result = tutor_router.invoke(q)
    print(f"'{q[:35]}...' → {result['subject']}")
# Output:
# '¿Cuánto es el 15% de 240?...' → math
# '¿Qué causó la Revolución Francesa?...' → history
# '¿Cómo funciona un átomo?...' → science
# '¿Cuál es la capital de Japón?...' → other

Ejercicio 4: Multi-level router (Medio)

Crea un router de 2 niveles. Nivel 1: clasifica en tech o business (determinístico). Nivel 2: si es tech, usa LLM para sub-clasificar en frontend, backend, o data. Implementa con StateGraph.

from dotenv import load_dotenv
load_dotenv()

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

class TechSubclass(BaseModel):
    area: Literal["frontend", "backend", "data"] = Field(
        description="Área técnica específica"
    )

class TwoLevelState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
    level1: str
    level2: str

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

def level1_router(state: TwoLevelState) -> dict:
    text = state["messages"][-1].content.lower()
    if any(kw in text for kw in ["código", "api", "base de datos", "react", "deploy", "python"]):
        return {"level1": "tech"}
    return {"level1": "business"}

def route_level1(state: TwoLevelState) -> str:
    return state["level1"]

def level2_tech_router(state: TwoLevelState) -> dict:
    structured = classifier.with_structured_output(TechSubclass)
    result = structured.invoke([
        SystemMessage(content=(
            "Clasifica esta pregunta técnica en:\n"
            "- frontend: UI, React, CSS, JavaScript, diseño web\n"
            "- backend: APIs, servidores, Python, bases de datos, autenticación\n"
            "- data: datos, análisis, ML, pandas, visualización"
        )),
        state["messages"][-1],
    ])
    return {"level2": result.area}

def route_level2(state: TwoLevelState) -> str:
    return state["level2"]

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

graph = StateGraph(TwoLevelState)
graph.add_node("level1", level1_router)
graph.add_node("level2_tech", level2_tech_router)
graph.add_node("business_agent", make_agent("Experto en negocios. Responde en español."))
graph.add_node("frontend_agent", make_agent("Experto en frontend. Responde en español."))
graph.add_node("backend_agent", make_agent("Experto en backend. Responde en español."))
graph.add_node("data_agent", make_agent("Experto en datos y ML. Responde en español."))

graph.add_edge(START, "level1")
graph.add_conditional_edges("level1", route_level1, {
    "tech": "level2_tech",
    "business": "business_agent",
})
graph.add_conditional_edges("level2_tech", route_level2, {
    "frontend": "frontend_agent",
    "backend": "backend_agent",
    "data": "data_agent",
})
for node in ["business_agent", "frontend_agent", "backend_agent", "data_agent"]:
    graph.add_edge(node, END)

app = graph.compile()

result = app.invoke({
    "messages": [HumanMessage(content="¿Cómo optimizo una query SQL que tarda 30 segundos?")],
    "level1": "",
    "level2": "",
})
print(f"Level 1: {result['level1']}, Level 2: {result['level2']}")
print(result["messages"][-1].content[:120])
# Output:
# Level 1: tech, Level 2: backend
# Para optimizar una query SQL lenta, primero necesitas identificar el cuello de botella...

Ejercicio 5: Router con fallback y confianza (Avanzado)

Implementa un router LLM que retorne una confianza. Si la confianza es < 0.6, el sistema debe pedir clarificación. Si es entre 0.6 y 0.8, debe rutear pero agregar un disclaimer. Si es > 0.8, rutea normalmente. Usa StateGraph.

from dotenv import load_dotenv
load_dotenv()

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

class SmartClassification(BaseModel):
    intent: Literal["code", "writing", "research", "unclear"] = Field(
        description="Intent del usuario"
    )
    confidence: float = Field(description="Confianza de 0.0 a 1.0")

class SmartRouterState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
    intent: str
    confidence: float
    confidence_tier: str

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

def smart_classifier(state: SmartRouterState) -> dict:
    structured = classifier.with_structured_output(SmartClassification)
    result = structured.invoke([
        SystemMessage(content=(
            "Clasifica la solicitud. Si es ambigua, usa 'unclear' con confianza baja. "
            "Sé honesto con tu nivel de confianza."
        )),
        state["messages"][-1],
    ])
    if result.confidence < 0.6:
        tier = "low"
    elif result.confidence < 0.8:
        tier = "medium"
    else:
        tier = "high"
    return {
        "intent": result.intent,
        "confidence": result.confidence,
        "confidence_tier": tier,
    }

def route_by_confidence(state: SmartRouterState) -> str:
    if state["confidence_tier"] == "low":
        return "clarification"
    return state["intent"]

def clarification_node(state: SmartRouterState) -> dict:
    return {"messages": [AIMessage(content=(
        "No estoy seguro de entender tu solicitud. ¿Podrías ser más específico? "
        "¿Necesitas ayuda con código, redacción, o investigación?"
    ))]}

def make_agent(prompt: str, add_disclaimer: bool = False):
    def agent(state: SmartRouterState) -> dict:
        model = init_chat_model("openai:gpt-4.1-mini")
        response = model.invoke(
            [SystemMessage(content=prompt)] + state["messages"]
        )
        content = response.content
        if add_disclaimer and state["confidence_tier"] == "medium":
            content = (
                "⚠️ *Nota: interpreté tu solicitud como una pregunta de "
                f"{state['intent']}. Si necesitas algo diferente, dímelo.*\n\n"
                + content
            )
        return {"messages": [AIMessage(content=content)]}
    return agent

graph = StateGraph(SmartRouterState)
graph.add_node("classify", smart_classifier)
graph.add_node("clarification", clarification_node)
graph.add_node("code", make_agent("Experto en código. Responde en español.", add_disclaimer=True))
graph.add_node("writing", make_agent("Escritor profesional. Responde en español.", add_disclaimer=True))
graph.add_node("research", make_agent("Investigador. Responde en español.", add_disclaimer=True))
graph.add_node("unclear", clarification_node)

graph.add_edge(START, "classify")
graph.add_conditional_edges("classify", route_by_confidence, {
    "code": "code", "writing": "writing",
    "research": "research", "unclear": "unclear",
    "clarification": "clarification",
})
for node in ["code", "writing", "research", "unclear", "clarification"]:
    graph.add_edge(node, END)

app = graph.compile()

for msg in ["Optimiza este loop en Python", "hmm", "algo sobre datos y textos creativos"]:
    result = app.invoke({
        "messages": [HumanMessage(content=msg)],
        "intent": "", "confidence": 0.0, "confidence_tier": "",
    })
    print(f"'{msg}' → intent={result['intent']}, "
          f"confidence={result['confidence']:.2f}, tier={result['confidence_tier']}")
    print(f"  Response: {result['messages'][-1].content[:80]}...")
    print()
# Output (ejemplo):
# 'Optimiza este loop en Python' → intent=code, confidence=0.95, tier=high
#   Response: Para optimizar un loop en Python, hay varias estrategias que puedes...
#
# 'hmm' → intent=unclear, confidence=0.15, tier=low
#   Response: No estoy seguro de entender tu solicitud. ¿Podrías ser más específico?...
#
# 'algo sobre datos y textos creativos' → intent=writing, confidence=0.65, tier=medium
#   Response: ⚠️ *Nota: interpreté tu solicitud como una pregunta de writing...

Ejercicio 6: Benchmark determinístico vs LLM (Avanzado)

Crea un test suite con 10 mensajes de ejemplo. Implementa un router determinístico y uno LLM para las mismas 3 categorías. Compara accuracy (contra labels manuales) y reporta cuántos el determinístico clasifica mal pero el LLM acierta.

from dotenv import load_dotenv
load_dotenv()

from typing import Literal
from pydantic import BaseModel, Field
from langchain.chat_models import init_chat_model

test_cases = [
    ("¿Cuánto cuesta el plan enterprise?", "billing"),
    ("Mi app se crashea al iniciar", "technical"),
    ("¿Tienen oficinas en México?", "general"),
    ("Me cobraron doble este mes", "billing"),
    ("El botón de login no responde", "technical"),
    ("¿Cómo cancelo mi suscripción?", "billing"),
    ("La página carga muy lento", "technical"),
    ("¿Puedo hablar con un humano?", "general"),
    ("Necesito un reembolso", "billing"),
    ("¿Qué lenguajes de programación soportan?", "general"),
]

def deterministic_classify(text: str) -> str:
    text_lower = text.lower()
    if any(kw in text_lower for kw in ["factura", "cobro", "pago", "precio", "plan", "cuesta"]):
        return "billing"
    elif any(kw in text_lower for kw in ["error", "no funciona", "crashea", "lento", "no responde"]):
        return "technical"
    return "general"

class DeptClass(BaseModel):
    department: Literal["billing", "technical", "general"] = Field(
        description="billing=pagos/costos/suscripciones, technical=errores/rendimiento, general=otro"
    )

classifier = init_chat_model("openai:gpt-4.1-mini")
structured = classifier.with_structured_output(DeptClass)

def llm_classify(text: str) -> str:
    result = structured.invoke(f"Clasifica esta solicitud de soporte: {text}")
    return result.department

det_correct = 0
llm_correct = 0
llm_wins = []

for text, label in test_cases:
    det_result = deterministic_classify(text)
    llm_result = llm_classify(text)

    det_ok = det_result == label
    llm_ok = llm_result == label
    det_correct += det_ok
    llm_correct += llm_ok

    if llm_ok and not det_ok:
        llm_wins.append((text, label, det_result))

    status = "✅" if det_ok else "❌"
    status_llm = "✅" if llm_ok else "❌"
    print(f"  Det {status} LLM {status_llm} | '{text[:40]}' → det={det_result}, llm={llm_result}, label={label}")

print(f"\nDeterminístico: {det_correct}/{len(test_cases)} ({det_correct/len(test_cases)*100:.0f}%)")
print(f"LLM:            {llm_correct}/{len(test_cases)} ({llm_correct/len(test_cases)*100:.0f}%)")
print(f"\nCasos donde LLM acertó y determinístico falló ({len(llm_wins)}):")
for text, label, det_result in llm_wins:
    print(f"  '{text}' → label={label}, det dijo={det_result}")
# Output (ejemplo):
# Determinístico: 7/10 (70%)
# LLM:            10/10 (100%)
#
# Casos donde LLM acertó y determinístico falló (3):
#   'Me cobraron doble este mes' → label=billing, det dijo=general
#   '¿Cómo cancelo mi suscripción?' → label=billing, det dijo=general
#   'Necesito un reembolso' → label=billing, det dijo=general

Resumen

En esta cápsula aprendiste:

• Un router toma UNA decisión ("¿quién maneja esto?") y delega — a diferencia de un supervisor que coordina múltiples pasos
• Los routers determinísticos (keywords, regex, metadata) son rápidos, baratos y predecibles — empieza siempre aquí
• Los routers LLM con with_structured_output manejan inputs ambiguos que las reglas no cubren
• Implementaste routers con StateGraph (nodos + conditional edges) y con Functional API (if/elif/else + dict lookup)
• El multi-level routing divide decisiones complejas en pasos jerárquicos (dominio → subtarea)
• El fallback con confianza evita que el router adivine cuando no debería — pide clarificación en vez de equivocarse
• La progresión natural es: determinístico → determinístico + LLM fallback → LLM completo

Próxima cápsula: Shared vs Isolated State — la decisión de diseño más difícil en multi-agent: qué puede ver y modificar cada agente.

Recursos adicionales

1. Multi-Agent Architectures — LangGraph Docs — Documentación oficial de patrones multi-agente incluyendo routers
2. How to build a multi-agent network — Tutorial práctico de routing entre agentes
3. Structured Output — LangChain Docs — Referencia de with_structured_output para clasificación
4. Conditional Edges — LangGraph Docs — Documentación de add_conditional_edges
5. How to create a router — Guía oficial para crear routers en LangGraph
6. Pydantic Models — Docs — Referencia de Pydantic para structured output schemas

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