Control Flow Nativo

Descripción de la cápsula

Esta es la killer feature de la Functional API: usas control flow estándar de Python — while, if/else, for, try/except — para dirigir la ejecución de tu workflow. No hay DSL nuevo, no hay edges condicionales, no hay métodos especiales para routing. Solo Python.

En la Graph API (Módulo 5), para hacer que un agente itere hasta que el modelo deje de pedir tools, necesitabas add_conditional_edges con una función router. Para hacer routing por tipo de intención, necesitabas otra add_conditional_edges. Para manejar errores, la lógica vivía dentro de cada nodo individual.

Con la Functional API, todo eso se convierte en un while True, un if/else, y un try/except. El código se lee como Python normal — porque es Python normal. LangGraph se encarga del checkpointing y streaming por debajo, pero tú escribes lógica de control como siempre la has escrito.

Esta cápsula te va a mostrar cada patrón de control flow, y va a comparar side-by-side con la Graph API para que entiendas exactamente cuánto código te ahorras (y cuándo la Graph API sigue siendo la mejor opción).

while loops: el loop del agente

El patrón más importante en agentes es el ReAct loop: el modelo genera una respuesta, si incluye tool calls las ejecutas, alimentas los resultados al modelo, y repites hasta que el modelo responda sin pedir tools.

Con la Functional API, esto es un while True:

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, ToolMessage
from langchain_core.tools import tool

@tool
def get_weather(city: str) -> str:
    """Obtiene el clima actual de una ciudad."""
    data = {"Madrid": "Soleado, 22°C", "CDMX": "Nublado, 18°C", "Buenos Aires": "Lluvioso, 15°C"}
    return data.get(city, f"Sin datos para {city}")

@tool
def get_population(city: str) -> str:
    """Obtiene la población de una ciudad."""
    data = {"Madrid": "3.2 millones", "CDMX": "9.2 millones", "Buenos Aires": "3.1 millones"}
    return data.get(city, f"Sin datos para {city}")

tools = [get_weather, get_population]
tool_map = {t.name: t for t in tools}

@task
def call_model(messages: list) -> object:
    """Llama al modelo con las tools disponibles."""
    model = init_chat_model("openai:gpt-4.1-mini")
    return model.bind_tools(tools).invoke(messages)

@task
def execute_tools(tool_calls: list) -> list:
    """Ejecuta las tool calls y retorna los resultados."""
    results = []
    for tc in tool_calls:
        output = tool_map[tc["name"]].invoke(tc["args"])
        results.append(ToolMessage(content=str(output), tool_call_id=tc["id"]))
    return results

@entrypoint()
def agent(user_message: str) -> str:
    messages = [HumanMessage(content=user_message)]
    
    while True:
        response = call_model(messages).result()
        
        if not response.tool_calls:
            return response.content
        
        tool_results = execute_tools(response.tool_calls).result()
        messages = messages + [response] + tool_results

result = agent.invoke("¿Qué clima hace en Madrid y cuánta gente vive ahí?")
print(result)
# Output esperado: En Madrid el clima es soleado con 22°C y tiene una
# población de aproximadamente 3.2 millones de habitantes.

Lee el while True como si fuera pseudocódigo:

1. Llama al modelo
2. Si no hay tool calls → retorna la respuesta (salimos del loop)
3. Si hay tool calls → ejecútalas y agrega los resultados a los mensajes
4. Repite

Ese es el ReAct loop completo. Sin edges, sin conditional edges, sin funciones router. Solo un while y un if.

if/else: routing por intención

Un caso común: clasificar la intención del usuario y routear a diferentes tareas según el resultado.

from dotenv import load_dotenv
load_dotenv()

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

@task
def classify_intent(message: str) -> str:
    """Clasifica la intención del mensaje."""
    model = init_chat_model("openai:gpt-4.1-mini")
    response = model.invoke(
        f"Clasifica la intención del siguiente mensaje en exactamente una palabra: "
        f"'code', 'question', o 'creative'.\n\nMensaje: {message}\n\nIntención:"
    )
    return response.content.strip().lower()

@task
def handle_code(message: str) -> str:
    """Genera código basado en el mensaje."""
    model = init_chat_model("openai:gpt-4.1-mini")
    response = model.invoke(f"Genera código Python para: {message}")
    return f"[Modo Código]\n{response.content}"

@task
def handle_question(message: str) -> str:
    """Responde una pregunta."""
    model = init_chat_model("openai:gpt-4.1-mini")
    response = model.invoke(f"Responde concisamente: {message}")
    return f"[Modo Q&A]\n{response.content}"

@task
def handle_creative(message: str) -> str:
    """Genera contenido creativo."""
    model = init_chat_model("openai:gpt-4.1-mini")
    response = model.invoke(f"Escribe algo creativo sobre: {message}")
    return f"[Modo Creativo]\n{response.content}"

@entrypoint()
def router_agent(message: str) -> str:
    intent = classify_intent(message).result()
    
    if intent == "code":
        return handle_code(message).result()
    elif intent == "creative":
        return handle_creative(message).result()
    else:
        return handle_question(message).result()

print(router_agent.invoke("Escribe una función que ordene una lista"))
# Output esperado: [Modo Código]
# ```python
# def sort_list(lst):
#     return sorted(lst)
# ...

print(router_agent.invoke("¿Cuántos planetas tiene el sistema solar?"))
# Output esperado: [Modo Q&A]
# El sistema solar tiene 8 planetas...

print(router_agent.invoke("Escribe un haiku sobre programación"))
# Output esperado: [Modo Creativo]
# Código que fluye / entre líneas de silencio / bugs se desvanecen

El routing es un if/elif/else estándar. Si necesitas agregar una nueva ruta, agregas un elif. No necesitas modificar edges ni crear nuevas funciones router.

for loops: iterar sobre múltiples items

Cuando necesitas procesar una colección de items, un for loop es la solución natural:

from dotenv import load_dotenv
load_dotenv()

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

@task
def analyze_sentiment(text: str) -> dict:
    """Analiza el sentimiento de un texto."""
    model = init_chat_model("openai:gpt-4.1-mini")
    response = model.invoke(
        f"Clasifica el sentimiento de este texto como 'positivo', 'negativo', o 'neutro'. "
        f"Responde solo con la clasificación.\n\nTexto: {text}"
    )
    sentiment = response.content.strip().lower()
    return {"text": text[:50], "sentiment": sentiment}

@entrypoint()
def batch_sentiment(reviews: list[str]) -> list[dict]:
    results = []
    for review in reviews:
        analysis = analyze_sentiment(review).result()
        results.append(analysis)
    return results

reviews = [
    "Este producto es increíble, superó mis expectativas",
    "Pésimo servicio, nunca más vuelvo a comprar aquí",
    "El envío llegó a tiempo, todo correcto",
    "La calidad es horrible, se rompió al primer uso",
]

output = batch_sentiment.invoke(reviews)
for item in output:
    print(f"  {item['sentiment']:>10} → {item['text']}")
# Output esperado:
#    positivo → Este producto es increíble, superó mis expectat
#    negativo → Pésimo servicio, nunca más vuelvo a comprar aqu
#     neutro → El envío llegó a tiempo, todo correcto
#    negativo → La calidad es horrible, se rompió al primer uso

Y con ejecución paralela usando el patrón de Futures:

from dotenv import load_dotenv
load_dotenv()

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

@task
def analyze_sentiment(text: str) -> dict:
    """Analiza el sentimiento de un texto."""
    model = init_chat_model("openai:gpt-4.1-mini")
    response = model.invoke(
        f"Clasifica el sentimiento como 'positivo', 'negativo', o 'neutro'. "
        f"Solo la clasificación.\n\nTexto: {text}"
    )
    return {"text": text[:50], "sentiment": response.content.strip().lower()}

@entrypoint()
def parallel_batch_sentiment(reviews: list[str]) -> list[dict]:
    futures = [analyze_sentiment(review) for review in reviews]
    results = [f.result() for f in futures]
    return results

reviews = [
    "Este producto es increíble, superó mis expectativas",
    "Pésimo servicio, nunca más vuelvo a comprar aquí",
    "El envío llegó a tiempo, todo correcto",
]

output = parallel_batch_sentiment.invoke(reviews)
for item in output:
    print(f"  {item['sentiment']:>10} → {item['text']}")
# Output esperado:
#    positivo → Este producto es increíble, superó mis expectat
#    negativo → Pésimo servicio, nunca más vuelvo a comprar aqu
#     neutro → El envío llegó a tiempo, todo correcto

La diferencia: en la versión secuencial, cada análisis espera al anterior. En la versión paralela, todos se lanzan a la vez y se recolectan los resultados después.

try/except: manejo de errores con degradación graceful

En producción, las APIs fallan. Los modelos retornan respuestas inesperadas. Los datos de entrada están corruptos. try/except te permite manejar estos casos sin que tu workflow entero colapse:

from dotenv import load_dotenv
load_dotenv()

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

@task
def fetch_primary_data(query: str) -> str:
    """Busca datos en la fuente principal."""
    model = init_chat_model("openai:gpt-4.1-mini")
    response = model.invoke(f"Responde en una oración: {query}")
    return response.content

@task
def fetch_fallback_data(query: str) -> str:
    """Fuente de respaldo — respuesta genérica."""
    return f"No se encontró información específica sobre '{query}'. Consulta fuentes adicionales."

@task
def generate_response(data: str, source: str) -> str:
    """Genera la respuesta final indicando la fuente."""
    return f"[Fuente: {source}] {data}"

@entrypoint()
def resilient_agent(query: str) -> str:
    try:
        data = fetch_primary_data(query).result()
        source = "principal"
    except Exception as e:
        print(f"  Fuente principal falló: {e}. Usando fallback...")
        data = fetch_fallback_data(query).result()
        source = "fallback"
    
    response = generate_response(data, source).result()
    return response

result = resilient_agent.invoke("¿Qué es la computación cuántica?")
print(result)
# Output esperado: [Fuente: principal] La computación cuántica es un paradigma
# de computación que utiliza qubits en lugar de bits clásicos...

El patrón es clásico: intenta la fuente principal, y si falla, usa el fallback. En la Graph API, este tipo de lógica requería manejar excepciones dentro de cada nodo individual, sin la posibilidad de hacer un flujo alternativo limpio desde el nivel del grafo.

Combinando try/except con while para retry

from dotenv import load_dotenv
load_dotenv()

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

@task
def unreliable_api_call(query: str) -> str:
    """Simula una API que puede fallar."""
    model = init_chat_model("openai:gpt-4.1-mini")
    response = model.invoke(f"Responde brevemente: {query}")
    return response.content

@entrypoint()
def retry_agent(query: str) -> str:
    max_attempts = 3
    
    for attempt in range(max_attempts):
        try:
            result = unreliable_api_call(query).result()
            return result
        except Exception as e:
            if attempt < max_attempts - 1:
                wait = 2 ** attempt
                print(f"  Intento {attempt + 1} falló: {e}. Esperando {wait}s...")
                time.sleep(wait)
            else:
                return f"Error: No se pudo obtener respuesta después de {max_attempts} intentos."

result = retry_agent.invoke("¿Qué es machine learning?")
print(result)
# Output esperado: Machine learning es una rama de la inteligencia artificial
# que permite a los sistemas aprender de datos sin ser programados explícitamente.

Un for loop con try/except y backoff exponencial. Control flow de Python puro que reemplaza lo que en un sistema basado en grafos requeriría nodos de retry, edges condicionales de vuelta al nodo que falló, y estado para trackear intentos.

Combinando control flow: while + if/else + try/except

El poder real aparece cuando combinas patrones. Aquí tienes un agente completo que usa los tres:

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, ToolMessage
from langchain_core.tools import tool

@tool
def search_docs(query: str) -> str:
    """Busca en la documentación interna."""
    docs = {
        "pricing": "Plan básico: $10/mes. Plan pro: $30/mes. Enterprise: contactar ventas.",
        "refund": "Política de reembolso: 30 días, sin preguntas.",
        "hours": "Horario de atención: lunes a viernes, 9:00 a 18:00 CT.",
    }
    for key, value in docs.items():
        if key in query.lower():
            return value
    return "No se encontró información relevante en la documentación."

@tool
def escalate_to_human(reason: str) -> str:
    """Escala el caso a un agente humano."""
    return f"Caso escalado. Razón: {reason}. Un agente te contactará en 24h."

tools = [search_docs, escalate_to_human]
tool_map = {t.name: t for t in tools}

@task
def call_support_model(messages: list) -> object:
    """Llama al modelo de soporte."""
    model = init_chat_model("openai:gpt-4.1-mini")
    system = SystemMessage(content=(
        "Eres un agente de soporte. Usa search_docs para buscar información. "
        "Si no puedes resolver, usa escalate_to_human. Responde en español."
    ))
    return model.bind_tools(tools).invoke([system] + messages)

@task
def run_tools(tool_calls: list) -> list:
    """Ejecuta las tool calls."""
    results = []
    for tc in tool_calls:
        try:
            output = tool_map[tc["name"]].invoke(tc["args"])
            results.append(ToolMessage(content=str(output), tool_call_id=tc["id"]))
        except Exception as e:
            results.append(ToolMessage(
                content=f"Error ejecutando {tc['name']}: {e}",
                tool_call_id=tc["id"]
            ))
    return results

@entrypoint()
def support_agent(user_message: str) -> str:
    messages = [HumanMessage(content=user_message)]
    max_iterations = 5
    iteration = 0
    
    while iteration < max_iterations:
        iteration += 1
        
        try:
            response = call_support_model(messages).result()
        except Exception as e:
            return f"Error del sistema: {e}. Por favor, intenta más tarde."
        
        if not response.tool_calls:
            return response.content
        
        tool_results = run_tools(response.tool_calls).result()
        messages = messages + [response] + tool_results
    
    return "Se alcanzó el límite de iteraciones. Escalando a un agente humano."

print(support_agent.invoke("¿Cuánto cuesta el plan pro?"))
# Output esperado: El plan pro cuesta $30 al mes. ¿Hay algo más en lo que
# pueda ayudarte?

print(support_agent.invoke("Quiero hablar con un humano, mi pedido está perdido"))
# Output esperado: He escalado tu caso a un agente humano. La razón registrada
# es que tu pedido está perdido. Te contactarán en las próximas 24 horas.

Este agente combina:

• while con límite de iteraciones (evita loops infinitos)
• if para decidir si terminó o necesita ejecutar tools
• try/except para manejar errores del modelo y de tools

Comparación side-by-side: Graph API vs Functional API

Veamos el mismo ReAct loop implementado con ambas APIs.

Graph API (StateGraph + conditional edges)

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 de una ciudad."""
    return f"Soleado, 22°C en {city}"

tools = [get_weather]
tool_map = {t.name: t for t in tools}

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

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

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}

def should_continue(state: State) -> str:
    if state["messages"][-1].tool_calls:
        return "tools"
    return END

graph_builder = StateGraph(State)
graph_builder.add_node("model", call_model)
graph_builder.add_node("tools", run_tools)
graph_builder.add_edge(START, "model")
graph_builder.add_conditional_edges("model", should_continue, {"tools": "tools", END: END})
graph_builder.add_edge("tools", "model")

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

result = graph.invoke({"messages": [HumanMessage(content="¿Clima en Madrid?")]})
print(result["messages"][-1].content)
# Output esperado: El clima en Madrid es soleado con una temperatura de 22°C.

Functional API (while loop)

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, ToolMessage
from langchain_core.tools import tool

@tool
def get_weather(city: str) -> str:
    """Obtiene el clima de una ciudad."""
    return f"Soleado, 22°C en {city}"

tools = [get_weather]
tool_map = {t.name: t for t in tools}

@task
def call_model(messages: list) -> object:
    model = init_chat_model("openai:gpt-4.1-mini")
    return model.bind_tools(tools).invoke(messages)

@task
def run_tools(tool_calls: list) -> list:
    results = []
    for tc in tool_calls:
        output = tool_map[tc["name"]].invoke(tc["args"])
        results.append(ToolMessage(content=str(output), tool_call_id=tc["id"]))
    return results

@entrypoint()
def agent(user_message: str) -> str:
    messages = [HumanMessage(content=user_message)]
    
    while True:
        response = call_model(messages).result()
        if not response.tool_calls:
            return response.content
        tool_results = run_tools(response.tool_calls).result()
        messages = messages + [response] + tool_results

result = agent.invoke("¿Clima en Madrid?")
print(result)
# Output esperado: El clima en Madrid es soleado con una temperatura de 22°C.

Diferencias clave

Tabla de equivalencias: Python ↔ Graph API

Cuándo Python control flow NO es suficiente

La Functional API no reemplaza a la Graph API en todos los casos. Hay escenarios donde la Graph API es claramente superior:

1. Topologías complejas con muchos nodos paralelos

Si tu workflow tiene 10+ nodos con dependencias cruzadas (A alimenta a C y D, B alimenta a D y E, C y D alimentan a F...), la Graph API expresa esas dependencias de forma más clara que anidar loops y condicionales.

2. Visualización es crítica

La Graph API genera diagramas con draw_mermaid_png(). Si tu equipo necesita ver la topología del workflow para entenderlo, la Graph API gana. La Functional API no produce un diagrama — tendrías que leer el código.

3. Múltiples puntos de re-entrada

Si tu workflow necesita "saltar" de un punto a otro que no es el paso siguiente (no es lineal), la Graph API con edges maneja esto de forma natural. Con control flow de Python, necesitarías flags y condicionales que se vuelven difíciles de mantener.

4. Sub-workflows reutilizables

La Graph API permite componer grafos como subgraphs. Un StateGraph compilado puede usarse como nodo dentro de otro grafo. Con la Functional API, la composición es a nivel de funciones (que también funciona, pero sin las garantías de aislamiento de estado que dan los subgraphs).

La regla práctica:

• ✅ Functional API: flujos secuenciales, loops simples, routing de 2-4 ramas, prototipos rápidos
• ✅ Graph API: topologías complejas, visualización necesaria, sub-workflows reutilizables, equipos grandes que necesitan diagramas

Control flow anidado: loops con condicionales

Los patrones reales combinan múltiples niveles de control flow. Veamos un agente que procesa una lista de documentos con lógica diferente según el tipo:

from dotenv import load_dotenv
load_dotenv()

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

@task
def classify_document(doc: str) -> str:
    """Clasifica un documento por tipo."""
    model = init_chat_model("openai:gpt-4.1-mini")
    response = model.invoke(
        f"Clasifica este texto como 'technical', 'business', o 'legal'. "
        f"Solo la clasificación.\n\nTexto: {doc}"
    )
    return response.content.strip().lower()

@task
def summarize_technical(doc: str) -> str:
    """Resume un documento técnico enfocándose en especificaciones."""
    model = init_chat_model("openai:gpt-4.1-mini")
    response = model.invoke(f"Resume las especificaciones técnicas clave: {doc}")
    return f"[TECH] {response.content}"

@task
def summarize_business(doc: str) -> str:
    """Resume un documento de negocio enfocándose en métricas."""
    model = init_chat_model("openai:gpt-4.1-mini")
    response = model.invoke(f"Resume las métricas y KPIs clave: {doc}")
    return f"[BIZ] {response.content}"

@task
def summarize_legal(doc: str) -> str:
    """Resume un documento legal enfocándose en obligaciones."""
    model = init_chat_model("openai:gpt-4.1-mini")
    response = model.invoke(f"Resume las obligaciones legales clave: {doc}")
    return f"[LEGAL] {response.content}"

@entrypoint()
def process_documents(documents: list[str]) -> list[str]:
    summaries = []
    
    for doc in documents:
        doc_type = classify_document(doc).result()
        
        if doc_type == "technical":
            summary = summarize_technical(doc).result()
        elif doc_type == "business":
            summary = summarize_business(doc).result()
        elif doc_type == "legal":
            summary = summarize_legal(doc).result()
        else:
            summary = f"[UNKNOWN] Tipo no reconocido: {doc_type}"
        
        summaries.append(summary)
    
    return summaries

docs = [
    "El servidor usa PostgreSQL 16 con 32GB RAM y replicación async.",
    "Q3 revenue: $2.4M, up 15% YoY. CAC reduced to $45.",
    "El contratista se obliga a entregar el software antes del 31 de diciembre.",
]

results = process_documents.invoke(docs)
for r in results:
    print(r)
# Output esperado:
# [TECH] El servidor utiliza PostgreSQL 16 con 32GB de RAM...
# [BIZ] Los ingresos del Q3 fueron $2.4M, un incremento del 15%...
# [LEGAL] El contratista tiene la obligación de entregar...

Un for loop que itera sobre documentos, con un if/elif/else dentro que routea a la task correcta según el tipo. Esto en la Graph API requeriría un nodo de clasificación, conditional edges a tres nodos diferentes de resumen, y luego edges de vuelta a un nodo acumulador. Más nodos, más edges, más código.

Troubleshooting

Problema 1: "Loop infinito — el agente nunca termina"

Síntoma: El workflow se ejecuta indefinidamente. Causa: El while True no tiene condición de salida, o el modelo sigue pidiendo tool calls en cada iteración. Solución: Agrega un límite de iteraciones:

@entrypoint()
def safe_agent(message: str) -> str:
    messages = [HumanMessage(content=message)]
    max_iterations = 10
    
    for i in range(max_iterations):
        response = call_model(messages).result()
        if not response.tool_calls:
            return response.content
        tool_results = run_tools(response.tool_calls).result()
        messages = messages + [response] + tool_results
    
    return "Límite de iteraciones alcanzado."

Problema 2: "El if/else no routea correctamente"

Síntoma: La clasificación retorna strings con espacios o capitalización inesperada, y ninguna rama del if se activa. Causa: El modelo retorna "Technical\n" o " Code " en vez de "technical". Solución: Normaliza la respuesta del modelo antes de comparar:

intent = classify_intent(message).result()
intent = intent.strip().lower()

if intent == "code":
    ...

Problema 3: "try/except captura errores que no debería"

Síntoma: El try/except esconde bugs legítimos (como TypeError o KeyError). Causa: Un except Exception genérico captura todo. Solución: Captura solo las excepciones esperadas:

from langchain_core.exceptions import OutputParserException

try:
    result = call_api(query).result()
except (ConnectionError, TimeoutError) as e:
    result = fallback(query).result()
# TypeError, KeyError, etc. se propagan normalmente

Problema 4: "El for loop es muy lento procesando muchos items"

Síntoma: Procesar 20 items toma mucho tiempo porque cada uno espera al anterior. Causa: Estás llamando .result() dentro del loop en cada iteración. Solución: Usa el patrón de Futures paralelos:

# ❌ Secuencial
for item in items:
    result = process(item).result()  # Bloquea en cada iteración

# ✅ Paralelo
futures = [process(item) for item in items]
results = [f.result() for f in futures]

Problema 5: "Quiero visualizar mi workflow pero no hay draw_mermaid_png()"

Síntoma: La Functional API no tiene método de visualización. Causa: La Functional API no construye un grafo explícito — el flujo vive en tu código Python. Solución: Para entender la topología de un workflow funcional, lee el código del @entrypoint. Si necesitas un diagrama, considera usar la Graph API para workflows complejos, o documenta el flujo manualmente con un diagrama Mermaid.

Ejercicios

Ejercicio 1: while loop con contador (Fácil)

Crea un agente que use un while loop para "refinar" una respuesta. En cada iteración, el modelo mejora su respuesta anterior. El loop termina cuando el modelo dice "FINAL:" al inicio de su respuesta, o después de 3 iteraciones.

from dotenv import load_dotenv
load_dotenv()

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

@task
def refine(prompt: str, previous: str, iteration: int) -> str:
    """Refina una respuesta previa."""
    model = init_chat_model("openai:gpt-4.1-mini")
    if previous:
        msg = (
            f"Tu respuesta anterior fue: '{previous}'. "
            f"Mejórala (iteración {iteration}). "
            f"Si ya es buena, empieza tu respuesta con 'FINAL:'. "
            f"Pregunta original: {prompt}"
        )
    else:
        msg = f"Responde brevemente: {prompt}"
    return model.invoke(msg).content

@entrypoint()
def refinement_agent(question: str) -> str:
    previous = ""
    max_iterations = 3
    iteration = 0
    
    while iteration < max_iterations:
        iteration += 1
        response = refine(question, previous, iteration).result()
        
        if response.startswith("FINAL:"):
            return response[6:].strip()
        
        previous = response
    
    return previous

result = refinement_agent.invoke("¿Qué es Docker en una oración?")
print(result)
# Output esperado: Docker es una plataforma de contenedorización que empaqueta
# aplicaciones y sus dependencias en contenedores portables y aislados.

Ejercicio 2: if/else para routing a 3 especialistas (Fácil)

Crea un workflow que clasifique un mensaje del usuario como "math", "history", o "science", y lo routee al especialista correcto (cada uno es una @task diferente que responde con un prefijo identificador).

from dotenv import load_dotenv
load_dotenv()

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

@task
def classify(message: str) -> str:
    """Clasifica el mensaje en math, history, o science."""
    model = init_chat_model("openai:gpt-4.1-mini")
    response = model.invoke(
        f"Clasifica esta pregunta como 'math', 'history', o 'science'. "
        f"Solo una palabra.\n\nPregunta: {message}"
    )
    return response.content.strip().lower()

@task
def math_expert(question: str) -> str:
    model = init_chat_model("openai:gpt-4.1-mini")
    response = model.invoke(f"Como experto en matemáticas, responde: {question}")
    return f"[Matemáticas] {response.content}"

@task
def history_expert(question: str) -> str:
    model = init_chat_model("openai:gpt-4.1-mini")
    response = model.invoke(f"Como experto en historia, responde: {question}")
    return f"[Historia] {response.content}"

@task
def science_expert(question: str) -> str:
    model = init_chat_model("openai:gpt-4.1-mini")
    response = model.invoke(f"Como experto en ciencias, responde: {question}")
    return f"[Ciencias] {response.content}"

@entrypoint()
def specialist_router(question: str) -> str:
    category = classify(question).result()
    
    if category == "math":
        return math_expert(question).result()
    elif category == "history":
        return history_expert(question).result()
    else:
        return science_expert(question).result()

print(specialist_router.invoke("¿Cuánto es la integral de x²?"))
# Output esperado: [Matemáticas] La integral de x² es (x³)/3 + C...

print(specialist_router.invoke("¿Quién fue Napoleón?"))
# Output esperado: [Historia] Napoleón Bonaparte fue un líder militar...

print(specialist_router.invoke("¿Cómo funciona la fotosíntesis?"))
# Output esperado: [Ciencias] La fotosíntesis es el proceso por el cual...

Ejercicio 3: for loop con procesamiento paralelo (Medio)

Recibe una lista de 4 ciudades. Para cada ciudad, una @task obtiene información simulada. Ejecuta todas en paralelo y retorna un reporte combinado.

from dotenv import load_dotenv
load_dotenv()

from langgraph.func import entrypoint, task

@task
def get_city_info(city: str) -> dict:
    """Obtiene información simulada de una ciudad."""
    data = {
        "Madrid": {"population": "3.2M", "country": "España", "temp": "22°C"},
        "CDMX": {"population": "9.2M", "country": "México", "temp": "18°C"},
        "Buenos Aires": {"population": "3.1M", "country": "Argentina", "temp": "15°C"},
        "Bogotá": {"population": "7.4M", "country": "Colombia", "temp": "14°C"},
    }
    info = data.get(city, {"population": "N/A", "country": "N/A", "temp": "N/A"})
    return {"city": city, **info}

@entrypoint()
def city_report(cities: list[str]) -> str:
    futures = [get_city_info(city) for city in cities]
    results = [f.result() for f in futures]
    
    report = "Reporte de ciudades:\n"
    for info in results:
        report += (
            f"  • {info['city']} ({info['country']}): "
            f"{info['population']} habitantes, {info['temp']}\n"
        )
    return report

output = city_report.invoke(["Madrid", "CDMX", "Buenos Aires", "Bogotá"])
print(output)
# Output esperado:
# Reporte de ciudades:
#   • Madrid (España): 3.2M habitantes, 22°C
#   • CDMX (México): 9.2M habitantes, 18°C
#   • Buenos Aires (Argentina): 3.1M habitantes, 15°C
#   • Bogotá (Colombia): 7.4M habitantes, 14°C

Ejercicio 4: try/except con fallback de 3 niveles (Medio)

Crea un workflow que intente 3 fuentes en orden: primaria, secundaria, y fallback local. Si una fuente falla, intenta la siguiente. Si todas fallan, retorna un mensaje de error.

from dotenv import load_dotenv
load_dotenv()

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

@task
def query_primary(question: str) -> str:
    """Fuente primaria — modelo principal."""
    model = init_chat_model("openai:gpt-4.1-mini")
    response = model.invoke(f"Responde concisamente: {question}")
    return f"[Primary] {response.content}"

@task
def query_secondary(question: str) -> str:
    """Fuente secundaria — modelo alternativo."""
    model = init_chat_model("openai:gpt-4.1-mini")
    response = model.invoke(f"Responde de forma simple: {question}")
    return f"[Secondary] {response.content}"

@task
def query_local_cache(question: str) -> str:
    """Fuente local — respuestas cacheadas."""
    cache = {
        "python": "Python es un lenguaje de programación de alto nivel.",
        "javascript": "JavaScript es un lenguaje para desarrollo web.",
    }
    for key, value in cache.items():
        if key in question.lower():
            return f"[Cache] {value}"
    return f"[Cache] No hay respuesta cacheada para: {question}"

@entrypoint()
def resilient_query(question: str) -> str:
    sources = [
        ("Primary", query_primary),
        ("Secondary", query_secondary),
        ("Cache", query_local_cache),
    ]
    
    for source_name, source_fn in sources:
        try:
            result = source_fn(question).result()
            return result
        except Exception as e:
            print(f"  {source_name} falló: {e}")
            continue
    
    return "Error: Todas las fuentes fallaron. Intenta más tarde."

result = resilient_query.invoke("¿Qué es Python?")
print(result)
# Output esperado: [Primary] Python es un lenguaje de programación de alto nivel,
# interpretado y de propósito general...

Ejercicio 5: ReAct agent completo con while + if + try/except (Avanzado)

Construye un agente de soporte técnico con 2 tools (check_status y restart_service). El agente usa un while loop (máximo 5 iteraciones), if/else para decidir si ejecutar tools o responder, y try/except para manejar errores en la ejecución de tools.

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, ToolMessage
from langchain_core.tools import tool

@tool
def check_status(service: str) -> str:
    """Verifica el estado de un servicio."""
    statuses = {
        "api": "running (99.9% uptime)",
        "database": "degraded (high latency: 500ms)",
        "cache": "down (last restart: 2h ago)",
    }
    return statuses.get(service.lower(), f"Servicio '{service}' no encontrado")

@tool
def restart_service(service: str) -> str:
    """Reinicia un servicio."""
    if service.lower() == "cache":
        return f"Servicio '{service}' reiniciado exitosamente. Estado: running."
    return f"Servicio '{service}' no necesita reinicio. Estado actual: running."

tools = [check_status, restart_service]
tool_map = {t.name: t for t in tools}

@task
def call_support_model(messages: list) -> object:
    model = init_chat_model("openai:gpt-4.1-mini")
    system = SystemMessage(content=(
        "Eres un agente de soporte técnico. "
        "Usa check_status para verificar servicios y restart_service para reiniciarlos. "
        "Responde en español."
    ))
    return model.bind_tools(tools).invoke([system] + messages)

@task
def execute_tool_calls(tool_calls: list) -> list:
    results = []
    for tc in tool_calls:
        try:
            output = tool_map[tc["name"]].invoke(tc["args"])
            results.append(ToolMessage(content=str(output), tool_call_id=tc["id"]))
        except Exception as e:
            results.append(ToolMessage(
                content=f"Error: {e}", tool_call_id=tc["id"]
            ))
    return results

@entrypoint()
def tech_support(user_message: str) -> str:
    messages = [HumanMessage(content=user_message)]
    
    for iteration in range(5):
        try:
            response = call_support_model(messages).result()
        except Exception as e:
            return f"Error del sistema en iteración {iteration + 1}: {e}"
        
        if not response.tool_calls:
            return response.content
        
        tool_results = execute_tool_calls(response.tool_calls).result()
        messages = messages + [response] + tool_results
    
    return "Se alcanzó el límite de iteraciones. Escalando a un ingeniero."

print(tech_support.invoke("El caché está caído, ¿puedes verificar y reiniciar?"))
# Output esperado: Verifiqué el estado del caché y estaba caído. Lo reinicié
# exitosamente y ahora está funcionando correctamente.

Ejercicio 6: Mismo problema con Graph API y Functional API (Avanzado)

Implementa un clasificador-router que reciba un mensaje, lo clasifique (pregunta vs comando), y lo routee a la task correcta. Hazlo dos veces: una con StateGraph + add_conditional_edges, y otra con @entrypoint + if/else. Compara la cantidad de código.

from dotenv import load_dotenv
load_dotenv()

# ========================================
# VERSIÓN 1: Graph API (StateGraph)
# ========================================
from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langchain_core.messages import AnyMessage, HumanMessage, AIMessage
from IPython.display import Image, display

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

def classify(state: State) -> dict:
    text = state["messages"][-1].content.lower()
    if any(w in text for w in ["?", "qué", "cómo", "cuándo"]):
        return {"intent": "question"}
    return {"intent": "command"}

def handle_question(state: State) -> dict:
    return {"messages": [AIMessage(content="[Q&A] Procesando tu pregunta...")]}

def handle_command(state: State) -> dict:
    return {"messages": [AIMessage(content="[CMD] Ejecutando tu comando...")]}

def route(state: State) -> str:
    if state["intent"] == "question":
        return "question"
    return "command"

builder = StateGraph(State)
builder.add_node("classify", classify)
builder.add_node("question", handle_question)
builder.add_node("command", handle_command)
builder.add_edge(START, "classify")
builder.add_conditional_edges("classify", route, {"question": "question", "command": "command"})
builder.add_edge("question", END)
builder.add_edge("command", END)

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

result = graph.invoke({"messages": [HumanMessage(content="¿Cómo funciona Python?")], "intent": ""})
print(f"Graph API: {result['messages'][-1].content}")
# Graph API: [Q&A] Procesando tu pregunta...

# ========================================
# VERSIÓN 2: Functional API
# ========================================
from langgraph.func import entrypoint, task

@task
def func_classify(message: str) -> str:
    text = message.lower()
    if any(w in text for w in ["?", "qué", "cómo", "cuándo"]):
        return "question"
    return "command"

@task
def func_handle_question(message: str) -> str:
    return f"[Q&A] Procesando tu pregunta..."

@task
def func_handle_command(message: str) -> str:
    return f"[CMD] Ejecutando tu comando..."

@entrypoint()
def func_router(message: str) -> str:
    intent = func_classify(message).result()
    
    if intent == "question":
        return func_handle_question(message).result()
    else:
        return func_handle_command(message).result()

result = func_router.invoke("¿Cómo funciona Python?")
print(f"Functional API: {result}")
# Functional API: [Q&A] Procesando tu pregunta...

Resumen

En esta cápsula aprendiste:

• La Functional API usa control flow estándar de Python (while, if/else, for, try/except) en lugar de edges y conditional edges
• while loops reemplazan conditional edges de vuelta al mismo nodo — perfecto para el ReAct loop de agentes
• if/else reemplaza add_conditional_edges — routing directo sin funciones router separadas
• for loops permiten iterar sobre colecciones dinámicas — con la opción de ejecución paralela usando Futures
• try/except habilita degradación graceful — fallbacks de múltiples niveles sin complejidad en la topología del grafo
• Los patrones se combinan: un agente real usa while + if + try/except en el mismo @entrypoint
• Graph API vs Functional API: la Functional API gana en flujos secuenciales y routing simple; la Graph API gana en topologías complejas, visualización, y sub-workflows reutilizables
• El código se lee como Python normal — esa es la propuesta de valor. LangGraph maneja checkpointing y streaming por debajo, pero la lógica de control es tuya

Próxima cápsula: aprenderás a combinar ambas APIs — usar un StateGraph como componente dentro de un @entrypoint, y viceversa — para tener lo mejor de ambos mundos.

Recursos adicionales

1. LangGraph Functional API — Conceptual Guide — Documentación oficial del control flow en Functional API
2. LangGraph Functional API — How-To Guide — Tutorial paso a paso con ejemplos de while, if/else, y for
3. Graph API vs Functional API — Comparación oficial entre ambas APIs
4. ReAct Pattern — LangGraph — El patrón ReAct implementado en LangGraph
5. Conditional Edges — Graph API — Referencia de conditional edges para la comparación
6. LangGraph Streaming — Cómo el streaming funciona con control flow nativo
7. Python Control Flow — Official Tutorial — Referencia de while, if/else, for, try/except en Python

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