自定义流通道

LangGraph 代理传输的不仅是消息和工具调用。一个服务端流转换器可以在协议流向客户端时对其进行检查或重写，并在一个命名的自定义通道上发布自己的结构化数据。前端通过两个选择器读取该通道：useExtension 用于获取最新载荷，而 useChannel 则作为原始事件的备用方案。以下示例是一个客户支持代理，其转换器在每个事件到达浏览器之前会编辑掉其中的 PII（电子邮件、电话号码、社会安全号码、信用卡号、IP 地址），并在一个 redaction-stats 通道上发布实时编辑计数。侧边面板会实时渲染这些计数。

自定义通道如何工作

自定义通道有两个端点。在服务端，一个 StreamTransformer 打开一个命名的 StreamChannel 并向其推送载荷。在客户端，一个选择器订阅匹配的 custom:<name> 通道，并将载荷暴露为响应式状态。转换器的 process 方法会对每个协议事件运行。它可以原地修改事件（此处是从 messages、tools 和 values 数据中清除 PII），并在有需要报告的内容时推送侧通道更新。客户端选择器（useExtension、useChannel）随 v1 前端 SDK 包（@langchain/react、@langchain/vue、@langchain/svelte、@langchain/angular）提供。

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

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; // 将（现在已编辑的）事件保留在流中。
    },
  };
};

构建代理时附加转换器：

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

The code examples use useStream<typeof myAgent> for type-safe stream state. See Type inference for Python or JavaScript backends.

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

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

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

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

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

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

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

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

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

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

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

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

返回值遵循每个框架的响应式模型：在 React 和 Svelte 中是普通值，在 Vue 中是 Ref（latest.value），在 Angular 中是信号（latest()）。在第一个载荷到达之前，该值是 undefined。可选的第三个 target 参数将订阅范围限定为命名空间，其方式与 useMessage(stream, node) 将消息范围限定为已发现的图节点相同。有关命名空间目标定位，请参见图执行。

使用 `useChannel` 缓冲原始事件

useChannel 是原始事件的备用方案。它订阅一个或多个通道，并返回底层协议事件的有界缓冲区，而不是单个解包后的值。当你需要历史记录而不是最新值（例如事件日志或审计追踪），或者你需要一个更高级别选择器未覆盖的通道时，可以使用它。传递完整的通道 ID（"custom:redaction-stats"）：

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

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

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

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

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

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

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

使用选项参数控制缓冲区：

const rawEvents = useChannel(
  stream,
  ["custom:redaction-stats"],
  undefined, // 目标命名空间
  { 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 计数、阻止的内容或策略命中。
进度报告：长时间运行的工具发出的完成百分比或步骤标签。
实时指标：运行期间累积的令牌使用量、延迟或成本。
来源与引用：当代理依据其回答时，检索到的文档被推送到侧边面板。
域事件：你的后端希望在不更改消息记录的情况下呈现的任何结构化更新。

Get started

Capabilities

Production

Frontend

LangGraph APIs

自定义通道如何工作

设置 `useStream`

使用 `useExtension` 读取最新载荷

使用 `useChannel` 缓冲原始事件

在 `useExtension` 和 `useChannel` 之间选择

用例

相关内容

​自定义通道如何工作

​设置 useStream

​使用 useExtension 读取最新载荷

​使用 useChannel 缓冲原始事件

​在 useExtension 和 useChannel 之间选择

​用例

​相关内容

自定义通道如何工作

设置 `useStream`

使用 `useExtension` 读取最新载荷

使用 `useChannel` 缓冲原始事件

在 `useExtension` 和 `useChannel` 之间选择

用例

相关内容