Стриминг агентов LangGraph в React без утечки внутренних деталей инструмента

Трансляция прогресса агента в браузер придает приложению на основе RAG ощущение живости, однако направление необработанного потока событий LangGraph непосредственно к клиенту приводит к двум отдельным проблемам. Поток теряет работоспособность при обработке объектов, которые JSON не может закодировать, а те события, что всё же доходят до клиента, раскрывают схемы данных, внутренние конечные точки и процессы принятия решений для любого, у кого открыта вкладка сети. В этой статье описывается небольшой слой очистки в FastAPI, который решает обе проблемы, а затем показано, как клиент на React преобразует отредактированные события в читаемые индикаторы статуса.

Почему необработанный состояние агента нарушает сериализацию в JSON

Во время выполнения состояние LangGraph представляет собой сложную структуру на Python, а не документ в формате JSON. Отправка результата функции astream_events в функцию json.dumps() или непосредственно в объект StreamingResponse FastAPI обычно проваливается из-за того, что события содержат:

• Классы сообщений LangChain, такие как AIMessage, HumanMessage и ToolMessage, которые представляют собой модели Pydantic (версия v1 или v2 в зависимости от версии LangChain) с собственными методами сериализации.
• Вложенные объекты Pydantic, хранящие промежуточные результаты рассуждений агента.
• Объекты исключений и информация об ошибках, возникающих при неудачных вызовах инструментов.

Когда код-кодер сталкивается с одним из этих случаев, он вызывает исключение, соединение Server-Sent Events прерывается, и фронтенд бесконечно ждет ответа, который так и не поступает.

Почему прямые вызовы инструментов представляют угрозу для безопасности

В агенте в стиле ReAct модель отправляет запрос к инструменту до его выполнения, и этот пакет данных часто является конфиденциальным. Инструмент преобразования текста в SQL может получать полную структуру таблицы, такой как underwriting_policies. Инструмент ценообразования может содержать внутренний URL и структуру параметров приватного микросервиса. Сам процесс размышлений может раскрывать бизнес-правила или инструкции системы. Если такие данные передаются в нефильтрованном виде, это позволяет злоумышленнику получить карту вашего бэкенда, поэтому пакеты данных необходимо обрабатывать на сервере перед их отправкой.

Генератор для очистки данных в FastAPI

Приведённый ниже генератор оборачивает поток событий. Для событий on_tool_start он генерирует новое событие tool_update, содержащее только имя инструмента, статус и маркер цензуры. Для фрагментов сообщений он преобразует экземпляры BaseMessage в словари с помощью функции model_dump(). Всё остальное кодируется по умолчанию с использованием параметра default=str; в случае неудачи сериализации генерируется общее событие ошибки вместо первичного исключения. Несмотря на первоначальную метку, этот фрагмент написан на 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"

Перед внедрением в производство следует устранить две проблемы. Во-первых, обрабатывается только функция on_tool_start. События вроде on_tool_end, цепочки событий, передающих состояние графа, а также фрагменты модели, в которых поля tool_call_chunks содержат аргументы, проходят дальше без обработки и передаются в неизменном виде. Использование списка разрешенных типов событий и полей, необходимых интерфейсу, безопаснее, чем список заблокированных. Во-вторых, следует заменить функцию print на структурированную систему логирования.

Эндпоинт инициирует работу графа с помощью функции astream_events по шаблону событий v2, пропускает данные через модуль очистки и возвращает их в формате 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")

Сопоставление скрытых событий с метками интерфейса в React

Поскольку клиент никогда не видит входные данные инструмента, система работает исключительно на основе его имени. Таблица сопоставлений преобразует внутренние имена, такие как policy_db_search, в текст, видимый пользователем, предусмотрен также универсальный вариант для неизвестных инструментов. Событие tool_update отображает мерцающий индикатор, а первый фрагмент события on_chat_model_stream удаляет его, поскольку агент начал отвечать.

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

Здесь требуется одно исправление: объект EventSource браузера отправляет только запросы типа GET и не может передавать тело сообщения, в то время как конечная точка — это маршрут типа POST, ожидающий сообщение. На практике следует либо считывать ответ с помощью fetch и читателя потока, либо использовать вспомогательную библиотеку, которая обрабатывает формат SSE через POST. Также убедитесь, что префикс /api соответствует настройкам прокси Vite.

Основные выводы

• События LangGraph содержат объекты, которые JSON не может закодировать; преобразуйте сообщения явно и ни в коем случае не позволяйте исключениям кодировщика достигать клиента.
• Аргументы инструментов, схемы и логика обработки являются конфиденциальной информацией на стороне сервера. Скройте их перед трансляцией, желательно с помощью списка разрешенных элементов, необходимых интерфейсу.
• Карта от имени инструмента к его метке позволяет сохранять информативность интерфейса без раскрытия внутренних деталей.
• Подбирайте способ передачи данных от клиента в соответствии с конечной точкой: для потоков типа GET подходит EventSource, тогда как для потоков типа POST требуется использование fetch.