LangGraph 이벤트 스트리밍

LangGraph 이벤트 스트리밍 (Event streaming)

메시지, 상태, 서브그래프, 출력, 확장을 위한 타입화된 프로젝션(projection)으로 LangGraph 실행을 스트리밍하는 방법이에요. 이벤트 스트리밍은 대부분의 LangGraph 애플리케이션 코드에서 권장되는 인프로세스 스트리밍 모델로, 실행 스트림 객체를 반환하는데 이걸 여러 방식으로 동시에 소비할 수 있어요.

예를 들어 메시지를 토큰 단위로 찍어내면서 최종 상태도 함께 얻고 싶을 때 쓰기 좋아요. 기본 개념인 스트리밍과의 차이부터, 프로젝션을 직접 만드는 방법까지 하나씩 풀어볼게요.

출처: 공식문서

빠른 시작 (Quickstart)

stream = graph.stream_events({
    "messages": [{"role": "user", "content": "What is 42 * 17?"}],
}, version="v3")

for message in stream.messages:
    for token in message.text:
        print(token, end="", flush=True)

final_state = stream.output

에이전트 서버 뒤에 배포된 그래프를 스트리밍하려면 LangSmith Streaming API 문서를 참고하세요.

구성 요소가 맞물리는 방식

스트리밍 스택은 두 개의 레이어로 이루어져요.

1. 스트리밍(Streaming) — Pregel 엔진에서 원시 그래프 실행 이벤트를 내보내요.
2. 이벤트 스트리밍(Event streaming) — 그 이벤트들을 정규화하고, 스트림 트랜스포머를 통과시킨 뒤, 타입화된 프로젝션으로 노출해요.

두 레이어 사이의 다리 역할을 하는 게 **이벤트 라우터(event router)**예요. 정규화된 Pregel 이벤트를 받아 각각 등록된 스트림 트랜스포머에 통과시키죠. 내장 트랜스포머는 stream.messages, stream.values, stream.subgraphs, stream.output 같은 표준 프로젝션을 만들고, 커스텀 트랜스포머는 stream.extensions 아래에 애플리케이션별 프로젝션을 추가할 수 있어요.

이벤트 스트리밍이 제공하는 것

이벤트 스트리밍은 원시 그래프 실행 이벤트를 updates, values, messages, custom, checkpoints, tasks, debug 같은 stream_mode 방식으로 노출하는 스트리밍의 한 단계 위에 위치해요.

스트림 상태 (Stream state)

stream.values로 각 스텝 이후의 전체 상태 스냅샷을 스트리밍할 수 있어요.

stream = graph.stream_events(input, version="v3")

for snapshot in stream.values:
    print(snapshot)

final_state = stream.output

모든 프로토콜 이벤트 스트리밍

각 이벤트 이름에서 : 앞부분은 안정적인 그래프·노드 이름이고, 접미사는 실행마다 달라지는 런타임 ID예요. 특정 하위 트리만 신경 쓴다면 원시 이벤트를 네임스페이스로 직접 필터링하면 되는데, stream.subgraphs가 중첩 그래프 실행에 대해 이미 그 역할을 해줘요.

채널과 이벤트 라이프사이클

나만의 프로젝션 만들기

내장 프로젝션(stream.messages, stream.values, stream.subgraphs, stream.output)과 제품별 프로젝션(LangChain의 stream.tool_calls, Deep Agents의 stream.subagents)은 모두 이와 같은 계약(contract)을 쓰는 트랜스포머예요.

트랜스포머가 작동하는 방식

이벤트 스트리밍은 LangGraph Pregel 엔진의 스트리밍 출력으로 시작해요. 런타임이 그 청크들을 프로토콜 이벤트로 정규화하고, 스트림 핸들러가 각 이벤트를 스트림 트랜스포머 스택에 통과시키죠.

Pregel modes → Events → Built-in projections → User transformers → Run projections

스트림 핸들러는 하나의 스트림에 대한 중앙 디스패처예요. 프로토콜 이벤트마다 여러 작업을 수행하고, 트랜스포머는 관찰용(observational)이라서 그래프 런타임에 콜백을 하지 않아요. 대신 이벤트를 소비해 파생 값을 StreamChannel, promise, 또는 다른 프로젝션 객체에 밀어 넣어요.

트랜스포머 형태

from langgraph.stream import ProtocolEvent, StreamTransformer


class MyTransformer(StreamTransformer):
    def init(self) -> dict:
        ...

    def process(self, event: ProtocolEvent) -> bool:
        ...

    def finalize(self) -> None:

필수 스트림 모드 선언

required_stream_modes는 스트림 중에 기본 그래프가 어떤 Pregel 스트림 모드를 만들어낼지 제어해요. 런타임은 등록된 모든 트랜스포머의 required_stream_modes의 합집합을 구해서, 그 합집합을 그래프의 .stream() 호출에 stream_mode 인자로 넘겨요.

내장: ToolCallTransformer

LangGraph는 ToolCallTransformer를 내장으로 제공해요. 등록하면 일반 StateGraph에서도 stream.tool_calls를 노출할 수 있어요.

관련 자료

• LangChain 에이전트 스트리밍 — ReAct 스타일 에이전트 메시지, 도구 호출, 미들웨어 업데이트
• Deep Agents 스트리밍 — 서브에이전트, 중첩 메시지, 서브에이전트 도구 호출
• LangSmith Streaming API

더 알아보기 (Learn more)

• LangGraph 스트리밍 (Streaming)
• LangGraph 프론트엔드 패턴
• Agent Protocol