Стрімування LangGraph Agents у 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 не може закодувати; необхідно явно конвертувати повідомлення та ніколи не дозволяти виникненню винятку кодувальника потрапити до клієнта.
• Аргументи інструментів, схеми та логіка обробки є конфіденційною інформацією на серверній стороні. Необхідно приховати їх перед трансляцією, бажано шляхом вказування списку того, що потрібно інтерфейсу.
• Карта відповідності імен інструментів їхнім міткам забезпечує інформативність інтерфейсу без розкриття внутрішніх деталей.
• Підберіть спосіб передачі даних клієнтом відповідно до кінцевої точки: EventSource підходить для потоків типу GET, тоді як для потоків типу POST потрібен fetch.