Fiabilidad de LangChain: Transmisión en tiempo real, agrupamiento, intentos repetidos, soluciones alternativas, middleware

Hasta ahora, el agente del servicio de asistencia ha tenido una libertad deliberadamente limitada. Puede consultar la asignación de servicios y los incidentes recientes, pero antes de redactar una nota de ticket debe detenerse, exponer los argumentos que pretende utilizar y esperar a una persona. Este arreglo funciona siempre y cuando cada llamada a modelos o herramientas responda de inmediato. En la práctica, no es así.

Una solicitud a un modelo puede tardar quince segundos. Un proveedor podría bloquear el tráfico durante un minuto completo debido a límites de uso. Un único ticket en un lote podría contener datos inválidos. Un modelo de respaldo podría responder de manera diferente al principal, y un usuario podría cerrar el navegador mientras una herramienta aprobada está en ejecución. Cada uno de estos casos merece un comportamiento definido respaldado por una prueba.

Este paso añade cinco restricciones alrededor del agente. Para una sola solicitud:

one request
  -> stream honest progress
  -> retry a small transient failure
  -> use a labeled fallback if attempts are exhausted
  -> stop at time and call limits

Para un montón de solicitudes no relacionadas:

many independent requests
  -> run a bounded batch
  -> keep one result or error beside each input

La regla constante: no oculte un fallo detrás de una respuesta alegre.

El ticket con el que seguimos probando

El incidente de prueba no cambia: las solicitudes de pago en la UE dejan de funcionar después del despliegue a las 14:05, con el mensaje “database connection refused” en los registros de checkout-api. El usuario pide al agente que busque al responsable y los incidentes recientes, sugiera el siguiente paso y añada una nota a INC-2048 después de que él dé su aprobación.

En el escenario ideal, el agente lee al responsable y los incidentes recientes, escribe una recomendación, propone una nota, espera la aprobación, escribe una vez y devuelve el resultado. Hagámoslo más difícil: envíe tres tickets más al mismo tiempo, haga que una llamada al proveedor falle dos veces, convierta un ticket en inválido y interrumpa la conexión con el cliente a mitad del primer pedido. Todo lo que sigue está diseñado para hacer frente exactamente a eso.

Dé primero nombres a los fallos

“La llamada falló” no indica qué hacer. Clasifique las fallas en categorías que indiquen acciones diferentes.

• Falla temporal del proveedor. Una reconexión, un tiempo de espera agotado o una respuesta del proveedor que pide claramente volver más tarde. Intente nuevamente varias veces con un intervalo entre cada intento.
• Límite de frecuencia. El proveedor está protegiendo una cuota o su capacidad. Siga cualquier información de reintentos que envíe, reduzca la velocidad de las solicitudes y disminuya la concurrencia. Intentar de inmediato solo empeora la situación.
• Entrada inválida. El ticket carece de ID, el esquema no coincide o el texto supera los límites establecidos por la aplicación. Responda con un error claro para el cliente y no envíe nuevamente la misma solicitud mal formada.
• Falla de autorización. La persona que realiza la llamada no tiene permiso para leer un manual de procedimientos ni escribir en el ticket. Deténgase ahí; no cambie las credenciales ni pida a otro modelo que solucione el rechazo.

Error del negocio de la herramienta. La solicitud ya está cerrada, no se permite la transición o existe ya una misma nota. Convierta la respuesta de la herramienta en un error de aplicación significativo, ya que intentar nuevamente podría resultar incorrecto.

Fallo en la validación de la salida del modelo. El proveedor devolvió datos que violan el contrato de salida. En el caso de un resultado de solo lectura, un intento de reparación puede ser aceptable, pero aún debe existir una ruta de fallo final.

Fallo desconocido. La aplicación no pudo clasificar el error. Cierre el fallo, registre el tipo de error e investigue; nunca trate cada excepción como temporal.

Juntas, estas categorías determinan qué fallos se intentarán nuevamente, cuáles podrán recurrir a alternativas y cuáles irán directamente al usuario que realizó la solicitud.

Diseño e instalación del proyecto

La captura de estado para este paso contiene:

langchain-helpdesk/
├── app.py
├── reliability.py
├── batch_runner.py
├── stream_view.py
├── middleware_config.py
└── tests/
    ├── test_reliability.py
    └── test_stream_view.py

Instale los paquetes:

python -m pip install -U \
  langchain \
  langchain-core \
  langgraph \
  pytest

Las pruebas nunca llaman a un proveedor.

La transmisión en tiempo real entrega contenido; no lo verifica

Una pantalla en blanco mientras se ejecutan las herramientas y el modelo trabaja es una mala experiencia. La transmisión en tiempo real soluciona esto emitiendo eventos a medida que avanza el trabajo. La transmisión en tiempo real del agente actual de LangChain le ofrece tres vistas:

• updates: pasos completados del agente
• messages: tokens del modelo más metadatos
• custom: datos de progreso definidos por su aplicación

Cada una contiene diferentes tipos de información, así que no muestre todas como si fueran la misma respuesta.

Un bucle básico de transmisión en tiempo real

for chunk in agent.stream(
    {
        "messages": [
            {
                "role": "user",
                "content": request_text,
            }
        ]
    },
    config,
    stream_mode=["updates", "custom"],
    version="v2",
):
    send_to_client(chunk)

Análisis detallado: agent.stream(...) inicia la misma ejecución que haría invoke(...), pero genera fragmentos mientras se ejecuta. El primer argumento es la actualización de estado, que en este caso es un mensaje del usuario. config almacena configuraciones específicas de la solicitud, como el ID del hilo, y los secretos no deben incluirse en él. stream_mode=["updates", "custom"] solicita datos de los pasos completados además de los mensajes de progreso personalizados, por lo que en este ejemplo no se devuelve texto token por token. version="v2" selecciona el formato actual de los fragmentos, que incluye un tipo, un espacio de nombres y un payload de datos en cada uno. send_to_client(chunk) corresponde a tu propio código, el cual puede emitir Server-Sent Events, un mensaje WebSocket o una línea de terminal.

Traducir fragmentos brutos para la interfaz de usuario

La captura de estado convierte los fragmentos antes de que lleguen a la interfaz de usuario:

@dataclass(frozen=True)
class ProgressEvent:
    kind: str
    label: str
    payload: Any

def readable_events(chunks):
    for chunk in chunks:
        chunk_type = chunk.get("type")
        data = chunk.get("data")        if chunk_type == "updates" and isinstance(data, dict):
            for step_name, step_data in data.items():
                yield ProgressEvent(
                    kind="step",
                    label=step_name,
                    payload=step_data,
                )

Ese adaptador proporciona a la interfaz de usuario un contrato pequeño y estable. Puede mostrar líneas como estas:

Checking service ownership...
Reading recent incidents...
Waiting for ticket-note approval...
Ticket note approved and written.

Lo que no debe mostrarse antes de que la ejecución haya finalizado realmente es lo siguiente:

Done.

El texto parcial no es el texto final

Los tokens de transmisión en tiempo real hacen que la respuesta parezca más rápida, pero no añaden nada a la seguridad ni a la precisión. El modelo puede comenzar una oración y luego dirigirse a otro lugar, una herramienta puede fallar después de que los tokens ya se hayan enviado, y una verificación de políticas puede vetar la acción propuesta.

Use estados de evento explícitos: running, waiting_for_approval, completed, failed y cancelled. Solo completed indica un resultado final.

Décida qué significa una desconexión

Un navegador cerrado no demuestra que el servidor se haya detenido. Defina la política con antelación:

• Cancele las tareas de solo lectura cuando exista una cancelación y nadie esté esperando la respuesta.
• Deje que una tarea duradera aprobada se complete si el usuario planea volver por ella.
• Nunca asuma que una operación de escritura no ocurrió solo porque su respuesta se perdió.

Busque la operación por su clave de idempotencia antes de intentar realizar la escritura nuevamente.

La transmisión en tiempo real y la cancelación están relacionadas, pero no constituyen una misma función.

Batching de entradas independientes

Los ejecutables de LangChain exponen batch y abatch. El valor predeterminado de batch ejecuta llamadas independientes en un pool de hilos, lo cual es adecuado para muchas operaciones dependientes de la red; además, un ejecutable específico del proveedor puede reemplazarlo por una API de batch nativa. El trabajador del servicio de atención al cliente recibe varios tickets no relacionados:

outputs = runnable.batch(
    tickets,
    config={"max_concurrency": 4},
    return_exceptions=True,
)

tickets es la lista ordenada de entradas. max_concurrency limita cuántas llamadas se ejecutan simultáneamente, y cuatro es solo un ejemplo; determine su valor según las cuotas del proveedor, la capacidad de la herramienta, el costo y la latencia medida. Con return_exceptions=True, si un elemento falla, eso ya no anula los resultados de los demás. Posteriormente, la aplicación asocia cada salida con su entrada correspondiente:

for ticket, output in zip(tickets, outputs, strict=True):
    if isinstance(output, BaseException):
        save_failure(ticket.ticket_id, type(output).__name__)
    else:
        save_recommendation(ticket.ticket_id, output)

Eso garantiza un solo resultado por ticket.

Omite el procesamiento en lotes paralelos en estos casos: el segundo elemento depende del primero; cada elemento actualiza el mismo registro sin control de concurrencia; el orden de ejecución es en sí una regla de negocio; el proveedor utiliza transacciones masivas con garantías diferentes; o un usuario puede enviar suficientes elementos como para agotar la capacidad compartida. El procesamiento en lotes solo redistribuye la carga, pero no la elimina.

Reintenta con el límite más seguro

Reintentar significa realizar la tarea nuevamente, así que ten cuidado con lo que repitas. Ejecutar de nuevo todo el flujo de trabajo puede hacer que se vuelvan a realizar las búsquedas, las llamadas al modelo y la escritura en el ticket. Un enfoque más seguro consiste en reintentar solo el paso de lectura que se sabe que falla de forma temporal. La captura de estado declara un error específico para ello:

class TemporaryProviderError(RuntimeError):
    """A short-lived error that may succeed when tried again."""

y envuelve un único elemento ejecutable:

primary_call = RunnableLambda(primary).with_retry(
    retry_if_exception_type=(TemporaryProviderError,),
    wait_exponential_jitter=False,
    stop_after_attempt=3,
)

RunnableLambda(primary) convierte una función normal de Python en un objeto ejecutable. retry_if_exception_type funciona como una lista de permisos, por lo que un ticket inválido no coincide y vuelve directamente al llamante. Al establecer wait_exponential_jitter=False, la prueba sin conexión se realiza de forma rápida y determinista. stop_after_attempt=3 cuenta cada intento, es decir, tres intentos en total y no tres reintentos adicionales al primero. Frente a un servicio de red real, el retroceso exponencial con fluctuaciones suele ser la mejor opción, ya que evita que todos los trabajadores accedan al servicio al mismo tiempo.

Las escrituras no son seguras para ser reintentadas ciegamente

Imagínese que el servicio de tickets agrega la nota y luego la conexión se interrumpe antes de que llegue la respuesta de éxito. El usuario ve un tiempo de espera, pero la escritura podría ya existir, y un intento de reintentar de forma ingenua agregaría la nota dos veces. Utilice una clave de idempotencia:

operation_id = "ticket-note:INC-2048:approval-7331"

El servicio de tickets guarda esa clave junto con la primera escritura exitosa. Si aparece la misma clave nuevamente, devuelve el resultado almacenado y no aplica el cambio una segunda vez. Cuando el sistema externo no puede ser idempotente, verifica su estado actual antes de intentarlo de nuevo y acepta que algunas situaciones de fallo no pueden resolverse por completo.

Una solución alternativa es seguir otro camino

Una vez que se agotan los intentos permitidos, un programa de solución alternativa puede hacerse cargo:

reliable = primary_call.with_fallbacks(
    [RunnableLambda(conservative_fallback)],
    exceptions_to_handle=(TemporaryProviderError,),
)

Solo se ejecuta para el error temporal especificado, y lo que devuelve es esto:

Recommendation(
    ticket_id=ticket.ticket_id,
    text=(
        "The recommendation service is temporarily unavailable. "
        "Keep the ticket open and compare the failed deployment "
        "with the last known-good release."
    ),
    source="fallback",
)

La respuesta admite lo que sucedió. No afirma que se buscaran los manuales de procedimiento ni que se actualizara el ticket.

Cambiar a otro modelo tampoco es lo mismo que reiniciar el primero. El segundo modelo puede comportarse de manera diferente al utilizar herramientas, tener un límite de contexto más reducido, soportar la salida estructurada de forma distinta, aplicar otros comportamientos de seguridad, diferir en latencia y precio, o estar sujeto a una política de datos diferente del proveedor. Trate la ruta de respaldo como una configuración independiente y pruébela por separado. A veces, el mejor respaldo es no utilizar ningún modelo en absoluto, sino simplemente un mensaje fijo y conservador o una cola para revisión humana.

Los tiempos de espera existen en varios niveles

Un único valor de tiempo de espera no es suficiente. Asigne presupuestos separados para la configuración de conexiones, una solicitud de modelo individual, una solicitud de herramienta individual, el funcionamiento completo del agente, la espera para obtener aprobación y la conexión con el cliente. El presupuesto total debe superar al de cualquier llamada individual, pero aún así necesita tener un límite.

Coloque el comportamiento de tiempo de espera junto a la operación que limita: los tiempos de espera del proveedor en el cliente del proveedor, los tiempos de espera de la herramienta en el cliente de la herramienta, y el plazo para las solicitudes web en el límite del servicio. Cuando se active alguno de ellos, registre qué capa puso fin al proceso. Un simple mensaje de “tiempo agotado” sin especificar el límite hace difícil la solución del problema.

El middleware envuelve las llamadas al agente

Los pasos anteriores utilizaron principalmente cadenas fijas, y el paso de aprobación introdujo un middleware de agente. El middleware también es un lugar adecuado para las reglas que deben abarcar muchas llamadas a modelos o herramientas: volver a intentar las fallas en los modelos seleccionados, limitar el número de llamadas a modelos, limitar el número de llamadas a herramientas, filtrar herramientas según el usuario actual, agregar metadatos de solicitud para registros y trazas, ocultar ciertos datos personales y detenerse cuando se agota el presupuesto de la aplicación. La captura de estado utiliza el middleware integrado:

def reliability_middleware():
    return [
        ModelRetryMiddleware(
            max_retries=2,
            retry_on=(
                TemporaryProviderError,
                TimeoutError,
                ConnectionError,
            ),
            on_failure="error",
            initial_delay=0.5,
            backoff_factor=2.0,
            max_delay=4.0,
            jitter=True,
        ),
        ModelCallLimitMiddleware(
            run_limit=8,
            exit_behavior="error",
        ),
        ToolCallLimitMiddleware(
            run_limit=6,
            exit_behavior="end",
        ),
    ]

ModelRetryMiddleware rodea las llamadas al modelo, y una llamada inicial más dos intentos de reintentar dan como máximo tres tentativas. ModelCallLimitMiddleware impide que un bucle de modelos se ejecute indefinidamente. ToolCallLimitMiddleware detiene el uso excesivo de herramientas durante una ejecución. Estos controles realizan tareas separadas: un límite de llamadas no sustituye a un tiempo de espera, un tiempo de espera no sustituye a la aprobación y la aprobación no sustituye a la autorización.

El orden es importante

Los ganchos que se ejecutan antes de una llamada lo hacen en orden de lista, aquellos que se ejecutan después lo hacen en orden inverso, y los ganchos anidados se estructuran de la misma manera que las llamadas a funciones. Cuando el middleware A envuelve a B, un reintento en A puede volver a ejecutar el trabajo de B, por lo que es necesario planificar el orden siempre que dos componentes puedan reintentar, cambiar modelos o modificar el estado. La aprobación de una escritura debe realizarse fuera de cualquier reintento que pueda repetir el efecto secundario ya aprobado.

Los ganchos personalizados son simplemente código

LangChain ofrece ganchos de tipo nodo (before_model, after_model) y ganchos de tipo envoltura (wrap_model_call, wrap_tool_call). Los ganchos de tipo nodo son adecuados para realizar verificaciones y actualizaciones de estado en puntos fijos. Los ganchos de tipo envoltura se utilizan para código que debe rodear una llamada, o que decide si el manejador se ejecuta cero, una o varias veces. Sin embargo, no toda regla de la aplicación debería convertirse en middleware. Las transiciones de estado de los tickets, las decisiones de autorización y las reglas de propiedad de datos son más fáciles de gestionar como servicios de aplicación con nombre específico.

Pruebas de los caminos de fallo sin conexión

El snapshot cuenta con cinco pruebas, ninguna de las cuales requiere un proveedor.

Un error temporal tiene éxito en el segundo intento:

result = runnable.invoke(ticket())

assert calls == 2
assert result.source == "primary"

Los intentos fallidos utilizan la solución de respaldo indicada:

assert calls == 3
assert result.source == "fallback"
assert "temporarily unavailable" in result.text

La entrada inválida no se vuelve a intentar ni se oculta:

with pytest.raises(InvalidTicketError):
    runnable.invoke(ticket())

assert calls == 1

Un fracaso en un lote no anula otro éxito:

assert results[0].status == "completed"
assert results[1].status == "failed"
assert results[1].error_type == "RuntimeError"

El progreso se mantiene separado de los tokens del modelo:

assert [(event.kind, event.label) for event in events] == [
    ("step", "model"),
    ("progress", "custom"),
    ("token", "model"),
]

Ejecútalos:

pytest -q

Resultado esperado:

5 passed

Estas pruebas demuestran las reglas de control locales. No prueban que los errores del proveedor se mapeen correctamente, lo cual requiere una pequeña prueba de integración con el SDK real del proveedor y una cuenta de prueba controlada.

Escribe la matriz de fallos

Incluye una tabla como esta en el código o la documentación antes del lanzamiento:

failure                         retry?   fallback?   client result
connection reset               yes      maybe       503 if exhausted
provider rate limit            yes      maybe       503 or queued
invalid ticket                 no       no          422
authorization denied           no       no          403
structured output invalid      limited  maybe       502 if exhausted
ticket already closed          no       no          409
unknown exception              no       no          500 + incident ID

Los códigos de estado exactos dependen de su API. Lo importante es que cada categoría de fallo tenga una política definida.

Todavía faltan cosas

El comportamiento en tiempo de ejecución es ahora más claro, pero esto sigue siendo un resumen didáctico desde la línea de comandos. No cuenta con una API HTTP autenticada, ni verificaciones de inicio o disponibilidad, ni carga de credenciales secretas, ni registros estructurados de solicitudes, ni almacenamiento duradero de tareas, ni configuración de despliegue, ni pruebas de integración con proveedores reales, ni conjunto de evaluación revisado antes del lanzamiento, ni revisión de trazas en producción. La versión que incluye todo esto vendrá después.

Situación actual del flujo de soporte técnico

El flujo ahora cuenta con estados visibles y una recuperación limitada:

request
  -> progress events
  -> small retry boundary
  -> labeled fallback when allowed
  -> model and tool call limits
  -> approval before write
  -> final result or named failure

batch
  -> bounded concurrency
  -> one outcome per ticket

Ningún control elimina la incertidumbre. El objetivo es mantenerla visible, delimitada y verificable. A continuación se presenta todo el flujo de trabajo del servicio de soporte detrás de una API, que incluye configuraciones, datos confidenciales, IDs de solicitud, tiempos de espera, registros y mecanismos de seguimiento, seguidos de pruebas deterministas, casos de evaluación y verificaciones de despliegue, además de una lista de lo que una empresa real debe proporcionar antes de que el servicio maneje tickets reales. Todo lo anterior fue verificado contra la documentación en Python de LangChain sobre transmisión en tiempo real, intentos repetibles ejecutables, soluciones alternativas, procesamiento por lotes, middleware y pruebas el 31 de agosto de 2026; las APIs del paquete y los tipos de error de los proveedores pueden cambiar.

Lecturas relacionadas

• LangChain @dynamic_prompt para instrucciones de sistema conscientes de la tarea — Crea un agente de recursos humanos que cambia las instrucciones del sistema relacionadas con licencias, beneficios y administración según el contexto en tiempo de ejecución, manteniendo al mismo tiempo la autorización en el código de la aplicación.
• Herramientas de Gate Agent por usuario con LangChain Wrap Middleware — Un agente calculador con tres niveles de acceso: el middleware de tipo wrap sobrescribe la lista de herramientas según el tipo de usuario en tiempo de ejecución, en lugar de crear tres agentes separados.