自定义流通道 - 以编程方式使用文档

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

  redactionStats: StreamChannel;
}> => {
  // Open a remote channel named "redaction-stats".
  const redactionStats = StreamChannel.remote("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.
    },
  };
};

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

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 句柄返回到这里。

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

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

  return ;
}

<script setup lang="ts">


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

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

<template>

</template>

<script lang="ts">


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

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

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

@Component({
  selector: "app-redaction-chat",
  template: `<app-redaction-stats-panel [stream]="stream" />`,
})

  stream = injectStream<typeof myAgent>({
    apiUrl: AGENT_URL,
    assistantId: "custom_stream_channel",
  });
}

使用以下方式读取最新负载 `useExtension`

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

传递裸通道名称（"redaction-stats"），而不是 custom: prefix:

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

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

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

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

返回值遵循各框架的响应式模型：React 和 Svelte 中为普通值， Vue 中为 Ref （latest.value），Angular 中为 signal (latest()）。该值 undefined 直到收到第一个 payload。

可选的第三个 target 参数将订阅限定到一个命名空间，与 useMessages(stream, node) 将消息限定到已发现的图形节点的方式相同。参见图形执行了解命名空间 targeting.

使用 `useChannel`

useChannel 是原始事件的应急方案。它订阅一个或多个通道，返回底层协议事件的有限缓冲区，而非单个未包装的值。当您需要历史记录而不是最新值时（如事件日志或审计跟踪），或当您需要一个没有更高级别选择器覆盖的通道时，请使用它。

传递完整的通道 id（"custom:redaction-stats"):

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

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

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

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

每个条目都是原始协议事件，因此 payload 位于 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` 与 `useChannel`

之间选择。两者都读取同一个自定义通道，但返回内容不同：

	`useExtension`	`useChannel`
返回	最新 payload（`T \	undefined`）
形状	解包后的类型化 payload	原始协议事件；解包 `event.params.data` 自行
订阅方式	通道名称（`"redaction-stats"`）	完整通道 id（`["custom:redaction-stats"]`)
使用场景	需要当前值	需要历史记录、日志或多通道
选项	—	`bufferSize`, `replay`

一种常见的模式是在同一通道上同时使用两者： useExtension 驱动实时摘要（当前总数），而 useChannel 支持可滚动的事件日志，记录整个线程中的每次更新。

用例

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

- **合规性和编辑统计**：已清除的PII、被阻止的内容或策略命中次数，如上面的示例所示。 - **进度报告**：长时间运行的工具发出的完成百分比或步骤标签。 - **实时指标**：运行期间累积的令牌使用量、延迟或成本。 - **来源和引用**：代理在提供答案时推送到侧边栏的检索文档。 - **领域事件**：后端想要呈现的任何结构化更新而无需更改消息记录。

自定义通道的工作原理

设置 useStream

使用以下方式读取最新负载 useExtension

使用 useChannel

在 useExtension 与 useChannel

用例

相关资源

设置 `useStream`

使用以下方式读取最新负载 `useExtension`

使用 `useChannel`

在 `useExtension` 与 `useChannel`