Transmitir LangGraph Agents a React sin revelar los detalles internos de la herramienta

Transmitir el progreso de un agente al navegador hace que una aplicación RAG parezca dinámica, pero enviar directamente el flujo de eventos sin procesar de LangGraph al cliente provoca dos problemas distintos. El flujo falla con objetos que JSON no puede codificar, y aquellos que sí logran pasar revelan esquemas, puntos de acceso internos y el razonamiento a cualquiera que tenga abierta la pestaña de red. Este artículo crea una pequeña capa de saneamiento en FastAPI que soluciona ambos problemas y muestra cómo un cliente React convierte esos eventos editados en insignias de estado legibles.

Por qué el estado bruto del agente rompe la serialización JSON

Durante la ejecución, el estado de LangGraph es una estructura en Python compleja, no un documento JSON. Enviar la salida de astream_events a json.dumps() o directamente a un StreamingResponse de FastAPI suele fallar porque los eventos contienen:

• Las clases de mensaje de LangChain como AIMessage, HumanMessage y ToolMessage, que son modelos Pydantic (v1 o v2, según la versión de LangChain) con sus propios métodos de serialización.
• Objetos Pydantic anidados que almacenan el razonamiento intermedio del agente.
• Objetos de excepción y registros de errores provenientes de llamadas a herramientas fallidas.

Cuando el codificador se encuentra con alguno de estos casos, genera una excepción, la conexión de Server-Sent Events se interrumpe y el frontend espera eternamente una respuesta que nunca llega.

Por qué las llamadas directas a herramientas representan un problema de seguridad

En un agente de estilo ReAct, el modelo emite una llamada a herramienta antes de que esta se ejecute, y ese payload suele ser sensible. Una herramienta de texto a SQL puede recibir el esquema completo de una tabla como underwriting_policies. Una herramienta de fijación de precios puede contener la URL interna y la estructura de parámetros de un microservicio privado. El propio proceso de razonamiento secuencial puede revelar reglas comerciales o instrucciones de los prompts del sistema. Al transmitirse sin filtrado, esto proporciona al atacante un mapa de tu backend; por lo tanto, los payloads deben eliminarse en el servidor antes de enviar cualquier cosa.

Un generador de sanitización en FastAPI

El generador que se muestra a continuación envuelve la secuencia de eventos. Para los eventos on_tool_start, emite un nuevo evento tool_update que contiene únicamente el nombre de la herramienta, un estado y un marcador de redacción. En cuanto a los fragmentos de mensaje, convierte las instancias BaseMessage en diccionarios mediante model_dump(). Todo lo demás se codifica con default=str como solución de respaldo, y cualquier fallo en la serialización genera un evento de error genérico en lugar de una excepción cruda. A pesar de su etiqueta original, el fragmento está escrito en Python.

import json
from typing import AsyncGenerator, Dict, Any
from langchain_core.messages import BaseMessage

async def sanitize_agent_stream(event_stream: AsyncGenerator[Dict[str, Any], None]) -> AsyncGenerator[str, None]:
    """
    Intercepts the LangGraph event stream, serializes objects, and scrubs tool payloads.
    """
    async for event in event_stream:
        # 1. Scrub Sensitive Tool Signatures
        if event.get("event") == "on_tool_start":
            tool_name = event.get("name")
            # We want to tell the UI a tool started, but hide the sensitive input
            scrubbed_event = {
                "event": "tool_update",
                "data": {
                    "tool": tool_name,
                    "status": "running",
                    "payload": "[REDACTED_FOR_SECURITY]" # Masked payload
                }
            }
            yield f"data: {json.dumps(scrubbed_event)}\n\n"
            continue

        # 2. Serialize LangChain Messages safely
        if "data" in event and "chunk" in event["data"]:
            chunk = event["data"]["chunk"]
            if isinstance(chunk, BaseMessage):
                # Safely convert to dict, ensuring content is serialized
                event["data"]["chunk"] = chunk.model_dump()

        # 3. Fallback Serialization for custom objects (simplified for example)
        try:
            # Attempt to serialize the cleaned event
            payload = json.dumps(event, default=str) # Fallback to string representation
            yield f"data: {payload}\n\n"
        except Exception as e:
            # Log error internally, do not stream raw exception to client
            print(f"Serialization Error: {e}")
            yield f"data: {json.dumps({'event': 'error', 'message': 'Internal processing error'})}\n\n"

Hay dos problemas que requieren atención antes de su uso en producción. Primero, solo se procesa on_tool_start. Eventos como on_tool_end, eventos en cadena que transportan el estado del grafo y los fragmentos de modelo cuyos tool_call_chunks contienen argumentos pasan directamente al método de respaldo y se transmiten sin cambios. Una lista de permisos, que solo envía los tipos de eventos y campos necesarios para la interfaz de usuario, es más segura que una lista de bloqueo. Segundo, reemplace print por un sistema de registro estructurado.

El endpoint inicia el grafo con astream_events utilizando el esquema de eventos v2, lo pasa por el sanitizador y lo devuelve como text/event-stream:

from fastapi import APIRouter
from fastapi.responses import StreamingResponse

router = APIRouter()

@router.post("/chat/stream")
async def stream_chat(request: ChatRequest):
    # Retrieve the LangGraph runnable
    graph = get_underwriting_graph()

    # Start the raw stream
    raw_stream = graph.astream_events(
        {"messages": [("user", request.message)]},
        version="v2"
    )

    # Pipe through the sanitization layer
    clean_stream = sanitize_agent_stream(raw_stream)

    return StreamingResponse(clean_stream, media_type="text/event-stream")

Asociación de eventos redactados con etiquetas en la UI en React

Dado que el cliente nunca ve las entradas de la herramienta, funciona únicamente con el nombre de esta. Una tabla de consulta asocia nombres internos como policy_db_search con texto visible para el usuario, y existe un valor por defecto genérico para herramientas desconocidas. Un evento tool_update muestra una insignia que parpadea, y el primer fragmento on_chat_model_stream la elimina, ya que el agente ha comenzado a responder.

import React, { useState, useEffect } from 'react';

// Map internal tool names to professional UI labels
const TOOL_UI_MAPPING: Record<string, string> = {
  actuarial_sandbox_tool: "Calculating Premium Factors...",
  policy_db_search: "Querying Policy Guidelines...",
  risk_model_evaluator: "Evaluating Risk Profile...",
};

export const AgentChatInterface = () => {
  const [activeTool, setActiveTool] = useState<string | null>(null);
  const [messages, setMessages] = useState<string[]>([]);

  useEffect(() => {
    const eventSource = new EventSource('/api/chat/stream');

    eventSource.onmessage = (event) => {
      const parsed = JSON.parse(event.data);

      if (parsed.event === 'tool_update') {
        // Display the user-friendly name instead of internal mechanics
        const uiLabel = TOOL_UI_MAPPING[parsed.data.tool] || "Processing Data...";
        setActiveTool(uiLabel);
      } else if (parsed.event === 'on_chat_model_stream') {
        setActiveTool(null); // Clear tool status when agent speaks
        // Append text chunk to messages...
      }
    };

    return () => eventSource.close();
  }, []);

  return (
    <div className="chat-container">
      {/* Active Tool Badge */}
      {activeTool && (
        <div className="tool-badge bg-blue-100 text-blue-800 px-3 py-1 rounded-full text-sm font-semibold animate-pulse">
          {activeTool}
        </div>
      )}

      {/* Chat Messages */}
      <div className="message-list">
        {/* Render messages here */}
      </div>
    </div>
  );
};

Hay una corrección que aplicar aquí: EventSource del navegador solo envía solicitudes GET y no puede enviar un cuerpo, mientras que el endpoint es una ruta POST que espera un mensaje. En la práctica, deberías leer la respuesta con fetch y un lector de flujo, o utilizar una biblioteca auxiliar que analice SSE a través de POST. También verifica que el prefijo /api coincida con la configuración del proxy de Vite.

Puntos clave

• Los eventos de LangGraph contienen objetos que JSON no puede codificar; convierta los mensajes de forma explícita y nunca permita que una excepción del codificador llegue al cliente.
• Los argumentos de las herramientas, los esquemas y el razonamiento son secretos del lado del servidor. Redactelos antes de transmitirlos, preferiblemente estableciendo una lista de lo que necesita la interfaz de usuario.
• Un mapa de nombre de herramienta a etiqueta mantiene la interfaz informativa sin revelar los detalles internos.
• Ajuste el transporte de su cliente al punto final: EventSource es adecuado para flujos GET, mientras que los flujos POST requieren fetch.

Lecturas relacionadas

• Anatomía de un equipo comercial de IA: Orquestando agentes con LangGraph y FastAPI — Una explicación detallada de una plataforma multiagente de código abierto: cómo se orquestan los agentes cofundador, gerente y especialista, comparten memoria, se detienen para obtener aprobación y generan informes.
• Incrustar un agente de LangChain en FastAPI: Herramientas, búsqueda manual, transmisión en tiempo real — Crea un asistente integrado en la aplicación con FastAPI y LangChain: un manual en PDF almacenado en ChromaDB que se expone como herramienta, contexto por usuario, historial con puntos de control y respuestas en transmisión en tiempo real.