@task: Tareas que Componen el Agente

Descripción de la cápsula

En la cápsula anterior aprendiste @entrypoint: el punto de entrada de tu workflow funcional. Pero un @entrypoint que hace todo en una sola función no aprovecha las garantías de LangGraph — checkpointing, streaming de progreso, ejecución paralela. Para eso necesitas descomponer tu workflow en tareas.

@task define unidades independientes de trabajo dentro de tu workflow. Cada tarea es:

• Checkpointable — si tu workflow crashea después de completar la tarea A, no necesitas re-ejecutarla. LangGraph guardó su resultado.
• Streamable — puedes reportar progreso al usuario mientras las tareas se ejecutan.
• Composable — puedes ejecutar múltiples tareas en paralelo y combinar sus resultados.

Piensa en @task como los bloques de construcción dentro de tu @entrypoint. El entrypoint es el plano del edificio; las tasks son los ladrillos.

Hay un concepto que necesitas internalizar antes de escribir código: @task no retorna el resultado directamente — retorna un Future. Si no entiendes esto, vas a pasar horas debuggeando por qué tu código "funciona" pero produce objetos extraños en vez de strings. Esta cápsula te va a dejar esto clarísimo.

Import

from langgraph.func import entrypoint, task

task vive en el mismo módulo que entrypoint. Los importas juntos porque siempre los usas juntos.

Uso básico: @task dentro de @entrypoint

Empecemos con un ejemplo completo que muestra el patrón fundamental:

from dotenv import load_dotenv
load_dotenv()

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

@task
def search_web(query: str) -> str:
    """Busca información en la web."""
    return f"Resultados para '{query}': LangChain es un framework para construir aplicaciones con LLMs. Soporta múltiples proveedores y tiene herramientas de orquestación."

@task
def summarize(text: str) -> str:
    """Genera un resumen del texto usando un LLM."""
    model = init_chat_model("openai:gpt-4.1-mini")
    response = model.invoke(f"Resume esto en una oración: {text}")
    return response.content

@entrypoint()
def research(topic: str) -> str:
    search_result = search_web(topic).result()
    summary = summarize(search_result).result()
    return summary

result = research.invoke("LangChain")
print(result)
# Output esperado: LangChain es un framework que permite construir aplicaciones
# con LLMs usando múltiples proveedores y herramientas de orquestación.

Paso a paso:

1. search_web está decorada con @task — es una unidad de trabajo independiente
2. summarize también es una @task — otra unidad independiente
3. Dentro de @entrypoint, llamas a cada task y llamas .result() para obtener el valor
4. El @entrypoint orquesta: primero busca, luego resume

La pregunta obvia: ¿por qué .result()? ¿Por qué no puedo usar el valor directamente?

El concepto de Future: @task NO retorna el resultado

Este es el concepto más importante de esta cápsula. Léelo dos veces si es necesario.

Cuando llamas a una función decorada con @task, no obtienes el resultado de la función. Obtienes un Future — un objeto que promete que el resultado va a estar disponible, pero no lo tiene todavía.

La analogía: JavaScript Promises

Si conoces JavaScript, un Future es exactamente como una Promise:

// JavaScript — Promise
const resultPromise = fetch("https://api.example.com/data");
// resultPromise NO es la data — es una Promise
// Necesitas: const data = await resultPromise;

# Python/LangGraph — Future
result_future = search_web("LangChain")
# result_future NO es el string — es un Future
# Necesitas: result = result_future.result()

¿Por qué Futures? Tres razones concretas

1. Checkpointing entre tareas

Si la tarea A completa y retorna su Future, LangGraph puede guardar ese resultado. Si la tarea B falla, puedes re-ejecutar el workflow y LangGraph salta la tarea A (ya tiene su resultado guardado). Sin Futures, LangGraph no tendría un punto de corte entre tareas.

2. Ejecución paralela

Cuando llamas a dos tasks sin pedir .result() inmediatamente, LangGraph puede ejecutarlas en paralelo:

@entrypoint()
def research(topic: str) -> str:
    future_a = search_web(topic)       # Lanza tarea A
    future_b = search_news(topic)      # Lanza tarea B (paralela)
    
    result_a = future_a.result()       # Espera resultado A
    result_b = future_b.result()       # Espera resultado B
    
    return f"{result_a}\n{result_b}"

Si @task retornara el resultado directamente, la segunda task no empezaría hasta que la primera termine. Los Futures permiten lanzar ambas y esperar después.

3. Streaming de progreso

LangGraph puede informar al usuario: "tarea search_web iniciada", "tarea search_web completada", "tarea summarize iniciada"... Esto es posible porque cada task es una unidad discreta con inicio y fin, gracias a los Futures.

Qué pasa sin .result()

Veamos qué ocurre cuando olvidas llamar .result():

from dotenv import load_dotenv
load_dotenv()

from langgraph.func import entrypoint, task

@task
def greet(name: str) -> str:
    return f"¡Hola, {name}!"

@entrypoint()
def my_workflow(name: str) -> str:
    result = greet(name)         # Sin .result()
    return f"Respuesta: {result}"

output = my_workflow.invoke("Ana")
print(output)
# Output: Respuesta: <langgraph.types.Future object at 0x...>

En lugar de "Respuesta: ¡Hola, Ana!", obtienes un objeto Future convertido a string. Tu código no lanza error — simplemente produce basura. Este es el bug más silencioso y frustrante de la Functional API.

La corrección:

@entrypoint()
def my_workflow(name: str) -> str:
    result = greet(name).result()  # ✅ Con .result()
    return f"Respuesta: {result}"

output = my_workflow.invoke("Ana")
print(output)
# Output esperado: Respuesta: ¡Hola, Ana!

Regla de oro: cada vez que llamas a una @task, la siguiente operación debe ser .result() (a menos que estés lanzando tareas en paralelo intencionalmente).

Ejemplo completo: workflow de investigación con múltiples tasks

Un ejemplo que justifica el uso de @task — cada tarea puede fallar independientemente, es costosa (llama a APIs), y podría ejecutarse en paralelo:

from dotenv import load_dotenv
load_dotenv()

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

@task
def decompose_query(topic: str) -> list[str]:
    """Descompone un tema en sub-preguntas de investigación."""
    model = init_chat_model("openai:gpt-4.1-mini")
    response = model.invoke(
        f"Genera exactamente 3 sub-preguntas de investigación sobre: {topic}. "
        f"Retorna solo las preguntas, una por línea, sin numeración."
    )
    questions = [q.strip() for q in response.content.strip().split("\n") if q.strip()]
    return questions[:3]

@task
def research_question(question: str) -> str:
    """Investiga una pregunta individual usando un LLM."""
    model = init_chat_model("openai:gpt-4.1-mini")
    response = model.invoke(
        f"Responde esta pregunta de investigación en 2-3 oraciones: {question}"
    )
    return response.content

@task
def synthesize(findings: list[str]) -> str:
    """Sintetiza múltiples hallazgos en un resumen coherente."""
    model = init_chat_model("openai:gpt-4.1-mini")
    combined = "\n\n".join(f"Hallazgo {i+1}: {f}" for i, f in enumerate(findings))
    response = model.invoke(
        f"Sintetiza estos hallazgos de investigación en un párrafo coherente:\n\n{combined}"
    )
    return response.content

@entrypoint()
def research_agent(topic: str) -> str:
    questions = decompose_query(topic).result()
    
    findings = []
    for question in questions:
        finding = research_question(question).result()
        findings.append(finding)
    
    summary = synthesize(findings).result()
    return summary

result = research_agent.invoke("El impacto de la IA en la educación")
print(result)
# Output esperado: Un párrafo coherente que sintetiza hallazgos sobre
# el impacto de la IA en la educación, cubriendo las 3 sub-preguntas.

¿Por qué cada función necesita @task aquí?

• decompose_query: llama a un LLM (costoso, puede fallar por rate limit)
• research_question: llama a un LLM por cada pregunta (puede fallar independientemente)
• synthesize: llama a un LLM (si falla, no queremos re-ejecutar las investigaciones)

Si decompose_query y las 3 llamadas a research_question completan pero synthesize falla, LangGraph puede re-ejecutar solo synthesize gracias al checkpointing por tarea.

Ejecución paralela con Futures

El ejemplo anterior investiga las preguntas en secuencia. Pero como research_question es una @task que retorna un Future, puedes lanzarlas todas y esperar después:

from dotenv import load_dotenv
load_dotenv()

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

@task
def search_source(source: str, query: str) -> str:
    """Busca en una fuente específica."""
    model = init_chat_model("openai:gpt-4.1-mini")
    response = model.invoke(
        f"Simula ser la fuente '{source}'. Responde brevemente: {query}"
    )
    return f"[{source}] {response.content}"

@task
def merge_results(results: list[str]) -> str:
    """Combina resultados de múltiples fuentes."""
    return "Resultados combinados:\n" + "\n".join(f"  • {r}" for r in results)

@entrypoint()
def multi_source_search(query: str) -> str:
    sources = ["Wikipedia", "ArXiv", "StackOverflow"]
    
    futures = []
    for source in sources:
        future = search_source(source, query)  # Lanza sin .result()
        futures.append(future)
    
    results = [f.result() for f in futures]  # Espera todos
    
    merged = merge_results(results).result()
    return merged

output = multi_source_search.invoke("¿Qué es RAG?")
print(output)
# Output esperado:
# Resultados combinados:
#   • [Wikipedia] RAG (Retrieval-Augmented Generation) es una técnica...
#   • [ArXiv] RAG combina recuperación de documentos con generación...
#   • [StackOverflow] RAG se usa para dar contexto relevante al LLM...

El patrón es claro:

1. Lanzar — llamas a las tasks en un loop sin .result()
2. Recolectar — guardas los Futures en una lista
3. Esperar — llamas .result() en cada Future cuando necesitas los valores

Cuándo usar @task vs funciones regulares

No todas las funciones necesitan ser tasks. Usar @task innecesariamente agrega overhead (serialización, checkpointing) sin beneficio.

Ejemplo de cuándo no usar @task:

from dotenv import load_dotenv
load_dotenv()

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

def format_as_markdown(text: str) -> str:
    """Función regular — simple, rápida, no falla."""
    return f"## Resultado\n\n{text}"

def validate_input(topic: str) -> str:
    """Función regular — validación trivial."""
    if len(topic.strip()) < 3:
        raise ValueError("El tema debe tener al menos 3 caracteres")
    return topic.strip()

@task
def generate_analysis(topic: str) -> str:
    """@task — llama a un LLM, puede fallar, es costosa."""
    model = init_chat_model("openai:gpt-4.1-mini")
    response = model.invoke(f"Analiza brevemente el tema: {topic}")
    return response.content

@entrypoint()
def analyze(topic: str) -> str:
    clean_topic = validate_input(topic)       # Función regular — sin .result()
    analysis = generate_analysis(clean_topic).result()  # @task — con .result()
    formatted = format_as_markdown(analysis)  # Función regular — sin .result()
    return formatted

result = analyze.invoke("Inteligencia Artificial")
print(result)
# Output esperado:
# ## Resultado
#
# La Inteligencia Artificial es un campo de la informática que busca
# crear sistemas capaces de realizar tareas que normalmente requieren
# inteligencia humana...

Las funciones regulares se llaman directamente (sin .result()). Solo las @task retornan Futures.

Tasks como unidades checkpointables

Una de las razones principales para usar @task es el checkpointing. Cuando configuras un checkpointer (lo verás en detalle en el Módulo 8), LangGraph guarda el resultado de cada tarea completada.

El concepto es simple:

Workflow: task_A → task_B → task_C

Primera ejecución:
  ✅ task_A completa → resultado guardado
  ✅ task_B completa → resultado guardado
  ❌ task_C falla (error de API)

Segunda ejecución (con checkpointer):
  ⏭️ task_A — resultado recuperado del checkpoint (no se re-ejecuta)
  ⏭️ task_B — resultado recuperado del checkpoint (no se re-ejecuta)
  ✅ task_C — se re-ejecuta (ahora funciona)

Para que el checkpointing funcione, necesitas dos cosas:

1. Que tus unidades de trabajo estén decoradas con @task
2. Que configures un checkpointer al compilar (Módulo 8)

Sin @task, LangGraph no tiene puntos de corte para guardar progreso. Todo el @entrypoint es una sola unidad — si falla en cualquier punto, se re-ejecuta todo.

from dotenv import load_dotenv
load_dotenv()

from langgraph.func import entrypoint, task

@task
def step_a() -> str:
    print("  Ejecutando step_a...")
    return "resultado_a"

@task
def step_b(input: str) -> str:
    print("  Ejecutando step_b...")
    return f"resultado_b (basado en {input})"

@task
def step_c(input: str) -> str:
    print("  Ejecutando step_c...")
    return f"resultado_c (basado en {input})"

@entrypoint()
def pipeline(start: str) -> str:
    a = step_a().result()
    b = step_b(a).result()
    c = step_c(b).result()
    return c

result = pipeline.invoke("inicio")
print(result)
# Output esperado:
#   Ejecutando step_a...
#   Ejecutando step_b...
#   Ejecutando step_c...
# resultado_c (basado en resultado_b (basado en resultado_a))

Cada print te muestra que la tarea se ejecutó. Con un checkpointer configurado (Módulo 8), en la segunda ejecución solo verías los print de las tareas que necesitan re-ejecutarse.

Restricciones de @task

Las tasks NO se pueden anidar

Una @task no puede llamar a otra @task internamente. Las tasks son planas — todas viven al mismo nivel dentro del @entrypoint:

from langgraph.func import entrypoint, task

@task
def outer_task(x: str) -> str:
    result = inner_task(x).result()  # ❌ ERROR: task dentro de task
    return result

@task
def inner_task(x: str) -> str:
    return x.upper()

La solución es llamar a ambas tasks desde el @entrypoint:

from dotenv import load_dotenv
load_dotenv()

from langgraph.func import entrypoint, task

@task
def process(x: str) -> str:
    return x.upper()

@task
def enrich(x: str) -> str:
    return f"[Enriched] {x}"

@entrypoint()
def workflow(text: str) -> str:
    processed = process(text).result()
    enriched = enrich(processed).result()
    return enriched

result = workflow.invoke("hola mundo")
print(result)
# Output esperado: [Enriched] HOLA MUNDO

Los valores de retorno deben ser serializables

Los resultados de @task se guardan para checkpointing y streaming. Eso significa que deben poder convertirse a JSON:

from langgraph.func import task

@task
def good_task() -> dict:
    return {"status": "ok", "items": [1, 2, 3]}  # ✅ Serializable

@task
def bad_task():
    import sqlite3
    return sqlite3.connect(":memory:")  # ❌ No serializable

Las tasks necesitan un @entrypoint padre

No puedes ejecutar una @task fuera de un @entrypoint. El entrypoint es el contexto que habilita checkpointing y streaming:

from langgraph.func import task

@task
def my_task(x: str) -> str:
    return x.upper()

# Esto NO funciona fuera de un @entrypoint:
# result = my_task("hola").result()  # ❌ Error: no hay contexto de entrypoint

Patrón: task con retry manual

Mientras no tengas el sistema de retry del Módulo 7, puedes implementar retry básico dentro de una task:

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 call_llm_with_retry(prompt: str, max_retries: int = 3) -> str:
    """Llama al LLM con reintentos manuales."""
    model = init_chat_model("openai:gpt-4.1-mini")
    for attempt in range(max_retries):
        try:
            response = model.invoke(prompt)
            return response.content
        except Exception as e:
            if attempt < max_retries - 1:
                wait = 2 ** attempt
                print(f"  Intento {attempt + 1} falló: {e}. Reintentando en {wait}s...")
                time.sleep(wait)
            else:
                raise RuntimeError(f"Falló después de {max_retries} intentos: {e}")

@entrypoint()
def robust_workflow(topic: str) -> str:
    result = call_llm_with_retry(f"Explica brevemente: {topic}").result()
    return result

output = robust_workflow.invoke("Futures en Python")
print(output)
# Output esperado: Los Futures en Python son objetos que representan
# el resultado de una operación asíncrona que aún no ha completado...

El retry vive dentro de la task porque la task es la unidad que puede fallar. El @entrypoint no necesita saber los detalles del retry.

Troubleshooting

Problema 1: "Mi código retorna un objeto Future en vez del valor"

Síntoma: En lugar de un string como "¡Hola!", obtienes algo como <langgraph.types.Future object at 0x...>. Causa: Olvidaste llamar .result() después de invocar una @task. Solución:

# ❌ Sin .result() — obtienes un Future
result = my_task("input")

# ✅ Con .result() — obtienes el valor real
result = my_task("input").result()

Problema 2: "Error: task llamada fuera de un entrypoint"

Síntoma: Error al intentar ejecutar una @task directamente. Causa: Las tasks solo pueden ejecutarse dentro de un @entrypoint. Solución: Asegúrate de que la llamada a la task está dentro de una función decorada con @entrypoint():

# ❌ Task suelta
output = my_task("data").result()

# ✅ Task dentro de entrypoint
@entrypoint()
def my_workflow(data: str) -> str:
    return my_task(data).result()

output = my_workflow.invoke("data")

Problema 3: "Error de serialización al retornar de una task"

Síntoma: Error indicando que el valor de retorno no se puede serializar. Causa: La task retorna un objeto no serializable (conexión de DB, función, generador). Solución: Retorna tipos primitivos, listas, diccionarios, o modelos Pydantic:

# ❌ Retorna objeto no serializable
@task
def bad() -> object:
    return open("file.txt")

# ✅ Retorna datos serializables
@task
def good() -> dict:
    with open("file.txt") as f:
        return {"content": f.read()}

Problema 4: "Intenté anidar tasks y obtuve un error"

Síntoma: Error al llamar una @task desde dentro de otra @task. Causa: Las tasks son planas — no se pueden anidar. Solución: Mueve ambas tasks al nivel del @entrypoint y orquesta desde ahí:

# ❌ Task dentro de task
@task
def outer(x):
    return inner(x).result()

# ✅ Ambas al mismo nivel, orquestadas por entrypoint
@entrypoint()
def workflow(x):
    a = task_a(x).result()
    b = task_b(a).result()
    return b

Problema 5: "Las tasks se ejecutan en secuencia aunque quiero paralelo"

Síntoma: Las tasks se ejecutan una tras otra, no en paralelo. Causa: Estás llamando .result() inmediatamente después de cada task. Solución: Lanza todas las tasks primero, luego llama .result():

# ❌ Secuencial — cada .result() bloquea antes de lanzar la siguiente
r1 = task_a("x").result()
r2 = task_b("y").result()

# ✅ Paralelo — lanza todas, luego espera
f1 = task_a("x")
f2 = task_b("y")
r1 = f1.result()
r2 = f2.result()

Ejercicios

Ejercicio 1: Tu primera @task (Fácil)

Crea una @task llamada translate que reciba un texto y retorne una "traducción simulada" (agrega el prefijo [EN] al texto). Úsala dentro de un @entrypoint que traduzca un saludo. No olvides .result().

from dotenv import load_dotenv
load_dotenv()

from langgraph.func import entrypoint, task

@task
def translate(text: str) -> str:
    """Simula una traducción agregando prefijo."""
    return f"[EN] {text}"

@entrypoint()
def translate_workflow(text: str) -> str:
    translated = translate(text).result()
    return translated

result = translate_workflow.invoke("Hola, ¿cómo estás?")
print(result)
# Output esperado: [EN] Hola, ¿cómo estás?

Ejercicio 2: Detectar el bug del Future (Fácil)

El siguiente código tiene un bug sutil. Identifícalo y corrígelo:

from langgraph.func import entrypoint, task

@task
def analyze(text: str) -> dict:
    word_count = len(text.split())
    return {"text": text, "word_count": word_count}

@entrypoint()
def analyze_workflow(text: str) -> str:
    analysis = analyze(text)
    return f"Palabras: {analysis['word_count']}"

from dotenv import load_dotenv
load_dotenv()

from langgraph.func import entrypoint, task

@task
def analyze(text: str) -> dict:
    word_count = len(text.split())
    return {"text": text, "word_count": word_count}

@entrypoint()
def analyze_workflow(text: str) -> str:
    analysis = analyze(text).result()  # ✅ Faltaba .result()
    return f"Palabras: {analysis['word_count']}"

result = analyze_workflow.invoke("LangGraph es un framework poderoso")
print(result)
# Output esperado: Palabras: 5

Ejercicio 3: Pipeline de 3 tasks secuenciales (Medio)

Crea un workflow con 3 tasks: extract_keywords (extrae las 3 palabras más largas de un texto), format_keywords (las pone en mayúsculas y las une con comas), y generate_report (crea un reporte con las keywords). Orquesta todo desde un @entrypoint.

from dotenv import load_dotenv
load_dotenv()

from langgraph.func import entrypoint, task

@task
def extract_keywords(text: str) -> list[str]:
    """Extrae las 3 palabras más largas como keywords."""
    words = text.split()
    sorted_words = sorted(words, key=len, reverse=True)
    return sorted_words[:3]

@task
def format_keywords(keywords: list[str]) -> str:
    """Formatea keywords en mayúsculas separadas por comas."""
    return ", ".join(kw.upper() for kw in keywords)

@task
def generate_report(original: str, formatted_kw: str) -> str:
    """Genera un reporte con el texto y sus keywords."""
    return f"Texto analizado: '{original}'\nKeywords detectadas: {formatted_kw}"

@entrypoint()
def keyword_pipeline(text: str) -> str:
    keywords = extract_keywords(text).result()
    formatted = format_keywords(keywords).result()
    report = generate_report(text, formatted).result()
    return report

result = keyword_pipeline.invoke("LangGraph permite construir aplicaciones inteligentes")
print(result)
# Output esperado:
# Texto analizado: 'LangGraph permite construir aplicaciones inteligentes'
# Keywords detectadas: APLICACIONES, INTELIGENTES, CONSTRUIR

Ejercicio 4: Búsqueda paralela en 3 fuentes (Medio)

Crea 3 tasks que simulen buscar en diferentes fuentes (Wikipedia, ArXiv, GitHub). Lanza las 3 en paralelo dentro del @entrypoint, recolecta los resultados, y combínalos en un solo string.

from dotenv import load_dotenv
load_dotenv()

from langgraph.func import entrypoint, task

@task
def search_wikipedia(query: str) -> str:
    """Simula búsqueda en Wikipedia."""
    return f"[Wikipedia] '{query}': Artículo enciclopédico con definición general y contexto histórico."

@task
def search_arxiv(query: str) -> str:
    """Simula búsqueda en ArXiv."""
    return f"[ArXiv] '{query}': Paper reciente sobre avances técnicos y benchmarks."

@task
def search_github(query: str) -> str:
    """Simula búsqueda en GitHub."""
    return f"[GitHub] '{query}': Repositorio con implementación de referencia y ejemplos."

@entrypoint()
def parallel_search(query: str) -> str:
    future_wiki = search_wikipedia(query)
    future_arxiv = search_arxiv(query)
    future_github = search_github(query)
    
    results = [
        future_wiki.result(),
        future_arxiv.result(),
        future_github.result(),
    ]
    
    return "Resultados de búsqueda:\n" + "\n".join(f"  • {r}" for r in results)

output = parallel_search.invoke("Retrieval Augmented Generation")
print(output)
# Output esperado:
# Resultados de búsqueda:
#   • [Wikipedia] 'Retrieval Augmented Generation': Artículo enciclopédico...
#   • [ArXiv] 'Retrieval Augmented Generation': Paper reciente...
#   • [GitHub] 'Retrieval Augmented Generation': Repositorio con implementación...

Ejercicio 5: @task vs función regular — refactorizar (Medio)

El siguiente código usa @task para todo, incluyendo operaciones triviales. Refactorízalo: mantén @task solo donde se justifica y convierte las demás en funciones regulares.

@task
def clean_input(text: str) -> str:
    return text.strip().lower()

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

@task
def add_timestamp(text: str) -> str:
    from datetime import datetime
    return f"[{datetime.now().isoformat()}] {text}"

from dotenv import load_dotenv
load_dotenv()

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

def clean_input(text: str) -> str:
    """Función regular — operación trivial, no falla, no costosa."""
    return text.strip().lower()

@task
def call_model(prompt: str) -> str:
    """@task — llama a API externa, costosa, puede fallar."""
    model = init_chat_model("openai:gpt-4.1-mini")
    return model.invoke(prompt).content

def add_timestamp(text: str) -> str:
    """Función regular — operación trivial, instantánea."""
    return f"[{datetime.now().isoformat()}] {text}"

@entrypoint()
def process(text: str) -> str:
    cleaned = clean_input(text)                   # Regular — sin .result()
    response = call_model(cleaned).result()       # @task — con .result()
    timestamped = add_timestamp(response)         # Regular — sin .result()
    return timestamped

result = process.invoke("  ¿Qué es LangGraph?  ")
print(result)
# Output esperado: [2026-03-08T...] LangGraph es un framework para construir
# aplicaciones con LLMs usando grafos de estado...

Ejercicio 6: Mini research agent con decompose + parallel search + synthesize (Avanzado)

Construye un workflow completo: una task descompone un tema en 2 preguntas, dos tasks investigan cada pregunta en paralelo (simulado), y una task final sintetiza los hallazgos. Usa el patrón de Futures paralelos.

from dotenv import load_dotenv
load_dotenv()

from langgraph.func import entrypoint, task

@task
def decompose(topic: str) -> list[str]:
    """Descompone un tema en 2 preguntas de investigación."""
    return [
        f"¿Cuál es la definición y origen de {topic}?",
        f"¿Cuáles son las aplicaciones prácticas de {topic}?",
    ]

@task
def investigate(question: str) -> str:
    """Investiga una pregunta (simulado)."""
    if "definición" in question:
        return f"Hallazgo: {question.split('de ')[-1].rstrip('?')} se refiere a un campo de la tecnología con raíces en los años 1950."
    return f"Hallazgo: Se aplica en medicina, finanzas, educación y transporte autónomo."

@task
def synthesize(findings: list[str]) -> str:
    """Sintetiza hallazgos en un resumen final."""
    summary = "Resumen de investigación:\n"
    for i, finding in enumerate(findings, 1):
        summary += f"  {i}. {finding}\n"
    summary += "Conclusión: Tema con amplio impacto en múltiples industrias."
    return summary

@entrypoint()
def mini_research(topic: str) -> str:
    questions = decompose(topic).result()
    
    futures = [investigate(q) for q in questions]
    findings = [f.result() for f in futures]
    
    report = synthesize(findings).result()
    return report

result = mini_research.invoke("Inteligencia Artificial")
print(result)
# Output esperado:
# Resumen de investigación:
#   1. Hallazgo: Inteligencia Artificial se refiere a un campo de la tecnología con raíces en los años 1950.
#   2. Hallazgo: Se aplica en medicina, finanzas, educación y transporte autónomo.
# Conclusión: Tema con amplio impacto en múltiples industrias.

Resumen

En esta cápsula aprendiste:

• @task define unidades independientes de trabajo dentro de un @entrypoint — cada task es checkpointable, streamable, y composable
• @task retorna un Future, no el resultado directamente — debes llamar .result() para obtener el valor real
• Un Future es como una Promise de JavaScript: el resultado está en camino pero no está listo todavía
• Los Futures habilitan: checkpointing (guardar progreso entre tasks), ejecución paralela (lanzar varias tasks y esperar después), y streaming (reportar progreso por task)
• Sin .result() obtienes un objeto Future convertido a string — el bug más silencioso de la Functional API
• Usa @task cuando la operación puede fallar (APIs), es costosa (LLMs), o puede ejecutarse en paralelo. Usa funciones regulares para operaciones triviales
• Las tasks son planas: no se pueden anidar (task dentro de task)
• Los valores de retorno deben ser serializables (strings, dicts, listas — no conexiones ni funciones)
• Para ejecución paralela: lanza las tasks sin .result(), guarda los Futures en una lista, y llama .result() al final

Próxima cápsula: Control Flow Nativo — aprenderás a usar while, if/else, for, y try/except de Python para dirigir la ejecución de tu workflow, sin edges ni conditional edges.

Recursos adicionales

1. LangGraph Functional API — Conceptual Guide — Documentación oficial de @entrypoint y @task
2. LangGraph Functional API — How-To Guide — Tutorial paso a paso con la Functional API
3. Python concurrent.futures — Future objects — Referencia de Futures en Python estándar (concepto similar)
4. JavaScript Promises — MDN — Referencia de Promises para la analogía
5. LangGraph Checkpointing — Cómo funciona el checkpointing con tasks (preview del Módulo 8)
6. LangGraph Streaming — Streaming de progreso por tarea
7. init_chat_model — LangChain — Referencia de init_chat_model usado en los ejemplos

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