Graph API vs Functional API: Comparación Profunda

Descripción de la cápsula

Ya conoces ambas APIs de LangGraph. En el Módulo 5 construiste workflows con StateGraph, nodos y edges. En las cápsulas anteriores de este módulo aprendiste @entrypoint, @task, .result(), y control flow con Python nativo. Ahora viene la pregunta inevitable: ¿cuándo uso cada una?

La respuesta no es "una es mejor que la otra." Son dos formas de expresar la misma intención — construir workflows con durabilidad, checkpointing y streaming. La Graph API te da un grafo explícito con nodos y edges que puedes visualizar. La Functional API te da funciones de Python con decoradores que se leen como código secuencial. Ambas usan el mismo runtime por debajo. La decisión correcta depende del problema, no de una preferencia general.

En esta cápsula vas a ver una tabla de comparación detallada, tres problemas resueltos side-by-side con ambas APIs, un framework de decisión concreto, y el camino de migración entre ambas.

Tabla de comparación detallada

La tabla es un punto de partida, no una sentencia. Vamos a ver en código cómo se manifiestan estas diferencias.

Side-by-side #1: Chatbot con tools

El patrón más común en agentes: un modelo que decide si llamar tools, ejecuta las tools, y vuelve al modelo hasta que tenga una respuesta final. El loop ReAct.

Versión Graph API

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")
    return {"messages": [model.bind_tools(tools_list).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:
    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 = StateGraph(State)
graph.add_node("model", call_model)
graph.add_node("tools", run_tools)
graph.add_edge(START, "model")
graph.add_conditional_edges("model", should_continue, {"tools": "tools", "end": END})
graph.add_edge("tools", "model")

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

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

Versión Functional API

from dotenv import load_dotenv
load_dotenv()

from langgraph.func import entrypoint, task
from langgraph.graph import add_messages
from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage, BaseMessage
from langchain_core.tools import tool

@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}
model = init_chat_model("openai:gpt-4.1-mini").bind_tools(tools_list)

@task
def call_model(messages: list[BaseMessage]):
    return model.invoke(messages)

@task
def call_tool(tool_call):
    return tool_map[tool_call["name"]].invoke(tool_call["args"])

@entrypoint()
def chatbot(messages: list[BaseMessage]):
    llm_response = call_model(messages).result()

    while llm_response.tool_calls:
        tool_futures = [call_tool(tc) for tc in llm_response.tool_calls]
        tool_results = [fut.result() for fut in tool_futures]
        messages = add_messages(messages, [llm_response, *tool_results])
        llm_response = call_model(messages).result()

    return llm_response.content

result = chatbot.invoke([HumanMessage(content="¿Qué clima hace en Madrid y cuánto es 15 * 7?")])
print(result)
# Output esperado: El clima en Madrid es soleado con 22°C. Y 15 × 7 = 105.

Análisis

Ambas versiones hacen exactamente lo mismo. La Graph API te da un diagrama visual del loop. La Functional API ejecuta las tools en paralelo con Futures y se lee como un while loop estándar. Ninguna es objetivamente mejor — depende de si valoras más la visualización o la legibilidad del código.

Side-by-side #2: Agente de investigación con routing

Un agente que clasifica la pregunta del usuario y la rutea a un especialista diferente según el tipo.

Versión Graph API

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 classify(state: State) -> dict:
    text = state["messages"][-1].content.lower()
    if any(w in text for w in ["código", "programa", "bug", "función"]):
        return {"intent": "code"}
    elif any(w in text for w in ["dato", "estadística", "número", "analiza"]):
        return {"intent": "data"}
    return {"intent": "general"}

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

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 = StateGraph(State)
graph.add_node("classify", classify)
graph.add_node("code_expert", make_specialist("Eres un experto en código. Responde en español."))
graph.add_node("data_analyst", make_specialist("Eres un analista de datos. Responde en español."))
graph.add_node("generalist", make_specialist("Eres un asistente general. Responde en español."))

graph.add_edge(START, "classify")
graph.add_conditional_edges("classify", route_intent, {
    "code": "code_expert", "data": "data_analyst", "general": "generalist"
})
for node in ["code_expert", "data_analyst", "generalist"]:
    graph.add_edge(node, END)

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

result = app.invoke({
    "messages": [HumanMessage(content="¿Cómo hago una función recursiva en Python?")],
    "intent": ""
})
print(f"Intent: {result['intent']}")
print(result["messages"][-1].content[:100])
# Output esperado:
# Intent: code
# Una función recursiva es aquella que se llama a sí misma. Aquí tienes un ejemplo...

Versión Functional API

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(w in text_lower for w in ["código", "programa", "bug", "función"]):
        return "code"
    elif any(w in text_lower for w in ["dato", "estadística", "número", "analiza"]):
        return "data"
    return "general"

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

SPECIALISTS = {
    "code": "Eres un experto en código. Responde en español.",
    "data": "Eres un analista de datos. Responde en español.",
    "general": "Eres un asistente general. Responde en español.",
}

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

result = research_agent.invoke("¿Cómo hago una función recursiva en Python?")
print(f"Intent: {result['intent']}")
print(result["answer"][:100])
# Output esperado:
# Intent: code
# Una función recursiva es aquella que se llama a sí misma. Aquí tienes un ejemplo...

Análisis

Con 3 rutas, ambas APIs funcionan bien. La Graph API te muestra el diagrama con las 3 rutas saliendo del clasificador — útil para presentar el flujo a un equipo. La Functional API condensa todo en un dict lookup + una llamada a @task.

Side-by-side #3: Pipeline multi-paso con evaluación

Un pipeline que genera contenido, lo evalúa, y lo regenera si no cumple el criterio de calidad. Incluye un loop de mejora condicional.

Versión Graph API

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langchain.chat_models import init_chat_model
from IPython.display import Image, display

class State(TypedDict):
    topic: str
    draft: str
    feedback: str
    quality: str
    iterations: int

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

def generate(state: State) -> dict:
    prompt = f"Escribe un párrafo sobre: {state['topic']}"
    if state.get("feedback"):
        prompt += f"\nMejora basándote en este feedback: {state['feedback']}"
    response = model.invoke(prompt)
    return {"draft": response.content, "iterations": state.get("iterations", 0) + 1}

def evaluate(state: State) -> dict:
    response = model.invoke(
        f"Evalúa este texto. Responde SOLO 'good' o 'needs_improvement' seguido de feedback breve.\n\n{state['draft']}"
    )
    text = response.content.strip().lower()
    if "good" in text[:10]:
        return {"quality": "good", "feedback": ""}
    return {"quality": "needs_improvement", "feedback": response.content}

def route_quality(state: State) -> str:
    if state["quality"] == "good" or state.get("iterations", 0) >= 3:
        return "done"
    return "regenerate"

graph = StateGraph(State)
graph.add_node("generate", generate)
graph.add_node("evaluate", evaluate)
graph.add_edge(START, "generate")
graph.add_edge("generate", "evaluate")
graph.add_conditional_edges("evaluate", route_quality, {
    "done": END, "regenerate": "generate"
})

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

result = app.invoke({"topic": "qué es un transformer", "draft": "", "feedback": "", "quality": "", "iterations": 0})
print(f"Iteraciones: {result['iterations']}")
print(f"Calidad: {result['quality']}")
print(result["draft"][:120])
# Output esperado:
# Iteraciones: 1-3 (depende de la evaluación)
# Calidad: good
# Un transformer es una arquitectura de redes neuronales propuesta en 2017...

Versión Functional API

from dotenv import load_dotenv
load_dotenv()

from langgraph.func import entrypoint, task
from langchain.chat_models import init_chat_model

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

@task
def generate_draft(topic: str, feedback: str = "") -> str:
    prompt = f"Escribe un párrafo sobre: {topic}"
    if feedback:
        prompt += f"\nMejora basándote en este feedback: {feedback}"
    return model.invoke(prompt).content

@task
def evaluate_draft(draft: str) -> dict:
    response = model.invoke(
        f"Evalúa este texto. Responde SOLO 'good' o 'needs_improvement' seguido de feedback breve.\n\n{draft}"
    )
    text = response.content.strip().lower()
    if "good" in text[:10]:
        return {"quality": "good", "feedback": ""}
    return {"quality": "needs_improvement", "feedback": response.content}

@entrypoint()
def content_pipeline(topic: str) -> dict:
    feedback = ""
    iterations = 0
    max_iterations = 3

    while iterations < max_iterations:
        draft = generate_draft(topic, feedback).result()
        evaluation = evaluate_draft(draft).result()
        iterations += 1

        if evaluation["quality"] == "good":
            break
        feedback = evaluation["feedback"]

    return {"draft": draft, "iterations": iterations, "quality": evaluation["quality"]}

result = content_pipeline.invoke("qué es un transformer")
print(f"Iteraciones: {result['iterations']}")
print(f"Calidad: {result['quality']}")
print(result["draft"][:120])
# Output esperado:
# Iteraciones: 1-3 (depende de la evaluación)
# Calidad: good
# Un transformer es una arquitectura de redes neuronales propuesta en 2017...

Análisis

Este ejemplo es donde la diferencia se siente más. La Graph API muestra un diagrama con la flecha circular evaluate → generate que comunica el patrón de forma visual. La Functional API expresa el mismo loop como un while con break — más familiar para cualquier desarrollador Python, pero invisible para quien mire el sistema desde fuera.

Framework de decisión

No memorices reglas — internaliza criterios. Cuando empieces un proyecto, hazte estas preguntas:

Usa Graph API cuando:

• ✅ Tu workflow tiene >5 nodos con routing complejo entre ellos
• ✅ Necesitas visualizar el flujo para debugging, documentación o comunicación con el equipo
• ✅ Múltiples equipos trabajan en diferentes nodos del mismo workflow
• ✅ La topología es no-trivial: múltiples branches, merge points, sub-workflows
• ✅ Necesitas Send API para crear workers dinámicamente

Usa Functional API cuando:

• ✅ Tu workflow es secuencial con branching simple (if/else)
• ✅ Quieres un prototipo rápido que funcione en minutos
• ✅ Prefieres que el código se lea como Python normal
• ✅ Tus developers vienen de Python y no conocen conceptos de grafos
• ✅ El flujo se expresa naturalmente como un loop con condiciones

Ambas funcionan cuando:

• ✅ Complejidad moderada (3-5 pasos con un branch)
• ✅ Preferencia personal del equipo
• ✅ El flujo podría crecer pero hoy es simple

La regla del "¿puedo dibujarlo fácil?"

Si puedes dibujar tu workflow en una servilleta en 10 segundos y se ve como una línea con un par de branches, la Functional API probablemente es más directa. Si necesitas 30 segundos y el dibujo tiene diamantes de decisión, merges, y flechas cruzadas, la Graph API te va a ahorrar dolores de cabeza.

Camino de migración

La buena noticia: no es una decisión permanente. Ambas APIs comparten el mismo runtime, así que migrar es refactorizar, no reescribir.

De Functional API a Graph API

La progresión natural:

1. Empiezas con @entrypoint + @task
   → Funciona, el código es limpio

2. Agregas más branches y loops
   → El if/elif/else crece, pero es manejable

3. Llegas a un punto de inflexión:
   → "Necesito visualizar este flujo para debuggear"
   → "Otro equipo necesita entender la topología"
   → "Tengo 8 tasks con routing complejo entre ellas"

4. Migras a StateGraph
   → Cada @task se convierte en un nodo
   → Cada if/else se convierte en un conditional edge
   → El while loop se convierte en un edge circular

Señales de que necesitas migrar

• ❌ Tu @entrypoint tiene más de 40 líneas de lógica de control
• ❌ Tienes if/elif/else anidados con más de 3 niveles
• ❌ Necesitas que otro equipo entienda el flujo sin leer el código
• ❌ El debugging requiere poner prints en 10 lugares diferentes

Señales de que NO necesitas migrar

• ✅ Tu flujo es un loop simple con 2-3 tasks
• ✅ El código se entiende en una lectura lineal
• ✅ Solo tú trabajas en este workflow
• ✅ No necesitas explicar el flujo visualmente a nadie

¿Se pueden mezclar?

Sí. Ambas APIs comparten el mismo runtime, así que puedes usar un StateGraph compilado dentro de un @entrypoint, o llamar un workflow funcional desde un nodo de un grafo. Esto es extremadamente útil cuando partes de tu sistema se expresan mejor como grafo y otras como funciones secuenciales.

Este patrón se cubre en detalle en la cápsula 07 de este módulo. Por ahora, lo importante es saber que no es una decisión de todo-o-nada — puedes combinarlas en el mismo proyecto.

Troubleshooting

Problema 1: "Elegí Functional API pero ahora tengo 6 niveles de if/else"

Síntoma: Tu @entrypoint creció hasta 80+ líneas con branching complejo y es difícil de seguir.

Solución: Es señal de que necesitas migrar a Graph API. Cada branch de tu if/else es un candidato a nodo independiente. Cada condición es un conditional edge. La migración es directa:

# Antes (Functional): if/else anidado
if intent == "code":
    if language == "python":
        result = python_expert(query).result()
    else:
        result = general_code(query).result()
elif intent == "data":
    result = data_analyst(query).result()

# Después (Graph): conditional edges explícitos
graph.add_conditional_edges("classify", route_intent, {
    "python_code": "python_expert",
    "general_code": "general_code",
    "data": "data_analyst",
})

Problema 2: "Elegí Graph API pero mi grafo es una línea recta"

Síntoma: Tu StateGraph tiene 4 nodos todos conectados con edges fijos: START → A → B → C → END. No hay conditional edges.

Solución: Un grafo lineal sin branching es un indicio de over-engineering. Migralo a Functional API donde se expresa como 3 llamadas secuenciales:

@entrypoint()
def pipeline(input_data: str) -> str:
    a = task_a(input_data).result()
    b = task_b(a).result()
    return task_c(b).result()

Problema 3: "No sé si mi flujo es 'suficientemente complejo' para Graph API"

Síntoma: Parálisis de análisis. Cada proyecto empieza con 20 minutos decidiendo qué API usar.

Solución: Empieza siempre con Functional API. Si en los primeros 30 minutos de desarrollo sientes que el control flow se complica, migra. La decisión no tiene que ser correcta desde el inicio — la migración es un refactor, no una reescritura.

Problema 4: "Necesito visualización pero prefiero la Functional API"

Síntoma: Quieres un diagrama del flujo para documentación, pero tu workflow es secuencial y la Functional API es más natural.

Solución: Puedes documentar el flujo con un diagrama Mermaid manual o usar herramientas externas. La visualización automática de draw_mermaid_png() es conveniente, pero no es la única forma de documentar un workflow. Si el único motivo para Graph API es la visualización, probablemente no justifica la migración.

Ejercicios

Ejercicio 1: ¿Qué API elegirías? (Fácil)

Para cada escenario, decide si usarías Graph API o Functional API. Justifica tu respuesta.

A) Un chatbot que responde preguntas de soporte usando 2 tools (buscar en FAQ, crear ticket).

B) Un sistema que recibe documentos, los clasifica en 5 categorías, y ejecuta un pipeline diferente para cada categoría. Cada pipeline tiene 3-4 pasos.

C) Un agente que traduce texto, evalúa la calidad de la traducción, y re-traduce si la calidad es baja.

D) Un workflow que busca información en 4 fuentes en paralelo y sintetiza los resultados.

Ejercicio 2: Convertir Graph API a Functional API (Fácil)

Convierte este workflow de Graph API a Functional API:

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langchain.chat_models import init_chat_model

class State(TypedDict):
    text: str
    cleaned: str
    summary: str

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

def clean_text(state: State) -> dict:
    cleaned = " ".join(state["text"].split())
    return {"cleaned": cleaned}

def summarize(state: State) -> dict:
    response = model.invoke(f"Resume en una oración: {state['cleaned']}")
    return {"summary": response.content}

graph = StateGraph(State)
graph.add_node("clean", clean_text)
graph.add_node("summarize", summarize)
graph.add_edge(START, "clean")
graph.add_edge("clean", "summarize")
graph.add_edge("summarize", END)

app = graph.compile()
result = app.invoke({"text": "  Python   es   un   lenguaje   de   programación   muy   popular  "})
print(result["summary"])

from dotenv import load_dotenv
load_dotenv()

from langgraph.func import entrypoint, task
from langchain.chat_models import init_chat_model

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

@task
def clean_text(text: str) -> str:
    return " ".join(text.split())

@task
def summarize(text: str) -> str:
    return model.invoke(f"Resume en una oración: {text}").content

@entrypoint()
def text_pipeline(text: str) -> str:
    cleaned = clean_text(text).result()
    summary = summarize(cleaned).result()
    return summary

result = text_pipeline.invoke("  Python   es   un   lenguaje   de   programación   muy   popular  ")
print(result)
# Output esperado: Python es un lenguaje de programación muy popular.

Ejercicio 3: Convertir Functional API a Graph API (Medio)

Convierte este workflow funcional a Graph API. Justifica por qué la migración tiene sentido.

from dotenv import load_dotenv
load_dotenv()

from langgraph.func import entrypoint, task
from langchain.chat_models import init_chat_model

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

@task
def classify(text: str) -> str:
    text_lower = text.lower()
    if any(w in text_lower for w in ["urgente", "error", "caído"]):
        return "urgent"
    elif any(w in text_lower for w in ["factura", "cobro", "pago"]):
        return "billing"
    return "general"

@task
def handle_urgent(text: str) -> str:
    return model.invoke(f"URGENTE. Responde rápido y directo: {text}").content

@task
def handle_billing(text: str) -> str:
    return model.invoke(f"Consulta de facturación. Sé preciso: {text}").content

@task
def handle_general(text: str) -> str:
    return model.invoke(f"Pregunta general. Sé amigable: {text}").content

@entrypoint()
def support_agent(text: str) -> dict:
    category = classify(text).result()
    if category == "urgent":
        response = handle_urgent(text).result()
    elif category == "billing":
        response = handle_billing(text).result()
    else:
        response = handle_general(text).result()
    return {"category": category, "response": response}

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langchain.chat_models import init_chat_model
from IPython.display import Image, display

class SupportState(TypedDict):
    text: str
    category: str
    response: str

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

def classify(state: SupportState) -> dict:
    text_lower = state["text"].lower()
    if any(w in text_lower for w in ["urgente", "error", "caído"]):
        return {"category": "urgent"}
    elif any(w in text_lower for w in ["factura", "cobro", "pago"]):
        return {"category": "billing"}
    return {"category": "general"}

def route_category(state: SupportState) -> str:
    return state["category"]

def handle_urgent(state: SupportState) -> dict:
    return {"response": model.invoke(f"URGENTE. Responde rápido y directo: {state['text']}").content}

def handle_billing(state: SupportState) -> dict:
    return {"response": model.invoke(f"Consulta de facturación. Sé preciso: {state['text']}").content}

def handle_general(state: SupportState) -> dict:
    return {"response": model.invoke(f"Pregunta general. Sé amigable: {state['text']}").content}

graph = StateGraph(SupportState)
graph.add_node("classify", classify)
graph.add_node("handle_urgent", handle_urgent)
graph.add_node("handle_billing", handle_billing)
graph.add_node("handle_general", handle_general)

graph.add_edge(START, "classify")
graph.add_conditional_edges("classify", route_category, {
    "urgent": "handle_urgent", "billing": "handle_billing", "general": "handle_general"
})
for node in ["handle_urgent", "handle_billing", "handle_general"]:
    graph.add_edge(node, END)

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

result = app.invoke({"text": "Mi servidor está caído, necesito ayuda urgente"})
print(f"Categoría: {result['category']}")
print(result["response"][:80])
# Output esperado:
# Categoría: urgent
# Entiendo la urgencia. Para diagnosticar el problema del servidor...

Ejercicio 4: Implementar el mismo problema en ambas APIs (Medio)

Implementa un workflow que reciba un tema, genere 3 preguntas sobre ese tema en paralelo (usando un LLM), y las combine en un quiz. Implementa en ambas APIs y compara.

from dotenv import load_dotenv
load_dotenv()

from langgraph.func import entrypoint, task
from langchain.chat_models import init_chat_model

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

@task
def generate_question(topic: str, difficulty: str) -> str:
    return model.invoke(
        f"Genera UNA pregunta de opción múltiple sobre '{topic}' con dificultad {difficulty}. "
        f"Incluye 4 opciones (A-D) y marca la correcta."
    ).content

@task
def format_quiz(questions: list) -> str:
    quiz = "# Quiz\n\n"
    for i, q in enumerate(questions, 1):
        quiz += f"## Pregunta {i}\n{q}\n\n"
    return quiz

@entrypoint()
def quiz_generator(topic: str) -> str:
    easy_fut = generate_question(topic, "fácil")
    medium_fut = generate_question(topic, "medio")
    hard_fut = generate_question(topic, "difícil")

    questions = [easy_fut.result(), medium_fut.result(), hard_fut.result()]
    return format_quiz(questions).result()

result = quiz_generator.invoke("Python decorators")
print(result[:200])
# Output esperado:
# # Quiz
#
# ## Pregunta 1
# ¿Qué es un decorator en Python?
# A) Una clase especial...

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

class QuizState(TypedDict):
    topic: str
    questions: Annotated[list[str], operator.add]
    quiz: str

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

def make_question_node(difficulty: str):
    def node(state: QuizState) -> dict:
        response = model.invoke(
            f"Genera UNA pregunta de opción múltiple sobre '{state['topic']}' con dificultad {difficulty}. "
            f"Incluye 4 opciones (A-D) y marca la correcta."
        )
        return {"questions": [response.content]}
    return node

def format_quiz(state: QuizState) -> dict:
    quiz = "# Quiz\n\n"
    for i, q in enumerate(state["questions"], 1):
        quiz += f"## Pregunta {i}\n{q}\n\n"
    return {"quiz": quiz}

graph = StateGraph(QuizState)
graph.add_node("easy", make_question_node("fácil"))
graph.add_node("medium", make_question_node("medio"))
graph.add_node("hard", make_question_node("difícil"))
graph.add_node("format", format_quiz)

graph.add_edge(START, "easy")
graph.add_edge(START, "medium")
graph.add_edge(START, "hard")
graph.add_edge("easy", "format")
graph.add_edge("medium", "format")
graph.add_edge("hard", "format")
graph.add_edge("format", END)

app = graph.compile()
result = app.invoke({"topic": "Python decorators", "questions": [], "quiz": ""})
print(result["quiz"][:200])
# Output esperado:
# # Quiz
#
# ## Pregunta 1
# ¿Qué es un decorator en Python?
# A) Una clase especial...

Ejercicio 5: Decision framework automatizado (Avanzado)

Crea una función recommend_api(requirements: dict) -> str que reciba un diccionario de requisitos y retorne "graph_api", "functional_api", o "either". Los requisitos son:

• num_nodes: número estimado de nodos/tasks
• has_complex_routing: bool (más de 2 destinos condicionales)
• needs_visualization: bool
• is_sequential: bool (flujo mayormente lineal)
• team_size: int (personas trabajando en el workflow)

Implementa la lógica de decisión y prueba con al menos 4 escenarios.

def recommend_api(requirements: dict) -> str:
    num_nodes = requirements.get("num_nodes", 3)
    has_complex_routing = requirements.get("has_complex_routing", False)
    needs_visualization = requirements.get("needs_visualization", False)
    is_sequential = requirements.get("is_sequential", True)
    team_size = requirements.get("team_size", 1)

    graph_signals = sum([
        num_nodes > 5,
        has_complex_routing,
        needs_visualization and num_nodes > 3,
        team_size > 2,
    ])

    functional_signals = sum([
        is_sequential,
        num_nodes <= 4,
        team_size <= 1,
        not has_complex_routing,
    ])

    if graph_signals >= 3:
        return "graph_api"
    if functional_signals >= 3 and graph_signals == 0:
        return "functional_api"
    return "either"


# Test 1: Pipeline simple
r1 = recommend_api({
    "num_nodes": 3, "has_complex_routing": False,
    "needs_visualization": False, "is_sequential": True, "team_size": 1
})
print(f"Pipeline simple: {r1}")
# Output esperado: functional_api

# Test 2: Sistema multi-ruta con equipo
r2 = recommend_api({
    "num_nodes": 8, "has_complex_routing": True,
    "needs_visualization": True, "is_sequential": False, "team_size": 4
})
print(f"Sistema multi-ruta: {r2}")
# Output esperado: graph_api

# Test 3: Loop evaluate-improve
r3 = recommend_api({
    "num_nodes": 3, "has_complex_routing": False,
    "needs_visualization": False, "is_sequential": True, "team_size": 1
})
print(f"Loop evaluate-improve: {r3}")
# Output esperado: functional_api

# Test 4: Complejidad moderada
r4 = recommend_api({
    "num_nodes": 5, "has_complex_routing": True,
    "needs_visualization": True, "is_sequential": False, "team_size": 2
})
print(f"Complejidad moderada: {r4}")
# Output esperado: either

Ejercicio 6: Refactoring challenge (Avanzado)

Tienes un @entrypoint que se salió de control. Identifica los problemas y decide: ¿refactorizas dentro de la Functional API o migras a Graph API?

@entrypoint()
def complex_agent(input_data: dict) -> dict:
    text = input_data["text"]
    mode = input_data["mode"]

    if mode == "research":
        sources = search_web(text).result()
        if len(sources) > 3:
            summaries = []
            for s in sources[:5]:
                summaries.append(summarize_source(s).result())
            combined = merge_summaries(summaries).result()
        else:
            combined = summarize_source(sources[0]).result()
        quality = evaluate_quality(combined).result()
        if quality["score"] < 0.7:
            combined = improve_text(combined, quality["feedback"]).result()
        return {"result": combined, "mode": "research"}
    elif mode == "code":
        spec = generate_spec(text).result()
        code = generate_code(spec).result()
        tests = generate_tests(code).result()
        test_result = run_tests(code, tests).result()
        if not test_result["passed"]:
            code = fix_code(code, test_result["errors"]).result()
        return {"result": code, "mode": "code"}
    elif mode == "translate":
        translation = translate(text).result()
        back_translation = translate(translation).result()
        if similarity(text, back_translation) < 0.8:
            translation = translate(text).result()
        return {"result": translation, "mode": "translate"}
    return {"result": "Modo no soportado", "mode": mode}

@entrypoint()
def complex_agent(input_data: dict) -> dict:
    mode = input_data["mode"]
    text = input_data["text"]

    handlers = {
        "research": research_workflow,
        "code": code_workflow,
        "translate": translate_workflow,
    }

    handler = handlers.get(mode)
    if handler is None:
        return {"result": "Modo no soportado", "mode": mode}

    result = handler(text).result()
    return {"result": result, "mode": mode}

Resumen

En esta cápsula aprendiste:

• Graph API y Functional API comparten el mismo runtime — la diferencia es cómo expresas el workflow, no cómo se ejecuta
• La Graph API te da estructura explícita: nodos, edges, conditional edges, visualización con draw_mermaid_png()
• La Functional API te da Python idiomático: while, if/else, for, Futures para paralelismo
• Viste 3 problemas resueltos side-by-side: chatbot con tools, agente con routing, pipeline con evaluación
• No hay una API "mejor" — hay criterios: complejidad de topología, necesidad de visualización, tamaño de equipo, naturaleza del flujo
• Framework de decisión: >5 nodos + routing complejo + equipo grande → Graph API. Flujo secuencial + branching simple + developer solo → Functional API
• El camino de migración va de Functional a Graph cuando la complejidad lo justifica — es refactor, no reescritura
• Se pueden mezclar ambas APIs en el mismo proyecto (cápsula 07)

Próxima cápsula: Patterns con Functional API — recetas reutilizables para los patrones más comunes que se resuelven de forma natural con @entrypoint y @task.

Recursos adicionales

1. Functional API Conceptual Guide — Documentación oficial de la Functional API
2. Graph API Conceptual Guide — Documentación oficial de la Graph API
3. Workflows and Agents Patterns — Patrones side-by-side oficiales en ambas APIs
4. How to use the Functional API — Tutorial práctico de la Functional API
5. How to build a ReAct agent from scratch — El patrón ReAct implementado con Graph API
6. Choosing between Graph API and Functional API — Guía oficial de decisión entre APIs
7. LangGraph StateGraph Reference — Referencia completa de StateGraph

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