Combinar Graph y Functional API

Descripción de la cápsula

No tienes que casarte con una sola API. LangGraph fue diseñado para que la Graph API y la Functional API coexistan en el mismo sistema — y esa es una de sus decisiones de diseño más potentes. Puedes usar un @task dentro de un nodo de StateGraph, llamar un grafo compilado desde un @entrypoint, o componer @entrypoints que se llaman entre sí. Cada parte de tu sistema usa la herramienta que mejor expresa su lógica.

¿Por qué querrías mezclar? Porque los sistemas reales no son uniformes. Tu flujo principal puede ser secuencial (perfecto para @entrypoint), pero uno de los pasos puede tener routing condicional complejo (perfecto para StateGraph). O tu grafo puede tener un nodo que internamente necesita ejecutar sub-operaciones checkpointables (perfecto para @task). La combinación no es un hack — es el caso de uso previsto.

En esta cápsula vas a aprender tres patrones de combinación, ver un sistema real que los usa juntos, y desarrollar el criterio para saber cuándo la combinación tiene sentido y cuándo es over-engineering. Este es el último paso antes del proyecto evolutivo — el AI Research Assistant que construirás en la siguiente cápsula usará exactamente este tipo de arquitectura híbrida a medida que evolucione.

Pattern 1: @task dentro de un StateGraph

El primer patrón es el más granular: usas @task dentro de una función que sirve como nodo de un StateGraph. El grafo maneja el routing y la estructura macro, mientras que @task te da sub-operaciones checkpointables dentro de un nodo individual.

Cuándo usar este patrón

• ✅ Tu workflow tiene routing condicional (necesita Graph API)
• ✅ Un nodo individual necesita ejecutar sub-pasos que quieres que sean checkpointables
• ✅ Quieres que las sub-operaciones dentro de un nodo puedan sobrevivir a crashes

El concepto

Normalmente, un nodo de StateGraph es una función normal. Todo lo que pasa dentro del nodo es una operación monolítica — si falla a la mitad, se repite desde el inicio. Pero si usas @task dentro del nodo, cada sub-operación queda registrada. Si el nodo falla después de completar 2 de 3 tasks, al resumir solo se re-ejecuta la tercera.

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator

from langchain.chat_models import init_chat_model
from langgraph.graph import StateGraph, START, END
from langgraph.func import entrypoint, task
from langgraph.checkpoint.memory import MemorySaver

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


@task
def extract_keywords(text: str) -> list[str]:
    """Extrae keywords de un texto usando el LLM."""
    response = model.invoke(
        f"Extrae 3-5 keywords del siguiente texto. "
        f"Retorna solo las keywords separadas por coma.\n\n{text}"
    )
    return [kw.strip() for kw in response.content.split(",")]


@task
def score_relevance(text: str, topic: str) -> float:
    """Calcula un score de relevancia de 0 a 1."""
    response = model.invoke(
        f"En una escala de 0.0 a 1.0, ¿qué tan relevante es este texto "
        f"para el tema '{topic}'? Responde SOLO con el número.\n\n{text}"
    )
    try:
        return float(response.content.strip())
    except ValueError:
        return 0.5


class AnalysisState(TypedDict):
    text: str
    topic: str
    keywords: list[str]
    relevance_score: float
    analysis: str


def analyze_node(state: AnalysisState) -> dict:
    """Nodo de StateGraph que internamente usa @tasks."""
    keywords_future = extract_keywords(state["text"])
    score_future = score_relevance(state["text"], state["topic"])

    keywords = keywords_future.result()
    score = score_future.result()

    return {
        "keywords": keywords,
        "relevance_score": score,
    }


def summarize_node(state: AnalysisState) -> dict:
    """Genera un resumen basado en el análisis."""
    kw_str = ", ".join(state["keywords"])
    response = model.invoke(
        f"Resume este texto en 2 oraciones. "
        f"Keywords identificadas: {kw_str}. "
        f"Relevancia al tema '{state['topic']}': {state['relevance_score']}\n\n"
        f"{state['text']}"
    )
    return {"analysis": response.content}


def route_by_relevance(state: AnalysisState) -> str:
    """Si la relevancia es alta, genera resumen. Si no, termina."""
    if state["relevance_score"] >= 0.5:
        return "summarize"
    return "__end__"


graph = StateGraph(AnalysisState)
graph.add_node("analyze", analyze_node)
graph.add_node("summarize", summarize_node)

graph.add_edge(START, "analyze")
graph.add_conditional_edges("analyze", route_by_relevance)
graph.add_edge("summarize", END)

app = graph.compile()

result = app.invoke({
    "text": "Python es un lenguaje de programación de alto nivel, interpretado y de propósito general. Es ampliamente usado en inteligencia artificial, data science y desarrollo web.",
    "topic": "inteligencia artificial",
})

print(f"Keywords: {result['keywords']}")
print(f"Relevancia: {result['relevance_score']}")
print(f"Análisis: {result['analysis']}")
# Output esperado:
# Keywords: ['Python', 'inteligencia artificial', 'data science', 'programación', 'desarrollo web']
# Relevancia: 0.8
# Análisis: Python es un lenguaje de propósito general muy utilizado en IA y data science...

La clave: analyze_node es un nodo normal del grafo, pero internamente lanza extract_keywords y score_relevance como @task. Si le agregas un checkpointer al grafo, esas tareas quedan registradas individualmente.

Lo que NO funciona directamente

Un detalle importante: @task está diseñado para ejecutarse dentro de un contexto de @entrypoint. Si tu StateGraph no tiene un @entrypoint como wrapper, las tareas se ejecutan pero sin el beneficio completo de checkpointing individual. Para obtener checkpointing real de las sub-tareas, necesitas el Pattern 2 o envolver el nodo en un @entrypoint.

La forma más práctica de activar esto es compilar el grafo con un checkpointer:

memory = MemorySaver()
app = graph.compile(checkpointer=memory)

result = app.invoke(
    {
        "text": "LangGraph permite construir agentes como grafos de estado.",
        "topic": "agentes de IA",
    },
    config={"configurable": {"thread_id": "analysis-1"}},
)
print(f"Keywords: {result['keywords']}")
print(f"Análisis: {result['analysis']}")
# Output esperado:
# Keywords: ['LangGraph', 'agentes', 'grafos', 'estado', 'IA']
# Análisis: LangGraph es un framework para construir agentes basados en grafos de estado...

Pattern 2: Grafo compilado dentro de @entrypoint

El segundo patrón es el inverso: tu flujo principal es un @entrypoint (secuencial, fácil de leer), pero uno de los pasos es lo suficientemente complejo como para justificar un StateGraph. Entonces llamas al grafo compilado como si fuera cualquier otra función.

Cuándo usar este patrón

• ✅ Tu flujo principal es secuencial (plan → ejecutar → analizar → reportar)
• ✅ Uno de los pasos tiene routing condicional o loops internos
• ✅ Quieres encapsular la complejidad de un paso en un grafo sin que "contamine" el flujo principal

Ejemplo: flujo secuencial con un paso complejo

Imagina un sistema que: 1) recibe una pregunta, 2) busca en múltiples fuentes (paso complejo con routing), 3) sintetiza la respuesta. El paso 2 es un grafo; los pasos 1 y 3 son funciones simples.

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator

from langchain.chat_models import init_chat_model
from langgraph.graph import StateGraph, START, END
from langgraph.func import entrypoint, task
from langgraph.checkpoint.memory import MemorySaver

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


# --- Paso complejo: grafo de búsqueda multi-fuente ---

class SearchState(TypedDict):
    query: str
    source_type: str
    results: Annotated[list[str], operator.add]


def classify_source(state: SearchState) -> dict:
    query_lower = state["query"].lower()
    if any(w in query_lower for w in ["código", "api", "función", "python"]):
        return {"source_type": "docs"}
    elif any(w in query_lower for w in ["último", "reciente", "noticia", "2026"]):
        return {"source_type": "news"}
    return {"source_type": "general"}


def search_docs(state: SearchState) -> dict:
    return {"results": [f"[DOCS] Documentación técnica sobre: {state['query']}"]}


def search_news(state: SearchState) -> dict:
    return {"results": [f"[NEWS] Últimas noticias sobre: {state['query']}"]}


def search_general(state: SearchState) -> dict:
    return {"results": [f"[WEB] Información general sobre: {state['query']}"]}


def route_source(state: SearchState) -> str:
    return {
        "docs": "search_docs",
        "news": "search_news",
        "general": "search_general",
    }[state["source_type"]]


search_graph = StateGraph(SearchState)
search_graph.add_node("classify", classify_source)
search_graph.add_node("search_docs", search_docs)
search_graph.add_node("search_news", search_news)
search_graph.add_node("search_general", search_general)

search_graph.add_edge(START, "classify")
search_graph.add_conditional_edges("classify", route_source)
search_graph.add_edge("search_docs", END)
search_graph.add_edge("search_news", END)
search_graph.add_edge("search_general", END)

compiled_search = search_graph.compile()


# --- Flujo principal: @entrypoint secuencial ---

@task
def plan_search(question: str) -> str:
    """Planifica qué buscar basado en la pregunta."""
    response = model.invoke(
        f"Dada esta pregunta, genera una query de búsqueda optimizada "
        f"(1 línea, sin explicación):\n\n{question}"
    )
    return response.content.strip()


@task
def execute_search(query: str) -> list[str]:
    """Ejecuta la búsqueda usando el grafo compilado."""
    result = compiled_search.invoke({"query": query})
    return result["results"]


@task
def synthesize(question: str, search_results: list[str]) -> str:
    """Sintetiza los resultados en una respuesta."""
    results_text = "\n".join(search_results)
    response = model.invoke(
        f"Pregunta original: {question}\n\n"
        f"Resultados de búsqueda:\n{results_text}\n\n"
        f"Genera una respuesta clara y completa."
    )
    return response.content


memory = MemorySaver()


@entrypoint(checkpointer=memory)
def research_flow(question: str) -> str:
    """Flujo principal de investigación."""
    optimized_query = plan_search(question).result()
    print(f"  Query optimizada: {optimized_query}")

    results = execute_search(optimized_query).result()
    print(f"  Resultados encontrados: {len(results)}")

    answer = synthesize(question, results).result()
    return answer


response = research_flow.invoke(
    "¿Cuáles son las últimas novedades en Python 3.13?",
    config={"configurable": {"thread_id": "research-1"}},
)
print(f"\nRespuesta:\n{response}")
# Output esperado:
#   Query optimizada: Python 3.13 nuevas características novedades
#   Resultados encontrados: 1
#
# Respuesta:
# Las últimas novedades de Python 3.13 incluyen mejoras en...

El @entrypoint se lee como Python normal: planificar → buscar → sintetizar. Pero el paso de búsqueda internamente usa un StateGraph con routing condicional. El flujo principal no sabe ni le importa que compiled_search es un grafo — lo trata como cualquier otra función.

Ventaja de este patrón

La legibilidad del flujo principal se mantiene limpia. Si lees research_flow, ves tres pasos secuenciales. No necesitas entender el grafo de búsqueda para entender el flujo general. La complejidad está encapsulada.

Pattern 3: @entrypoint llamando a otro @entrypoint

El tercer patrón es composición modular: un @entrypoint que llama a otro @entrypoint. Cada módulo de tu sistema es un workflow independiente con su propio checkpointing, y el workflow principal los orquesta.

Cuándo usar este patrón

• ✅ Tu sistema tiene módulos independientes que podrían ejecutarse solos
• ✅ Cada módulo necesita su propio checkpointing y manejo de errores
• ✅ Quieres reutilizar módulos en diferentes flujos
• ✅ Equipos diferentes trabajan en módulos diferentes

Ejemplo: pipeline de contenido modular

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from langgraph.func import entrypoint, task
from langgraph.checkpoint.memory import MemorySaver

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


# --- Módulo 1: Investigación ---

@task
def gather_info(topic: str) -> str:
    response = model.invoke(
        f"Investiga brevemente sobre: {topic}. "
        f"Da 3 puntos clave en formato lista."
    )
    return response.content


@entrypoint()
def research_module(topic: str) -> dict:
    """Módulo de investigación independiente."""
    info = gather_info(topic).result()
    return {"topic": topic, "findings": info}


# --- Módulo 2: Análisis ---

@task
def analyze_findings(findings: str) -> str:
    response = model.invoke(
        f"Analiza estos hallazgos e identifica la tendencia principal:\n\n"
        f"{findings}"
    )
    return response.content


@entrypoint()
def analysis_module(research_data: dict) -> dict:
    """Módulo de análisis independiente."""
    analysis = analyze_findings(research_data["findings"]).result()
    return {
        "topic": research_data["topic"],
        "findings": research_data["findings"],
        "analysis": analysis,
    }


# --- Módulo 3: Redacción ---

@task
def write_report(topic: str, findings: str, analysis: str) -> str:
    response = model.invoke(
        f"Escribe un mini-reporte (3 párrafos) sobre '{topic}'.\n\n"
        f"Hallazgos:\n{findings}\n\n"
        f"Análisis:\n{analysis}"
    )
    return response.content


@entrypoint()
def writing_module(analysis_data: dict) -> str:
    """Módulo de redacción independiente."""
    report = write_report(
        analysis_data["topic"],
        analysis_data["findings"],
        analysis_data["analysis"],
    ).result()
    return report


# --- Orquestador principal ---

@entrypoint()
def content_pipeline(topic: str) -> str:
    """Pipeline completo: investigar → analizar → redactar."""
    research_data = research_module.invoke(topic)
    print(f"  Investigación completada: {len(research_data['findings'])} chars")

    analysis_data = analysis_module.invoke(research_data)
    print(f"  Análisis completado: {len(analysis_data['analysis'])} chars")

    report = writing_module.invoke(analysis_data)
    print(f"  Reporte generado: {len(report)} chars")

    return report


result = content_pipeline.invoke("el impacto de la IA generativa en educación")
print(f"\n{'=' * 60}")
print(result)
# Output esperado:
#   Investigación completada: ~200 chars
#   Análisis completado: ~150 chars
#   Reporte generado: ~400 chars
#
# ============================================================
# [Mini-reporte de 3 párrafos sobre IA generativa en educación]

Cada módulo (research_module, analysis_module, writing_module) es un @entrypoint completo que podría ejecutarse de forma independiente. El content_pipeline los compone llamando a .invoke() de cada uno.

La diferencia con funciones normales

¿Por qué no usar funciones normales en vez de @entrypoint para cada módulo? Porque @entrypoint te da:

• ✅ Checkpointing individual por módulo
• ✅ Streaming del progreso de cada módulo
• ✅ Resume automático si uno de los módulos falla
• ✅ Interfaz estándar (.invoke(), .stream()) para cada módulo

Si un módulo es simple (una sola llamada al modelo), usa @task. Si es un workflow con múltiples pasos, usa @entrypoint.

Arquitectura real: sistema de investigación híbrido

Ahora veamos un ejemplo más realista que combina los tres patrones. Un sistema de investigación donde:

• Flujo principal: @entrypoint secuencial (plan → research → analyze → write)
• Paso de research: StateGraph con routing entre fuentes
• Operaciones individuales: @task para cada búsqueda (checkpointables, paralelas)

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator

from langchain.chat_models import init_chat_model
from langgraph.graph import StateGraph, START, END
from langgraph.func import entrypoint, task
from langgraph.checkpoint.memory import MemorySaver

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


# --- @tasks individuales para búsquedas ---

@task
def search_academic(query: str) -> str:
    """Busca en fuentes académicas (mock)."""
    response = model.invoke(
        f"Simula un resultado de búsqueda académica sobre: {query}. "
        f"Incluye un título de paper y un hallazgo clave."
    )
    return f"[ACADEMIC] {response.content}"


@task
def search_web(query: str) -> str:
    """Busca en la web general (mock)."""
    return f"[WEB] Resultados web para '{query}': información general disponible en múltiples fuentes."


@task
def search_news(query: str) -> str:
    """Busca en noticias recientes (mock)."""
    return f"[NEWS] Últimas noticias sobre '{query}': desarrollos recientes reportados por medios especializados."


# --- StateGraph para el paso de research con routing ---

class ResearchStepState(TypedDict):
    query: str
    depth: str
    results: Annotated[list[str], operator.add]


def assess_depth(state: ResearchStepState) -> dict:
    """Determina la profundidad de búsqueda necesaria."""
    query_lower = state["query"].lower()
    if any(w in query_lower for w in ["detallado", "profundo", "completo", "exhaustivo"]):
        return {"depth": "deep"}
    return {"depth": "standard"}


def standard_search(state: ResearchStepState) -> dict:
    """Búsqueda estándar: solo web."""
    web_result = search_web(state["query"]).result()
    return {"results": [web_result]}


def deep_search(state: ResearchStepState) -> dict:
    """Búsqueda profunda: web + académico + noticias en paralelo."""
    web_future = search_web(state["query"])
    academic_future = search_academic(state["query"])
    news_future = search_news(state["query"])

    return {
        "results": [
            web_future.result(),
            academic_future.result(),
            news_future.result(),
        ]
    }


def route_depth(state: ResearchStepState) -> str:
    return "deep_search" if state["depth"] == "deep" else "standard_search"


research_step = StateGraph(ResearchStepState)
research_step.add_node("assess", assess_depth)
research_step.add_node("standard_search", standard_search)
research_step.add_node("deep_search", deep_search)

research_step.add_edge(START, "assess")
research_step.add_conditional_edges("assess", route_depth)
research_step.add_edge("standard_search", END)
research_step.add_edge("deep_search", END)

compiled_research = research_step.compile()


# --- @entrypoint principal ---

@task
def plan_research(topic: str) -> list[str]:
    """Descompone el tema en sub-queries."""
    response = model.invoke(
        f"Descompone este tema de investigación en 2-3 sub-preguntas específicas. "
        f"Retorna solo las preguntas, una por línea.\n\nTema: {topic}"
    )
    queries = [q.strip() for q in response.content.strip().split("\n") if q.strip()]
    return queries[:3]


@task
def synthesize_report(topic: str, all_results: list[str]) -> str:
    """Sintetiza todos los resultados en un reporte."""
    results_text = "\n".join(f"- {r}" for r in all_results)
    response = model.invoke(
        f"Genera un reporte de investigación sobre '{topic}'.\n\n"
        f"Fuentes recopiladas:\n{results_text}\n\n"
        f"El reporte debe tener: resumen ejecutivo (2 oraciones), "
        f"3 hallazgos principales, y una conclusión."
    )
    return response.content


memory = MemorySaver()


@entrypoint(checkpointer=memory)
def hybrid_research(topic: str) -> str:
    """Sistema de investigación híbrido: @entrypoint + StateGraph + @tasks."""
    print(f"  Investigando: {topic}")

    queries = plan_research(topic).result()
    print(f"  Sub-queries generadas: {len(queries)}")

    all_results = []
    for query in queries:
        search_result = compiled_research.invoke({"query": query})
        all_results.extend(search_result["results"])
        print(f"    ✓ '{query[:40]}...' → {len(search_result['results'])} resultados")

    report = synthesize_report(topic, all_results).result()
    print(f"  Reporte generado: {len(report)} caracteres")

    return report


result = hybrid_research.invoke(
    "análisis detallado del impacto de LLMs en desarrollo de software",
    config={"configurable": {"thread_id": "hybrid-1"}},
)
print(f"\n{'=' * 60}")
print(result)
# Output esperado:
#   Investigando: análisis detallado del impacto de LLMs en desarrollo de software
#   Sub-queries generadas: 3
#     ✓ '¿Cómo han cambiado los LLMs las prácti...' → 3 resultados
#     ✓ '¿Qué herramientas de codificación basa...' → 3 resultados
#     ✓ '¿Cuáles son los riesgos y limitaciones...' → 3 resultados
#   Reporte generado: ~500 caracteres
#
# ============================================================
# [Reporte con resumen ejecutivo, hallazgos y conclusión]

Nota cómo la keyword "detallado" en el topic activa la búsqueda profunda (3 fuentes en paralelo) en vez de la estándar (solo web). El @entrypoint controla el flujo macro, el StateGraph maneja el routing de búsqueda, y los @task ejecutan las búsquedas individuales.

Cuándo combinar tiene sentido vs cuándo es over-engineering

Sí combina cuando:

• ✅ Tu flujo principal es secuencial pero un paso tiene routing complejo
• ✅ Diferentes partes del sistema tienen diferentes necesidades de checkpointing
• ✅ Quieres equipos trabajando en módulos independientes
• ✅ Un módulo existente (como un StateGraph ya probado) necesita integrarse en un flujo nuevo
• ✅ Necesitas mezclar ejecución paralela (@task futures) con routing condicional (StateGraph)

No combines cuando:

• ❌ Todo tu flujo es secuencial → usa solo @entrypoint + @task
• ❌ Todo tu flujo tiene routing complejo → usa solo StateGraph
• ❌ Estás combinando "por si acaso" o "para estar preparado" → YAGNI
• ❌ Cada "módulo" es una sola llamada al LLM → no necesita ser un @entrypoint propio
• ❌ Tu equipo es de una persona y el sistema tiene 3 nodos → la modularidad no te da beneficio

La regla práctica

Pregúntate: "¿este paso tiene una complejidad diferente al flujo principal?" Si la respuesta es sí, encapsúlalo con la API que mejor lo exprese. Si todo tiene la misma complejidad, usa una sola API.

Comparación: Pure Graph vs Pure Functional vs Hybrid

Progresión recomendada

1. Empieza con @entrypoint + @task (Functional API)
   → La mayoría de workflows empiezan secuenciales

2. Si un paso necesita routing condicional, extráelo a StateGraph
   → El grafo encapsula esa complejidad

3. Si necesitas módulos independientes, usa @entrypoint como wrappers
   → Cada módulo tiene su propio ciclo de vida

4. Resultado: sistema híbrido que usa cada herramienta donde brilla

Preview: el AI Research Assistant y la arquitectura híbrida

El proyecto que construirás en la siguiente cápsula (y que evolucionará hasta el Módulo 12) empieza simple con Functional API. Pero a medida que crece, naturalmente va a necesitar combinar:

El Research Assistant va a evolucionar de pure Functional (M6) a un sistema híbrido (M7+). Esto no es un accidente de diseño — es la progresión natural. Empieza simple, agrega complejidad solo cuando el problema lo requiere.

Troubleshooting

1. @task dentro de StateGraph no se checkpointea individualmente

Causa: Las @task necesitan un contexto de @entrypoint o un checkpointer activo para registrar sus resultados individualmente. Si tu StateGraph se compila sin checkpointer, las tareas se ejecutan pero como funciones normales.

Solución: Compila el grafo con un checkpointer y pasa un thread_id:

memory = MemorySaver()
app = graph.compile(checkpointer=memory)
result = app.invoke(input_data, config={"configurable": {"thread_id": "my-thread"}})

2. Error de serialización al pasar datos entre @entrypoint y StateGraph

Causa: El StateGraph espera un diccionario con claves específicas (tu TypedDict), pero le estás pasando un tipo diferente (string, lista, etc.).

Solución: Asegúrate de que lo que pasas a compiled_graph.invoke() coincide exactamente con el esquema del TypedDict:

# ❌ Error: pasando un string donde se espera un dict
result = compiled_search.invoke("mi query")

# ✅ Correcto: pasando un dict con las claves del TypedDict
result = compiled_search.invoke({"query": "mi query"})

3. El @entrypoint no puede acceder al estado interno del StateGraph

Causa: El StateGraph compilado retorna su estado final como diccionario. El @entrypoint recibe este diccionario, no el estado tipado.

Solución: Accede a los resultados del grafo por clave de diccionario:

@entrypoint()
def my_flow(topic: str) -> str:
    graph_result = compiled_graph.invoke({"query": topic})
    results = graph_result["results"]  # ← accede por clave
    return synthesize(results).result()

4. Deadlock al combinar @entrypoint con StateGraph con checkpointer compartido

Causa: Si el @entrypoint y el StateGraph compilado usan el mismo checkpointer con el mismo thread_id, pueden competir por locks.

Solución: Usa thread_ids diferentes para cada nivel, o deja que el grafo interno se compile sin checkpointer propio:

compiled_search = search_graph.compile()  # sin checkpointer propio

@entrypoint(checkpointer=memory)
def main_flow(topic: str) -> str:
    result = compiled_search.invoke({"query": topic})  # sin thread_id
    return result["results"]

5. No sé si necesito hybrid o estoy sobre-complicando

Causa: Falta de criterio claro. La tentación de usar la combinación "porque puedo" es fuerte.

Solución: Empieza con Functional API pura. Si en algún punto necesitas routing condicional que un if/else no expresa bien, extrae ese paso a un StateGraph. Si nunca llegas a ese punto, no necesitas hybrid. El dolor te guía, no la anticipación.

Ejercicios

Ejercicio 1: Identificar el patrón correcto (Básico)

Para cada escenario, decide si usarías: (A) Pure Functional, (B) Pure Graph, o (C) Hybrid. Justifica.

Escenario 1: Un pipeline que traduce un texto a 3 idiomas en paralelo y luego concatena los resultados.

Escenario 2: Un sistema de soporte que clasifica tickets (billing/tech/general), rutea a un handler especializado, y dentro del handler técnico ejecuta diagnóstico en 3 pasos checkpointables.

Escenario 3: Un chatbot que procesa mensajes secuencialmente: detectar idioma → traducir si es necesario → responder → traducir respuesta de vuelta.

Ejercicio 2: Convertir un StateGraph a hybrid (Básico)

Tienes este StateGraph puramente lineal. Conviértelo a un @entrypoint con @task, manteniendo el mismo comportamiento.

from typing import TypedDict
from langgraph.graph import StateGraph, START, END

class PipeState(TypedDict):
    text: str
    cleaned: str
    summarized: str
    formatted: str

def clean(state):
    return {"cleaned": state["text"].strip().lower()}

def summarize(state):
    return {"summarized": f"Resumen de: {state['cleaned'][:50]}"}

def format_output(state):
    return {"formatted": f"📄 {state['summarized']}"}

g = StateGraph(PipeState)
g.add_node("clean", clean)
g.add_node("summarize", summarize)
g.add_node("format", format_output)
g.add_edge(START, "clean")
g.add_edge("clean", "summarize")
g.add_edge("summarize", "format")
g.add_edge("format", END)
app = g.compile()

from langgraph.func import entrypoint, task


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


@task
def summarize(cleaned: str) -> str:
    return f"Resumen de: {cleaned[:50]}"


@task
def format_output(summarized: str) -> str:
    return f"📄 {summarized}"


@entrypoint()
def text_pipeline(text: str) -> str:
    cleaned = clean(text).result()
    summarized = summarize(cleaned).result()
    formatted = format_output(summarized).result()
    return formatted


result = text_pipeline.invoke("  Este Es Un Texto De EJEMPLO Para Procesar  ")
print(result)
# Output esperado:
# 📄 Resumen de: este es un texto de ejemplo para procesar

Ejercicio 3: Integrar un grafo existente en un @entrypoint (Intermedio)

Tienes este StateGraph de clasificación de sentimiento que ya funciona. Intégralo dentro de un @entrypoint que: 1) recibe una lista de textos, 2) clasifica cada uno con el grafo, 3) retorna un resumen de cuántos son positivos, negativos y neutros.

from typing import TypedDict
from langgraph.graph import StateGraph, START, END

class SentimentState(TypedDict):
    text: str
    sentiment: str

def classify_sentiment(state: SentimentState) -> dict:
    text_lower = state["text"].lower()
    positive_words = ["bueno", "excelente", "genial", "increíble", "perfecto", "amor"]
    negative_words = ["malo", "terrible", "horrible", "pésimo", "odio", "error"]
    pos = sum(1 for w in positive_words if w in text_lower)
    neg = sum(1 for w in negative_words if w in text_lower)
    if pos > neg:
        return {"sentiment": "positive"}
    elif neg > pos:
        return {"sentiment": "negative"}
    return {"sentiment": "neutral"}

sentiment_graph = StateGraph(SentimentState)
sentiment_graph.add_node("classify", classify_sentiment)
sentiment_graph.add_edge(START, "classify")
sentiment_graph.add_edge("classify", END)
compiled_sentiment = sentiment_graph.compile()

from langgraph.func import entrypoint, task


@task
def analyze_text(text: str) -> str:
    """Clasifica un texto usando el grafo de sentimiento."""
    result = compiled_sentiment.invoke({"text": text})
    return result["sentiment"]


@entrypoint()
def batch_sentiment(texts: list[str]) -> dict:
    """Clasifica una lista de textos y resume resultados."""
    futures = [analyze_text(text) for text in texts]
    sentiments = [f.result() for f in futures]

    summary = {
        "positive": sentiments.count("positive"),
        "negative": sentiments.count("negative"),
        "neutral": sentiments.count("neutral"),
        "total": len(sentiments),
        "details": list(zip(texts, sentiments)),
    }
    return summary


texts = [
    "Este producto es excelente, me encanta",
    "Terrible experiencia, todo salió mal",
    "El paquete llegó ayer por la tarde",
    "Increíble servicio, genial atención",
    "Horrible error en el sistema de pagos",
]

result = batch_sentiment.invoke(texts)
print(f"Total: {result['total']}")
print(f"Positivos: {result['positive']}")
print(f"Negativos: {result['negative']}")
print(f"Neutros: {result['neutral']}")
for text, sent in result["details"]:
    print(f"  [{sent:>8}] {text[:50]}")
# Output esperado:
# Total: 5
# Positivos: 2
# Negativos: 2
# Neutros: 1
#   [positive] Este producto es excelente, me encanta
#   [negative] Terrible experiencia, todo salió mal
#   [ neutral] El paquete llegó ayer por la tarde
#   [positive] Increíble servicio, genial atención
#   [negative] Horrible error en el sistema de pagos

Ejercicio 4: Diseñar una arquitectura híbrida (Intermedio)

Sin codificar completamente, diseña la arquitectura para un sistema de generación de contenido que:

1. Recibe un tema y un formato deseado (blog, tweet thread, email newsletter)
2. Investiga el tema (búsqueda en 2-3 fuentes)
3. Genera un borrador en el formato correcto
4. Valida el borrador (longitud apropiada, tono correcto)
5. Si la validación falla, regenera (máximo 2 intentos)

Define: qué partes usarían @entrypoint, cuáles @task, cuáles StateGraph, y por qué.

Arquitectura propuesta:

1. FLUJO PRINCIPAL: @entrypoint (secuencial)
   → content_generator(topic, format)
   Razón: el flujo macro es secuencial (investigar → generar → validar)

2. PASO DE INVESTIGACIÓN: @task (paralelo)
   → search_source_1(topic), search_source_2(topic), search_source_3(topic)
   Razón: búsquedas paralelas sin routing condicional. Futures pattern.

3. PASO DE GENERACIÓN: @task
   → generate_draft(topic, format, research_results)
   Razón: una sola operación. No necesita ser un módulo propio.

4. PASO DE VALIDACIÓN + RETRY: StateGraph
   → validate_node → route(pass/fail) → regenerate_node → validate_node (loop)
   Razón: tiene un loop condicional (validar → ¿ok? → si no, regenerar → validar).
   El retry con máximo 2 intentos se modela como un loop con contador en el estado.
   Conditional edge: si validation_pass=True → END, si attempts < 2 → regenerate.

   StateGraph(DraftState):
     - validate: revisa longitud y tono
     - regenerate: genera nuevo borrador con feedback
     - route: pass → END, fail+attempts<2 → regenerate, fail+attempts>=2 → END

5. COMPOSICIÓN:
   @entrypoint content_generator:
     research = [search_1(topic), search_2(topic)]  # @tasks paralelas
     draft = generate_draft(topic, format, results)  # @task
     final = compiled_validation_graph.invoke(draft)  # StateGraph con loop

Por qué hybrid:
- El flujo principal es secuencial → @entrypoint
- Las búsquedas son paralelas sin routing → @task con futures
- La validación tiene un loop condicional → StateGraph
- Pure Functional no maneja bien el loop de retry
- Pure Graph haría el flujo principal innecesariamente verbose

Ejercicio 5: Implementar nested @entrypoints con error handling (Avanzado)

Implementa dos @entrypoint modules y un @entrypoint orquestador:

• Module A (translator): Recibe un texto y un idioma destino. Retorna la traducción (mock: agrega un prefijo con el idioma).
• Module B (quality_checker): Recibe un texto y retorna {"passed": True/False, "reason": "..."}. Falla (retorna passed: False) si el texto tiene menos de 10 caracteres.
• Orquestador: Traduce el texto, verifica calidad. Si falla la calidad, intenta traducir de nuevo con un prompt mejorado. Máximo 2 intentos.

from langgraph.func import entrypoint, task


@task
def do_translation(text: str, target_lang: str, attempt: int) -> str:
    """Mock de traducción. En intento 2 genera texto más largo."""
    prefix = f"[{target_lang.upper()}]"
    if attempt > 1:
        return f"{prefix} (mejorado) Traducción completa y detallada de: {text}"
    return f"{prefix} {text}"


@entrypoint()
def translator(params: dict) -> str:
    """Módulo de traducción."""
    result = do_translation(
        params["text"],
        params["target_lang"],
        params.get("attempt", 1),
    ).result()
    return result


@task
def check_quality(text: str) -> dict:
    """Verifica calidad: mínimo 10 caracteres."""
    if len(text) < 10:
        return {"passed": False, "reason": f"Muy corto ({len(text)} chars, mínimo 10)"}
    return {"passed": True, "reason": "OK"}


@entrypoint()
def quality_checker(text: str) -> dict:
    """Módulo de verificación de calidad."""
    return check_quality(text).result()


@entrypoint()
def translate_with_quality(params: dict) -> dict:
    """Orquestador: traduce y verifica calidad, con retry."""
    text = params["text"]
    target_lang = params["target_lang"]
    max_attempts = 2

    for attempt in range(1, max_attempts + 1):
        translation = translator.invoke({
            "text": text,
            "target_lang": target_lang,
            "attempt": attempt,
        })
        print(f"  Intento {attempt}: '{translation}'")

        quality = quality_checker.invoke(translation)
        print(f"  Calidad: {quality}")

        if quality["passed"]:
            return {
                "translation": translation,
                "attempts": attempt,
                "quality": quality,
            }

        print(f"  Retry: {quality['reason']}")

    return {
        "translation": translation,
        "attempts": max_attempts,
        "quality": quality,
        "warning": "Calidad no alcanzada tras máximo de intentos",
    }


result = translate_with_quality.invoke({
    "text": "Hola",
    "target_lang": "en",
})
print(f"\nResultado: {result}")
# Output esperado:
#   Intento 1: '[EN] Hola'
#   Calidad: {'passed': False, 'reason': 'Muy corto (10 chars, mínimo 10)'}
#   Retry: Muy corto (10 chars, mínimo 10)
#   Intento 2: '[EN] (mejorado) Traducción completa y detallada de: Hola'
#   Calidad: {'passed': True, 'reason': 'OK'}
#
# Resultado: {'translation': '[EN] (mejorado) ...', 'attempts': 2, 'quality': {'passed': True, ...}}

result2 = translate_with_quality.invoke({
    "text": "Este es un texto largo que debería pasar",
    "target_lang": "fr",
})
print(f"\nResultado 2: {result2}")
# Output esperado:
#   Intento 1: '[FR] Este es un texto largo...'
#   Calidad: {'passed': True, 'reason': 'OK'}
#
# Resultado 2: {'translation': '[FR] Este es un texto...', 'attempts': 1, ...}

Ejercicio 6: Refactorizar de hybrid a pure Functional (Challenge)

Tienes este sistema hybrid. El StateGraph tiene un routing de solo 2 caminos sin loops. Refactoriza todo a pure Functional API (solo @entrypoint + @task), reemplazando el conditional edge por un if/else. Compara la legibilidad de ambas versiones.

from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.func import entrypoint, task

class ProcessState(TypedDict):
    data: str
    category: str
    result: str

def categorize(state: ProcessState) -> dict:
    if len(state["data"]) > 20:
        return {"category": "complex"}
    return {"category": "simple"}

def handle_simple(state: ProcessState) -> dict:
    return {"result": f"Simple: {state['data'].upper()}"}

def handle_complex(state: ProcessState) -> dict:
    return {"result": f"Complex: análisis profundo de '{state['data'][:20]}...'"}

def route_category(state: ProcessState) -> str:
    return "handle_complex" if state["category"] == "complex" else "handle_simple"

g = StateGraph(ProcessState)
g.add_node("categorize", categorize)
g.add_node("handle_simple", handle_simple)
g.add_node("handle_complex", handle_complex)
g.add_edge(START, "categorize")
g.add_conditional_edges("categorize", route_category)
g.add_edge("handle_simple", END)
g.add_edge("handle_complex", END)
process_graph = g.compile()

from langgraph.func import entrypoint, task


@task
def categorize(data: str) -> str:
    if len(data) > 20:
        return "complex"
    return "simple"


@task
def handle_simple(data: str) -> str:
    return f"Simple: {data.upper()}"


@task
def handle_complex(data: str) -> str:
    return f"Complex: análisis profundo de '{data[:20]}...'"


@entrypoint()
def process_data(data: str) -> str:
    category = categorize(data).result()

    if category == "complex":
        result = handle_complex(data).result()
    else:
        result = handle_simple(data).result()

    return result


print(process_data.invoke("Hola"))
# Output esperado: Simple: HOLA

print(process_data.invoke("Este es un texto suficientemente largo para ser complejo"))
# Output esperado: Complex: análisis profundo de 'Este es un texto suf...'

Resumen

En esta cápsula aprendiste:

• Pattern 1: @task dentro de StateGraph — usa el grafo para routing macro y @task para sub-operaciones checkpointables dentro de un nodo
• Pattern 2: Grafo compilado dentro de @entrypoint — flujo principal secuencial con un paso complejo encapsulado como grafo. La complejidad no contamina la legibilidad del flujo
• Pattern 3: @entrypoint llamando a @entrypoint — composición modular donde cada módulo es un workflow independiente con su propio ciclo de vida
• La combinación no es un hack — es un caso de uso previsto. LangGraph fue diseñado para que ambas APIs coexistan
• Cuándo combinar: cuando diferentes partes del sistema tienen diferentes necesidades (routing vs secuencial, checkpointing granular, equipos independientes)
• Cuándo NO combinar: cuando todo es secuencial, cuando todo es routing, o cuando anticipas necesidades futuras que no existen hoy
• El AI Research Assistant evolucionará de pure Functional (M6) a hybrid (M7+) — exactamente como se recomienda: empezar simple, agregar complejidad cuando el problema lo requiere

Próxima cápsula: Proyecto — construirás la primera versión del AI Research Assistant usando Functional API. Es el inicio de un proyecto que evolucionará durante los próximos 6 módulos.

Recursos adicionales

1. LangGraph Functional API Conceptual Guide — Documentación oficial de @entrypoint y @task
2. LangGraph Low-Level Concepts — Conceptos de StateGraph, nodos, edges y estado
3. LangGraph Checkpointing Guide — Cómo funciona el checkpointing en ambas APIs
4. LangGraph How-To: Subgraphs — Composición de grafos dentro de grafos
5. LangGraph Tutorials — Tutoriales oficiales que cubren ambas APIs
6. YAGNI Principle (Martin Fowler) — El principio de diseño que guía cuándo agregar complejidad

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