自定义流通道（Custom stream channels）

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

LangGraph 智能体流式传输的内容不仅限于消息和工具调用。服务端流转换器（stream transformer）可以在协议流向客户端时检查或重写协议，并在命名的自定义通道（custom channel）上发布自己的结构化数据。前端使用两个选择器读取该通道：用于获取最新有效载荷的 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）一起提供。

INFO

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

typescript

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

export const createRedactionStatsTransformer = (): StreamTransformer<{
  redactionStats: StreamChannel<RedactionStatsEvent>;
}> => {
  // 打开名为 "redaction-stats" 的远程通道。
  const redactionStats = StreamChannel.remote<RedactionStatsEvent>("redaction-stats");
  const counts = emptyCounts();

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

    process(event: ProtocolEvent): boolean {
      // 原地脱敏 event.params.data 并统计发现的内容。
      const delta = redactInPlace(event, counts);
      if (Object.keys(delta).length > 0) {
        // 在通道上发布载荷。
        redactionStats.push({
          kind: "update",
          at: Date.now(),
          delta,
          counts: { ...counts },
          total: totalRedactions(counts),
        });
      }
      return true; // 在流中保留（现在已脱敏的）事件。
    },
  };
};

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

typescript

import { createAgent } from "langchain";

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

载荷类型是转换器推送的任何内容。下面的客户端示例读取此形状：

typescript

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 句柄。

INFO

代码示例使用 useStream<typeof myAgent> 来实现类型安全的流状态。请参阅用于 Python 或 JavaScript 后端的类型推断。

ReactVueSvelteAngular

tsx

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

vue

<script setup lang="ts">
import { useStream } from "@langchain/vue";

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

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

<template>
  <RedactionStatsPanel :stream="stream" />
</template>

svelte

<script lang="ts">
  import { useStream } from "@langchain/svelte";

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

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

<RedactionStatsPanel {stream} />

typescript

import { Component } from "@angular/core";
import { injectStream } from "@langchain/angular";

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

@Component({
  selector: "app-redaction-chat",
  template: `<app-redaction-stats-panel [stream]="stream" />`,
})
export class RedactionChatComponent {
  stream = injectStream<typeof myAgent>({
    apiUrl: AGENT_URL,
    assistantId: "custom_stream_channel",
  });
}

使用 `useExtension` 读取最新载荷

useExtension 订阅 custom:<name> 通道并返回转换器推送的最新载荷（已解包且类型化）。当 UI 仅需要当前值（例如实时计数器、进度百分比或状态徽章）时，它是符合人体工程学的选择。

传递纯通道名称（"redaction-stats"），而不是 custom: 前缀：

ReactVueSvelteAngular

tsx

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

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

vue

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

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

svelte

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

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

typescript

import { injectExtension } from "@langchain/angular";

const latest = injectExtension<RedactionStatsEvent>(stream, "redaction-stats");
// latest()?.total

返回值遵循每个框架的响应式模型：React 和 Svelte 中的普通值、Vue 中的 Ref（latest.value）以及 Angular 中的 signal（latest()）。在第一个载荷到达之前，该值为 undefined。

可选的第三个 target 参数将订阅范围限定为命名空间，就像 useMessages(stream, node) 将消息范围限定为已发现的图节点一样。有关命名空间定位，请参阅图执行。

使用 `useChannel` 缓冲原始事件

useChannel 是原始事件的安全出口。它订阅一个或多个通道，并返回底层协议事件的有界缓冲区，而不是单个解包的值。当你需要历史记录而不是最新值时（例如事件日志或审计跟踪），或者当你需要没有更高级别选择器涵盖的通道时，请使用它。

传递完整的通道 ID（"custom:redaction-stats"）：

ReactVueSvelteAngular

tsx

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

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

vue

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

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

svelte

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

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

typescript

import { injectChannel } from "@langchain/angular";

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

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

typescript

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

使用 options 参数控制缓冲区：

typescript

const rawEvents = useChannel(
  stream,
  ["custom:redaction-stats"],
  undefined, // 目标命名空间
  { bufferSize: 200, replay: true },
);

选项	默认值	效果
`bufferSize`	`"default"`	缓冲事件的最大数量。一旦达到上限，较旧的事件将被丢弃。
`replay`	`true`	在选择器挂载时重放在通道上已看到的事件，而不是仅重放实时事件。

INFO

对于常见情况，优先使用更高级别的选择器（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 使用量、延迟或成本。
来源和引用：在智能体论证其答案时，将检索到的文档推送到侧面板。
领域事件：你的后端想要公开的任何结构化更新，而无需更改消息对话记录。

自定义流通道（Custom stream channels） ​

自定义通道的工作原理 ​

设置 useStream ​

使用 useExtension 读取最新载荷 ​

使用 useChannel 缓冲原始事件 ​

在 useExtension 与 useChannel 之间进行选择 ​

使用案例 ​

自定义流通道（Custom stream channels）

自定义通道的工作原理

设置 `useStream`

使用 `useExtension` 读取最新载荷

使用 `useChannel` 缓冲原始事件

在 `useExtension` 与 `useChannel` 之间进行选择

使用案例