Стрімування агентаў LangGraph у React без вытэкання внутрашніх дадзенняў інструмента

Адрасаванне працы агента ў браузер прабуджае вярнючы ўражанне жывага прыемніка дадзенняў у RAG-практыкі, але прымусовая перадача неапранутага струму ляўкавых падзеяў LangGraph адразу да кліента спрычыняе два окалічныя проблемы. Струм зупініваецца, калі патрапляюць об’екты, якія JSON не можа закодаваць, а тые падзеі, якія такім чынам праходзяць, апавярожняюць схемы, внутранія канчаткі і прычыны дзеяння для будзь-каго, у каго ачынены роздзіл сеті. У этай статыце практыкуецца невялікі шар санітазацыі ў FastAPI, який рашае обе проблемы, а пасля паказвае, як кліент на базе React ператварае апранутыя падзеі на чытальныя індикаторы статусу.

Чаму неапрануты стан агента наражае на проблемы сэrialізацію JSON

У час выканання стан LangGraph ёсць збагачанай структурой на Python, а не докумэнтом JSON. Перадача выходных даных з astream_events у json.dumps() або безпосередна ў FastAPI StreamingResponse зазвычай не вяршыцца, таму што падзеі маюць:

• Класы паведамленняў 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 маюць аргументы, праходзяць далей да запаснага варіанта і передаюцца без змян. Спіс дазволеных значэнняў, які перадае толькі тыпы адбывання і поля, якія патрэбны UI, є безпечнейшым за спіс заблокаваных значэнняў. Другі момент — заменіць 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")

Спаўненне адбыванняў, якія былі змінены, з абозначэннямі UI у 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.