> ## Documentation Index
> Fetch the complete documentation index at: https://langchain-docs-zh.mintlify.site/llms.txt
> Use this file to discover all available pages before exploring further.

# 自定义流通道

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

LangGraph智能体可流式输出超过消息和工具调用。服务器端的**流式转换器**能在协议流向客户端时进行协议检查或重写，并在命名的**自定义频道**上发布结构化数据。前端通过两种选择器读取该频道：@\[`useExtension`]用于获取最新载荷，@\[`useChannel`]作为原始事件逃逸通道。

下例展示了一个客户支持智能体的转换器，该转换器会在事件到达浏览器前擦除PII（电子邮件、电话号码、SSN、卡号、IP地址），并在`redaction-stats`频道上发布实时擦除计数。侧边栏会动态渲染这些计数。

交互示例暂不可用（上游文档源缺失：custom-stream-channel）

## 自定义频道的工作原理

自定义频道包含两端。在服务器端，@\[`StreamTransformer`]会打开命名的@\[`StreamChannel`]并推送载荷。在客户端，选择器会订阅匹配的`custom:<name>`频道并暴露载荷为响应式状态。

转换器的@\[`process`]方法会在每个协议事件时执行。它可以原地修改事件（此处从`messages`、`tools`、`values`数据中擦除PII），并在有需要时推送侧通道更新。

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

:::python

> [!NOTE]
>
> 流式转换器和`StreamChannel`需要`langgraph>=1.2`。

:::

:::js

> [!NOTE]
>
> 流式转换器和`StreamChannel`需要`@langchain/langgraph>=1.3.1`。

:::

:::python

```python theme={null}
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()
```

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

```python theme={null}
from langchain.agents import create_agent

agent = create_agent(
    model="anthropic:claude-haiku-4-5",
    tools=[...],
    transformers=[create_redaction_stats_transformer],
)
```

:::

:::js

```ts theme={null}
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.
    },
  };
};
```

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

```ts theme={null}
import { createAgent } from "langchain";

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

:::

载荷类型由转换器决定。客户端示例下方会展示该结构：

```ts theme={null}
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`]句柄。

> [!NOTE]
>
> 代码示例使用 `useStream<typeof myAgent>` 进行类型安全的流状态管理。请参考 [Python](https://docs.langchain.com/oss/python/langchain/frontend/overview#type-inference) 或 [JavaScript](https://docs.langchain.com/oss/javascript/langchain/frontend/overview#type-inference) 后端的类型推断说明。

```tsx React theme={null}
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 Vue theme={null}
<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 Svelte theme={null}
<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} />
```

```ts Angular theme={null}
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:`前缀：

```tsx React theme={null}
import { useExtension } from "@langchain/react";

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

```vue Vue theme={null}
import { useExtension } from "@langchain/vue";

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

```svelte Svelte theme={null}
import { useExtension } from "@langchain/svelte";

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

```ts Angular theme={null}
import { injectExtension } from "@langchain/angular";

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

返回值遵循各框架的响应式模型：React和Svelte返回普通值，Vue（`latest.value`）返回`Ref`，Angular返回`latest()`。初始值为`undefined`，直到首个载荷到达。

可选的第三个`target`参数会限定订阅命名空间，与`useMessages(stream, node)`对消息限定图节点命名空间方式相同。
见[图执行](https://langchain-docs-zh.mintlify.site/langchain/oss/langgraph/frontend/graph-execution.md)了解命名空间定位。

## 使用`useChannel`缓冲原始事件

`useChannel`是原始事件逃逸通道。它订阅一个或多个频道，返回底层协议事件的有限缓冲区而非单个解包值。当需要历史记录（如事件日志或审计轨迹）或需要多个频道时使用。

传入完整频道ID（`"custom:redaction-stats"`）：

```tsx React theme={null}
import { useChannel } from "@langchain/react";

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

```vue Vue theme={null}
import { useChannel } from "@langchain/vue";

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

```svelte Svelte theme={null}
import { useChannel } from "@langchain/svelte";

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

```ts Angular theme={null}
import { injectChannel } from "@langchain/angular";

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

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

```ts theme={null}
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;
}
```

通过选项参数控制缓冲区：

```ts theme={null}
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使用量、延迟或成本
* **来源与引用**：智能体支撑答案时推送的检索文档到侧边栏
* **领域事件**：后端需要暴露但无需修改消息转录的结构化更新

## 相关文档

* [概述](https://langchain-docs-zh.mintlify.site/langchain/oss/langgraph/frontend/overview.md) — LangGraph前端流API及架构
* [图执行](https://langchain-docs-zh.mintlify.site/langchain/oss/langgraph/frontend/graph-execution.md) — 命名空间限定选择器用于多节点流水线

***

> 官方原文：[src/oss/langgraph/frontend/custom-stream-channels.mdx](https://github.com/langchain-ai/docs/blob/main/src/oss/langgraph/frontend/custom-stream-channels.mdx) · 非官方翻译，译文由 AI 生成，以英文原文为准。


This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.