Módulo 6: Functional API

@entrypoint: Definir un Agente como Función

Descripción de la cápsula

@entrypoint es el equivalente funcional de StateGraph + compile(). Donde en la Graph API definías un StateGraph con nodos, edges, estado tipado, compilabas, y obtenías un grafo ejecutable — con @entrypoint decoras una función Python y obtienes lo mismo: un objeto ejecutable con invoke, stream, checkpointing, y soporte para interrupts.

La diferencia fundamental: con StateGraph, el flujo está en la estructura del grafo. Con @entrypoint, el flujo está en el código Python dentro de la función. No defines nodos ni edges — usas variables, condicionales, loops, y llamadas a funciones. LangGraph construye el grafo implícitamente en runtime.

En esta cápsula aprenderás a definir un workflow con @entrypoint, entender qué hace por debajo, ejecutarlo con invoke y stream, y compararlo side-by-side con la Graph API para que veas exactamente dónde están las diferencias.


Import

from langgraph.func import entrypoint

Un solo import. entrypoint viene del módulo langgraph.func, que es el hogar de la Functional API.


Anatomía básica de @entrypoint

from langgraph.func import entrypoint

@entrypoint()
def my_agent(query: str) -> str:
    return f"Respuesta a: {query}"

result = my_agent.invoke("¿Qué es LangGraph?")
print(result)
# Output esperado: Respuesta a: ¿Qué es LangGraph?

Tres cosas pasaron aquí:

  1. Decoraste la función con @entrypoint() — LangGraph envuelve tu función en un objeto Pregel (el mismo runtime que usan los grafos compilados)
  2. La función recibe un solo argumento posicionalquery: str. Si necesitas pasar múltiples datos, usa un diccionario
  3. Ejecutaste con .invoke() — la misma interfaz que un grafo compilado con graph.compile()

La función parece Python normal. Pero al decorarla con @entrypoint(), LangGraph le da:

  • ✅ Ejecución gestionada (no es una simple llamada a función)
  • ✅ Soporte para streaming con .stream()
  • ✅ Checkpointing si pasas un checkpointer
  • ✅ Soporte para interrupt() (human-in-the-loop)

Qué hace @entrypoint por debajo

Cuando decoras una función con @entrypoint(), LangGraph no la ejecuta directamente. Crea un objeto Pregel — el mismo tipo de objeto que produce graph_builder.compile() en la Graph API.

Tu código:                         Lo que LangGraph construye:
                                   
@entrypoint()                      ┌──────────────────────────┐
def my_agent(query):  ───────────▶ │  Pregel (runtime object)  │
    ...                            │  - invoke()               │
    return result                  │  - stream()               │
                                   │  - checkpointing          │
                                   │  - interrupt support       │
                                   └──────────────────────────┘

Esto significa que my_agent ya no es una función Python regular. Es un Pregel instance con métodos como invoke, stream, ainvoke, y astream. No puedes llamarla directamente como my_agent("hola") — necesitas my_agent.invoke("hola").

from langgraph.func import entrypoint

@entrypoint()
def my_agent(query: str) -> str:
    return f"Procesando: {query}"

print(type(my_agent))
# Output esperado: <class 'langgraph.pregel.Pregel'>

result = my_agent.invoke("test")
print(result)
# Output esperado: Procesando: test

El parámetro de entrada

@entrypoint requiere que tu función acepte un solo argumento posicional como input del workflow. Si necesitas pasar múltiples datos, usa un diccionario:

Un solo valor

from langgraph.func import entrypoint

@entrypoint()
def simple_agent(query: str) -> str:
    return f"Respuesta a: {query}"

result = simple_agent.invoke("¿Qué es Python?")
print(result)
# Output esperado: Respuesta a: ¿Qué es Python?

Múltiples valores con diccionario

from langgraph.func import entrypoint

@entrypoint()
def research_agent(inputs: dict) -> str:
    query = inputs["query"]
    max_results = inputs.get("max_results", 5)
    language = inputs.get("language", "es")
    return f"Buscando '{query}' (max: {max_results}, idioma: {language})"

result = research_agent.invoke({
    "query": "prompt engineering",
    "max_results": 10,
    "language": "es"
})
print(result)
# Output esperado: Buscando 'prompt engineering' (max: 10, idioma: es)

Los inputs y outputs deben ser JSON-serializable (dict, list, str, int, float, bool, None). No puedes pasar objetos Python custom como Pydantic models o clases propias como input directo.


Retorno: lo que produce el workflow

El valor que retorna tu función @entrypoint es el output del workflow. Es lo que obtienes al llamar invoke:

from langgraph.func import entrypoint

@entrypoint()
def agent_with_dict_output(query: str) -> dict:
    return {
        "answer": f"Respuesta a: {query}",
        "confidence": 0.95,
        "sources": ["fuente_1", "fuente_2"]
    }

result = agent_with_dict_output.invoke("¿Qué es RAG?")
print(result["answer"])
# Output esperado: Respuesta a: ¿Qué es RAG?
print(result["confidence"])
# Output esperado: 0.95
print(result["sources"])
# Output esperado: ['fuente_1', 'fuente_2']

El retorno también debe ser JSON-serializable. Si necesitas retornar datos structured complejos, usa diccionarios anidados.


Ejecutar un @entrypoint

invoke — ejecución completa

from langgraph.func import entrypoint

@entrypoint()
def my_agent(query: str) -> str:
    return f"Resultado para: {query}"

result = my_agent.invoke("¿Cómo funciona LangGraph?")
print(result)
# Output esperado: Resultado para: ¿Cómo funciona LangGraph?

invoke ejecuta el workflow completo y retorna el resultado final. Igual que graph.invoke() en la Graph API.

stream — ejecución con streaming

from langgraph.func import entrypoint, task

@task
def step_one(query: str) -> str:
    return f"Paso 1 completado para: {query}"

@task
def step_two(intermediate: str) -> str:
    return f"Paso 2: procesando '{intermediate}'"

@entrypoint()
def my_agent(query: str) -> str:
    result_1 = step_one(query).result()
    result_2 = step_two(result_1).result()
    return result_2

for chunk in my_agent.stream("mi pregunta"):
    print(chunk)
    print("---")
# Output esperado:
# {'step_one': 'Paso 1 completado para: mi pregunta'}
# ---
# {'step_two': "Paso 2: procesando 'Paso 1 completado para: mi pregunta'"}
# ---
# {'my_agent': "Paso 2: procesando 'Paso 1 completado para: mi pregunta'"}
# ---

stream emite un chunk por cada @task completado, más un chunk final con el resultado del @entrypoint. Cada chunk es un diccionario donde la key es el nombre de la task/entrypoint.

Esto es útil para dar feedback progresivo al usuario: "Paso 1 completado... Paso 2 en progreso... Resultado final."


Ejecución durable: preview de checkpointing

Cuando pasas un checkpointer al @entrypoint, LangGraph guarda el resultado de cada @task completado. Si el workflow crashea a mitad de ejecución, puedes retomarlo sin re-ejecutar las tasks que ya terminaron:

from langgraph.func import entrypoint, task
from langgraph.checkpoint.memory import InMemorySaver

@task
def expensive_api_call(query: str) -> str:
    """Simula una llamada costosa a una API externa."""
    return f"Resultado de API para: {query}"

@task
def process_result(raw_result: str) -> str:
    """Procesa el resultado de la API."""
    return f"Procesado: {raw_result}"

@entrypoint(checkpointer=InMemorySaver())
def durable_agent(query: str) -> str:
    raw = expensive_api_call(query).result()
    processed = process_result(raw).result()
    return processed

config = {"configurable": {"thread_id": "session-001"}}
result = durable_agent.invoke("LangGraph durability", config)
print(result)
# Output esperado: Procesado: Resultado de API para: LangGraph durability

El thread_id identifica la sesión. Si la ejecución se interrumpe después de expensive_api_call pero antes de process_result, al reinvocar con el mismo thread_id, LangGraph recupera el resultado guardado de expensive_api_call y continúa desde donde se quedó.

No re-ejecuta work que ya completó. Esto es durable execution — y lo profundizarás en el Módulo 8.


Parámetros inyectables

@entrypoint soporta parámetros que LangGraph inyecta automáticamente en runtime. Los declaras como keyword-only arguments (después de *):

from typing import Any
from langchain_core.runnables import RunnableConfig
from langgraph.func import entrypoint
from langgraph.checkpoint.memory import InMemorySaver

@entrypoint(checkpointer=InMemorySaver())
def my_agent(
    query: str,
    *,
    previous: Any = None,
    config: RunnableConfig
) -> str:
    history = previous or "Sin historial"
    thread = config["configurable"]["thread_id"]
    return f"[{thread}] {history}{query}"

config = {"configurable": {"thread_id": "demo-001"}}

result_1 = my_agent.invoke("primera pregunta", config)
print(result_1)
# Output esperado: [demo-001] Sin historial → primera pregunta

result_2 = my_agent.invoke("segunda pregunta", config)
print(result_2)
# Output esperado: [demo-001] [demo-001] Sin historial → primera pregunta → segunda pregunta
ParámetroTipoQué recibe
previousAnyEl valor retornado por la invocación anterior en el mismo thread
configRunnableConfigLa configuración de runtime (incluye thread_id)
storeBaseStoreAcceso a long-term memory (se cubre en Módulo 8)
writerStreamWriterPara streaming custom en async Python < 3.11

previous te da acceso al estado de la invocación anterior sin definir TypedDict ni reducers — es la forma en que la Functional API maneja memoria a corto plazo.


Comparación side-by-side: @entrypoint vs StateGraph

El mismo problema — un chatbot simple — resuelto con ambas APIs:

Graph API (StateGraph)

from typing import TypedDict, Annotated
import operator
from dotenv import load_dotenv
load_dotenv()

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

class ChatState(TypedDict):
    messages: Annotated[list[str], operator.add]

def respond(state: ChatState) -> dict:
    model = init_chat_model("openai:gpt-4.1-mini")
    last_message = state["messages"][-1]
    response = model.invoke(last_message)
    return {"messages": [response.content]}

graph_builder = StateGraph(ChatState)
graph_builder.add_node("respond", respond)
graph_builder.add_edge(START, "respond")
graph_builder.add_edge("respond", END)

graph = graph_builder.compile()
result = graph.invoke({"messages": ["¿Qué es Python?"]})
print(result["messages"][-1])
# Output esperado: Python es un lenguaje de programación...

Líneas de código de setup: ~15 (TypedDict, StateGraph, add_node, add_edge × 2, compile)

Functional API (@entrypoint)

from dotenv import load_dotenv
load_dotenv()

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

@entrypoint()
def chat_agent(message: str) -> str:
    model = init_chat_model("openai:gpt-4.1-mini")
    response = model.invoke(message)
    return response.content

result = chat_agent.invoke("¿Qué es Python?")
print(result)
# Output esperado: Python es un lenguaje de programación...

Líneas de código de setup: ~5 (import, decorador, función, invoke)

Análisis de la comparación

AspectoGraph APIFunctional API
SetupTypedDict + StateGraph + edges + compileUn decorador
EstadoTypedDict con reducers explícitosSin estado explícito (variables locales)
FlujoDefinido por edgesDefinido por el código de la función
Verbosidad~15 líneas de setup~5 líneas de setup
Visualizacióndraw_mermaid_png() muestra el grafoNo disponible
EscalabilidadMejor con múltiples nodos y routingMejor para flujos lineales

Para un chatbot simple de un solo paso, la Functional API es claramente más concisa. Pero a medida que el workflow crece (múltiples nodos, conditional routing, visualización), la Graph API empieza a justificarse.

No hay respuesta universal. Hay contexto.


Cuándo @entrypoint solo es suficiente

Para workflows que son una sola función con lógica interna, @entrypoint sin @task funciona perfectamente:

from dotenv import load_dotenv
load_dotenv()

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

@entrypoint()
def classifier_agent(message: str) -> dict:
    model = init_chat_model("openai:gpt-4.1-mini")
    classification = model.invoke(
        f"Clasifica este mensaje: técnico, creativo, o informativo. "
        f"Responde SOLO la categoría.\n\nMensaje: {message}"
    )
    category = classification.content.strip().lower()
    return {"category": category, "original_message": message}

result = classifier_agent.invoke("¿Cómo implemento un API REST en Python?")
print(f"Categoría: {result['category']}")
# Output esperado: Categoría: técnico

Funciona sin @task porque toda la lógica está en una sola función. No hay pasos independientes que necesiten checkpointing individual o paralelización.


Cuándo necesitas @task

@entrypoint solo no es suficiente cuando tu workflow tiene múltiples pasos que se benefician de:

  • Checkpointing individual: cada paso se guarda, permitiendo retomar desde el último exitoso
  • Streaming por paso: el usuario ve progreso a medida que cada task completa
  • Paralelización: múltiples tasks ejecutándose en paralelo
  • Retry por paso: si un paso falla, solo reintenta ese paso, no todo el workflow
from langgraph.func import entrypoint, task

@task
def fetch_data(source: str) -> dict:
    return {"source": source, "data": f"Datos de {source}"}

@task
def analyze_data(data: dict) -> str:
    return f"Análisis de {data['source']}: datos procesados"

@task
def generate_report(analyses: list[str]) -> str:
    return f"Reporte final con {len(analyses)} análisis"

@entrypoint()
def research_pipeline(query: str) -> str:
    sources = ["web", "papers", "docs"]
    data_results = [fetch_data(s).result() for s in sources]
    analyses = [analyze_data(d).result() for d in data_results]
    report = generate_report(analyses).result()
    return report

for chunk in research_pipeline.stream("mi investigación"):
    print(chunk)
# Output esperado: un chunk por cada @task ejecutado (3 fetch + 3 analyze + 1 report + 1 entrypoint = 8 chunks)

Cada @task emite su propio chunk en el stream. Si fetch_data("papers") falla y tienes checkpointing, fetch_data("web") no se re-ejecuta al retomar. Esa granularidad no es posible con @entrypoint solo.

La siguiente cápsula profundiza en @task, Futures, y .result().


Troubleshooting

Problema 1: "TypeError: 'Pregel' object is not callable"

Síntoma: Intentas llamar la función directamente: my_agent("hola"). Causa: @entrypoint transforma la función en un objeto Pregel. Ya no es una función Python regular. Solución:

# Incorrecto — la función ya no es callable directamente
result = my_agent("hola")

# Correcto — usa invoke
result = my_agent.invoke("hola")

Problema 2: "Tasks can only be called from within an entrypoint"

Síntoma: Error al llamar un @task fuera de un @entrypoint. Causa: Los @task requieren el contexto de ejecución de un @entrypoint para funcionar (checkpointing, streaming). Solución:

from langgraph.func import entrypoint, task

@task
def my_task(x: int) -> int:
    return x * 2

# Incorrecto — @task fuera de @entrypoint
# result = my_task(5).result()  # Error

# Correcto — @task dentro de @entrypoint
@entrypoint()
def my_workflow(x: int) -> int:
    return my_task(x).result()

print(my_workflow.invoke(5))
# Output esperado: 10

Problema 3: "SerializationError — inputs must be JSON-serializable"

Síntoma: Error al pasar objetos Python custom como input a invoke. Causa: @entrypoint requiere que inputs y outputs sean JSON-serializable para checkpointing. Solución:

# Incorrecto — objeto custom como input
class Query:
    def __init__(self, text, lang):
        self.text = text
        self.lang = lang

# result = my_agent.invoke(Query("hola", "es"))  # Error

# Correcto — diccionario JSON-serializable
result = my_agent.invoke({"text": "hola", "lang": "es"})

Problema 4: "El streaming no muestra pasos intermedios"

Síntoma: stream() solo muestra el resultado final, no pasos intermedios. Causa: No estás usando @task para los pasos intermedios. Sin @task, todo el código corre dentro del @entrypoint como un solo bloque. Solución: Envuelve cada paso que quieras ver en el stream con @task. Solo los @task emiten chunks individuales en stream().

Problema 5: "El @entrypoint no recuerda conversaciones anteriores"

Síntoma: Cada invocación empieza desde cero, sin memoria de invocaciones previas. Causa: No estás usando un checkpointer ni el parámetro previous. Solución: Pasa checkpointer=InMemorySaver() al decorador, declara previous: Any = None como keyword-only argument, y usa el mismo thread_id en la config entre invocaciones.


Ejercicios

Ejercicio 1: Tu primer @entrypoint (Fácil)

Crea un @entrypoint que reciba un nombre (string) y retorne un saludo personalizado con el formato "Bienvenido al Research Lab, {nombre}. Tu sesión ha iniciado." Ejecútalo con invoke y verifica el resultado.

Ver solución
from langgraph.func import entrypoint

@entrypoint()
def welcome_agent(name: str) -> str:
    return f"Bienvenido al Research Lab, {name}. Tu sesión ha iniciado."

result = welcome_agent.invoke("Ana")
print(result)
# Output esperado: Bienvenido al Research Lab, Ana. Tu sesión ha iniciado.

print(type(welcome_agent))
# Output esperado: <class 'langgraph.pregel.Pregel'>

Explicación: @entrypoint() transforma la función en un objeto Pregel. En vez de llamarla directamente, usas .invoke() con el input como argumento.

Ejercicio 2: @entrypoint con input de diccionario (Fácil)

Crea un @entrypoint que reciba un diccionario con las keys topic, depth (str: "shallow" o "deep"), y language (str). La función debe retornar un diccionario con query (formateado como "Investigar {topic} a nivel {depth}") y config (con language y timestamp simulado).

Ver solución
from langgraph.func import entrypoint

@entrypoint()
def configure_research(inputs: dict) -> dict:
    topic = inputs["topic"]
    depth = inputs.get("depth", "shallow")
    language = inputs.get("language", "es")

    return {
        "query": f"Investigar {topic} a nivel {depth}",
        "config": {
            "language": language,
            "timestamp": "2026-03-08T10:00:00Z"
        }
    }

result = configure_research.invoke({
    "topic": "prompt engineering",
    "depth": "deep",
    "language": "es"
})
print(result["query"])
# Output esperado: Investigar prompt engineering a nivel deep
print(result["config"])
# Output esperado: {'language': 'es', 'timestamp': '2026-03-08T10:00:00Z'}

Explicación: Cuando necesitas múltiples inputs, usa un diccionario como argumento. @entrypoint solo acepta un argumento posicional.

Ejercicio 3: Streaming con @task (Medio)

Crea un workflow con @entrypoint y tres @task en secuencia: parse_query (extrae keywords del query), search_sources (simula búsqueda), y summarize (genera resumen). Usa stream para observar la ejecución paso a paso y cuenta cuántos chunks recibe el stream.

Ver solución
from langgraph.func import entrypoint, task

@task
def parse_query(query: str) -> list[str]:
    words = query.lower().split()
    keywords = [w for w in words if len(w) > 3]
    return keywords

@task
def search_sources(keywords: list[str]) -> list[dict]:
    results = []
    for kw in keywords:
        results.append({"keyword": kw, "source": f"https://example.com/{kw}"})
    return results

@task
def summarize(results: list[dict]) -> str:
    sources_text = ", ".join(r["keyword"] for r in results)
    return f"Resumen basado en {len(results)} fuentes: {sources_text}"

@entrypoint()
def research_workflow(query: str) -> str:
    keywords = parse_query(query).result()
    results = search_sources(keywords).result()
    summary = summarize(results).result()
    return summary

chunk_count = 0
for chunk in research_workflow.stream("¿Cómo funciona prompt engineering en producción?"):
    chunk_count += 1
    for key, value in chunk.items():
        print(f"Chunk #{chunk_count} [{key}]: {value}")
    print("---")

print(f"\nTotal chunks: {chunk_count}")
# Output esperado:
# Chunk #1 [parse_query]: ['cómo', 'funciona', 'prompt', 'engineering', 'producción?']
# ---
# Chunk #2 [search_sources]: [{'keyword': 'cómo', 'source': '...'}, ...]
# ---
# Chunk #3 [summarize]: 'Resumen basado en 5 fuentes: ...'
# ---
# Chunk #4 [research_workflow]: 'Resumen basado en 5 fuentes: ...'
# ---
# Total chunks: 4

Explicación: El stream emite un chunk por cada @task completado (3 tasks = 3 chunks) más un chunk final con el resultado del @entrypoint (total = 4 chunks).

Ejercicio 4: Side-by-side Graph API vs Functional API (Medio)

Implementa un pipeline ETL (Extract, Transform, Load) de dos formas: con StateGraph (Graph API) y con @entrypoint + @task (Functional API). El pipeline recibe un número, lo multiplica por 2 (extract), le suma 10 (transform), y lo formatea como string (load). Compara la cantidad de líneas de cada implementación.

Ver solución
# ====== GRAPH API ======
from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END

class ETLState(TypedDict):
    messages: Annotated[list[str], operator.add]
    value: int

def extract(state: ETLState) -> dict:
    new_val = state["value"] * 2
    return {"messages": [f"Extract: {state['value']}{new_val}"], "value": new_val}

def transform(state: ETLState) -> dict:
    new_val = state["value"] + 10
    return {"messages": [f"Transform: {state['value']}{new_val}"], "value": new_val}

def load(state: ETLState) -> dict:
    return {"messages": [f"Load: resultado = {state['value']}"]}

graph_builder = StateGraph(ETLState)
graph_builder.add_node("extract", extract)
graph_builder.add_node("transform", transform)
graph_builder.add_node("load", load)
graph_builder.add_edge(START, "extract")
graph_builder.add_edge("extract", "transform")
graph_builder.add_edge("transform", "load")
graph_builder.add_edge("load", END)

graph = graph_builder.compile()
result_graph = graph.invoke({"messages": [], "value": 5})
print(f"Graph API: {result_graph['messages']}")
# Graph API: ['Extract: 5 → 10', 'Transform: 10 → 20', 'Load: resultado = 20']

# ====== FUNCTIONAL API ======
from langgraph.func import entrypoint, task

@task
def extract_fn(value: int) -> int:
    return value * 2

@task
def transform_fn(value: int) -> int:
    return value + 10

@task
def load_fn(value: int) -> str:
    return f"Resultado = {value}"

@entrypoint()
def etl_pipeline(value: int) -> str:
    extracted = extract_fn(value).result()
    transformed = transform_fn(extracted).result()
    loaded = load_fn(transformed).result()
    return loaded

result_func = etl_pipeline.invoke(5)
print(f"Functional API: {result_func}")
# Functional API: Resultado = 20

# Comparación
print(f"\nGraph API: ~25 líneas de código")
print(f"Functional API: ~18 líneas de código")
print(f"Mismo resultado, diferente expresividad")

Explicación: Ambas producen el mismo resultado (5 → 10 → 20). La Graph API requiere definir TypedDict, StateGraph, add_node × 3, add_edge × 4, y compile. La Functional API define las tasks como funciones y las llama secuencialmente en el entrypoint.

Ejercicio 5: Short-term memory con previous (Challenge)

Crea un @entrypoint con InMemorySaver que mantenga un historial de conversación usando el parámetro previous. El workflow recibe un mensaje del usuario, lo agrega al historial (que viene de previous), y retorna el historial completo como lista de strings. Verifica que después de 3 invocaciones en el mismo thread, el historial tiene los 3 mensajes.

Ver solución
from typing import Any
from langgraph.func import entrypoint
from langgraph.checkpoint.memory import InMemorySaver

@entrypoint(checkpointer=InMemorySaver())
def conversation_agent(message: str, *, previous: Any = None) -> list[str]:
    history = previous if previous is not None else []
    history.append(message)
    return history

config = {"configurable": {"thread_id": "convo-001"}}

result_1 = conversation_agent.invoke("Hola, ¿cómo estás?", config)
print(f"Turno 1: {result_1}")
# Output esperado: Turno 1: ['Hola, ¿cómo estás?']

result_2 = conversation_agent.invoke("Quiero investigar sobre RAG", config)
print(f"Turno 2: {result_2}")
# Output esperado: Turno 2: ['Hola, ¿cómo estás?', 'Quiero investigar sobre RAG']

result_3 = conversation_agent.invoke("Específicamente sobre chunking strategies", config)
print(f"Turno 3: {result_3}")
# Output esperado: Turno 3: ['Hola, ¿cómo estás?', 'Quiero investigar sobre RAG', 'Específicamente sobre chunking strategies']

assert len(result_3) == 3, f"Esperaba 3 mensajes, obtuve {len(result_3)}"
print("\n✅ Historial mantiene los 3 mensajes correctamente")

different_thread = {"configurable": {"thread_id": "convo-002"}}
result_new = conversation_agent.invoke("Nuevo thread", different_thread)
print(f"\nNuevo thread: {result_new}")
# Output esperado: Nuevo thread: ['Nuevo thread']

assert len(result_new) == 1, "Thread diferente debe empezar vacío"
print("✅ Threads diferentes tienen historiales independientes")

Explicación: previous recibe el retorno de la invocación anterior en el mismo thread_id. Cada invocación agrega el nuevo mensaje al historial y lo retorna. Threads diferentes mantienen historiales independientes porque el checkpointing es por thread_id.


Resumen

En esta cápsula aprendiste:

  • @entrypoint es el equivalente funcional de StateGraph + compile() — transforma una función Python en un objeto Pregel ejecutable con invoke y stream
  • La función decorada recibe un solo argumento posicional (usa diccionario para múltiples datos). Inputs y outputs deben ser JSON-serializable
  • @entrypoint no ejecuta la función directamente — crea un runtime object. Llamas my_agent.invoke(), no my_agent()
  • Streaming funciona a nivel de @task: cada task completada emite un chunk. Sin @task, el stream solo muestra el resultado final
  • Checkpointing se habilita con checkpointer=InMemorySaver(). Los resultados de cada @task se guardan, permitiendo retomar desde el último exitoso
  • Parámetros inyectables (previous, config, store, writer) se declaran como keyword-only arguments. previous da acceso al retorno de la invocación anterior en el mismo thread
  • Comparado con StateGraph, @entrypoint requiere menos código de setup pero no soporta visualización. Para flujos lineales o secuenciales, la Functional API es más concisa. Para topologías complejas con branching visual, la Graph API es más clara

Próxima cápsula: @task: Tareas que Componen el Agente — aprenderás cómo @task define unidades de trabajo independientes, qué son los Futures (como Promises en JavaScript), y cuándo usar @task vs código directo dentro del @entrypoint.


Recursos adicionales

  1. Functional API Overview — Documentación oficial con explicación completa de @entrypoint y @task
  2. @entrypoint Reference — Referencia completa del decorador con todos los parámetros
  3. How to use the Functional API — Guía práctica con ejemplos de streaming, memory, y human-in-the-loop
  4. Choosing between Graph API and Functional API — Criterios oficiales para elegir entre ambas APIs
  5. Introducing the LangGraph Functional API — Blog post de lanzamiento con motivación de diseño
  6. Persistence in LangGraph — Cómo funciona el checkpointing que habilita @entrypoint
  7. InMemorySaver Reference — Referencia del checkpointer in-memory usado en los ejemplos
  8. Pregel — LangGraph Reference — Referencia del objeto runtime que @entrypoint produce

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