将 LangGraph Agents 流式传输到 React 中，同时避免泄露工具内部实现细节

将智能体的处理进度流式传输到浏览器中，能让RAG应用显得更加生动，但直接将LangGraph的原始事件流发送给客户端会导致两个问题：首先，JSON无法编码的对象会使得数据流中断；其次，即便部分数据成功传输，也会让任何打开网络标签页的人看到系统的架构结构、内部接口以及推理过程。本文在FastAPI中构建了一个简单的过滤层来解决这两个问题，同时展示了React客户端如何将经过处理后的事件转换为可读的状态标识。

为何原始智能体状态会导致JSON序列化失败

在运行过程中，LangGraph的状态是一种复杂的Python数据结构，而非JSON文档。将astream_events的输出传递给json.dumps()或直接放入FastAPI的StreamingResponse中通常会失败，因为这些事件中包含：

• LangChain中的消息类，如AIMessage、HumanMessage和ToolMessage，它们都是Pydantic模型（版本取决于你的LangChain版本，可能是v1或v2），并拥有各自的序列化方法。
• 用于存储代理中间推理结果的嵌套Pydantic对象。
• 来自工具调用失败的异常对象及回溯信息。

当编码器遇到这些情况时会产生异常，导致Server-Sent Events连接中断，前端将永远等待永远不会到来的响应。

为何直接调用工具会带来安全问题

在 ReAct 风格的智能体中，模型会在工具执行之前发出工具调用请求，而该请求中的数据往往包含敏感信息。文本转 SQL 工具可能会收到如 underwriting_policies 这类表的完整结构信息；定价工具则可能携带私有微服务的内部 URL 及参数格式。思维链本身也可能暴露业务规则或系统提示指令。如果这些信息未经过滤就直接传输，攻击者就能获取到你后端的架构图，因此必须在服务器端对数据进行处理后再发送。

FastAPI 中的净化生成器

下方的生成器用于封装事件流。对于on_tool_start事件，它会发出一个新的tool_update事件，其中仅包含工具名称、状态以及屏蔽标记。对于消息块，则通过model_dump()将BaseMessage实例转换为字典形式。其余内容则默认使用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替换为结构化日志记录。

该接口使用v2事件规范通过astream_events启动图处理，将其传递给净化器后以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中将已屏蔽的事件映射为UI标签

由于客户端永远无法看到工具的输入参数，因此只需通过工具名称即可实现功能。查找表会将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配合流式读取器来获取响应，要么使用能够通过POST方式解析SSE的辅助库。同时还需确认/api前缀与您的Vite代理配置一致。

关键要点

• LangGraph事件中包含JSON无法编码的对象；需手动转换消息，绝不能让编码异常传递到客户端。
• 工具参数、架构及推理逻辑属于服务器端机密，应在数据流传输前进行遮蔽处理，最好通过白名单方式仅允许UI访问所需内容。
• 工具名称与标签的映射关系既能保证界面信息清晰，又不会暴露内部结构。
• 需将客户端传输方式与端点匹配：EventSource适用于GET流，而POST流则需使用fetch。