Pasar trabajo entre agentes LangGraph sin corromper el estado compartido
+
- 1State that survives a restart
- 2Hand work between agents
- 3Governed tools an agent can call
- 🏆Engineer the context, not the prompt
Tienes dos agentes. Uno reserva citas. Otro gestiona facturación.
Llega un ticket que necesita los dos: cambia la fecha de mi instalación, y mi factura parece incorrecta. Así que lo envías a los dos.
El primero reserva el martes. El segundo ve que la cuenta está en mora y la congela. Los dos terminan casi en el mismo momento, y los dos guardan lo que decidieron.
Solo se guarda uno de ellos. ¿Cuál? El que terminó primero — que depende de lo lenta que estuviera una llamada de API ese día. Así que reservas una cita en una cuenta congelada, o congelas una cuenta a la que acabas de prometerle un ingeniero. Después parece que se tomó una única decisión limpia.
La Parte 1 construyó un agente que ejecuta sus pasos en un orden fijo. Este post tiene varios: un supervisor que elige quién trabaja en qué, un agente que pasa el trabajo a otro a mitad de camino, y dos agentes puestos en el camino del otro a propósito — para descubrir qué hace LangGraph cuando no están de acuerdo.
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 · langchain-openai 1.6.0 · langfuse 4.14.5 contra el servidor
Langfuse v3.205.1 OSS · imagen postgres:17. El contenedor lg-checkpoints y el
virtualenv de la Parte 1 se reutilizan sin cambios.
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
- El contenedor de Postgres y el virtualenv de la Parte 1
- Una instancia de Langfuse para la última sección — mira Observa producción con Langfuse
Alguien tiene que decidir quién trabaja en esto
Dos agentes significa que algo tiene que elegir entre ellos. Ese algo se llama supervisor, y es simplemente otro agente cuyo único trabajo es elegir al siguiente.
La Parte 1 conectó sus pasos de antemano con add_edge — paso uno, luego dos, luego tres.
Un supervisor no puede hacer eso, porque no sabe adónde va el trabajo hasta que lo lee.
En su lugar el nodo devuelve un Command:
from langgraph.types import Command
def supervisor(state: State) -> Command[Literal["scheduler", "billing"]]:
nxt = "billing" if "invoice" in state["ticket"].lower() else "scheduler"
print(f"supervisor routing to {nxt}", flush=True)
return Command(goto=nxt, update={"completed": [f"supervisor->{nxt}"]})
Command hace dos trabajos a la vez: goto nombra el siguiente nodo, y update lleva un
cambio de estado consigo. No hay ningún add_edge desde supervisor a nada — el
enrutamiento ocurre en tiempo de ejecución, desde dentro de la función.
Guarda todo como lg_super.py:
import sys
from operator import add
from typing import Annotated, Literal
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.postgres import PostgresSaver
from langgraph.types import Command
THREAD = sys.argv[1] if len(sys.argv) > 1 else "sup-1"
TICKET = sys.argv[2] if len(sys.argv) > 2 else "reschedule my install"
class State(TypedDict):
ticket: str
completed: Annotated[list[str], add]
def supervisor(state: State) -> Command[Literal["scheduler", "billing"]]:
nxt = "billing" if "invoice" in state["ticket"].lower() else "scheduler"
print(f"supervisor routing to {nxt}", flush=True)
return Command(goto=nxt, update={"completed": [f"supervisor->{nxt}"]})
def scheduler(state: State):
print("scheduler running", flush=True)
return {"completed": ["scheduler"]}
def billing(state: State):
print("billing running", flush=True)
return {"completed": ["billing"]}
builder = StateGraph(State)
builder.add_node("supervisor", supervisor)
builder.add_node("scheduler", scheduler)
builder.add_node("billing", billing)
builder.add_edge(START, "supervisor")
builder.add_edge("scheduler", END)
builder.add_edge("billing", END)
if "--draw" in sys.argv:
print(builder.compile().get_graph().draw_mermaid())
raise SystemExit
with PostgresSaver.from_conn_string(DB_URI) as cp:
cp.setup()
graph = builder.compile(checkpointer=cp)
config = {"configurable": {"thread_id": THREAD}}
print("FINAL:", graph.invoke({"ticket": TICKET, "completed": []}, config), flush=True)
La rama --draw imprime el grafo y sale antes de tocar Postgres, así que puedes inspeccionar
la topología sin base de datos. Se gana su sitio en la sección siguiente.
~/.venv/bin/python ~/lg_super.py sup-1 "reschedule my install"
supervisor routing to scheduler
scheduler running
FINAL: {'ticket': 'reschedule my install', 'completed': ['supervisor->scheduler', 'scheduler']}
Y la otra rama, para mostrar que el enrutamiento depende de los datos y no está fijo en el código:
~/.venv/bin/python ~/lg_super.py sup-2 "please fix the invoice on my account"
supervisor routing to billing
billing running
FINAL: {'ticket': 'please fix the invoice on my account', 'completed': ['supervisor->billing', 'billing']}
La anotación que no es una pista de tipos
Esa anotación de retorno parece typing normal que podrías quitar:
def supervisor(state: State) -> Command[Literal["scheduler", "billing"]]:
Quítala y el código se ejecuta igual — mismo enrutamiento, misma salida. Así que es opcional. Excepto que no lo es.
sed 's/ -> Command\[Literal\["scheduler", "billing"\]\]//' ~/lg_super.py > ~/lg_super_noann.py
Pídele a las dos versiones que se dibujen:
~/.venv/bin/python ~/lg_super.py --draw
~/.venv/bin/python ~/lg_super_noann.py --draw
Con la anotación, las dos rutas posibles están ahí como aristas condicionales punteadas:
__start__ --> supervisor;
supervisor -.-> billing;
supervisor -.-> scheduler;
billing --> __end__;
scheduler --> __end__;
Sin ella, desaparecen, y el grafo afirma algo falso:
__start__ --> supervisor;
supervisor --> __end__;
scheduler y billing siguen declarados como nodos, y ahora nada llega a ellos. El
diagrama dice que el supervisor va directo al final.
Figura 1. El mismo código, con una anotación de diferencia. El grafo de abajo está mal.
La razón es mecánica: goto se decide dentro del cuerpo de la función, donde nada puede
inspeccionarlo. La anotación es la única declaración de dónde tiene permitido un
supervisor mandar trabajo. Omítela y tu código sigue funcionando mientras cada diagrama, y
cualquier otra cosa que lea la topología, miente en silencio.
draw_ascii() necesita un paquete que LangGraph no incluyeget_graph().draw_ascii() lanza
ImportError: Install grandalf to draw graphs: 'pip install grandalf'.
draw_mermaid() no tiene dependencias extra y es lo que usan los ejemplos de arriba.
Traspasar sin preguntarle al supervisor
Un supervisor decide una vez, al principio. Eso es un problema cuando el descubrimiento ocurre más tarde.
Toma un ticket que pide cambiar la fecha de una instalación, en una cuenta que resulta estar en mora. El supervisor ve una petición de agenda y enruta en consecuencia. El agente de agenda empieza a trabajar, lee la cuenta, y encuentra un problema que le pertenece a facturación.
El agente de agenda también puede enrutar — es un nodo, y los nodos devuelven Command.
Copia lg_super.py a lg_handoff.py, reemplaza scheduler con la versión de abajo, y borra
la línea builder.add_edge("scheduler", END) — el nodo decide su propia salida ahora:
def scheduler(state: State) -> Command[Literal["billing", "__end__"]]:
if "past due" in state["ticket"].lower():
print("scheduler: account is past due, handing off to billing", flush=True)
return Command(goto="billing", update={"completed": ["scheduler(handoff)"]})
print("scheduler: booking it", flush=True)
return Command(goto=END, update={"completed": ["scheduler(booked)"]})
~/.venv/bin/python ~/lg_handoff.py handoff-3 "reschedule my install, account is past due"
~/.venv/bin/python ~/lg_handoff.py handoff-4 "reschedule my install"
supervisor -> scheduler
scheduler: account is past due, handing off to billing
billing running
FINAL: {'ticket': '...past due', 'completed': ['supervisor', 'scheduler(handoff)', 'billing']}
supervisor -> scheduler
scheduler: booking it
FINAL: {'ticket': 'reschedule my install', 'completed': ['supervisor', 'scheduler(booked)']}
Figura 2. El supervisor nunca supo que facturación estaría implicada.
Dos detalles que vale la pena guardar. goto=END funciona desde dentro de un Command, así
que un nodo puede terminar la ejecución por sí mismo. Y "__end__" es la forma en cadena de
END en ese Literal — la anotación necesita el valor literal, no la constante.
Esta es la diferencia entre los dos patrones publicados. Un supervisor decide desde fuera y necesita conocer cada precondición por adelantado. Un traspaso deja que el agente que hace el trabajo redirija en cuanto aprende algo. La mayoría de los sistemas reales quieren los dos, que es lo que tiene este grafo.
Enviar trabajo a dos agentes a la vez
Hasta aquí el supervisor elige un agente. A veces quieres los dos — revisar el calendario y
la cuenta al mismo tiempo, en vez de esperar a uno antes de empezar el otro. goto acepta
una lista precisamente para eso:
return Command(goto=["scheduler", "billing"], update={"completed": ["supervisor"]})
Los dos se ejecutan en el mismo super-paso. Que es donde se pone interesante, porque ahora pueden no estar de acuerdo.
Copia lg_super.py a lg_conflict.py. Dale al estado dos tipos de campo:
class State(TypedDict):
completed: Annotated[list[str], add] # has a reducer
decision: str # no reducer
Y haz que cada agente escriba un valor distinto en el que no tiene reducer:
def scheduler(state: State):
return {"completed": ["scheduler"], "decision": "book tuesday"}
def billing(state: State):
return {"completed": ["billing"], "decision": "refund first"}
~/.venv/bin/python ~/lg_conflict.py conflict-2
supervisor sending work to BOTH
billing deciding
scheduler deciding
Traceback (most recent call last):
...
File ".../langgraph/pregel/_loop.py", line 692, in after_tick
self.updated_channels = apply_writes(
File ".../langgraph/channels/last_value.py", line 64, in update
raise InvalidUpdateError(msg)
langgraph.errors.InvalidUpdateError: At key 'decision': Can receive only one value
per step. Use an Annotated key to handle multiple values.
Figura 3. Los dos agentes terminaron. El grafo se negó a fusionar sus respuestas.
Lee la traza con atención, porque tres cosas de ella importan.
Los dos agentes se ejecutaron. billing deciding y scheduler deciding se imprimieron
los dos. El trabajo se completó; el fallo vino después, en after_tick → apply_writes.
Esto no son dos agentes chocando en vuelo. Es el grafo negándose a reconciliarlos en el
límite del paso.
completed estuvo bien. Los dos agentes le añadieron valores en el mismo paso y nada se
quejó, porque add dice qué significan dos valores. Solo falló decision.
LangGraph no elige un ganador. Ni "el último gana", ni una elección silenciosa. Que es el comportamiento correcto — la alternativa es un agente de reservas que a veces reserva el martes y a veces emite un reembolso según qué tarea terminó primero por casualidad.
Fíjate también en el orden: billing se imprimió antes que scheduler, al revés de la
lista de goto. Las ramas paralelas terminan en el orden en que terminen.
Declarar la política
El arreglo no es evitar el desacuerdo. Es decir qué significa un desacuerdo. Copia
lg_conflict.py a lg_resolve.py y añade el reducer:
PRIORITY = {"refund first": 2, "book tuesday": 1}
def prefer_higher_priority(old: str, new: str) -> str:
"""The policy: a billing hold outranks a scheduling decision."""
winner = max([old, new], key=lambda v: PRIORITY.get(v, 0))
print(f" reducer: '{old}' vs '{new}' -> '{winner}'", flush=True)
return winner
class State(TypedDict):
completed: Annotated[list[str], add]
decision: Annotated[str, prefer_higher_priority]
~/.venv/bin/python ~/lg_resolve.py resolve-2
reducer: '' vs '' -> ''
supervisor sending work to BOTH
billing deciding
scheduler deciding
reducer: '' vs 'refund first' -> 'refund first'
reducer: 'refund first' vs 'book tuesday' -> 'refund first'
FINAL: {'completed': ['supervisor', 'billing', 'scheduler'], 'decision': 'refund first'}
Figura 4. El print dentro del reducer hace visible la fusión.
Mismo grafo, mismo desacuerdo, ningún error. Y la traza muestra tres cosas que la documentación no deja claras:
El reducer se ejecuta por pares, plegando hacia la izquierda. No una vez con los dos valores — dos veces, acumulando. Primero fusiona el estado existente con la respuesta de facturación, luego fusiona ese resultado con la del agente de agenda.
Por tanto debe ser independiente del orden. Como las ramas terminan en orden arbitrario,
un reducer del tipo "toma el más nuevo" produce respuestas distintas en ejecuciones
distintas. max por prioridad no.
Se dispara antes de que se ejecute nada. Ese primer reducer: '' vs '' -> '' es el
estado inicial siendo escrito, fusionando el valor por defecto con el decision: "" pasado a
invoke. Tu reducer ve entrada vacía y tiene que tolerarla.
Caerse a mitad de un paso paralelo
La Parte 1 mostró una ejecución matada reanudando desde su último nodo completado. El trabajo paralelo plantea una pregunta más afilada: si una rama termina y su hermana muere, ¿se pierde el trabajo terminado?
Copia lg_conflict.py a lg_partial.py. Quita el campo decision, añade
RESUME = "--resume" in sys.argv, pasa None if RESUME else {...} a invoke como en la
Parte 1, y reemplaza los dos especialistas por un nodo instantáneo y otro lo bastante lento
para matarlo:
def fast(state: State):
print("fast running — finishes immediately", flush=True)
return {"completed": ["fast"]}
def slow(state: State):
print("slow running — 30s window, KILL ME HERE", flush=True); time.sleep(30)
return {"completed": ["slow"]}
~/.venv/bin/python ~/lg_partial.py partial-2
Pulsa Ctrl+C durante la ventana. Hacen falta dos pulsaciones, y la ejecución termina con algo
que parece mucho peor que un KeyboardInterrupt:
File ".../langgraph/pregel/_runner.py", line 613, in commit
self.put_writes()(task.id, task.writes)
...
RuntimeError: cannot schedule new futures after shutdown
Esa es la escritura del checkpoint de fast siendo confirmada contra un executor que ya se
ha cerrado. Se lee como pérdida de datos.
No lo es:
~/.venv/bin/python ~/lg_partial.py partial-2 --resume
BEFORE values={'completed': ['supervisor', 'fast']} next=('slow',)
slow running — 30s window, KILL ME HERE
FINAL: {'completed': ['supervisor', 'fast', 'slow']}
Figura 5. fast recuperado, no se volvió a ejecutar, y next nombra solo la rama que
murió.
fast está en el estado recuperado, aparece exactamente una vez en el resultado final, y
next=('slow',) nombra solo la rama que nunca terminó. La escritura ya había aterrizado en
checkpoint_writes — la misma tabla que impidió que step_one se repitiera en la Parte 1
ahora protege a una hermana completada.
Vale la pena documentarlo precisamente porque el error sugiere lo contrario. Un
RuntimeError en una ruta de cierre es ruido; el estado es lo que hay que comprobar.
Un modelo como enrutador
Cada decisión de enrutamiento hasta aquí ha sido "invoice" in ticket — coincidencia de
palabras clave. Ese es el enfoque que medimos desmoronándose en
el enrutador por complejidad del gateway, donde un
plural basta para no encontrar una palabra clave.
Así que deja que el modelo decida. Copia lg_super.py a lg_traced.py — mismo grafo, un
nodo cambiado:
llm = ChatOpenAI(
model="gemma4",
base_url="https://inference.wiline.com/v1",
api_key=os.environ["WEC_API_KEY"],
temperature=0,
)
def supervisor(state: State) -> Command[Literal["scheduler", "billing"]]:
r = llm.invoke(
"Route this support ticket to exactly one team. "
"Answer with one word, either scheduler or billing, nothing else.\n\n"
f"Ticket: {state['ticket']}"
)
raw = (r.content or "").strip().lower()
nxt = "billing" if "billing" in raw else "scheduler"
print(f"supervisor: model said {raw!r} -> {nxt}", flush=True)
return Command(goto=nxt, update={"completed": [f"supervisor->{nxt}"]})
Las credenciales vienen del entorno, que el .env del gateway ya contiene:
set -a; . ~/llm-gateway/.env; set +a; ~/.venv/bin/python ~/lg_traced.py traced-1
El ticket es deliberadamente ambiguo — "my bill looks wrong and I need a new install date" menciona los dos:
supervisor: model said 'billing' -> billing
billing running
FINAL: {'ticket': 'my bill looks wrong and I need a new install date', 'completed': ['supervisor->billing', 'billing']}
Una palabra, sin frase envolvente, y el parseo se mantuvo. Eligió facturación — defendible, ya que una cuenta sin pagar bloquea la instalación, que es la misma conclusión a la que llegó antes el traspaso leyendo la cuenta.
Lo que costó esa palabra
Pedirle a un modelo que elija no es gratis, y la factura no está donde la buscarías. Activar
el trazado muestra a dónde fueron el tiempo y los tokens. Es una línea, como en la Parte 1 —
CallbackHandler() lee sus credenciales del entorno y va 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]}
Figura 6. Tres cosas enmarcadas: lo que costó el enrutamiento, la decisión como dato, y
la carga llegando.
El árbol de spans es el grafo:
LangGraph 8.66s
supervisor 8.53s
ChatOpenAI 8.49s 52 → 412 (Σ 464)
billing 0.01s
Tres hallazgos están en ese cuadro.
Una respuesta de una palabra costó 412 tokens de salida. Confirmado contra la API en vez de leído en la pantalla:
set -a; . ~/llm-gateway/.env; set +a
curl -s -u "$LANGFUSE_PUBLIC_KEY:$LANGFUSE_SECRET_KEY" \
"$LANGFUSE_HOST/api/public/observations?limit=8" \
| jq '.data[] | select(.name=="ChatOpenAI") | {model, usage}'
{
"model": "gemma4",
"usage": {
"unit": "TOKENS",
"input": 52,
"output": 412,
"total": 464
}
}
La respuesta visible tiene siete caracteres. La factura es de 412 tokens de salida. Lo que sea que el modelo generó antes de decidirse, lo pagaste — y una llamada de enrutamiento parece lo más barato del grafo.
El supervisor es el nodo caro, no los especialistas. billing tardó 6ms y ningún token.
El enrutador tardó 8,49s y 464. En un grafo multi-agente, las decisiones de orquestación son
el centro de coste.
La decisión de enrutamiento queda registrada como dato. La salida del supervisor en el
rastro es el Command serializado:
{"graph": null, "update": {"completed": ["supervisor->billing"]}, "resume": null, "goto": "billing"}
Y la entrada de billing contiene ["supervisor->billing"] — el update de ese Command
llegando como estado del agente que lo recibe. Traspaso, coste y carga en un solo cuadro.
scheduler no aparece en ninguna parte de este rastro — ni como span omitido, ni como nodo
desactivado. El panel Graph dibuja __start__ → supervisor → billing → __end__. No puedes
saber desde un rastro qué alternativas existían, solo cuál se ejecutó, que es una segunda
razón por la que importa la anotación Literal que falta: el grafo dibujado es el único
sitio donde tus alternativas están escritas.
Lo que viene
Parte 3: las herramientas que un agente tiene permitido llamar. Todo lo anterior confía en que cada nodo haga solo lo que debe — el próximo post pone las herramientas detrás de un gateway con permisos por herramienta, para que las capacidades de un agente se concedan en vez de asumirse.
Resolución de problemas
ImportError: Install grandalf to draw graphs
draw_ascii() necesita un paquete del que LangGraph no depende. Usa draw_mermaid().
El diagrama del grafo muestra al supervisor yendo directo a __end__
Al nodo de enrutamiento le falta su anotación de retorno -> Command[Literal[...]]. El
código sigue funcionando; solo la topología está mal.
InvalidUpdateError: At key 'x': Can receive only one value per step
Dos nodos escribieron la misma clave en un super-paso y esa clave no tiene reducer. O le das
uno con Annotated[T, fn], o dejas de despachar los dos nodos en paralelo.
RuntimeError: cannot schedule new futures after shutdown después de Ctrl+C
La escritura del checkpoint de una rama completada compitió con el cierre del executor.
Comprueba graph.get_state(config).values antes de asumir que se perdió algo — en las
pruebas la escritura había aterrizado todas las veces.
El reducer produce resultados distintos en ejecuciones distintas
No es independiente del orden. Las ramas paralelas terminan en orden arbitrario, así que un reducer que dependa de qué valor llega segundo es no determinista por construcción.
Lecturas adicionales
- Parte 1 — checkpoints de un agente LangGraph — estado que sobrevive a una caída
- Enrutar por complejidad — lo que el enrutamiento por palabras clave no ve, medido
- Trazado a nivel de componente para llamadas a herramientas de agentes — encontrar qué paso falló