自定义流通道
通过服务端流转换器(Stream Transformer)将自定义数据推送到命名通道,并在前端使用
useExtension和useChannel读取。
LangGraph Agent 的流式输出远不止消息(messages)和工具调用(tool calls)。通过服务端的 流转换器(stream transformer),你可以检查甚至改写流向客户端的协议事件,并在一个命名的 自定义通道(custom channel) 上发布你自己的结构化数据。前端则通过两个选择器来读取该通道的数据:useExtension 用于获取最新的载荷(payload),而 useChannel 则是原始事件的"逃生舱",当你需要事件历史而非单值时使用。
接下来,我们通过一个客服 Agent 的实际场景来理解这套机制。这个 Agent 的转换器会在每个事件到达浏览器之前,对其中的 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)。
INFO
流转换器和 StreamChannel 需要 @langchain/langgraph>=1.3.1。
下面是创建脱敏统计转换器的完整代码。它打开了一个名为 redaction-stats 的远程通道,并在每次脱敏时推送一条带时间戳的更新载荷:
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.
},
};
};关键点:
process返回true表示"保留这个事件继续流式传输"。转换器的作用是透明地增强流,而不是拦截它。即便你已经脱敏了内容,消息和工具调用仍然会正常到达前端。
创建好转换器后,在构建 Agent 时将其附加进去即可:
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 句柄。
INFO
下面的代码示例使用 useStream<typeof myAgent> 来获得类型安全的流状态。关于后端类型推断,请参阅 Python 或 JavaScript 的类型推断说明。
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?.deltaimport { useExtension } from "@langchain/vue";
const latest = useExtension<RedactionStatsEvent>(stream, "redaction-stats");
// latest.value?.totalimport { useExtension } from "@langchain/svelte";
const latest = useExtension<RedactionStatsEvent>(stream, "redaction-stats");
// latest?.totalimport { 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 是原始事件的逃生舱。它订阅一个或多个通道,返回底层协议事件的有界缓冲区(bounded buffer),而非单个解包值。当你需要事件历史而非最新值时(例如事件日志或审计跟踪),或者需要读取高层选择器未覆盖的通道时,就该用它了。
传入时使用完整的通道 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.valueimport { 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;
}通过 options 参数控制缓冲行为:
const rawEvents = useChannel(
stream,
["custom:redaction-stats"],
undefined, // target namespace
{ 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 使用量、延迟或成本。
- 来源与引用:随着 Agent 为答案补充依据,将被检索到的文档推送到侧面板。
- 领域事件:后端希望呈现的任何结构化更新,而无需修改消息记录本身。
核心思路:如果你的数据"不是一条消息",但又需要在 UI 上展示,自定义通道就是正确的抽象层。它让你在不污染对话历史的前提下,把后端信号干净地映射到前端的独立视图。
相关阅读
本文基于 LangGraph 官方文档 翻译并二次创作。