自定义流通道

将自定义服务器端数据流式传输到前端，并使用 useExtension 和 useChannel 读取。

LangGraph智能体可流式输出超过消息和工具调用。服务器端的流式转换器能在协议流向客户端时进行协议检查或重写，并在命名的自定义频道上发布结构化数据。前端通过两种选择器读取该频道：@[useExtension]用于获取最新载荷，@[useChannel]作为原始事件逃逸通道。

下例展示了一个客户支持智能体的转换器，该转换器会在事件到达浏览器前擦除PII（电子邮件、电话号码、SSN、卡号、IP地址），并在redaction-stats频道上发布实时擦除计数。侧边栏会动态渲染这些计数。

自定义频道的工作原理

自定义频道包含两端。在服务器端，@[StreamTransformer]会打开命名的@[StreamChannel]并推送载荷。在客户端，选择器会订阅匹配的custom:<name>频道并暴露载荷为响应式状态。

转换器的@[process]方法会在每个协议事件时执行。它可以原地修改事件（此处从messages、tools、values数据中擦除PII），并在有需要时推送侧通道更新。

客户端选择器（useExtension、useChannel）随v1前端SDK包（@langchain/react、@langchain/vue、@langchain/svelte、@langchain/angular）一起发布。

:::python

流式转换器和StreamChannel需要langgraph>=1.2。

:::

:::js

流式转换器和StreamChannel需要@langchain/langgraph>=1.3.1。

:::

:::python

import time

from langgraph.stream import ProtocolEvent, StreamChannel, StreamTransformer


class RedactionStatsTransformer(StreamTransformer):
    def __init__(self, scope: tuple[str, ...] = ()) -> None:
        super().__init__(scope)
        # Open a channel named "redaction-stats".
        self.redaction_stats = StreamChannel("redaction-stats")
        self.counts = empty_counts()

    def init(self) -> dict[str, StreamChannel]:
        return {"redactionStats": self.redaction_stats}

    def process(self, event: ProtocolEvent) -> bool:
        # Redact event["params"]["data"] in place and tally what was found.
        delta = redact_in_place(event, self.counts)
        if delta:
            # Publish a payload on the channel.
            self.redaction_stats.push(
                {
                    "kind": "update",
                    "at": int(time.time() * 1000),
                    "delta": delta,
                    "counts": dict(self.counts),
                    "total": sum(self.counts.values()),
                }
            )
        return True  # Keep the (now-redacted) event in the stream.


def create_redaction_stats_transformer() -> RedactionStatsTransformer:
    return RedactionStatsTransformer()

在构建智能体时附加转换器：

from langchain.agents import create_agent

agent = create_agent(
    model="anthropic:claude-haiku-4-5",
    tools=[...],
    transformers=[create_redaction_stats_transformer],
)

:::

:::js

import { StreamChannel } from "@langchain/langgraph";
import type { ProtocolEvent, StreamTransformer } from "@langchain/langgraph";

export const createRedactionStatsTransformer = (): StreamTransformer<{
  redactionStats: StreamChannel<RedactionStatsEvent>;
}> => {
  // Open a remote channel named "redaction-stats".
  const redactionStats = StreamChannel.remote<RedactionStatsEvent>("redaction-stats");
  const counts = emptyCounts();

  return {
    init: () => ({ redactionStats }),

    process(event: ProtocolEvent): boolean {
      // Redact event.params.data in place and tally what was found.
      const delta = redactInPlace(event, counts);
      if (Object.keys(delta).length > 0) {
        // Publish a payload on the channel.
        redactionStats.push({
          kind: "update",
          at: Date.now(),
          delta,
          counts: { ...counts },
          total: totalRedactions(counts),
        });
      }
      return true; // Keep the (now-redacted) event in the stream.
    },
  };
};

在构建智能体时附加转换器：

import { createAgent } from "langchain";

const agent = createAgent({
  model: "anthropic:claude-haiku-4-5",
  tools: [...],
  streamTransformers: [createRedactionStatsTransformer],
});

:::

载荷类型由转换器决定。客户端示例下方会展示该结构：

type PiiType = "email" | "phone" | "ssn" | "credit_card" | "ip_address";

type RedactionStatsEvent = {
  kind: "update";
  at: number;
  delta: Partial<Record<PiiType, number>>;
  counts: Record<PiiType, number>;
  total: number;
};

配置useStream

按常规方式连接@[useStream]。自定义频道选择器会使用返回的相同@[stream]句柄。

代码示例使用 useStream<typeof myAgent> 进行类型安全的流状态管理。请参考 Python 或 JavaScript 后端的类型推断说明。

React

import { useStream } from "@langchain/react";

const AGENT_URL = "http://localhost:2024";

export function RedactionChat() {
  const stream = useStream<typeof myAgent>({
    apiUrl: AGENT_URL,
    assistantId: "custom_stream_channel",
  });

  return <RedactionStatsPanel stream={stream} />;
}

使用useExtension读取最新载荷

useExtension订阅一个custom:<name>频道，并返回转换器推送的最新载荷（已解包且类型化）。当UI只需要当前值时（如实时计数、进度百分比或状态徽章）这是最便捷的选择。

传入裸频道名称（"redaction-stats"），而非custom:前缀：

React

import { useExtension } from "@langchain/react";

const latest = useExtension<RedactionStatsEvent>(stream, "redaction-stats");
// latest?.total, latest?.counts.email, latest?.delta

返回值遵循各框架的响应式模型：React和Svelte返回普通值，Vue（latest.value）返回Ref，Angular返回latest()。初始值为undefined，直到首个载荷到达。

可选的第三个target参数会限定订阅命名空间，与useMessages(stream, node)对消息限定图节点命名空间方式相同。 见图执行了解命名空间定位。

使用useChannel缓冲原始事件

useChannel是原始事件逃逸通道。它订阅一个或多个频道，返回底层协议事件的有限缓冲区而非单个解包值。当需要历史记录（如事件日志或审计轨迹）或需要多个频道时使用。

传入完整频道ID（"custom:redaction-stats"）：

React

import { useChannel } from "@langchain/react";

const rawEvents = useChannel(stream, ["custom:redaction-stats"]);

每个条目是原始协议事件，因此载荷位于event.params.data下。需自行解包：

function parseRedactionStatsEvents(rawEvents: Event[]): RedactionStatsEvent[] {
  const out: RedactionStatsEvent[] = [];
  for (const event of rawEvents) {
    const data = event.params?.data;
    const payload = data?.payload ?? data;
    if (payload?.kind === "update") out.push(payload);
  }
  return out;
}

通过选项参数控制缓冲区：

const rawEvents = useChannel(
  stream,
  ["custom:redaction-stats"],
  undefined, // target namespace
  { bufferSize: 200, replay: true },
);

选项	默认	效果
bufferSize	"default"	最大缓冲事件数，达到上限后旧事件会被丢弃
replay	true	播放已订阅频道在挂载时已存在的所有事件，而非仅实时事件

优先使用高阶选择器（useExtension、useMessages、useToolCalls、useValues）处理常见场景。它们返回类型化解包值，仅跟踪渲染内容。当明确需要原始事件流时使用useChannel。

选择useExtension和useChannel

两者读取同一自定义频道但返回不同内容：

	useExtension	useChannel
返回值	最新载荷（T | undefined）	有限原始事件缓冲区（Event[]）
数据结构	解包且类型化的载荷	原始协议事件；需自行解包event.params.data
订阅方式	频道名称（"redaction-stats"）	完整频道ID（["custom:redaction-stats"]）
使用场景	需要当前值（如实时计数）	需要历史记录、日志或多频道
选项	—	bufferSize、replay

常见模式是在同一频道上同时使用两者：useExtension驱动实时摘要（当前统计值），而useChannel支持滚动事件日志记录线程内所有更新。

应用场景

自定义频道适用于无法干净映射到消息、工具调用或图状态的任何服务器端信号：

• 合规与擦除统计：已擦除PII、过滤内容或策略触发的计数（如示例所示）
• 进度报告：长时间运行工具的百分比完成或步骤标签
• 实时指标：运行期间累计的token使用量、延迟或成本
• 来源与引用：智能体支撑答案时推送的检索文档到侧边栏
• 领域事件：后端需要暴露但无需修改消息转录的结构化更新

相关文档

• 概述 — LangGraph前端流API及架构
• 图执行 — 命名空间限定选择器用于多节点流水线

---

官方原文：src/oss/langgraph/frontend/custom-stream-channels.mdx · 非官方翻译，译文由 AI 生成，以英文原文为准。