LangChain 中文文档

外观

LangGraph 智能体流式输出的内容不仅仅是消息和工具调用。服务端流式转换器（stream transformer）可以在协议流向客户端时对其进行检查或重写，并将自身结构化数据发布到命名的自定义通道上。前端通过两个选择器读取该通道：useExtension 用于获取最新负载，useChannel 则是访问原始事件的逃生通道（escape hatch）。

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

import { PatternEmbed } from "/snippets/pattern-embed.jsx"

自定义通道的工作原理 ​

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

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

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

INFO

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

INFO

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

python
import time

from langgraph.stream import ProtocolEvent, StreamChannel, StreamTransformer

class RedactionStatsTransformer(StreamTransformer):
    def __init__(self, scope: tuple[str, ...] = ()) -> None:
        super().__init__(scope)
        # 打开一个名为 "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:
        # 就地脱敏 event["params"]["data"] 并统计发现的内容。
        delta = redact_in_place(event, self.counts)
        if delta:
            # 在通道上发布负载。
            self.redaction_stats.push(
                {
                    "kind": "update",
                    "at": int(time.time() * 1000),
                    "delta": delta,
                    "counts": dict(self.counts),
                    "total": sum(self.counts.values()),
                }
            )
        return True  # 将（现已脱敏的）事件保留在流中。

def create_redaction_stats_transformer() -> RedactionStatsTransformer:
    return RedactionStatsTransformer()

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

python
from langchain.agents import create_agent

agent = create_agent(
    model="anthropic:claude-haiku-4-5",
    tools=[...],
    transformers=[create_redaction_stats_transformer],
)
ts
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; // 将（现已脱敏的）事件保留在流中。
    },
  };
};

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

ts
import { createAgent } from "langchain";

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

负载类型就是转换器推送的任何类型。下面的客户端示例读取这种结构：

ts
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

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

tsx
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} />
ts
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> 通道，并返回转换器推送的最新负载（已经解包并带有类型）。当界面只需要当前值时（例如实时计数器、进度百分比或状态徽章），这是更符合人体工程学的选择。

传入不带 custom: 前缀的裸通道名称（"redaction-stats"）：

tsx
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
ts
import { 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 是获取原始事件的逃生通道（escape hatch）。它订阅一个或多个通道，并返回底层协议事件的有界缓冲区，而不是单个已解包的值。当您需要历史记录而非最新值时（例如事件日志或审计追踪），或者需要某个没有更高级选择器覆盖的通道时，请使用它。

传入完整的通道 id（"custom:redaction-stats"）：

tsx
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"]);
ts
import { injectChannel } from "@langchain/angular";

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

每个条目都是一个原始协议事件，因此负载位于 event.params.data 之下。请自行解包：

ts
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 参数控制缓冲区：

ts
const rawEvents = useChannel(
  stream,
  ["custom:redaction-stats"],
  undefined, // 目标命名空间
  { bufferSize: 200, replay: true },
);
选项默认值效果
bufferSize"default"缓冲事件的最大数量。达到上限后，较早的事件会被丢弃。
replaytrue在选择器挂载时重放通道上已看到的事件，而不仅仅是实时事件。

INFO

在常见场景下，优先使用更高级的选择器（useExtension、useMessages、useToolCalls、useValues）。它们返回带类型的、已解包的值，并且只追踪您渲染的内容。当您特别需要原始事件流时，才使用 useChannel。

在 useExtension 和 useChannel 之间选择 ​

两者读取相同的自定义通道，但返回的内容不同：

useExtensionuseChannel
返回内容最新负载（T | undefined）原始事件的有界缓冲区（Event[]）
形式已解包的、带类型的负载原始协议事件；自行解包 event.params.data
订阅依据通道名称（"redaction-stats"）完整的通道 id（["custom:redaction-stats"]）
适用场景需要当前值需要历史记录、日志或多个通道
选项—bufferSize、replay

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

使用场景 ​

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

  • 合规与脱敏统计：被清洗的 PII、被阻止的内容或策略命中的计数，如上面的示例所示。
  • 进度报告：由长时间运行的工具发出的完成百分比或步骤标签。
  • 实时指标：运行期间累计的 token 用量、延迟或成本。
  • 来源与引用：智能体为答案提供依据时，将检索到的文档推送到侧边面板。
  • 领域事件：您的后端希望在不变更消息记录的情况下暴露的任何结构化更新。

相关 ​

  • 概述 — LangGraph 前端流式输出 API 与架构。
  • 图执行 — 用于多节点流水线的命名空间限定选择器。