Flujos de agentes avanzados de LangChain: planes en tiempo real, llamadas a herramientas y tokens.
Sustituir una invocación de bloqueo por un flujo para que los usuarios vean el funcionamiento de un agente avanzado: los tres modos de flujo, los nombres de espacio para subagentes y los tokens de filtrado y resumen.
Fuentes: LangChain Deep Agents: Your AI Agent Is Not Frozen. It Just Cannot Tell You It Is Working. II
Pocas cosas erosionan la confianza más rápido que una pantalla congelada. Cuando un usuario no puede saber si un agente está pensando, atascado o ya inactivo, asume lo peor. La transmisión en tiempo real reemplaza ese silencio con una actividad continua: el plan, las llamadas a herramientas y las palabras aparecen a medida que ocurren, y las personas pueden ver que algo está sucediendo.
Por qué invoke hace que un agente avanzado parezca inactivo
Intente ejecutar un agente avanzado en una tarea real mientras alguien lo observa. La llamada pasa por invoke, y luego la pantalla se queda inmóvil durante cuarenta segundos. Dentro, el agente está leyendo archivos, elaborando un plan, utilizando herramientas y razonando al respecto. El usuario solo ve una caja inmóvil y un indicador de carga, lo cual podría significar un trabajo intenso o una caída silenciosa. Dado que ambos estados se ven idénticos, las personas vuelven a cargar la página y presentan un ticket indicando que está atascado.
Nada se colgó. El agente simplemente no tenía un canal para narrarse a sí mismo. Los bloques invoke están diseñados así: ejecutan todo el bucle de forma invisible y devuelven una única respuesta final, lo cual es perfecto para un script o una tarea cron que nadie está supervisando. Sin embargo, es la elección incorrecta cuando hay un humano al otro lado.
La solución propuesta deja el bucle intacto. Reemplaza la llamada bloqueante por un flujo en tiempo real, de modo que los planes, las llamadas a herramientas y los tokens se muestran al instante.
De invoke a stream
El streaming está estrechamente relacionado con invoke. En lugar de esperar a que se alcance un estado final, se recorre agent.stream(...) y un fragmento de datos llega cada vez que el bucle realiza alguna acción. Una versión asíncrona, agent.astream(...), tiene la misma estructura y es la que utilizaría un servidor web.
El modelo mental es sencillo. El LangGraph duradero que subyace a un agente avanzado emite un evento en cada paso, y el flujo en tiempo real transmite esos eventos a medida que se generan. No se trata de un agente diferente ni de un bucle distinto, sino simplemente de una vista del agente que ya conoces.
Hay un detalle que debe resolverse primero, ya que todos los ejemplos dependen de él: el formato de flujo v2, activado con version="v2". Cada bloque es un pequeño diccionario con tres claves:
typeindica qué tipo de evento es.ns, el espacio de nombres, indica quién lo generó.datacontiene la información del payload.
Esas tres claves permanecen iguales independientemente de lo que se transmita, lo que mantiene el código legible en lugar de convertirlo en un enredo de desempaquetado de tuplas.
Tres modos, tres alturas
Un flujo no es un solo manguero de incendios. La configuración stream_mode determina el tipo de eventos que se reciben, y las tres opciones comunes corresponden a tres niveles de detalle.
updates es la vista por pasos. Llega un evento cada vez que finaliza un nodo del gráfico, lo que en la práctica significa que el modelo ha realizado una acción o que se ha ejecutado una herramienta. Es el nivel adecuado para un registro de progreso, suficiente para mostrar cómo el agente avanza en su trabajo sin saturar a nadie con demasiados datos, y constituye un valor por defecto razonable para una fuente clara que indique “esto es lo que está haciendo el agente”.
messages representa la vista por tokens. La salida del modelo llega poco a poco a medida que se genera, del mismo modo en que se escribe cuando un asistente de chat teclea. Úsala cuando alguien quiere leer el texto generado por el agente tal como aparece. Aquí también llegan las llamadas de las herramientas de streaming, por lo que puedes anunciar “se está llamando a read_file” en el momento en que el modelo inicia la llamada y no después de que se complete.
values es la vista por instantáneas. Después de cada paso recibes todo el estado del agente, no solo lo que ha cambiado. Es más pesado y se necesita con menos frecuencia, pero es la opción adecuada cuando es importante tener una visión completa en todo momento, por ejemplo, para mostrar en tiempo real la lista de tareas o los archivos en los que está trabajando el agente.
Al pasar una lista de modos se devuelven varios a la vez, que es cómo suelen funcionar las interfaces de usuario pulidas: updates controla un indicador de paso mientras que messages proporciona el texto visible. El aspecto básico es este, un registro de progreso claro generado por updates:
for chunk in agent.stream(
{”messages”: [{”role”: “user”, “content”: “Summarize the repo’s failing tests”}]},
stream_mode=”updates”,
version=”v2”,
):
if chunk[”type”] == “updates”:
for node_name in chunk[”data”]:
print(f”step: {node_name}”)
Lo que se imprime es una lista de los pasos del grafo a medida que se completan, una línea por cada turno del modelo y por cada ejecución de herramienta. Una pantalla que parecía congelada ahora avanza visiblemente. Si en su lugar se establece stream_mode en "messages", se obtiene el texto en sí, listo para imprimir mientras el agente escribe.
Diferenciar al agente principal de sus subagentes
Un agente avanzado rara vez trabaja solo durante mucho tiempo. Tan pronto como asigna una subtarea a un subagente, dos o más agentes generan salida al mismo tiempo. Si todo se envía a un único flujo, el resultado es un desorden: parte de una oración del coordinador, unos pocos tokens de un investigador, una llamada a herramienta de quién sabe dónde. El campo ns soluciona este problema. Cada fragmento lleva uno cuando se activa la transmisión de subgrafos con subgraphs=True.
La regla es fácil de aplicar. Si el espacio de nombres está vacío, el agente principal emitió el evento. Si contiene un segmento tools:, fue un subagente quien lo hizo, y se indican los nombres de dichos segmentos. A continuación, se puede ordenar la salida por origen: el texto del coordinador aquí, y el trabajo de cada subagente en su propia sección.
for chunk in agent.stream(
{”messages”: [{”role”: “user”, “content”: “Research and summarize quantum computing”}]},
stream_mode=”messages”,
subgraphs=True, # ①
version=”v2”,
):
if chunk[”type”] == “messages”: # ②
token, metadata = chunk[”data”] # ③
is_subagent = any(s.startswith(”tools:”) for s in chunk[”ns”]) # ④
source = “subagent” if is_subagent else “main” # ⑤
if token.content:
print(f”[{source}] {token.content}”, end=”“, flush=True) # ⑥
Seis detalles son importantes en ese fragmento. Primero, subgraphs=True es lo que le da significado a ns, ya que sin él los eventos de subagente no cuentan con un espacio de nombres distintivo. Segundo, la rama solo conserva los eventos messages porque este bucle dirige tokens y no pasos. Tercero, la carga útil de cada fragmento se desempaca en un token junto con sus metadatos. Cuarto, la prueba de espacio de nombres realiza el trabajo real: un segmento tools: en cualquier parte de ns marca al token como proveniente de un subagente. Quinto, el valor booleano resultante se convierte en una etiqueta de origen legible. Sexto, imprimir con esa etiqueta como prefijo transforma un flujo desordenado en dos canales con atributos.
Cada token se etiqueta con su generador antes de ser impreso, por lo que cuando un coordinador lo transfiere a un investigador, aparecen como dos flujos con nombre. Una interfaz web completa es un proyecto más complejo, pero el espacio de nombres de cada fragmento es la misma señal que leería una interfaz frontal de React.
Los tokens aislados que no provienen del agente
Esto sorprende a las personas en su primera ejecución prolongada en streaming. Para gestionar su propio contexto, el sistema condensa los mensajes antiguos una vez que la conversación se acerca al límite del modelo. El paso de condensación también es una llamada al modelo, por lo que emite tokens que se integran en el flujo messages junto a la respuesta real del agente. Si ese flujo llega directamente al usuario, verán que el agente repite repentinamente un resumen de su propia historia, lo que parece un error.
Afortunadamente, los tokens están etiquetados. Cada evento messages cuenta con metadatos, y los tokens de resumen llevan un valor lc_source de "summarization". Omita esos y conserve el resto.
for chunk in agent.stream(
{”messages”: [{”role”: “user”, “content”: “...”}]},
stream_mode=”messages”,
version=”v2”,
):
token, metadata = chunk[”data”]
if metadata.get(”lc_source”) == “summarization”:
continue
# ... handle real agent tokens here
Basta con verificar los metadatos para distinguir al agente que habla con el usuario del sistema que se comunica consigo mismo. Solo se muestra el primero. Se trata de un trabajo sencillo que genera una mejora notable en la calidad del resultado.
Observar una ejecución versus mantener una conversación
A menudo se confunden la transmisión en tiempo real y la conversación, por lo que es necesario establecer una distinción clara. Todo lo anterior se refiere a la observación de cómo se desarrolla una ejecución. Cada llamada a stream, al igual que cada llamada a invoke, es por defecto de uso único. El agente realiza la tarea, la termina y la olvida, por lo que cualquier seguimiento a través de una nueva llamada comienza sin recordar el intercambio anterior.
Mantener una conversación abierta a través de turnos, de modo que la opción “arreglar también la otra prueba” funcione porque el agente recuerda la primera corrección, es un mecanismo separado. Este depende de hilos y un puntero de verificación, la capa de persistencia donde el grafo duradero escribe su estado. Diferencia importante: el flujo en tiempo real brinda visibilidad dentro de una ejecución, mientras que los hilos aseguran continuidad entre ejecuciones. Son independientes, y con frecuencia se necesitarán ambos.
Una pequeña interfaz de terminal que muestra el progreso
A continuación, reúna todo en una consola que pueda ejecutarse realmente. Debería imprimir una línea por paso a medida que el agente avanza y transmitir en tiempo real el texto que escribe; por eso se solicitan ambos modos al mismo tiempo y se toma una decisión según el valor de type del fragmento.
from deepagents import create_deep_agent
agent = create_deep_agent(
model="anthropic:claude-sonnet-4-6",
system_prompt="You are a code-maintenance assistant for a small Python repo.",
) # ①
prompt = "List the files you'd look at first to find a failing test, and explain why."
for chunk in agent.stream(
{"messages": [{"role": "user", "content": prompt}]},
stream_mode=["updates", "messages"], # ②
version="v2",
):
if chunk["type"] == "updates": # ③
for node_name in chunk["data"]:
print(f"\n[step] {node_name}")
elif chunk["type"] == "messages": # ④
token, metadata = chunk["data"]
if metadata.get("lc_source") == "summarization": # ⑤
continue
if token.content:
print(token.content, end="", flush=True) # ⑥
El código es breve, pero cada parte tiene una función específica. La estructura no ha cambiado, ya que create_deep_agent crea el agente exactamente como antes. Dos modos solicitados en una sola llamada permiten que un único bucle proporcione tanto un indicador de paso como un flujo de texto. Cuando un fragmento corresponde a un evento updates, el bucle imprime una línea [step] para el nodo que acaba de completarse. Cuando se trata de un evento messages, el bucle imprime el texto generado. La verificación de resumen del apartado anterior se encuentra dentro de esa rama, de modo que las tareas de mantenimiento del sistema nunca llegan al terminal, y los tokens vacíos se omiten antes de imprimirse.
El resultado es una línea [step] cada vez que el agente avanza, entrelazada con su texto que llega carácter por carácter, y el filtro ya está en vigor para que nada se filtre. No es bonito, pero es honesto. Un usuario puede ver que el agente está activo y observar cómo piensa.
Las versiones más recientes de Deep Agents incluyen una API de nivel superior para esto: agent.stream_events(...) en version="v3", con proyecciones tipadas como stream.subagents. La documentación orienta a las nuevas aplicaciones hacia esta opción, ya que cada proyección es su propio iterador ordenado y no hay necesidad de tomar decisiones basadas en stream_mode ni de decodificar nombres de espacio por sí mismo. Aún así, vale la pena aprender primero el enfoque de stream_mode, ya que las proyecciones se basan en estos mismos eventos. Consulte la documentación actual antes de decidirse por una interfaz compleja.
Una lista de verificación para aplicar hoy
- Comience buscando cada
invokepara el cual haya una persona esperando, y cámbielos aagent.stream(...)en modoupdates. Un registro aproximado de pasos sigue siendo mejor que una pantalla en blanco. - Examine un fragmento real. Ejécutelo con
version="v2", imprímalo y lea los campostype,nsydatapara conocer su estructura con precisión. - Antes de que los usuarios reales vean los resultados, elimine cualquier fragmento
messagescuya metadatos indiquenlc_source == "summarization". - Si el agente asigna tareas a subagentes, establezca
subgraphs=Truey ordene los resultados por espacio de nombres, con un canal por agente. - Mantenga una consola sencilla como la anterior en el repositorio. Le permite observar cómo se ejecuta cada nueva funcionalidad en lugar de tener que adivinar a partir del último mensaje.
Qué cambió y qué no
Resultado final: una llamada de bloqueo se convirtió en una transmisión en vivo. stream_mode determina el nivel de detalle que se obtiene, ns indica quién generó cada evento, el filtro de resumen elimina los comentarios propios del sistema, y ahora debería quedar claro que observar una ejecución y recordarla son problemas diferentes.
Nada cambió en el agente ni en su bucle. La única adición fue una ventana, y esa ventana marca la diferencia entre una herramienta en la que la gente confía y una que actualizan en pánico. Cuando un agente muestra su trabajo, los usuarios creen que está funcionando.
Así que la próxima vez que alguien informe que el agente “se queda atascado”, omite el perfilador e intenta primero una transmisión en vivo. El agente estaba hablando todo el tiempo; necesitaba un canal.
Leer más
- El sistema de ficheros de su agente es una decisión: explicación de los backends de agentes profundos — Cómo los backends de LangChain Deep Agents deciden qué se mantiene: cinco funciones integradas, el enrutamiento de CompositeBackend, la trampa del espacio de nombres para almacenes multiusuario y los patrones de sandbox.
- Agentes Profundos Gestionados en beta: una forma basada en directorios para desplegar agentes de LangChain — Los Agentes Profundos Gestionados de LangChain convierten una carpeta de proyecto en un agente alojado. Qué hace cada archivo, cómo funciona el despliegue de mda y las limitaciones de la versión beta que deben conocerse.