이벤트 스트리밍

이벤트 스트리밍 (Event streaming)

LangChain 에이전트 실행에서 실시간 업데이트를 스트리밍해요.

LangChain 에이전트는 LangGraph 위에 구축되므로, 메시지·툴 호출·상태·커스텀 업데이트에 대한 에이전트 중심 프로젝션과 함께 동일한 스트리밍 스택을 지원해요.

대부분의 애플리케이션·프론트엔드 사용 사례에서는 stream_events(..., version="v3")를 통한 이벤트 스트리밍을 사용하세요. 이벤트 스트리밍은 타입이 지정된 프로젝션을 가진 run 객체를 반환하므로, stream-mode 튜플을 파싱하는 대신 각 프로젝션을 독립적으로 소비할 수 있어요.

import { createAgent, tool } from "langchain";
import * as z from "zod";

const getWeather = tool(
  async ({ city }) => `It's always sunny in ${city}!`,
  {
    name: "get_weather",
    description: "Get weather for a city.",
    schema: z.object({ city: z.string() }),
  }
);

const agent = createAgent({
  model: "gpt-5-nano",
  tools: [getWeather],
});

const stream = await agent.streamEvents(
  { messages: [{ role: "user", content: "What is the weather in SF?" }] },
  { version: "v3" }
);

for await (const message of stream.messages) {
  for await (const delta of message.text) {
    process.stdout.write(delta);
  }
}

const finalState = await stream.output;

스트리밍할 수 있는 것 (What you can stream)

stream.messages는 메시지 스트림을 생성해요. 각 메시지 스트림은 .text, .reasoning, .toolCalls, .output, .usage를 노출해요. 비동기 프로젝션은 실시간 델타를 위해 반복하거나 최종 값을 위해 await할 수 있어요.

에이전트 메시지 (Agent messages)

각 LLM 호출의 모델 출력을 원할 때 stream.messages를 사용하세요.

const stream = await agent.streamEvents(input, { version: "v3" });

for await (const message of stream.messages) {
  process.stdout.write(`[${message.node}] `);
  for await (const delta of message.text) {
    process.stdout.write(delta);
  }

  const fullMessage = await message.output;
  console.log(fullMessage.content);

  const usage = await message.usage;
  if (usage) {
    console.log(usage);
  }
}

message.output은 제공자별 콘텐츠 블록을 포함한 확정된 AI 메시지를 제공해요. TypeScript에서는 토큰 개수나 기타 사용 메타데이터만 필요할 때 message.usage를 사용하고, Python에서는 message.output.usage_metadata에서 사용 정보를 읽어요.

추론 콘텐츠 (Reasoning content)

추론 콘텐츠는 텍스트 콘텐츠와 같은 형태를 사용하지만, 선택한 모델이 추론 블록을 방출할 때만 사용할 수 있어요.

const stream = await agent.streamEvents(input, { version: "v3" });

for await (const message of stream.messages) {
  for await (const delta of message.reasoning) {
    process.stdout.write(`[thinking] ${delta}`);
  }

  for await (const delta of message.text) {
    process.stdout.write(delta);
  }
}

모델 구성 세부 사항은 추론 가이드와 제공자의 통합 페이지를 참고하세요.

툴 호출 (Tool calls)

유용한 툴 호출 프로젝션은 두 가지예요:

• message.tool_calls는 모델이 툴 호출을 생성하는 동안 툴 호출 인자 청크를 스트리밍해요.
• stream.tool_calls는 툴 호출이 시작된 후 툴 실행의 수명 주기를 스트리밍해요.

const stream = await agent.streamEvents(input, { version: "v3" });

await Promise.all([
  (async () => {
    for await (const message of stream.messages) {
      for await (const chunk of message.toolCalls) {
        console.log("tool call chunk", chunk);
      }
    }
  })(),
  (async () => {
    for await (const call of stream.toolCalls) {
      console.log(call.name, call.input);
      console.log(await call.output, await call.error);
    }
  })(),
]);

서브 에이전트 스트리밍 (Streaming sub-agents)

createAgent 호출이 다른 이름 있는 createAgent(일반적으로 래핑 툴을 통해)를 호출하면, 내부 에이전트의 이벤트는 중첩 네임스페이스로 흘러요. createAgent에 전달하는 name이 스트림에서 그 내부 에이전트를 식별하므로, 에이전트별로 필터링·레이블링할 수 있어요.

이름 있는 서브 에이전트는 전용 stream.subagents 프로젝션에 표시돼요. 각 핸들은 내부 에이전트 자체의 .messages, .toolCalls, .output과 함께 .name(전달한 name=), .cause(서브 에이전트를 디스패치한 툴 호출), 중첩 .subagents를 노출해요. 이름 있는 createAgent 실행만 여기에 나타나므로 일반 서브그래프를 필터링할 필요가 없어요.

import { createAgent, tool } from "langchain";
import { z } from "zod";

const getWeather = tool(
  async ({ city }) => `It's always sunny in ${city}!`,
  { name: "get_weather", schema: z.object({ city: z.string() }) }
);

const weatherAgent = createAgent({
  model: "openai:gpt-5.5",
  tools: [getWeather],
  name: "weather_agent",
});

const callWeather = tool(
  async ({ query }) => {
    const result = await weatherAgent.invoke({
      messages: [{ role: "user", content: query }],
    });
    return result.messages.at(-1)?.text ?? "";
  },
  { name: "call_weather", schema: z.object({ query: z.string() }) }
);

const supervisor = createAgent({
  model: "openai:gpt-5.5",
  tools: [callWeather],
  name: "supervisor",
});

const stream = await supervisor.streamEvents(
  { messages: [{ role: "user", content: "What's the weather in Boston?" }] },
  { version: "v3" }
);

for await (const subagent of stream.subagents) {
  process.stdout.write(`${subagent.name}: `);
  for await (const message of subagent.messages) {
    for await (const token of message.text) {
      process.stdout.write(token);
    }
  }
  process.stdout.write("\n");
}
//Output: "weather_agent: The weather in Boston is sunny!"

툴에서 호출된 일반 StateGraph 서브그래프도 stream.subgraphs에 표시돼요 — .compile(name=...)에 name=을 설정하면 subagent.graph_name에 레이블이 표시돼요.

stream.subagents는 이름 있는 createAgent 서브 에이전트의 집중된 보기이고, stream.subgraphs는 모든 중첩 그래프를 포괄해요. UI에 맞는 것을 사용하세요.

상태와 최종 출력 (State and final output)

stream.values는 상태 스냅샷에, stream.output은 최종 에이전트 상태에 사용하세요.

const stream = await agent.streamEvents(input, { version: "v3" });

for await (const snapshot of stream.values) {
  console.log(snapshot);
}

const finalState = await stream.output;

여러 프로젝션 (Multiple projections)

JavaScript에서 여러 프로젝션을 원할 때 동시 소비자(concurrent consumers)를 사용하세요:

const stream = await agent.streamEvents(input, { version: "v3" });

await Promise.all([
  (async () => {
    for await (const message of stream.messages) {
      console.log(await message.text);
    }
  })(),
  (async () => {
    for await (const call of stream.toolCalls) {
      console.log(call.name, call.input);
    }
  })(),
]);

타입 프로젝션으로 노출되지 않은 채널에 접근하거나 전체 이벤트 엔벨로프를 검사하려면 원시 프로토콜 이벤트를 반복하세요:

for await (const event of stream) {
  console.log(event.method, event.params.namespace, event.params.data);
}

커스텀 업데이트 (Custom updates)

애플리케이션이 내장되지 않은 프로젝션(검색 진행 상황, 아티팩트, 도메인별 이벤트 등)을 필요로 할 때 커스텀 스트림 트랜스포머를 사용하세요.

const stream = await agent.streamEvents(input, {
  version: "v3",
  transformers: [toolActivityTransformer],
});

for await (const activity of stream.extensions.toolActivity) {
  console.log(activity);
}

미들웨어에 트랜스포머 등록 (Register transformers on middleware)

미들웨어 등록 트랜스포머는 [email protected] 이상이 필요해요.

미들웨어는 훅과 툴과 함께 스트림 트랜스포머 팩토리를 선언할 수 있어요. 팩토리 형태는 언어마다 달라요:

createMiddleware에 streamTransformers를 팩토리 튜플로 전달하세요. 각 팩토리는 () => StreamTransformer<any> 형태(인자 0개)이며 스코프당 한 번 호출돼요. 호출마다 새 트랜스포머를 반환하면 각 서브그래프가 격리 상태를 유지해요.

import { createAgent, createMiddleware } from "langchain";

const toolActivityMiddleware = createMiddleware({
  name: "ToolActivityMiddleware",
  streamTransformers: [toolActivityTransformer],
});

const agent = createAgent({
  model: "gpt-5-nano",
  tools: [getWeather],
  middleware: [toolActivityMiddleware],
});

컴파일 시 createAgent는 미들웨어 등록 팩토리를 자체 streamTransformers 옵션에 전달된 모든 것과 병합해요. 컴파일된 그래프의 최종 순서는:

1. 내장 ToolCallTransformer.
2. 미들웨어 등록 팩토리, 미들웨어 순서대로.
3. createAgent의 호출자 제공 streamTransformers.

이렇게 하면 내장 툴 호출 프로젝션이 소비자 트랜스포머 앞에 오고, 호출자가 제공한 항목이 최종 발언권을 가져요.

트랜스포머 계약은 직접 프로젝션 만들기를 참고하세요.

관련 (Related)

• Streaming이 저수준 Pregel 스트림 모드를 다뤄요.
• 직접 프로젝션 만들기가 애플리케이션 특화 프로젝션 작성법을 다뤄요.
• 프론트엔드 스트리밍 패턴이 스트리밍된 상태 위에 구축된 UI 사용 사례를 보여줘요.

출처: 문서

본문

이 페이지는 LangChain 에이전트의 이벤트 스트리밍을 다뤄요. streamEvents(..., version="v3")로 타입 프로젝션(stream.messages, message.text·reasoning·toolCalls·output·usage, stream.values·output·subgraphs·subagents·toolCalls·extensions)을 받아 독립적으로 소비할 수 있어요. 서브 에이전트·상태·최종 출력·다중 프로젝션·커스텀 스트림 트랜스포머(미들웨어 등록 포함)를 지원해요.

더 알아보기 (Learn more)

• LangChain JavaScript 이벤트 스트리밍 문서