Diffuser les agents LangGraph vers React sans divulguer les détails internes de l’outil

La diffusion des progrès d’un agent vers le navigateur donne l’impression que l’application RAG est dynamique, mais l’envoi direct du flux d’événements bruts de LangGraph vers le client entraîne deux problèmes distincts. Le flux s’arrête en raison d’objets que JSON ne peut pas encoder, et ce qui parvient à être transmis expose les schémas, les points de terminaison internes ainsi que le processus de raisonnement à quiconque a la page réseau ouverte. Cet article crée une petite couche de nettoyage dans FastAPI qui résout ces deux problèmes, puis montre comment un client React transforme ces événements modifiés en badges d’état lisible.

Pourquoi l’état brut de l’agent perturbe la sérialisation JSON

Pendant l’exécution, l’état de LangGraph est une structure Python complexe, et non un document JSON. L’envoi du résultat de astream_events vers json.dumps() ou directement dans une StreamingResponse de FastAPI échoue généralement parce que les événements contiennent :

• Les classes de messages LangChain telles que AIMessage, HumanMessage et ToolMessage, qui sont des modèles Pydantic (v1 ou v2, selon la version de LangChain) disposant de leurs propres méthodes de sérialisation.
• Des objets Pydantic imbriqués contenant les raisonnements intermédiaires de l’agent.
• Des objets d’exception et des traces d’erreurs provenant de appels d’outils infructueux.

Lorsque l’encodeur rencontre l’un de ces éléments, il génère une exception, la connexion Server-Sent Events est interrompue, et l’interface utilisateur attend indéfiniment une réponse qui ne vient jamais.

Pourquoi les appels d’outils bruts posent un problème de sécurité

Dans un agent de type ReAct, le modèle émet une requête d’outil avant que l’outil ne s’exécute, et ce chargement est souvent sensible. Un outil texte vers SQL peut recevoir l’ensemble du schéma d’une table telle que underwriting_policies. Un outil de tarification peut contenir l’URL interne ainsi que la structure des paramètres d’un microservice privé. Le processus de réflexion elle-même peut révéler des règles commerciales ou des instructions de prompt système. Si ces données sont transmises sans filtrage, elles fournissent à un attaquant une carte de votre backend ; par conséquent, les chargements doivent être supprimés sur le serveur avant tout envoi.

Un générateur de nettoyage dans FastAPI

Le générateur ci-dessous enrobe le flux d’événements. Pour les événements on_tool_start, il émet un nouvel événement tool_update ne contenant que le nom de l’outil, un état et un marqueur de censure. Pour les blocs de messages, il convertit les instances BaseMessage en dictionnaires à l’aide de model_dump(). Tout le reste est codé avec default=str en tant que solution de rechange, et toute erreur de sérialisation génère un événement d’erreur générique plutôt qu’une exception brute. Ce fragment est en Python, malgré son étiquette initiale.

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"

Deux problèmes nécessitent une attention avant utilisation en production. Premièrement, seuls les événements on_tool_start sont filtrés. Des événements tels que on_tool_end, les événements en chaîne transportant l’état du graphe, ainsi que les blocs de modèle dont tool_call_chunks contiennent des arguments, parviennent tous au mécanisme de secours et sont transmis tels quels. Une liste d’autorisation, ne transférant que les types d’événements et les champs nécessaires à l’interface utilisateur, est plus sûre qu’une liste de blocage. Deuxièmement, il convient de remplacer print par un système d’journalisation structuré.

L’endpoint initialise le graphe avec astream_events en utilisant le schéma d’événements v2, le fait passer par le filtre de nettoyage, puis le renvoie sous forme 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")

Mappage des événements masqués vers les étiquettes de l’UI dans React

Puisque le client ne voit jamais les entrées de l’outil, cela fonctionne uniquement à partir du nom de l’outil. Une table de correspondance associe les noms internes tels que policy_db_search à du texte destiné à l’utilisateur, avec une solution de repli générique pour les outils inconnus. Un événement tool_update affiche une étiquette clignotante, et le premier chunk de on_chat_model_stream la supprime, car l’agent a commencé à répondre.

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

Une correction s’impose ici : EventSource du navigateur ne génère que des requêtes GET et ne peut pas envoyer de corps, tandis que l’endpoint est une route POST qui attend un message. En pratique, vous devriez soit lire la réponse avec fetch et un lecteur de flux, soit utiliser une bibliothèque d’aide qui analyse SSE via POST. Vérifiez également que le préfixe /api correspond à votre configuration de proxy Vite.

Points clés

• Les événements LangGraph contiennent des objets que JSON ne peut pas encoder ; convertissez les messages explicitement et empêchez à tout prix qu’une exception d’encodage parvienne au client.
• Les arguments des outils, leurs schémas ainsi que leur logique de raisonnement constituent des secrets côté serveur. Masquez-les avant la diffusion, de préférence en définissant une liste autorisée des éléments nécessaires à l’interface utilisateur.
• Un tableau de correspondance entre le nom de l’outil et son étiquette permet à l’interface d’être informative sans révéler les détails internes.
• Adaptez le mécanisme de transport de votre client au point de terminaison : EventSource convient aux flux GET, tandis que les flux POST nécessitent fetch.