Checkpoints de un agente LangGraph en una instancia WEC para que las caídas no te cuesten nada

++

0/4

🎯 Skill path0/4 earned

Agent orchestration with LangGraph

• 1State that survives a restart
• 2Hand work between agents
• 3Governed tools an agent can call
• 🏆Engineer the context, not the prompt

Finish this tutorial, then mark it complete below to earn the skill.

Tu agente lleva cuarenta segundos trabajando en la petición de un cliente. Ha leído el ticket, ha consultado la cuenta y ha revisado los calendarios de tres ingenieros. Está a punto de reservar la cita.

Entonces el proceso muere. Un despliegue que sale, el host que se queda sin memoria, alguien que reinicia el contenedor — da igual cuál de ellos.

El agente no continúa donde lo dejó, porque no hay nada desde donde continuar. Todo lo que aprendió vivía en variables dentro de un proceso que ya no existe. El cliente sigue esperando. Ejecútalo de nuevo y pagas todo ese trabajo por segunda vez. Y la parte que más debería preocuparte: nadie puede decir si la cita se reservó en el último segundo antes de morir.

Un agente es un modelo dentro de un bucle. Como script de Python normal, ese bucle es exactamente igual de frágil que el proceso que lo contiene.

Aquí construimos uno que guarda su estado en Postgres después de cada paso, para que una caída no cueste nada.

Reproducibilidad

Versiones verificadas: Python 3.10.12 · langgraph 1.2.11 · langgraph-checkpoint 4.2.0 · langgraph-checkpoint-postgres 3.1.2 · psycopg 3.3.4 · psycopg-pool 3.3.1 · langchain-openai 1.6.0 · langfuse 4.14.5 contra el servidor Langfuse v3.205.1 OSS · imagen postgres:17.

El problema, antes de la solución

Lo que quieres es una ejecución reanudable: el resultado de cada paso escrito en algún sitio duradero en el momento en que termina, para que un proceso nuevo pueda continuar desde ahí.

Eso es lo que hace un checkpointer.

Por qué un grafo y no un bucle

Un bucle con un try/except y unas cuantas escrituras a base de datos podría hacer esto para cuatro pasos fijos. Deja de funcionar en el momento en que el agente elige su propio camino: cuando el modelo escoge la siguiente herramienta, "¿dónde estamos?" no tiene una respuesta que puedas escribir.

LangGraph te obliga a declarar el trabajo como un grafo — nodos con nombre, aristas explícitas. Eso parece ceremonia hasta que quieres durabilidad.

Un bucle	Un grafo
Los pasos están implícitos en el flujo de control	Los pasos son nodos con nombre
"¿Dónde estoy?" no tiene respuesta	La posición es un valor que puedes leer
El estado vive en variables locales	El estado es un objeto declarado
No hay nada que guardar	Hay un momento obvio para guardar: los límites de nodo

El runtime sabe cuándo empieza y termina un nodo, así que tiene un momento obvio para guardar. El estado es un objeto declarado, así que hay algo concreto que escribir.

Eso es lo que significa "una máquina de estados determinista alrededor de un modelo no determinista": la salida es impredecible, el siguiente paso no. Más sobre ese debate en "Loop Engineering Is Dead".

Lo que vas a construir

Tres propiedades que los ejemplos habituales no tienen:

• estado en Postgres, no en el proceso
• una puerta de aprobación humana en la que el grafo puede quedarse días
• cada nodo trazado como un span, con conteo de tokens y latencia por nodo

Requisitos previos

• Una instancia WEC — despliega una si aún no lo has hecho
• Una clave de la API de Inferencia de WEC — créala en el portal
• Una instancia de Langfuse — mira Observa producción con Langfuse

Paso 1 — Instalar, y por qué tres paquetes

python3 -m venv .venv
source .venv/bin/activate
pip install "langgraph" "langgraph-checkpoint-postgres" "psycopg[binary,pool]"

Paquete	Qué hace	Por qué lo necesitas aquí
langgraph	El runtime del grafo — nodos, aristas, estado	La orquestación en sí
langgraph-checkpoint-postgres	Escribe el estado en Postgres	El objetivo de este tutorial
psycopg[binary,pool]	Driver de Postgres + pool de conexiones	PostgresSaver requiere un pool

Vale la pena detenerse en la fila del medio. La persistencia por defecto de LangGraph es en memoria. La durabilidad es opcional, en un paquete aparte que tienes que saber que hay que instalar. Esa es precisamente la razón por la que tantos ejemplos publicados pierden el estado en silencio: nunca instalaron esto, y nada les avisó.

Tres nombres arrastran más de tres paquetes:

./.venv/bin/pip list | grep -iE "langgraph|psycopg" && python3 --version

Output

langgraph                     1.2.11
langgraph-checkpoint          4.2.0
langgraph-checkpoint-postgres 3.1.2
langgraph-prebuilt            1.1.0
langgraph-sdk                 0.4.3
psycopg                       3.3.4
psycopg-binary                3.3.4
psycopg-pool                  3.3.1
Python 3.10.12

langgraph-checkpoint es el que hay que notar: es la interfaz abstracta, y -postgres es una implementación de ella. Esa separación es la razón por la que cambiar Postgres por SQLite o Redis más adelante cambia una línea.

El extra pool tampoco es decoración. Instala psycopg a secas y el fallo llega al conectar, no al instalar — que es un sitio mucho más confuso donde encontrarlo.

Figura 1. Ocho paquetes desde tres nombres — la interfaz de checkpoint y su implementación en Postgres llegan por separado.

Paso 2 — Una base de datos para checkpoints, no para la aplicación

docker run -d --name lg-checkpoints \
  -e POSTGRES_USER=langgraph -e POSTGRES_PASSWORD=changeme -e POSTGRES_DB=checkpoints \
  -p 127.0.0.1:5434:5432 postgres:17

Tres decisiones deliberadas:

Una base de datos dedicada. Los checkpoints son intensivos en escritura — una fila por transición de nodo, por hilo, para siempre. Mezclados en la base de datos de tu aplicación se convierten en una tabla que te da miedo truncar. Separados, son desechables.

127.0.0.1:5434:5432, no -p 5434:5432. La forma corta escucha en todas las interfaces, y Docker escribe sus propias reglas de iptables que se saltan UFW — así que un firewall que parece correcto no está protegiendo este puerto. Lo cubrimos en Endurecer redes Docker.

El número de la izquierda es el puerto del host, el de la derecha el del contenedor. En una instancia vacía usa 5432:5432. Si ya autoalojas algo con base de datos, ese puerto probablemente esté ocupado — elige cualquiera libre.

postgres:17 fijado. latest significa que alguien que lea esto en seis meses ejecutará algo que nunca probaste.

Estar en marcha no es lo mismo que estar listo:

docker exec lg-checkpoints pg_isready -U langgraph -d checkpoints

Output

/var/run/postgresql:5432 - accepting connections

Un contenedor informa Up en el instante en que arranca, mientras Postgres dentro todavía se está inicializando. Conectar demasiado pronto da un error de conexión que parece un problema de configuración y no lo es.

Paso 3 — El grafo, deliberadamente sin modelo

Ningún LLM en este paso. Si algo falla ahora, es culpa del checkpointer y no del modelo — y separar esas dos cosas es la mayor parte de lo que consiste depurar un agente.

El objeto de estado

from operator import add
from typing import Annotated
from typing_extensions import TypedDict

class State(TypedDict):
    completed: Annotated[list[str], add]

Esto declara qué recuerda el grafo. TypedDict da la forma; la parte Annotated[..., add] es la mitad interesante.

Por defecto, cuando un nodo devuelve un valor para una clave, ese valor reemplaza lo que había. Annotated[list[str], add] adjunta un reducer — una función que combina el valor viejo con el nuevo en vez de sobrescribirlo. Con operator.add sobre una lista, eso significa añadir al final.

Declaración	step_one devuelve ["a"], luego step_two devuelve ["b"]
completed: list[str]	["b"] — el primer resultado se pierde
completed: Annotated[list[str], add]	["a", "b"] — acumula

Equivócate en esto y tu estado olvida en silencio todo excepto el último nodo.

Los nodos

def step_one(state: State):
    print("step_one running", flush=True); time.sleep(2)
    return {"completed": ["step_one"]}

def step_two(state: State):
    print("step_two running — 30s window, KILL ME HERE", flush=True); time.sleep(30)
    return {"completed": ["step_two"]}

Un nodo es una función normal. Recibe el estado actual y devuelve una actualización parcial — solo las claves que cambió, no el objeto entero. El runtime fusiona esa actualización usando los reducers que declaraste.

El sleep de treinta segundos en step_two representa un paso lento — una llamada a un modelo, una API externa, un trabajo por lotes. Es también lo que te da tiempo para matar el proceso en el momento justo en el paso siguiente.

Conectar el grafo

builder = StateGraph(State)
for name, fn in [("step_one", step_one), ("step_two", step_two), ("step_three", step_three)]:
    builder.add_node(name, fn)
builder.add_edge(START, "step_one")
builder.add_edge("step_one", "step_two")
builder.add_edge("step_two", "step_three")
builder.add_edge("step_three", END)

StateGraph(State) vincula el esquema. add_node registra una función bajo un nombre — ese nombre es el que aparece en los checkpoints y en los rastros más adelante. add_edge declara el orden, con START y END como los centinelas que marcan la entrada y la salida.

Este grafo es una línea recta. La ramificación llega en la parte 2.

Adjuntar el checkpointer

with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
    checkpointer.setup()
    graph = builder.compile(checkpointer=checkpointer)
    config = {"configurable": {"thread_id": THREAD}}
    result = graph.invoke({"completed": []}, config)

Cuatro líneas, cada una con su trampa.

with ... as checkpointer. from_conn_string está decorado con @contextmanager en el código fuente — cede un valor, no lo devuelve. Asignarlo directamente te entrega un objeto gestor de contexto donde esperabas un saver, y el error llega mucho más tarde que el fallo.

También abre la conexión con autocommit=True, prepare_threshold=0 y row_factory=dict_row. Si pasas tu propia conexión tienes que configurar eso tú, o .setup() puede parecer que funcionó y no guardar nada. Por qué hace falta cada uno no está documentado en ninguna parte — mira la issue #4937, cerrada sin respuesta.

.setup() es obligatorio en el primer uso. Crea las tablas y ejecuta las migraciones.

compile(checkpointer=...). Sin este argumento el grafo se ejecuta perfectamente y no guarda nada. No hay ninguna advertencia. Es la razón más probable por la que alguien cree que los checkpoints "no funcionan".

thread_id. El identificador de una conversación o de una ejecución. Cada checkpoint está limitado a él, y reanudar significa pasar el mismo. Tiene que quedarse por debajo de 255 caracteres.

Todo junto, eso es lg_agent.py:

lg_agent.py

import sys, time
from operator import add
from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.postgres import PostgresSaver

DB_URI = "postgresql://langgraph:[email protected]:5434/checkpoints"
THREAD = sys.argv[1] if len(sys.argv) > 1 else "ticket-1"
RESUME = "--resume" in sys.argv

class State(TypedDict):
    completed: Annotated[list[str], add]

def step_one(state: State):
    print("step_one running", flush=True); time.sleep(2)
    return {"completed": ["step_one"]}

def step_two(state: State):
    print("step_two running — 30s window, KILL ME HERE", flush=True); time.sleep(30)
    return {"completed": ["step_two"]}

def step_three(state: State):
    print("step_three running", flush=True)
    return {"completed": ["step_three"]}

builder = StateGraph(State)
for name, fn in [("step_one", step_one), ("step_two", step_two), ("step_three", step_three)]:
    builder.add_node(name, fn)
builder.add_edge(START, "step_one")
builder.add_edge("step_one", "step_two")
builder.add_edge("step_two", "step_three")
builder.add_edge("step_three", END)

with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
    checkpointer.setup()
    graph = builder.compile(checkpointer=checkpointer)
    config = {"configurable": {"thread_id": THREAD}}

    snap = graph.get_state(config)
    print(f"BEFORE  values={snap.values}  next={snap.next}", flush=True)

    result = graph.invoke(None if RESUME else {"completed": []}, config)
    print("FINAL:", result, flush=True)

    snap = graph.get_state(config)
    print(f"AFTER   values={snap.values}  next={snap.next}", flush=True)

RESUME es lo que hace que --resume funcione: en una reanudación la entrada es None en vez de un estado nuevo, que es lo que cubre el paso siguiente.

python lg_agent.py ticket-41

Después de la primera ejecución existen cuatro tablas:

docker exec lg-checkpoints psql -U langgraph -d checkpoints -c '\dt'

Output

 public | checkpoint_blobs      | table | langgraph
 public | checkpoint_migrations | table | langgraph
 public | checkpoint_writes     | table | langgraph
 public | checkpoints           | table | langgraph

Tabla	Contiene
checkpoints	Una fila por límite de nodo — el índice de instantáneas
checkpoint_blobs	Los valores de estado serializados
checkpoint_writes	Escrituras pendientes de nodos que terminaron en un super-paso
checkpoint_migrations	Registro de la versión del esquema

checkpoint_writes es la que hace que funcione la recuperación tras una caída, como muestra el paso siguiente.

Figura 2. .setup() escribe el esquema por ti — sin DDL manual.

Leer la posición

snap = graph.get_state(config)
print(f"BEFORE  values={snap.values}  next={snap.next}")

get_state devuelve un StateSnapshot. Importan dos campos: values es el estado ahora mismo, y next es la tupla de nodos que quedan por ejecutar. next=() significa que el hilo terminó o nunca empezó; cualquier otra cosa nombra lo que está pendiente.

next es cómo respondes "¿dónde está esta ejecución?" — la pregunta que un bucle no puede responder.

Paso 4 — Mátalo, y mira cómo vuelve

Arranca un hilo nuevo y pulsa Ctrl+C durante la ventana de treinta segundos.

python lg_agent.py ticket-42

Output

BEFORE  values={}  next=()
step_one running
step_two running — 30s window, KILL ME HERE
^CKeyboardInterrupt

Figura 3. Una caída real — la traza viene de dentro de step_two, que por tanto nunca se completó.

Ahora reanuda el mismo thread_id:

python lg_agent.py ticket-42 --resume

Output

BEFORE  values={'completed': ['step_one']}  next=('step_two',)
step_two running — 30s window, KILL ME HERE
step_three running
FINAL: {'completed': ['step_one', 'step_two', 'step_three']}
AFTER   values={'completed': [...]}  next=()

Figura 4. La línea BEFORE es la prueba — estado recuperado de Postgres por un proceso completamente nuevo.

Lee esa primera línea con atención, porque es el tutorial entero:

• values muestra el resultado de step_one, recuperado de Postgres por un proceso que no existía cuando se produjo
• next=('step_two',) — el grafo sabe exactamente qué nodo no terminó
• step_one no se ejecutó otra vez. Se ejecutó una sola vez a lo largo de tres invocaciones

Ese último punto es lo que compra checkpoint_writes: LangGraph conserva las escrituras completadas de un super-paso, así que reanudar no repite trabajo que ya tuvo éxito.

La llamada de reanudación en sí es:

graph.invoke(None, config)

None como entrada significa "sin entrada nueva — continúa desde el checkpoint". Vale la pena saberlo: esto está documentado en la página de interrupciones para breakpoints estáticos, pero no en las páginas de persistencia o de ejecución duradera, que es donde mirarías cuando tu proceso acaba de morir. Funciona; simplemente no está escrito donde lo necesitas.

Mátalo una segunda vez y reanuda otra vez. Funciona, porque leer un checkpoint no lo consume.

Paso 5 — Pausa para una persona

Una caída es una parada accidental. Una aprobación es una deliberada — y una vez que puedes sobrevivir a la primera, la segunda es casi gratis.

El agente está a punto de escribir una cita en el calendario de un cliente. Eso es irreversible y lo ve una persona real, así que quieres que alguien lo revise antes.

Sin estado duradero, esperar a una persona significa mantener un proceso abierto. Con el estado en Postgres el agente se detiene, el proceso sale, y la aprobación llega cuando llegue — diez minutos o el lunes por la mañana.

from langgraph.types import interrupt, Command

def approval(state: State):
    decision = interrupt({"question": "Approve step_two?", "so_far": state["completed"]})
    return {"completed": ["approval"], "approved": decision}

Guárdalo como lg_agent2.py y ejecútalo:

python lg_agent2.py booking-7

interrupt() hace dos cosas a la vez: detiene el grafo, y entrega su argumento a quien lo haya llamado. Ese argumento es tu pregunta para la persona — cualquier valor serializable a JSON.

Output

step_one running
RESULT: {'completed': ['step_one'], 'approved': '',
         '__interrupt__': [Interrupt(value={'question': 'Approve step_two?',
         'so_far': ['step_one']}, id='29f211cf5a73ea78e0f694e1bd38822b')]}
AFTER   next=('approval',)  interrupts=(Interrupt(...),)

Ninguna excepción, código de salida 0. Nota la diferencia con el Paso 4: la caída producía una traza, esto produce un retorno limpio.

La pausa aparece en dos sitios, y la diferencia importa:

Dónde	Uso
__interrupt__ en el resultado de invoke	Quien acaba de ejecutar el grafo
snapshot.interrupts	Cualquier otro proceso — un panel que muestre la pregunta

El id correlaciona una respuesta con la pausa correcta, que importa en cuanto hay más de una aprobación pendiente.

El proceso ya no existe; el estado está en Postgres. Una hora o una semana después:

python lg_agent2.py booking-7 yes

que llama a:

graph.invoke(Command(resume="yes"), config)

Output

approval resumed with: yes
step_two running (approved=yes)
AFTER   next=()  interrupts=()

Command(resume=...) es distinto de invoke(None, ...). None dice "continúa"; Command(resume=X) dice "continúa, y interrupt() debería devolver X". Ese valor vino de la línea de comandos, pasó por el retorno de interrupt(), entró en el estado, y lo leyó el nodo siguiente.

Figura 5. Pausa y reanudación — dos procesos distintos, una sola ejecución.

La parte que te va a morder

Al reanudar, LangGraph vuelve a ejecutar el nodo interrumpido desde arriba. Esta vez interrupt() devuelve tu valor en vez de pausar — pero todo lo que está por encima en esa función ya se ejecutó una segunda vez.

Pon un print antes de la interrupción, luego ejecuta un hilo nuevo y apruébalo:

python lg_agent2.py booking-8

python lg_agent2.py booking-8 approve

Aparece en las dos ejecuciones, para una sola aprobación:

Output

run 1:  step_one running
        SIDE EFFECT — before interrupt
        RESULT: {... '__interrupt__': [Interrupt(...)]}

run 2:  SIDE EFFECT — before interrupt
        approval resumed with: approve
        step_two running (approved=approve)

Figura 6. Una aprobación, dos ejecuciones de todo lo que está antes de la interrupción.

Si esa línea fuera send_email() o charge_card(), una aprobación le cobra al cliente dos veces. Nada lanza una excepción. El estado final es correcto. Solo se duplicó el efecto secundario — que es la razón por la que esto es fácil de desplegar y difícil de notar.

No pongas nada antes de interrupt() en ese nodo. Mueve los efectos secundarios después, o dale a la interrupción un nodo propio que no haga nada más. El mismo razonamiento aplica a reanudar tras una caída: un nodo siempre puede ejecutarse más de una vez, así que cualquier cosa con una consecuencia externa debería ser idempotente o estar aislada.

Paso 6 — Un modelo real, y rastros

Dos paquetes más: uno para hablar con el modelo, otro para trazarlo.

pip install langchain-openai langfuse

llm = ChatOpenAI(
    model="Qwen2.5-3B-Instruct",
    base_url="https://inference.wiline.com/v1",
    api_key=os.environ["WEC_API_KEY"],
    temperature=0,
)

ChatOpenAI no es específico de OpenAI — habla el protocolo de OpenAI, que WEC Inference implementa. Apuntar base_url ahí es toda la integración. temperature=0 mantiene las ejecuciones comparables mientras pruebas.

La llamada va dentro de un nodo como cualquier otro trabajo:

def write(state: State):
    r = llm.invoke(f"Write one sentence about {state['topic']}.")
    return {"completed": ["write"], "draft": r.content}

Necesitas un proyecto de Langfuse y un par de claves de API para esto — creados en Organization → Project, luego Settings → API Keys, como cubrimos en Observa producción con Langfuse.

El trazado se adjunta por invocación, en el mismo diccionario de configuración que el id del hilo:

from langfuse.langchain import CallbackHandler
handler = CallbackHandler()
config = {"configurable": {"thread_id": THREAD}, "callbacks": [handler]}

configurable son los ajustes propios de LangGraph; callbacks es el gancho de observadores de LangChain. Langfuse se suscribe a los eventos de inicio y fin de nodo y construye spans a partir de ellos — que es la razón por la que el rastro refleja tu grafo sin ninguna instrumentación extra.

Las credenciales vienen del entorno — en el SDK v4 CallbackHandler() no acepta argumentos, así que esta es la única forma de configurarlo:

export WEC_API_KEY='...'
export LANGFUSE_PUBLIC_KEY='pk-lf-...'
export LANGFUSE_SECRET_KEY='sk-lf-...'
export LANGFUSE_HOST='http://localhost:3001'

Todo junto como lg_llm.py:

import os, sys
from operator import add
from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.postgres import PostgresSaver
from langchain_openai import ChatOpenAI
from langfuse.langchain import CallbackHandler

DB_URI = "postgresql://langgraph:[email protected]:5434/checkpoints"
THREAD = sys.argv[1] if len(sys.argv) > 1 else "draft-1"

llm = ChatOpenAI(
    model="Qwen2.5-3B-Instruct",
    base_url="https://inference.wiline.com/v1",
    api_key=os.environ["WEC_API_KEY"],
    temperature=0,
)

class State(TypedDict):
    topic: str
    completed: Annotated[list[str], add]
    draft: str

def plan(state: State):
    print("plan running", flush=True)
    return {"completed": ["plan"]}

def write(state: State):
    r = llm.invoke(f"Write one sentence about {state['topic']}.")
    print("model said:", r.content, flush=True)
    return {"completed": ["write"], "draft": r.content}

builder = StateGraph(State)
builder.add_node("plan", plan)
builder.add_node("write", write)
builder.add_edge(START, "plan")
builder.add_edge("plan", "write")
builder.add_edge("write", END)

handler = CallbackHandler()

with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
    checkpointer.setup()
    graph = builder.compile(checkpointer=checkpointer)
    config = {"configurable": {"thread_id": THREAD}, "callbacks": [handler]}
    result = graph.invoke(
        {"topic": "edge computing", "completed": [], "draft": ""}, config
    )
    print("FINAL:", result, flush=True)

python lg_llm.py draft-1

Output

LangGraph            2.18s
  plan               0.00s
  write              2.14s
    ChatOpenAI       2.13s   36 -> 37 (73 tokens)

Figura 7. Árbol de spans, topología del grafo y conteo de tokens — desde un solo callback.

Casi todo es la llamada al modelo. Eso deja unos 40ms para el grafo y el trazado, y la diferencia se mantuvo cerca de 40ms incluso en ejecuciones que tardaron el doble. plan, que no llama a nada, informa 0,00s.

La conclusión útil: la orquestación no es tu problema de latencia. Si un agente se siente lento, el grafo no es la razón.

Resolución de problemas (errores reales)

address already in use frente a port is already allocated. Parecen intercambiables y no lo son. El primero significa que un proceso del host tiene el puerto; el segundo que lo tiene otro contenedor — la contabilidad propia de Docker, antes de intentar el bind. Diagnostica con:

docker ps --format '{{.Names}}\t{{.Ports}}' && sudo ss -ltnp | grep :543

Un docker run fallido bloquea el nombre. El contenedor se crea, luego muere, y sigue conservando su nombre — así que tu reintento falla de forma distinta a la primera vez. docker rm <nombre> antes de reintentar.

ModuleNotFoundError: No module named 'langchain' al importar el CallbackHandler de Langfuse. Ni LangGraph ni langchain-openai necesitan el paquete paraguas langchain; Langfuse lo importa puramente como centinela de versión. El arreglo es instalar un paquete que nada más en tu stack usa.

Vale la pena distinguir dos hilos del proyecto. #9758 reportó el fallo de importación en octubre de 2025 y está cerrado — pero todavía se reproduce en la versión del SDK de arriba, así que léelo como historia y no como un arreglo que llegó. #13651 está abierto y propone el remedio real: leer langchain_core.__version__ en vez de exigir el paquete paraguas.

Langfuse se desactiva en silencio. Olvida una variable de entorno y recibes una línea de advertencia, y luego una ejecución completamente normal — código de salida 0, salida correcta, ningún rastro:

Output

Authentication error: Langfuse client initialized without public_key.
Client will be disabled.

En el SDK v4 CallbackHandler() no acepta argumentos en el constructor, así que las variables de entorno son la única vía de configuración — y que falte una no lanza ninguna excepción. Comprueba que el cliente está habilitado al arrancar en vez de confiar en la ausencia de errores.

Deriva de versiones. La mayoría del material sobre LangGraph + Langfuse apunta al SDK v3 de Langfuse. En v4 el parámetro update_trace ya no existe y lanza TypeError, y las credenciales ya no pueden pasarse al constructor.

🎯

Finished this tutorial?

Mark it complete to earn State that survives a restart on your skill path.

Lo que viene

Parte 2: pasar trabajo entre agentes — y qué le ocurre al estado compartido cuando no están de acuerdo.

Lecturas relacionadas en este sitio:

• Trazado a nivel de componente para llamadas a herramientas de agentes — encontrar qué paso de un agente falló, no solo que falló
• Autoaloja el agente Hermes con memoria persistente — un modelo de persistencia distinto, memoria entre conversaciones en vez de estado dentro de una ejecución
• "Loop Engineering Is Dead" — por qué la industria pasó de bucles a grafos