Strumieniowanie agentów LangGraph do React bez ujawniania wewnętrznych szczegółów narzędzia

Przesyłanie postępów agenta do przeglądarki sprawia, że aplikacja typu RAG wydaje się żywa, ale przekazywanie surowego strumienia zdarzeń LangGraph bezpośrednio do klienta powoduje dwa odrębne problemy. Strumień zawiesza się przy obiektach, których JSON nie może zekodować, a to, co uda się przesłać, ujawnia schematy, wewnętrzne punkty końcowe oraz procesy rozumowania każdemu, kto ma otwartą kartę sieci. Ten artykuł opisuje budowę niewielkiej warstwy oczyszczania w FastAPI, która rozwiązuje oba te problemy, a następnie pokazuje, jak klient React przekształca zredagowane zdarzenia w czytelne ikony stanu.

Dlaczego surowy stan agenta psuje serializację JSON

Podczas wykonywania stan LangGraph jest złożoną strukturą w Pythonie, a nie dokumentem JSON. Przesyłanie wyniku funkcji astream_events do json.dumps() lub bezpośrednio do FastAPI StreamingResponse zazwyczaj kończy się niepowodzeniem, ponieważ zdarzenia zawierają:

• Klasy wiadomości LangChain takie jak AIMessage, HumanMessage i ToolMessage, które są modelami Pydantic (wersja 1 lub 2, w zależności od wersji LangChain) posiadającymi własne metody serializacji.
• Zagnieżdżone obiekty Pydantic przechowujące pośrednie rozumowanie agenta.
• Obiekty wyjątków oraz informacje o błędach po nieudanych wywołaniach narzędzi.

Gdy kod kodujący napotka jeden z tych przypadków, wywołuje błąd, połączenie typu Server-Sent Events zostaje przerwane, a interfejs użytkownika czeka w nieskończoność na odpowiedź, która nigdy nie przychodzi.

Dlaczego surowe wywołania narzędzi stanowią problem bezpieczeństwa

W agencie w stylu ReAct model wysyła żądanie użycia narzędzia przed jego uruchomieniem, a ten payload jest często wrażliwy. Narzędzie przekształcające tekst na SQL może otrzymać pełny schemat tabeli, takiej jak underwriting_policies. Narzędzie do ustalania cen może zawierać wewnętrzną adresację URL oraz strukturę parametrów prywatnego mikrosłużby. Sam proces rozważania może ujawnić zasady biznesowe lub instrukcje systemu. Jeśli dane te są przekazywane w niesprawdzonej formie, atakujący może uzyskać mapę twojego backendu, dlatego payloady muszą zostać usunięte na serwerze przed wysłaniem jakichkolwiek danych.

Generator służący do oczyszczania w FastAPI

Generator poniżej otacza strumień zdarzeń. Dla zdarzeń on_tool_start emituje nowe zdarzenie tool_update, które zawiera jedynie nazwę narzędzia, status oraz marker cenzury. W przypadku fragmentów wiadomości konwertuje instancje BaseMessage na słowniki za pomocą model_dump(). Wszystko inne jest kodowane z użyciem default=str jako rozwiązania awaryjnego, a błąd serializacji powoduje powstanie ogólnego zdarzenia błędu zamiast surowej wyjątku. Choć oryginalna etykieta wskazuje inaczej, ten fragment jest napisany w języku 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"

Dwa problemy wymagają uwagi przed użyciem w produkcji. Po pierwsze, przetwarzany jest tylko on_tool_start. Zdarzenia takie jak on_tool_end, zdarzenia łańcuchowe przenoszące stan grafu oraz fragmenty modelu, których tool_call_chunks zawierają argumenty, trafiają do mechanizmu awaryjnego i są przekazywane bez zmian. Lista dopuszczalnych zdarzeń, która przekazuje tylko te typy zdarzeń i pola, których potrzebuje interfejs użytkownika, jest bezpieczniejsza niż lista blokująca. Po drugie, należy zastąpić print strukturalnym logowaniem.

Koniec punktu końcowego uruchamia graf za pomocą astream_events według schematu zdarzeń v2, przekazuje go przez narzędzie sanitujące i zwraca w formacie 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")

Mapowanie zredagowanych zdarzeń na etykiety w interfejsie użytkownika w React

Ponieważ klient nigdy nie widzi danych wprowadzanych przez narzędzie, funkcjonuje ono na podstawie samego nazwy narzędzia. Tabela odnoszeń mapuje wewnętrzne nazwy, takie jak policy_db_search, na tekst przeznaczony dla użytkownika, przy czym istnieje ogólny rozwiązanie awaryjne dla nieznanych narzędzi. Zdarzenie tool_update wyświetla pulsującą ikonkę, a pierwszy fragment danych on_chat_model_stream ją usuwa, ponieważ agent zaczął odpowiadać.

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

Tutaj wymagana jest jedna korekta: EventSource przeglądarki wysyła jedynie żądania GET i nie może przesyłać ciała wiadomości, podczas gdy koniec ścieżki to ruta POST oczekująca wiadomości. W praktyce należy albo odczytać odpowiedź za pomocą fetch i czytnika strumieni, albo użyć biblioteki pomocniczej, która przetwarza SSE przez POST. Należy również upewnić się, że przedrostek /api pasuje do konfiguracji proxy w Vite.

Główne wnioski

• Wydarzenia LangGraph zawierają obiekty, których JSON nie może zakodować; konwertuj wiadomości wyraźnie i nigdy nie pozwól, by wyjątek kodera dotarł do klienta.
• Argumenty narzędzia, schematy oraz procesy rozumowania to tajemnice po stronie serwera. Zaszyfrowuj je przed transmisją, najlepiej poprzez ustalenie listy uprawnień dla tego, czego potrzebuje interfejs użytkownika.
• Mapa od nazwy narzędzia do etykiety zapewnia, że interfejs będzie informacyjny, nie ujawniając jednocześnie szczegółów wewnętrznych.
• Dostosuj transport klienta do punktu końcowego: EventSource nadaje się do strumieni typu GET, natomiast strumienie POST wymagają użycia fetch.

Literatura pokrewna

• Anatomia zespołu biznesowego w AI: Orkiestracja agentów za pomocą LangGraph i FastAPI — Przewodnik po platformie open-source do zarządzania wieloma agentami: jak ko-fundatorzy, menedżerowie i specjaliści są koordynowani, dzielą się informacjami, czekają na zatwierdzenie i składają raporty.
• Wbudowanie agenta LangChain w FastAPI: Narzędzia, poszukiwanie ręczne, strumieniowanie — Stwórz asystenta wewnątrz aplikacji przy użyciu FastAPI i LangChain: PDF z instrukcjami przechowywany w ChromaDB dostępny jako narzędzie, kontekst dla każdego użytkownika, zapisywana historia działań oraz odpowiedzi w formie strumienia.