Patterns con Functional API

Descripción de la cápsula

Ya entiendes la Functional API y cómo se compara con la Graph API. Ahora vamos a lo práctico: patterns reutilizables — recetas que resuelven problemas comunes de forma natural con @entrypoint y @task. Estos patterns son los que vas a copiar y adaptar en tus proyectos reales.

Cada pattern resuelve un tipo de problema arquitectónico: loops de ejecución de tools, razonamiento multi-paso, paralelismo con Futures, extracción de datos estructurados, y manejo de errores con fallback. Son los building blocks que hacen que la Functional API brille — flujos que se expresan como Python limpio con superpoderes de durabilidad y checkpointing.

Pattern 1: Tool execution loop

El pattern más fundamental para agentes: el modelo decide si necesita llamar tools, las ejecuta, y vuelve al modelo. Es el loop ReAct implementado con un while y @task.

Por qué funciona bien con Functional API

En la Graph API, este pattern requiere un conditional edge que evalúe si hay tool calls y un edge circular de vuelta al nodo del modelo. Con la Functional API, es un while loop — exactamente como lo pensarías antes de conocer LangGraph.

Implementación completa

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

@tool
def search_web(query: str) -> str:
    """Busca información en la web sobre un tema."""
    return f"Resultados para '{query}': Python 3.12 incluye mejoras en tipado, rendimiento y f-strings."

@tool
def get_date() -> str:
    """Obtiene la fecha actual."""
    return "8 de marzo de 2026"

tools_list = [search_web, get_date]
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 execute_tool(tool_call: dict):
    return tool_map[tool_call["name"]].invoke(tool_call["args"])

@entrypoint()
def react_agent(user_message: str) -> str:
    messages = [
        SystemMessage(content="Eres un asistente de investigación. Usa las tools disponibles."),
        HumanMessage(content=user_message),
    ]

    while True:
        response = call_model(messages).result()

        if not response.tool_calls:
            return response.content

        tool_futures = [execute_tool(tc) for tc in response.tool_calls]
        tool_results = [fut.result() for fut in tool_futures]

        messages = add_messages(messages, [response, *tool_results])

result = react_agent.invoke("¿Qué novedades tiene Python 3.12? ¿Y qué fecha es hoy?")
print(result)
# Output esperado: Python 3.12 incluye mejoras en tipado, rendimiento y f-strings.
# La fecha de hoy es 8 de marzo de 2026.

Anatomía del pattern

1. Setup: Modelo con tools vinculadas, mapa de tools por nombre
2. Loop: while True con break implícito en return
3. Decisión: if not response.tool_calls: return — si el modelo no pide tools, terminamos
4. Ejecución paralela: Las tools se lanzan como Futures y se recolectan después
5. Acumulación: add_messages actualiza el historial de conversación

Las tools se ejecutan en paralelo automáticamente — si el modelo pide search_web y get_date en la misma respuesta, ambas se lanzan sin esperar a que una termine antes de empezar la otra.

Pattern 2: Multi-step reasoning

Descomponer un problema complejo en pasos: analizar → razonar → sintetizar. Cada paso como un @task independiente con su propio checkpoint.

Por qué funciona bien con Functional API

El razonamiento multi-paso es inherentemente secuencial: cada paso depende del anterior. La Functional API lo expresa como una secuencia de llamadas @task — sin nodos, sin edges, solo funciones que se llaman una después de otra.

Implementación completa

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 decompose_question(question: str) -> list:
    response = model.invoke(
        f"Descompone esta pregunta compleja en 2-3 sub-preguntas más simples. "
        f"Responde SOLO con las sub-preguntas, una por línea.\n\n"
        f"Pregunta: {question}"
    )
    sub_questions = [q.strip() for q in response.content.strip().split("\n") if q.strip()]
    return sub_questions

@task
def reason_about(sub_question: str) -> str:
    response = model.invoke(
        f"Responde esta pregunta de forma concisa y técnica (2-3 oraciones):\n{sub_question}"
    )
    return response.content

@task
def synthesize(question: str, partial_answers: list) -> str:
    context = "\n".join(f"- {a}" for a in partial_answers)
    response = model.invoke(
        f"Usando estas respuestas parciales, genera una respuesta completa y coherente.\n\n"
        f"Pregunta original: {question}\n\n"
        f"Respuestas parciales:\n{context}"
    )
    return response.content

@entrypoint()
def chain_of_thought(question: str) -> dict:
    sub_questions = decompose_question(question).result()

    partial_answers = []
    for sq in sub_questions:
        answer = reason_about(sq).result()
        partial_answers.append(answer)

    final_answer = synthesize(question, partial_answers).result()

    return {
        "question": question,
        "sub_questions": sub_questions,
        "partial_answers": partial_answers,
        "final_answer": final_answer,
    }

result = chain_of_thought.invoke(
    "¿Por qué los transformers reemplazaron a las RNNs en procesamiento de lenguaje natural?"
)
print(f"Sub-preguntas: {len(result['sub_questions'])}")
for i, sq in enumerate(result["sub_questions"], 1):
    print(f"  {i}. {sq}")
print(f"\nRespuesta final: {result['final_answer'][:150]}...")
# Output esperado:
# Sub-preguntas: 2-3
#   1. ¿Cuáles eran las limitaciones de las RNNs?
#   2. ¿Qué ventajas introduce la arquitectura transformer?
#   3. ¿Qué evidencia empírica mostró la superioridad de los transformers?
# Respuesta final: Los transformers reemplazaron a las RNNs principalmente por...

Checkpointing automático

Cada @task guarda su resultado en un checkpoint. Si el workflow falla en synthesize, al resumir no re-ejecuta decompose_question ni los reason_about que ya completaron — recupera sus resultados del checkpoint. En un razonamiento multi-paso con 5 sub-preguntas donde cada una requiere una llamada al LLM, esto ahorra tiempo y tokens.

Pattern 3: Parallel task execution con Futures

Lanzar múltiples tasks sin esperar a que cada una termine, y recoger todos los resultados al final. El pattern de paralelismo que hace a la Functional API práctica para workloads I/O-bound.

Por qué funciona bien con Functional API

Llamar un @task retorna un Future inmediatamente — la tarea se ejecuta en background. Si llamas 3 tasks sin .result(), las 3 pueden correr en paralelo. Cuando finalmente llamas .result(), bloqueas hasta que el resultado esté listo. Este pattern es natural para búsquedas en múltiples fuentes, procesamiento de múltiples documentos, o cualquier operación I/O-bound.

Implementación completa

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 search_wikipedia(topic: str) -> str:
    response = model.invoke(
        f"Simula una búsqueda en Wikipedia sobre '{topic}'. "
        f"Genera un párrafo informativo como si fuera un artículo de Wikipedia."
    )
    return f"[Wikipedia] {response.content}"

@task
def search_arxiv(topic: str) -> str:
    response = model.invoke(
        f"Simula una búsqueda en arXiv sobre '{topic}'. "
        f"Genera un resumen técnico como si fuera un abstract de paper."
    )
    return f"[arXiv] {response.content}"

@task
def search_news(topic: str) -> str:
    response = model.invoke(
        f"Simula una búsqueda de noticias recientes sobre '{topic}'. "
        f"Genera un resumen periodístico breve."
    )
    return f"[News] {response.content}"

@task
def synthesize_results(topic: str, sources: list) -> str:
    context = "\n\n".join(sources)
    response = model.invoke(
        f"Sintetiza estas fuentes en un resumen ejecutivo de 3-4 oraciones sobre '{topic}':\n\n{context}"
    )
    return response.content

@entrypoint()
def parallel_research(topic: str) -> dict:
    wiki_future = search_wikipedia(topic)
    arxiv_future = search_arxiv(topic)
    news_future = search_news(topic)

    wiki = wiki_future.result()
    arxiv = arxiv_future.result()
    news = news_future.result()

    summary = synthesize_results(topic, [wiki, arxiv, news]).result()

    return {
        "topic": topic,
        "sources": [wiki, arxiv, news],
        "summary": summary,
    }

result = parallel_research.invoke("large language models fine-tuning")
print(f"Fuentes recolectadas: {len(result['sources'])}")
for source in result["sources"]:
    print(f"  {source[:80]}...")
print(f"\nResumen: {result['summary'][:150]}...")
# Output esperado:
# Fuentes recolectadas: 3
#   [Wikipedia] Los modelos de lenguaje grandes (LLMs) son redes neuronales...
#   [arXiv] Presentamos un análisis comparativo de técnicas de fine-tuning...
#   [News] Las empresas de tecnología están adoptando técnicas de fine-tuning...
# Resumen: El fine-tuning de LLMs ha evolucionado desde...

La mecánica de Futures

wiki_future = search_wikipedia(topic)    # Retorna Future (task se lanza)
arxiv_future = search_arxiv(topic)       # Retorna Future (task se lanza)
news_future = search_news(topic)         # Retorna Future (task se lanza)
# Las 3 tasks están ejecutándose en paralelo

wiki = wiki_future.result()    # Bloquea hasta que wiki termine
arxiv = arxiv_future.result()  # Bloquea hasta que arxiv termine
news = news_future.result()    # Bloquea hasta que news termine

La diferencia clave: si llamas .result() inmediatamente después de cada task, pierdes el paralelismo — cada task espera a la anterior. El pattern es: lanzar todas las tasks primero, recoger todos los resultados después.

# ❌ Secuencial (no aprovecha paralelismo)
wiki = search_wikipedia(topic).result()    # Espera
arxiv = search_arxiv(topic).result()       # Espera
news = search_news(topic).result()         # Espera

# ✅ Paralelo (lanza todas, recoge después)
wiki_fut = search_wikipedia(topic)         # Lanza
arxiv_fut = search_arxiv(topic)            # Lanza
news_fut = search_news(topic)              # Lanza
wiki = wiki_fut.result()                   # Recoge
arxiv = arxiv_fut.result()                 # Recoge
news = news_fut.result()                   # Recoge

Pattern 4: Structured output extraction

Usar Pydantic + with_structured_output() dentro de un @task para extraer datos estructurados de texto libre. Garantiza que el output del LLM tenga exactamente la estructura que tu código espera.

Implementación completa

from dotenv import load_dotenv
load_dotenv()

from pydantic import BaseModel, Field
from langgraph.func import entrypoint, task
from langchain.chat_models import init_chat_model

class ProductReview(BaseModel):
    sentiment: str = Field(description="'positive', 'negative', o 'neutral'")
    score: float = Field(description="Score de 0.0 a 1.0")
    key_points: list[str] = Field(description="Lista de puntos clave mencionados")
    recommendation: str = Field(description="Resumen de una oración")

model = init_chat_model("openai:gpt-4.1-mini")
structured_model = model.with_structured_output(ProductReview)

@task
def analyze_review(review_text: str) -> dict:
    result = structured_model.invoke(
        f"Analiza esta reseña de producto y extrae información estructurada:\n\n{review_text}"
    )
    return result.model_dump()

@task
def aggregate_reviews(analyses: list) -> dict:
    avg_score = sum(a["score"] for a in analyses) / len(analyses)
    all_points = []
    for a in analyses:
        all_points.extend(a["key_points"])

    sentiment_counts = {}
    for a in analyses:
        s = a["sentiment"]
        sentiment_counts[s] = sentiment_counts.get(s, 0) + 1

    return {
        "total_reviews": len(analyses),
        "average_score": round(avg_score, 2),
        "sentiment_distribution": sentiment_counts,
        "all_key_points": all_points,
    }

@entrypoint()
def review_analyzer(reviews: list) -> dict:
    analysis_futures = [analyze_review(r) for r in reviews]
    analyses = [fut.result() for fut in analysis_futures]
    summary = aggregate_reviews(analyses).result()
    return summary

reviews = [
    "Excelente producto, la batería dura todo el día. La cámara es increíble. Muy recomendado.",
    "Regular. El diseño es bonito pero el rendimiento no convence. Se calienta mucho.",
    "Pésimo servicio post-venta. El producto llegó defectuoso y no me dieron solución.",
]

result = review_analyzer.invoke(reviews)
print(f"Reviews analizadas: {result['total_reviews']}")
print(f"Score promedio: {result['average_score']}")
print(f"Sentimientos: {result['sentiment_distribution']}")
print(f"Puntos clave: {result['all_key_points'][:3]}")
# Output esperado:
# Reviews analizadas: 3
# Score promedio: 0.5
# Sentimientos: {'positive': 1, 'neutral': 1, 'negative': 1}
# Puntos clave: ['Batería duradera', 'Cámara de calidad', 'Se calienta mucho']

Por qué .model_dump() en el @task

El @task necesita retornar valores JSON-serializables para checkpointing. El objeto ProductReview de Pydantic no se serializa automáticamente por LangGraph, pero .model_dump() lo convierte a un diccionario Python estándar que sí es serializable. Es un paso extra, pero garantiza que el checkpoint funcione correctamente.

Pattern 5: Error handling con fallback

Usar try/except dentro del @entrypoint para manejar errores de forma elegante: reintentar con un modelo diferente, degradar la respuesta, o tomar un camino alternativo. Python nativo para resiliencia.

Implementación completa

from dotenv import load_dotenv
load_dotenv()

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

@task
def call_primary_model(prompt: str) -> str:
    model = init_chat_model("openai:gpt-4.1-mini")
    return model.invoke(prompt).content

@task
def call_fallback_model(prompt: str) -> str:
    model = init_chat_model("openai:gpt-4.1-nano")
    return model.invoke(prompt).content

@task
def validate_response(response: str, min_length: int) -> dict:
    if len(response) < min_length:
        return {"valid": False, "reason": f"Respuesta muy corta ({len(response)} chars, mínimo {min_length})"}
    if not any(c.isalpha() for c in response):
        return {"valid": False, "reason": "Respuesta no contiene texto legible"}
    return {"valid": True, "reason": "OK"}

@entrypoint()
def resilient_agent(request: dict) -> dict:
    prompt = request["prompt"]
    min_length = request.get("min_length", 50)
    max_retries = request.get("max_retries", 2)

    for attempt in range(max_retries + 1):
        try:
            if attempt == 0:
                response = call_primary_model(prompt).result()
            else:
                response = call_fallback_model(prompt).result()

            validation = validate_response(response, min_length).result()

            if validation["valid"]:
                return {
                    "response": response,
                    "attempt": attempt + 1,
                    "model_used": "primary" if attempt == 0 else "fallback",
                    "status": "success",
                }

        except Exception as e:
            if attempt == max_retries:
                return {
                    "response": f"Error después de {max_retries + 1} intentos: {str(e)}",
                    "attempt": attempt + 1,
                    "model_used": "none",
                    "status": "error",
                }
            continue

    return {
        "response": response,
        "attempt": max_retries + 1,
        "model_used": "fallback",
        "status": "degraded",
    }

result = resilient_agent.invoke({
    "prompt": "Explica qué es machine learning en 3 oraciones.",
    "min_length": 50,
    "max_retries": 2,
})
print(f"Status: {result['status']}")
print(f"Modelo usado: {result['model_used']}")
print(f"Intento: {result['attempt']}")
print(f"Respuesta: {result['response'][:120]}...")
# Output esperado:
# Status: success
# Modelo usado: primary
# Intento: 1
# Respuesta: Machine learning es una rama de la inteligencia artificial...

Anatomía del pattern

1. Loop de reintentos: for attempt in range(max_retries + 1) — Python nativo para retry
2. Escalación de modelo: Intento 0 usa el modelo principal, reintentos usan el fallback
3. Validación: Un @task que verifica la calidad de la respuesta antes de aceptarla
4. Fallback graceful: Si todo falla, retorna un error informativo en vez de crashear

Con la Graph API, este pattern requiere un nodo de validación, un conditional edge para retry, otro para fallback, y lógica de conteo de intentos en el estado. Con la Functional API, es un for loop con try/except — el pattern que cualquier developer Python ya conoce.

Cuándo los patterns no alcanzan

Estos patterns cubren la mayoría de los workflows secuenciales y de complejidad moderada. Pero hay señales claras de que necesitas migrar a la Graph API:

• ❌ Necesitas más de 3 branches condicionales que convergen en un punto de merge — la lógica de merge se complica con variables locales
• ❌ Quieres Send API para crear workers dinámicos basados en datos del runtime — la Functional API no tiene equivalente directo
• ❌ El workflow tiene sub-workflows que se componen como sub-grafos reutilizables por otros equipos
• ❌ Necesitas visualización automática del flujo para documentación o monitoring — draw_mermaid_png() solo existe en Graph API
• ❌ Quieres control granular de streaming por nodo, con stream_mode="messages" o custom stream events

Si te encuentras en alguno de estos escenarios, no forces la Functional API. Es como usar un for loop cuando necesitas recursión — técnicamente posible, pero el código se vuelve frágil e ilegible. La Graph API existe para estos casos.

Troubleshooting

Problema 1: "Mis tasks no se ejecutan en paralelo"

Síntoma: Lanzas 3 tasks pero el tiempo total es igual a la suma de las 3, no al máximo.

Causa: Estás llamando .result() inmediatamente después de cada task:

# ❌ Secuencial disfrazado
a = task_a(x).result()
b = task_b(y).result()
c = task_c(z).result()

Solución: Separa el lanzamiento de la recolección:

# ✅ Paralelo real
fut_a = task_a(x)
fut_b = task_b(y)
fut_c = task_c(z)
a, b, c = fut_a.result(), fut_b.result(), fut_c.result()

Problema 2: "Mi @task retorna un objeto que no se serializa"

Síntoma: Error de serialización al ejecutar un workflow con checkpointer.

Causa: El @task retorna un objeto Python complejo (clase custom, modelo Pydantic sin convertir, objeto con métodos).

Solución: Convierte a tipos serializables antes de retornar:

@task
def my_task(input: str) -> dict:
    result = some_complex_operation(input)
    # ❌ return result  (objeto complejo)
    # ✅ Convertir a dict/list/str
    return {"value": str(result), "score": result.score}

Para modelos Pydantic, usa .model_dump().

Problema 3: "El retry loop no recupera tasks previos del checkpoint"

Síntoma: Al resumir un workflow después de un error, las tasks que ya completaron se re-ejecutan.

Causa: Las tasks dentro de un try/except que fallan no guardan resultado. Pero las que completaron exitosamente sí se recuperan del checkpoint al resumir.

Solución: Asegúrate de que el error ocurre en una task específica, no en lógica fuera de tasks. La regla: todo efecto secundario (llamadas a APIs, I/O) debe estar dentro de un @task para que el checkpoint funcione correctamente.

Problema 4: "El @entrypoint no acepta múltiples argumentos"

Síntoma: TypeError: entrypoint function must accept exactly one positional argument.

Causa: @entrypoint solo acepta un argumento posicional (el input del workflow). Si necesitas pasar múltiples valores, usa un dict:

# ❌ Múltiples argumentos
@entrypoint()
def my_workflow(topic: str, max_results: int) -> str:
    ...

# ✅ Un solo dict como input
@entrypoint()
def my_workflow(config: dict) -> str:
    topic = config["topic"]
    max_results = config.get("max_results", 5)
    ...

Ejercicios

Ejercicio 1: Tool loop básico (Fácil)

Implementa un agente con el Pattern 1 (tool execution loop) que tenga una sola tool: lookup_capital(country: str) -> str que retorna la capital de un país. Prueba con "¿Cuál es la capital de Francia?" y "¿Cuáles son las capitales de España, Alemania y Japón?".

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

CAPITALS = {
    "francia": "París", "españa": "Madrid", "alemania": "Berlín",
    "japón": "Tokio", "brasil": "Brasilia", "méxico": "Ciudad de México",
}

@tool
def lookup_capital(country: str) -> str:
    """Busca la capital de un país."""
    return CAPITALS.get(country.lower(), f"No tengo datos para {country}")

tools_list = [lookup_capital]
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 execute_tool(tool_call: dict):
    return tool_map[tool_call["name"]].invoke(tool_call["args"])

@entrypoint()
def capital_agent(question: str) -> str:
    messages = [
        SystemMessage(content="Ayuda al usuario a encontrar capitales de países."),
        HumanMessage(content=question),
    ]

    while True:
        response = call_model(messages).result()
        if not response.tool_calls:
            return response.content
        tool_futures = [execute_tool(tc) for tc in response.tool_calls]
        tool_results = [fut.result() for fut in tool_futures]
        messages = add_messages(messages, [response, *tool_results])

print(capital_agent.invoke("¿Cuál es la capital de Francia?"))
print(capital_agent.invoke("¿Cuáles son las capitales de España, Alemania y Japón?"))
# Output esperado:
# La capital de Francia es París.
# Las capitales son: España → Madrid, Alemania → Berlín, Japón → Tokio.

Ejercicio 2: Chain of thought con pasos variables (Fácil)

Adapta el Pattern 2 (multi-step reasoning) para que la cantidad de sub-preguntas sea configurable. El input debe ser un dict con "question" y "num_steps". Si num_steps es 2, descompone en 2 sub-preguntas; si es 4, en 4.

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 decompose(question: str, num_steps: int) -> list:
    response = model.invoke(
        f"Descompone esta pregunta en exactamente {num_steps} sub-preguntas. "
        f"Responde SOLO con las sub-preguntas, una por línea.\n\n"
        f"Pregunta: {question}"
    )
    questions = [q.strip() for q in response.content.strip().split("\n") if q.strip()]
    return questions[:num_steps]

@task
def answer_subquestion(sub_question: str) -> str:
    return model.invoke(f"Responde concisamente (2-3 oraciones): {sub_question}").content

@task
def synthesize(question: str, answers: list) -> str:
    context = "\n".join(f"- {a}" for a in answers)
    return model.invoke(
        f"Sintetiza una respuesta completa.\n\nPregunta: {question}\n\nRespuestas parciales:\n{context}"
    ).content

@entrypoint()
def configurable_cot(config: dict) -> dict:
    question = config["question"]
    num_steps = config.get("num_steps", 3)

    sub_questions = decompose(question, num_steps).result()
    answers = [answer_subquestion(sq).result() for sq in sub_questions]
    final = synthesize(question, answers).result()

    return {"sub_questions": sub_questions, "answers": answers, "final_answer": final}

result = configurable_cot.invoke({
    "question": "¿Cómo impacta la inteligencia artificial en la educación?",
    "num_steps": 2,
})
print(f"Sub-preguntas ({len(result['sub_questions'])}):")
for sq in result["sub_questions"]:
    print(f"  - {sq}")
print(f"\nRespuesta: {result['final_answer'][:150]}...")
# Output esperado:
# Sub-preguntas (2):
#   - ¿De qué formas la IA está cambiando los métodos de enseñanza?
#   - ¿Cuáles son los riesgos y desafíos de usar IA en educación?
# Respuesta: La inteligencia artificial impacta la educación...

Ejercicio 3: Búsqueda paralela con timeout simulado (Medio)

Implementa el Pattern 3 (parallel execution) con 4 fuentes de búsqueda. Una de las fuentes debe simular un fallo (raise exception). Usa try/except para manejar el fallo y sintetizar con las fuentes que sí respondieron.

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 search_source_a(topic: str) -> str:
    return f"[Source A] Información general sobre {topic} encontrada."

@task
def search_source_b(topic: str) -> str:
    return f"[Source B] Datos técnicos sobre {topic} recopilados."

@task
def search_source_c(topic: str) -> str:
    raise ConnectionError(f"Source C no disponible para '{topic}'")

@task
def search_source_d(topic: str) -> str:
    return f"[Source D] Noticias recientes sobre {topic} encontradas."

@task
def synthesize(topic: str, results: list) -> str:
    context = "\n".join(results)
    return model.invoke(
        f"Sintetiza estas fuentes sobre '{topic}':\n\n{context}"
    ).content

@entrypoint()
def resilient_search(topic: str) -> dict:
    futures = {
        "a": search_source_a(topic),
        "b": search_source_b(topic),
        "c": search_source_c(topic),
        "d": search_source_d(topic),
    }

    results = []
    errors = []
    for name, future in futures.items():
        try:
            results.append(future.result())
        except Exception as e:
            errors.append(f"Source {name}: {str(e)}")

    summary = synthesize(topic, results).result()

    return {
        "sources_ok": len(results),
        "sources_failed": len(errors),
        "errors": errors,
        "summary": summary,
    }

result = resilient_search.invoke("quantum computing")
print(f"Fuentes exitosas: {result['sources_ok']}")
print(f"Fuentes fallidas: {result['sources_failed']}")
print(f"Errores: {result['errors']}")
print(f"Resumen: {result['summary'][:100]}...")
# Output esperado:
# Fuentes exitosas: 3
# Fuentes fallidas: 1
# Errores: ["Source c: Source C no disponible para 'quantum computing'"]
# Resumen: Basándonos en las fuentes disponibles sobre quantum computing...

Ejercicio 4: Extracción estructurada de múltiples entidades (Medio)

Usa el Pattern 4 (structured output) para extraer entidades de un texto en paralelo: personas, organizaciones y ubicaciones. Cada extracción como un @task separado con su propio modelo Pydantic. Combina los resultados.

from dotenv import load_dotenv
load_dotenv()

from pydantic import BaseModel, Field
from langgraph.func import entrypoint, task
from langchain.chat_models import init_chat_model

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

class People(BaseModel):
    names: list[str] = Field(description="Nombres de personas mencionadas")

class Organizations(BaseModel):
    names: list[str] = Field(description="Nombres de organizaciones mencionadas")

class Locations(BaseModel):
    names: list[str] = Field(description="Nombres de ubicaciones mencionadas")

@task
def extract_people(text: str) -> list:
    extractor = model.with_structured_output(People)
    result = extractor.invoke(f"Extrae todos los nombres de personas de este texto:\n\n{text}")
    return result.names

@task
def extract_organizations(text: str) -> list:
    extractor = model.with_structured_output(Organizations)
    result = extractor.invoke(f"Extrae todas las organizaciones de este texto:\n\n{text}")
    return result.names

@task
def extract_locations(text: str) -> list:
    extractor = model.with_structured_output(Locations)
    result = extractor.invoke(f"Extrae todas las ubicaciones de este texto:\n\n{text}")
    return result.names

@entrypoint()
def entity_extractor(text: str) -> dict:
    people_fut = extract_people(text)
    orgs_fut = extract_organizations(text)
    locs_fut = extract_locations(text)

    return {
        "people": people_fut.result(),
        "organizations": orgs_fut.result(),
        "locations": locs_fut.result(),
    }

text = (
    "Satya Nadella, CEO de Microsoft, anunció en Seattle que la empresa invertirá "
    "en inteligencia artificial junto con OpenAI. Sam Altman confirmó la colaboración "
    "desde las oficinas de San Francisco."
)

result = entity_extractor.invoke(text)
print(f"Personas: {result['people']}")
print(f"Organizaciones: {result['organizations']}")
print(f"Ubicaciones: {result['locations']}")
# Output esperado:
# Personas: ['Satya Nadella', 'Sam Altman']
# Organizaciones: ['Microsoft', 'OpenAI']
# Ubicaciones: ['Seattle', 'San Francisco']

Ejercicio 5: Pipeline completo con todos los patterns (Avanzado)

Combina los patterns 1, 3, 4 y 5 en un solo workflow: un agente que recibe un tema, busca en 3 fuentes en paralelo (Pattern 3), extrae datos estructurados de cada fuente (Pattern 4), maneja errores con fallback (Pattern 5), y usa un LLM para sintetizar el resultado final. El loop de tools (Pattern 1) no aplica aquí directamente, pero el loop de retry sí.

from dotenv import load_dotenv
load_dotenv()

from pydantic import BaseModel, Field
from langgraph.func import entrypoint, task
from langchain.chat_models import init_chat_model

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

class SourceSummary(BaseModel):
    key_facts: list[str] = Field(description="Hechos clave encontrados")
    confidence: float = Field(description="Confianza de 0.0 a 1.0")
    source_type: str = Field(description="Tipo de fuente: 'academic', 'news', o 'general'")

@task
def search_source(topic: str, source_name: str, source_type: str) -> str:
    return model.invoke(
        f"Simula una búsqueda en {source_name} sobre '{topic}'. "
        f"Genera un párrafo informativo como si fuera de una fuente {source_type}."
    ).content

@task
def extract_structured(text: str, source_type: str) -> dict:
    extractor = model.with_structured_output(SourceSummary)
    result = extractor.invoke(
        f"Extrae hechos clave y evalúa la confianza de este texto (fuente tipo '{source_type}'):\n\n{text}"
    )
    return result.model_dump()

@task
def synthesize_final(topic: str, structured_data: list) -> str:
    all_facts = []
    for sd in structured_data:
        for fact in sd["key_facts"]:
            all_facts.append(f"[{sd['source_type']}, confianza {sd['confidence']}] {fact}")
    context = "\n".join(all_facts)
    return model.invoke(
        f"Genera un resumen ejecutivo sobre '{topic}' basado en estos hechos:\n\n{context}"
    ).content

SOURCES = [
    {"name": "Wikipedia", "type": "general"},
    {"name": "arXiv", "type": "academic"},
    {"name": "TechCrunch", "type": "news"},
]

@entrypoint()
def research_pipeline(topic: str) -> dict:
    search_futures = [
        search_source(topic, s["name"], s["type"]) for s in SOURCES
    ]

    raw_results = []
    errors = []
    for i, fut in enumerate(search_futures):
        try:
            raw_results.append({"text": fut.result(), "type": SOURCES[i]["type"]})
        except Exception as e:
            errors.append(f"{SOURCES[i]['name']}: {str(e)}")

    if not raw_results:
        return {"error": "Todas las fuentes fallaron", "errors": errors}

    extract_futures = [
        extract_structured(r["text"], r["type"]) for r in raw_results
    ]
    structured_data = [fut.result() for fut in extract_futures]

    summary = synthesize_final(topic, structured_data).result()

    return {
        "topic": topic,
        "sources_used": len(raw_results),
        "sources_failed": len(errors),
        "structured_data": structured_data,
        "summary": summary,
    }

result = research_pipeline.invoke("retrieval-augmented generation")
print(f"Fuentes: {result['sources_used']} OK, {result['sources_failed']} fallidas")
print(f"Hechos extraídos: {sum(len(sd['key_facts']) for sd in result['structured_data'])}")
print(f"Resumen: {result['summary'][:150]}...")
# Output esperado:
# Fuentes: 3 OK, 0 fallidas
# Hechos extraídos: 6-9
# Resumen: RAG (Retrieval-Augmented Generation) es una técnica que combina...

Ejercicio 6: Evaluator-optimizer loop (Avanzado)

Implementa un workflow que genere un resumen de un texto, lo evalúe con criterios específicos (longitud, claridad, completitud), y lo regenere si no pasa la evaluación. Máximo 3 iteraciones. Usa Pydantic para la evaluación estructurada.

from dotenv import load_dotenv
load_dotenv()

from pydantic import BaseModel, Field
from langgraph.func import entrypoint, task
from langchain.chat_models import init_chat_model

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

class Evaluation(BaseModel):
    passes: bool = Field(description="True si el resumen cumple todos los criterios")
    clarity_score: float = Field(description="Claridad de 0.0 a 1.0")
    completeness_score: float = Field(description="Completitud de 0.0 a 1.0")
    feedback: str = Field(description="Feedback específico para mejorar")

@task
def generate_summary(text: str, feedback: str = "") -> str:
    prompt = f"Resume este texto en 3-4 oraciones claras y completas:\n\n{text}"
    if feedback:
        prompt += f"\n\nFeedback de la iteración anterior (mejora esto): {feedback}"
    return model.invoke(prompt).content

@task
def evaluate_summary(original: str, summary: str) -> dict:
    evaluator = model.with_structured_output(Evaluation)
    result = evaluator.invoke(
        f"Evalúa este resumen comparándolo con el texto original.\n\n"
        f"Original:\n{original}\n\nResumen:\n{summary}\n\n"
        f"Criterios: claridad >= 0.7, completitud >= 0.7, longitud entre 50-200 palabras."
    )
    return result.model_dump()

@entrypoint()
def summarize_with_eval(text: str) -> dict:
    feedback = ""
    max_iterations = 3

    for iteration in range(max_iterations):
        summary = generate_summary(text, feedback).result()
        evaluation = evaluate_summary(text, summary).result()

        if evaluation["passes"]:
            return {
                "summary": summary,
                "iterations": iteration + 1,
                "final_evaluation": evaluation,
                "status": "approved",
            }

        feedback = evaluation["feedback"]

    return {
        "summary": summary,
        "iterations": max_iterations,
        "final_evaluation": evaluation,
        "status": "max_iterations_reached",
    }

text = (
    "Los modelos de lenguaje grandes (LLMs) han transformado el campo del procesamiento "
    "del lenguaje natural. Basados en la arquitectura transformer, estos modelos aprenden "
    "patrones estadísticos del lenguaje a partir de grandes cantidades de texto. Su capacidad "
    "para generar texto coherente, responder preguntas y realizar tareas de razonamiento "
    "los ha convertido en herramientas fundamentales en la industria tecnológica. Sin embargo, "
    "presentan desafíos como alucinaciones, sesgo y alto costo computacional."
)

result = summarize_with_eval.invoke(text)
print(f"Status: {result['status']}")
print(f"Iteraciones: {result['iterations']}")
print(f"Claridad: {result['final_evaluation']['clarity_score']}")
print(f"Completitud: {result['final_evaluation']['completeness_score']}")
print(f"Resumen: {result['summary']}")
# Output esperado:
# Status: approved
# Iteraciones: 1-2
# Claridad: 0.8+
# Completitud: 0.8+
# Resumen: Los LLMs, basados en la arquitectura transformer, han revolucionado...

Resumen

En esta cápsula aprendiste 5 patterns reutilizables para la Functional API:

• Pattern 1: Tool execution loop — El loop ReAct implementado con while True y @task. Las tools se ejecutan en paralelo con Futures. Es el pattern base para cualquier agente
• Pattern 2: Multi-step reasoning — Descomponer → razonar → sintetizar. Cada paso como un @task con checkpoint automático. Si falla en el paso 3, los pasos 1 y 2 no se re-ejecutan
• Pattern 3: Parallel task execution — Lanzar múltiples @task sin .result(), recoger después. La diferencia entre secuencial y paralelo es dónde pones el .result()
• Pattern 4: Structured output — Pydantic + with_structured_output() dentro de @task. Usa .model_dump() para que el output sea serializable para checkpointing
• Pattern 5: Error handling con fallback — try/except + for loop para retry con escalación de modelo. Python nativo para resiliencia sin conditional edges

Cuando estos patterns no alcanzan — branches complejos, Send API, sub-workflows reutilizables, visualización automática — es momento de migrar a la Graph API.

Próxima cápsula: Mezclando APIs — cómo usar un StateGraph dentro de un @entrypoint y viceversa, para aprovechar lo mejor de ambas APIs en el mismo proyecto.

Recursos adicionales

1. Functional API Conceptual Guide — Conceptos fundamentales de @entrypoint y @task
2. Workflows and Agents Patterns — Patterns oficiales (parallelization, routing, evaluator-optimizer)
3. How to use the Functional API — Tutorial práctico con ejemplos completos
4. LangGraph Persistence — Cómo funciona el checkpointing que hace posibles estos patterns
5. Pydantic with_structured_output — Structured output con LangChain y Pydantic
6. LangGraph Error Handling — Manejo de errores y retry en LangGraph
7. Python Futures (concurrent.futures) — Referencia de Futures en Python (concepto análogo al de @task)

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