LangGraph Agents in React streamen, ohne die internen Tools preiszugeben

Das Streamen des Fortschritts eines Agenten in den Browser verleiht einer RAG-Anwendung das Gefühl von Lebendigkeit, doch das Direktübertragen des rohen Ereignisstroms von LangGraph an den Client führt zu zwei separaten Problemen. Der Stream bricht bei Objekten ab, die JSON nicht kodieren kann, und alles, was dennoch durchkommt, macht Schemata, interne Endpunkte sowie die Denkprozesse für jeden sichtbar, der das Netzwerkfenster geöffnet hat. Dieser Artikel erstellt eine kleine Sanitierungsschicht in FastAPI, die beides behebt, und zeigt anschließend, wie ein React-Client die redigierten Ereignisse in lesbare Statusanzeigen umwandelt.

Warum der rohe Agentenzustand die JSON-Serialisierung stört

Während der Ausführung ist der Zustand von LangGraph eine komplexe Python-Struktur, kein JSON-Dokument. Das Senden der Ausgabe von astream_events an json.dumps() oder direkt in eine FastAPI-StreamingResponse scheitert in der Regel, weil die Ereignisse Folgendes enthalten:

• LangChain-Nachrichtsklassen wie AIMessage, HumanMessage und ToolMessage, die Pydantic-Modelle sind (Version v1 oder v2, je nach LangChain-Version) und über eigene Serialisierungsverfahren verfügen.
• Eingebettete Pydantic-Objekte, die das Zwischenergebnis der Agentenlogik enthalten.
• Exception-Objekte sowie Tracebacks von fehlgeschlagenen Tool-Aufrufen.

Sobald der Encoder auf eines dieser Objekte stößt, wird eine Ausnahme ausgelöst, die Server-Sent Events-Verbindung wird unterbrochen, und die Frontend-Anwendung wartet endlos auf eine Antwort, die niemals eintrifft.

Warum rohe Tool-Aufrufe ein Sicherheitsproblem darstellen

In einem ReAct-stiligen Agenten sendet das Modell einen Tool-Aufruf aus, bevor das Tool ausgeführt wird, und dieser Payload ist oft sensibel. Ein Text-zu-SQL-Tool kann das vollständige Schema einer Tabelle wie underwriting_policies erhalten. Ein Preismodul kann die interne URL sowie die Parameterstruktur eines privaten Microservices enthalten. Der Chain-of-Thought selbst kann Geschäftsregeln oder Anweisungen für Systemanfragen offenbaren. Wenn dieser unfiltert übertragen wird, erhält ein Angreifer eine Übersicht Ihres Backends – daher müssen die Payloads auf dem Server entfernt werden, bevor etwas gesendet wird.

Ein Sanitisierungs-Generator in FastAPI

Der Generator unten umhüllt den Ereignisstrom. Für on_tool_start-Ereignisse gibt er ein neues tool_update-Ereignis aus, das lediglich den Toolnamen, einen Status sowie einen Redaktionsmarker enthält. Für Nachrichtsteile wandelt er BaseMessage-Instanzen mithilfe von model_dump() in Wörterbücher um. Alles Weitere wird als fallback mit default=str kodiert, und bei einer Serialisierungsfehler entsteht ein generisches Fehlerereignis anstelle einer rohen Ausnahme. Der Codeausschnitt ist zwar unter einem anderen Label gelistet, handelt sich aber um 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"

Zwei Probleme erfordern vor der Produktionseinsatz eine Behandlung. Erstens wird nur on_tool_start gefiltert. Ereignisse wie on_tool_end, Kettevorgänge, die den Graphenzustand übertragen, sowie Modellblöcke, deren tool_call_chunks Argumente enthalten, gelangen unverändert zum Fallback und werden weitergestreamt. Eine Allowlist, die nur die vom UI benötigten Ereignistypen und Felder weiterleitet, ist sicherer als eine Blocklist. Zweitens sollte print durch strukturiertes Logging ersetzt werden.

Der Endpunkt startet den Graphen mit astream_events unter Verwendung des v2-Ereignisschemas, leitet ihn durch den Sanitizer und gibt ihn als text/event-stream zurück:

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")

Kopplung von redigierten Ereignissen an UI-Bezeichnungen in React

Weil der Client die Eingaben der Tools niemals sieht, funktioniert es bereits anhand des Toolnamens allein. Eine Abfrageschlüsseltabelle ordnet interne Namen wie policy_db_search benutzerfreundlichem Text zu und bietet außerdem einen allgemeinen Ersatz für unbekannte Tools. Ein tool_update-Ereignis zeigt ein pulsierendes Symbol an, und der erste on_chat_model_stream-Datensatz löscht dieses, da der Agent mit der Beantwortung begonnen hat.

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>
  );
};

Es gibt hier eine Korrektur zu machen: Der EventSource des Browsers sendet nur GET-Anfragen und kann keinen Datenkörper übertragen, während das Endpunkt eine POST-Route ist, die eine Nachricht erwartet. In der Praxis sollte man entweder die Antwort mit fetch und einem Stream-Reader lesen oder eine Hilfsbibliothek verwenden, die SSE über POST parsen kann. Überprüfen Sie außerdem, ob das /api-Präfix mit Ihrer Vite-Proxy-Konfiguration übereinstimmt.

Haupterkenntnisse

• LangGraph-Ereignisse enthalten Objekte, die JSON nicht kodieren kann; konvertieren Sie die Nachrichten explizit und lassen Sie niemals eine Encoder-Exception zum Client gelangen.
• Tool-Argumente, Schemata und Berechnungslogiken sind Server-seitige Geheimnisse. Entfernen Sie diese vor dem Streamen, am besten indem Sie nur das freigeben, was die Benutzeroberfläche benötigt.
• Eine Karte von Tool-Name zu Label sorgt dafür, dass die Benutzeroberfläche informativ bleibt, ohne interne Details preiszugeben.
• Passen Sie den Transport Ihres Clients zum Endpunkt an: EventSource eignet sich für GET-Streams, während POST-Streams fetch erfordern.