커스텀 스트림 채널

커스텀 스트림 채널 (Custom stream channels)

커스텀 서버 측 데이터를 프론트엔드로 스트리밍하고 useExtension과 useChannel로 읽어요.

LangGraph 에이전트는 메시지와 도구 호출보다 더 많은 것을 스트리밍합니다. 서버 측 **스트림 트랜스포머(stream transformer)**는 프로토콜이 클라이언트로 흐를 때 이를 검사하거나 다시 쓰고, 자체 구조화 데이터를 이름 있는 **커스텀 채널(custom channel)**에 게시할 수 있어요. 프론트엔드는 두 개의 셀렉터로 그 채널을 읽습니다: 최신 페이로드용 useExtension, 원시 이벤트 탈출구로 useChannel.

출처: 문서

본문

아래 예시는 브라우저에 도달하기 전에 모든 이벤트에서 PII(이메일, 전화번호, SSN, 카드 번호, IP)를 가리는 트랜스포머를 가진 고객 지원 에이전트이며, 실행 중인 마스킹 횟수를 redaction-stats 채널에 게시합니다. 사이드 패널이 그 횟수를 실시간으로 렌더링해요.

커스텀 채널이 어떻게 동작하는가 (How custom channels work)

커스텀 채널에는 두 끝이 있습니다. 서버에서 StreamTransformer가 이름 있는 StreamChannel을 열고 페이로드를 밀어 넣습니다. 클라이언트에서 셀렉터는 일치하는 custom:<name> 채널을 구독하고 페이로드를 반응형 상태로 노출합니다.

트랜스포머의 process 메서드는 모든 프로토콜 이벤트에 대해 실행됩니다. 이벤트를 제자리에서 변경할 수 있고(여기서는 messages, tools, values 데이터의 PII를 가림), 보고할 것이 있을 때마다 사이드 채널 업데이트를 밀어 넣을 수 있어요.

클라이언트 측 셀렉터(useExtension, useChannel)는 v1 프론트엔드 SDK 패키지(@langchain/react, @langchain/vue, @langchain/svelte, @langchain/angular)와 함께 제공됩니다.

스트림 트랜스포머와 `StreamChannel`은 `langgraph>=1.2`가 필요합니다.

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

에이전트를 만들 때 트랜스포머를 연결하세요:

from langchain.agents import create_agent

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

페이로드 타입은 트랜스포머가 밀어 넣는 것이 무엇이든 간에 정해집니다. 아래 클라이언트 예시는 이 형태를 읽습니다:

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 설정 (Setting up useStream)

useStream을 평소처럼 연결하세요. 커스텀 채널 셀렉터는 여기서 반환된 것과 같은 stream 핸들을 받습니다.

코드 예시는 타입 안전한 스트림 상태를 위해 `useStream`를 사용합니다. 타입 추론은 [Python](/oss/python/langchain/frontend/overview#type-inference) 또는 [JavaScript](/oss/javascript/langchain/frontend/overview#type-inference) 백엔드를 참고하세요.

React:

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

Angular:

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으로 최신 페이로드 읽기 (Read the latest payload with useExtension)

useExtension은 custom:<name> 채널을 구독하고 트랜스포머가 밀어 넣은 가장 최근 페이로드를 이미 풀고(unwrap) 타입이 지정된 상태로 반환합니다. UI가 라이브 카운터, 진행 퍼센트, 상태 배지 같은 현재 값만 필요할 때 인체공학적인 선택입니다.

custom: 프리픽스가 아닌 순수 채널 이름("redaction-stats")을 전달하세요:

React:

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

Angular:

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

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

반환 값은 각 프레임워크의 반응형 모델을 따릅니다. React·Svelte에서는 일반 값, Vue에서는 Ref(latest.value), Angular에서는 시그널(latest())이에요. 첫 페이로드가 도착할 때까지 값은 undefined입니다.

선택적인 세 번째 target 인자는 구독을 네임스페이스로 범위를 한정합니다. useMessages(stream, node)가 발견된 그래프 노드에 메시지를 범위 한정하는 것과 같아요. 네임스페이스 지정은 Graph execution을 참고하세요.

useChannel로 원시 이벤트 버퍼링 (Buffer raw events with useChannel)

useChannel은 원시 이벤트 탈출구입니다. 하나 이상의 채널을 구독하고, 단일 풀린 값 대신 기반 프로토콜 이벤트의 경계 있는 버퍼를 반환합니다. 최신 값 대신 이력이 필요할 때 — 이벤트 로그나 감사 추적 — 또는 더 높은 수준의 셀렉터가 다루지 않는 채널이 필요할 때 사용하세요.

전체 채널 ID("custom:redaction-stats")를 전달합니다:

React:

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"]);

Angular:

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, // target namespace
  { bufferSize: 200, replay: true },
);

옵션	기본값	효과
bufferSize	"default"	버퍼링되는 최대 이벤트 수. 상한에 도달하면 오래된 이벤트가 버려집니다.
replay	true	셀렉터가 마운트될 때 채널에서 이미 본 이벤트를 재생할지, 라이브 이벤트만 볼지.

일반적인 경우에는 더 높은 수준의 셀렉터(`useExtension`, `useMessages`, `useToolCalls`, `useValues`)를 선호하세요. 타입이 지정되고 풀린 값을 반환하며 렌더링하는 것만 추적합니다. 원시 이벤트 스트림이 구체적으로 필요할 때만 `useChannel`을 사용하세요.

useExtension vs useChannel 선택 (Choosing between useExtension and useChannel)

둘 다 같은 커스텀 채널을 읽지만 반환하는 것에서 다릅니다:

	useExtension	useChannel
반환	최신 페이로드 (T | undefined)	원시 이벤트의 경계 있는 버퍼 (Event[])
형태	풀리고(unwrapped) 타입 지정된 페이로드	원시 프로토콜 이벤트; event.params.data를 직접 풀기
구독 기준	채널 이름 ("redaction-stats")	전체 채널 ID (["custom:redaction-stats"])
사용 시점	현재 값이 필요할 때	이력, 로그, 또는 여러 채널이 필요할 때
옵션	—	bufferSize, replay

같은 채널에 둘 다 사용하는 것이 흔한 패턴입니다. useExtension이 라이브 요약(현재 총계)을 구동하고, useChannel이 스레드 전체의 모든 업데이트에 대한 스크롤 이벤트 로그를 담당합니다.

사용 사례 (Use cases)

커스텀 채널은 메시지, 도구 호출, 그래프 상태에 깔끔하게 매핑되지 않는 모든 서버 측 신호에 맞습니다:

• 컴플라이언스·마스킹 통계: 위 예시처럼 가려진 PII, 차단된 콘텐츠, 정책 적중 횟수.
• 진행 보고: 장기 실행 도구가 방출하는 완료 퍼센트나 단계 라벨.
• 라이브 메트릭: 실행 중 누적되는 토큰 사용량, 지연 시간, 비용.
• 출처와 인용: 에이전트가 답을 근거 지을 때 검색된 문서를 사이드 패널로 push.
• 도메인 이벤트: 메시지 대화록을 바꾸지 않고 백엔드가 표면화하려는 모든 구조화 업데이트.

더 알아보기 (Learn more)

• 개요 — LangGraph 프론트엔드 스트림 API와 아키텍처.
• Graph execution — 다중 노드 파이프라인용 네임스페이스 범위 셀렉터.
• event streaming — 서버 측 스트림 트랜스포머와 StreamChannel.