自定义流通道

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

LangGraph 智能体流式传输的内容不仅仅是消息和工具调用。服务端 流转换器 可以在协议流向客户端的过程中检查或重写它，并在具名的 自定义通道 上发布自己的结构化数据。 前端使用两个选择器读取该通道：useExtension 用于获取最新 载荷，useChannel 作为原始事件的逃生舱。

下面的示例是一个客户支持智能体，其转换器会对每个事件进行 PII 脱敏 （电子邮件、电话号码、SSN、卡号、IP），在其 到达浏览器之前，并在一个 redaction-stats 通道上发布实时的脱敏计数。侧边面板会实时渲染这些计数。

在线演示

此处为官方文档中的交互式演示（PatternEmbed），静态站点无法呈现。请前往官方原文查看。

自定义通道如何工作

自定义通道有两端。在服务端，StreamTransformer 打开一个 具名的 StreamChannel 并向其推送载荷。在客户端，一个选择器 订阅匹配的 custom:<name> 通道，并将载荷暴露为 响应式状态。

转换器的 process 方法会针对每个协议事件运行。它可以原地修改 该事件（此处是从 messages、tools 和 values 数据中清除 PII），并在有内容需要报告时推送侧通道更新。

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

NOTE

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

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],
)

载荷类型就是转换器所推送的任何内容。下面的客户端示例 读取这个结构：

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 后端。

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

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

Angular

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: 前缀：

React

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

Angular

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

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

返回值遵循各个框架的响应式模型：在 React 和 Svelte 中是普通值， 在 Vue 中是 Ref（latest.value），在 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"]);

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"]);

Angular

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

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

每个条目都是一个原始协议事件，因此载荷位于 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;
}

使用 options 参数控制缓冲区：

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

选项	默认值	效果
bufferSize	"default"	缓冲事件的最大数量。达到上限后，较旧的事件会被丢弃。
replay	true	当选择器挂载时，重放通道上已经见过的事件，而不仅仅是实时事件。

NOTE

对于常见情况，优先使用更高级的选择器（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 与架构。
• 图执行 —— 面向多节点管道的 命名空间作用域选择器。

---