스트리밍

스트리밍 (Streaming)

새 애플리케이션에는 이벤트 스트리밍을 권장해요. 이것은 LangGraph v1.2에서 도입된 타입 기반 프로젝션 API예요. 이벤트 스트리밍은 프로젝션별(메시지, 값, 서브그래프, 출력)로 별도의 이터레이터를 제공해, stream_mode 청크에서 분기하는 대신 독립적으로 소비할 수 있게 해줘요.

이 페이지는 LangGraph의 stream-mode API를 다뤄요. updates, values, messages, custom, checkpoints, tasks, debug 같은 stream mode를 통해 그래프 실행을 노출해요. 그래프 런타임 이벤트나 특정 stream-mode 출력에 직접 접근해야 할 때 사용해요.

출처: 문서

본문

시작하기 (Get started)

기본 사용법 (Basic usage)

LangGraph 그래프는 stream(동기)과 astream(비동기) 메서드를 노출해 스트리밍된 출력을 이터레이터로 내보내요. 하나 이상의 stream mode를 전달해 어떤 데이터를 받을지 제어해요.

for chunk in graph.stream(
    {"topic": "ice cream"},
    stream_mode=["updates", "custom"],  # [!code highlight]
    version="v2",  # [!code highlight]
):
    if chunk["type"] == "updates":
        for node_name, state in chunk["data"].items():
            print(f"Node {node_name} updated: {state}")
    elif chunk["type"] == "custom":
        print(f"Status: {chunk['data']['status']}")

Status: thinking of a joke...
Node generate_joke updated: {'joke': 'Why did the ice cream go to school? To get a sundae education!'}

전체 예시 (Full example)

from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.config import get_stream_writer


class State(TypedDict):
    topic: str
    joke: str


def generate_joke(state: State):
    writer = get_stream_writer()
    writer({"status": "thinking of a joke..."})
    return {"joke": f"Why did the {state['topic']} go to school? To get a sundae education!"}

graph = (
    StateGraph(State)
    .add_node(generate_joke)
    .add_edge(START, "generate_joke")
    .add_edge("generate_joke", END)
    .compile()
)

for chunk in graph.stream(
    {"topic": "ice cream"},
    stream_mode=["updates", "custom"],
    version="v2",
):
    if chunk["type"] == "updates":
        for node_name, state in chunk["data"].items():
            print(f"Node {node_name} updated: {state}")
    elif chunk["type"] == "custom":
        print(f"Status: {chunk['data']['status']}")

Status: thinking of a joke...
Node generate_joke updated: {'joke': 'Why did the ice cream go to school? To get a sundae education!'}

LangSmith로 스트리밍 이벤트를 디버깅하고, 토큰 단위 LLM 출력을 살펴보며, 지연 시간을 모니터링해요. 트레이싱 퀵스타트를 따라 설정해요.

스트림 출력 형식 (v2) (Stream output format)

LangGraph 1.1 이상이 필요해요. 이 페이지의 모든 예시는 version="v2"를 사용해요.

stream()이나 astream()에 version="v2"를 전달하면 통일된 출력 형식을 얻을 수 있어요. 모든 청크는 stream mode, 모드 수, 서브그래프 설정과 무관하게 일관된 모양의 StreamPart 딕셔너리예요.

{
    "type": "values" | "updates" | "messages" | "custom" | "checkpoints" | "tasks" | "debug",
    "ns": (),           # namespace tuple, populated for subgraph events
    "data": ...,        # the actual payload (type varies by stream mode)
}

각 stream mode에는 대응하는 TypedDict가 있어요. ValuesStreamPart, UpdatesStreamPart, MessagesStreamPart, CustomStreamPart, CheckpointStreamPart, TasksStreamPart, DebugStreamPart예요. 이 타입들은 langgraph.types에서 가져올 수 있어요. 유니온 타입 StreamPart은 part["type"]에 대한 분리 유니온(disjoint union)이라 편집기와 타입 체커에서 완전한 타입 내로잉을 켤 수 있어요.

v1(기본값)에서는 출력 형식이 스트리밍 옵션에 따라 달라져요(단일 모드는 raw data, 여러 모드는 (mode, data) 튜플, 서브그래프는 (namespace, data) 튜플). v2에서는 형식이 항상 같아요.

# v2 (new)
for chunk in graph.stream(inputs, stream_mode="updates", version="v2"):
    print(chunk["type"])  # "updates"
    print(chunk["ns"])    # ()
    print(chunk["data"])  # {"node_name": {"key": "value"}}

# v1 (current default)
for chunk in graph.stream(inputs, stream_mode="updates"):
    print(chunk)  # {"node_name": {"key": "value"}}

v2 형식은 타입 내로잉도 가능하게 해요. chunk["type"]으로 청크를 필터링하면 올바른 페이로드 타입을 얻을 수 있어요. 각 분기는 part["data"]를 해당 모드의 특정 타입으로 좁혀요.

for part in graph.stream(
    {"topic": "ice cream"},
    stream_mode=["values", "updates", "messages", "custom"],
    version="v2",
):
    if part["type"] == "values":
        # ValuesStreamPart — full state snapshot after each step
        print(f"State: topic={part['data']['topic']}")
    elif part["type"] == "updates":
        # UpdatesStreamPart — only the changed keys from each node
        for node_name, state in part["data"].items():
            print(f"Node `{node_name}` updated: {state}")
    elif part["type"] == "messages":
        # MessagesStreamPart — (message_chunk, metadata) from LLM calls
        msg, metadata = part["data"]
        print(msg.content, end="", flush=True)
    elif part["type"] == "custom":
        # CustomStreamPart — arbitrary data from get_stream_writer()
        print(f"Progress: {part['data']['progress']}%")

스트림 모드 (Stream modes)

다음 stream mode들을 목록으로 stream 또는 astream 메서드에 하나 이상 전달해요.

Mode	Type	Description
values	ValuesStreamPart	Full state after each step.
updates	UpdatesStreamPart	State updates after each step. Multiple updates in the same step are streamed separately.
messages	MessagesStreamPart	2-tuples of (LLM token, metadata) from LLM calls.
custom	CustomStreamPart	Custom data emitted from nodes via get_stream_writer.
checkpoints	CheckpointStreamPart	Checkpoint events (same format as get_state()). Requires a checkpointer.
tasks	TasksStreamPart	Task start/finish events with results and errors. Requires a checkpointer.
debug	DebugStreamPart	All available info — combines checkpoints and tasks with extra metadata.

그래프 상태 (Graph state)

updates와 values stream mode를 사용해 그래프가 실행되는 동안 그 상태를 스트리밍해요.

• updates는 그래프의 각 단계 후 상태에 대한 업데이트를 스트리밍해요.
• values는 그래프의 각 단계 후 상태의 전체 값을 스트리밍해요.

from typing import TypedDict
from langgraph.graph import StateGraph, START, END


class State(TypedDict):
  topic: str
  joke: str


def refine_topic(state: State):
    return {"topic": state["topic"] + " and cats"}


def generate_joke(state: State):
    return {"joke": f"This is a joke about {state['topic']}"}

graph = (
  StateGraph(State)
  .add_node(refine_topic)
  .add_node(generate_joke)
  .add_edge(START, "refine_topic")
  .add_edge("refine_topic", "generate_joke")
  .add_edge("generate_joke", END)
  .compile()
)

updates

노드들이 각 단계 후 반환하는 상태 업데이트만 스트리밍하는 데 사용해요. 스트리밍된 출력에는 노드 이름과 업데이트가 포함돼요.

for chunk in graph.stream(
    {"topic": "ice cream"},
    stream_mode="updates",  # [!code highlight]
    version="v2",  # [!code highlight]
):
    if chunk["type"] == "updates":
        for node_name, state in chunk["data"].items():
            print(f"Node `{node_name}` updated: {state}")

Node `refine_topic` updated: {'topic': 'ice cream and cats'}
Node `generate_joke` updated: {'joke': 'This is a joke about ice cream and cats'}

values

각 단계 후 그래프의 전체 상태를 스트리밍하는 데 사용해요.

for chunk in graph.stream(
    {"topic": "ice cream"},
    stream_mode="values",  # [!code highlight]
    version="v2",  # [!code highlight]
):
    if chunk["type"] == "values":
        print(f"topic: {chunk['data']['topic']}, joke: {chunk['data']['joke']}")

topic: ice cream, joke:
topic: ice cream and cats, joke:
topic: ice cream and cats, joke: This is a joke about ice cream and cats

LLM 토큰 (LLM tokens)

messages 스트리밍 모드를 사용해 노드, 도구, 서브그래프, 태스크를 포함한 그래프의 어떤 부분이든 Large Language Model(LLM) 출력을 토큰 단위로 스트리밍해요.

messages 모드의 스트리밍된 출력은 (message_chunk, metadata) 튜플이에요.

• message_chunk: LLM의 토큰 또는 메시지 세그먼트.
• metadata: 그래프 노드와 LLM 호출에 대한 세부 정보를 담은 딕셔너리.

LLM이 LangChain 통합으로 제공되지 않는다면 custom 모드를 사용해 출력을 스트리밍할 수 있어요. 모든 LLM과 함께 사용하기를 참고해요.

Python < 3.11에서의 비동기에는 수동 config 필요 Python < 3.11과 함께 비동기 코드를 쓸 때는 올바른 스트리밍을 위해 ainvoke()에 RunnableConfig를 명시적으로 전달해야 해요. Python < 3.11에서의 비동기를 참고하거나 Python 3.11+로 업그레이드해요.

from dataclasses import dataclass

from langchain.chat_models import init_chat_model
from langgraph.graph import StateGraph, START


@dataclass
class MyState:
    topic: str
    joke: str = ""


model = init_chat_model(model="gpt-5.4-mini")

def call_model(state: MyState):
    """Call the LLM to generate a joke about a topic"""
    # Note that message events are emitted even when the LLM is run using .invoke rather than .stream
    model_response = model.invoke(  # [!code highlight]
        [
            {"role": "user", "content": f"Generate a joke about {state.topic}"}
        ]
    )
    return {"joke": model_response.content}

graph = (
    StateGraph(MyState)
    .add_node(call_model)
    .add_edge(START, "call_model")
    .compile()
)

# The "messages" stream mode streams LLM tokens with metadata
# Use version="v2" for a unified StreamPart format
for chunk in graph.stream(
    {"topic": "ice cream"},
    stream_mode="messages",  # [!code highlight]
    version="v2",  # [!code highlight]
):
    if chunk["type"] == "messages":
        message_chunk, metadata = chunk["data"]
        if message_chunk.content:
            print(message_chunk.content, end="|", flush=True)

LLM 호출로 필터링 (Filter by LLM invocation)

LLM 호출에 tags를 연결하면 LLM 호출별로 스트리밍된 토큰을 필터링할 수 있어요.

from langchain.chat_models import init_chat_model

# model_1 is tagged with "joke"
model_1 = init_chat_model(model="gpt-5.4-mini", tags=['joke'])
# model_2 is tagged with "poem"
model_2 = init_chat_model(model="gpt-5.4-mini", tags=['poem'])

graph = ... # define a graph that uses these LLMs

# The stream_mode is set to "messages" to stream LLM tokens
# The metadata contains information about the LLM invocation, including the tags
async for chunk in graph.astream(
    {"topic": "cats"},
    stream_mode="messages",  # [!code highlight]
    version="v2",  # [!code highlight]
):
    if chunk["type"] == "messages":
        msg, metadata = chunk["data"]
        # Filter the streamed tokens by the tags field in the metadata to only include
        # the tokens from the LLM invocation with the "joke" tag
        if metadata["tags"] == ["joke"]:
            print(msg.content, end="|", flush=True)

확장 예시: 태그로 필터링

from typing import TypedDict

from langchain.chat_models import init_chat_model
from langgraph.graph import START, StateGraph

# The joke_model is tagged with "joke"
joke_model = init_chat_model(model="gpt-5.4-mini", tags=["joke"])
# The poem_model is tagged with "poem"
poem_model = init_chat_model(model="gpt-5.4-mini", tags=["poem"])


class State(TypedDict):
      topic: str
      joke: str
      poem: str


async def call_model(state, config):
      topic = state["topic"]
      print("Writing joke...")
      # Note: Passing the config through explicitly is required for python < 3.11
      # Since context var support wasn't added before then: https://docs.python.org/3/library/asyncio-task.html#creating-tasks
      # The config is passed through explicitly to ensure the context vars are propagated correctly
      # This is required for Python < 3.11 when using async code. Please see the async section for more details
      joke_response = await joke_model.ainvoke(
            [{"role": "user", "content": f"Write a joke about {topic}"}],
            config,
      )
      print("\n\nWriting poem...")
      poem_response = await poem_model.ainvoke(
            [{"role": "user", "content": f"Write a short poem about {topic}"}],
            config,
      )
      return {"joke": joke_response.content, "poem": poem_response.content}


graph = (
      StateGraph(State)
      .add_node(call_model)
      .add_edge(START, "call_model")
      .compile()
)

# The stream_mode is set to "messages" to stream LLM tokens
# The metadata contains information about the LLM invocation, including the tags
async for chunk in graph.astream(
      {"topic": "cats"},
      stream_mode="messages",
      version="v2",
):
    if chunk["type"] == "messages":
        msg, metadata = chunk["data"]
        if metadata["tags"] == ["joke"]:
            print(msg.content, end="|", flush=True)

스트림에서 메시지 생략 (Omit messages from the stream)

nostream 태그를 사용하면 LLM 출력을 스트림에서 완전히 제외할 수 있어요. nostream으로 태그된 호출은 여전히 실행되고 출력을 만들지만, 그 토큰은 messages 모드에서 emit되지 않아요.

이런 경우에 유용해요.

• 내부 처리를 위해 LLM 출력이 필요하지만(예: 구조화 출력) 클라이언트로 스트리밍하고 싶지 않을 때
• 같은 내용을 다른 채널(예: 커스텀 UI 메시지)로 스트리밍하고 messages 스트림에서 중복 출력을 피하고 싶을 때

from typing import Any, TypedDict

from langchain_anthropic import ChatAnthropic
from langgraph.graph import START, StateGraph

stream_model = ChatAnthropic(model_name="claude-haiku-4-5-20251001")
internal_model = ChatAnthropic(model_name="claude-haiku-4-5-20251001").with_config(
    {"tags": ["nostream"]}
)


class State(TypedDict):
    topic: str
    answer: str
    notes: str


def answer(state: State) -> dict[str, Any]:
    r = stream_model.invoke(
        [{"role": "user", "content": f"Reply briefly about {state['topic']}"}]
    )
    return {"answer": r.content}


def internal_notes(state: State) -> dict[str, Any]:
    # Tokens from this model are omitted from stream_mode="messages" because of nostream
    r = internal_model.invoke(
        [{"role": "user", "content": f"Private notes on {state['topic']}"}]
    )
    return {"notes": r.content}


graph = (
    StateGraph(State)
    .add_node("write_answer", answer)
    .add_node("internal_notes", internal_notes)
    .add_edge(START, "write_answer")
    .add_edge("write_answer", "internal_notes")
    .compile()
)

initial_state: State = {"topic": "AI", "answer": "", "notes": ""}
stream = graph.stream_events(initial_state, version="v3")

Example trace 보기: 이 예시의 공개 LangSmith run을 열어보세요.

노드로 필터링 (Filter by node)

특정 노드에서만 토큰을 스트리밍하려면 stream_mode="messages"를 사용하고 스트리밍된 메타데이터의 langgraph_node 필드로 출력을 필터링해요.

# The "messages" stream mode streams LLM tokens with metadata
# Use version="v2" for a unified StreamPart format
for chunk in graph.stream(
    inputs,
    stream_mode="messages",  # [!code highlight]
    version="v2",  # [!code highlight]
):
    if chunk["type"] == "messages":
        msg, metadata = chunk["data"]
        # Filter the streamed tokens by the langgraph_node field in the metadata
        # to only include the tokens from the specified node
        if msg.content and metadata["langgraph_node"] == "some_node_name":
            ...

확장 예시: 특정 노드에서 LLM 토큰 스트리밍

from typing import TypedDict
from langgraph.graph import START, StateGraph
from langchain_openai import ChatOpenAI

model = ChatOpenAI(model="gpt-5.4-mini")


class State(TypedDict):
      topic: str
      joke: str
      poem: str


def write_joke(state: State):
      topic = state["topic"]
      joke_response = model.invoke(
            [{"role": "user", "content": f"Write a joke about {topic}"}]
      )
      return {"joke": joke_response.content}


def write_poem(state: State):
      topic = state["topic"]
      poem_response = model.invoke(
            [{"role": "user", "content": f"Write a short poem about {topic}"}]
      )
      return {"poem": poem_response.content}


graph = (
      StateGraph(State)
      .add_node(write_joke)
      .add_node(write_poem)
      # write both the joke and the poem concurrently
      .add_edge(START, "write_joke")
      .add_edge(START, "write_poem")
      .compile()
)

# The "messages" stream mode streams LLM tokens with metadata
# Use version="v2" for a unified StreamPart format
for chunk in graph.stream(
    {"topic": "cats"},
    stream_mode="messages",  # [!code highlight]
    version="v2",  # [!code highlight]
):
    if chunk["type"] == "messages":
        msg, metadata = chunk["data"]
        # Filter the streamed tokens by the langgraph_node field in the metadata
        # to only include the tokens from the write_poem node
        if msg.content and metadata["langgraph_node"] == "write_poem":
            print(msg.content, end="|", flush=True)

커스텀 데이터 (Custom data)

LangGraph 노드나 도구 안에서 커스텀 사용자 정의 데이터를 보내려면 다음 단계를 따라요.

1. get_stream_writer를 사용해 stream writer에 접근하고 커스텀 데이터를 emit해요.
2. .stream()이나 .astream()을 호출할 때 stream_mode="custom"을 설정해 스트림에서 커스텀 데이터를 받아요. 여러 모드를 결합할 수 있지만(예: ["updates", "custom"]), 적어도 하나는 "custom"이어야 해요.

Python < 3.11 비동기에는 get_stream_writer 없음 Python < 3.11에서 실행되는 비동기 코드에서는 get_stream_writer가 동작하지 않아요. 대신 노드나 도구에 writer 파라미터를 추가하고 수동으로 전달해요. 사용 예시는 Python < 3.11에서의 비동기를 참고해요.

node

from typing import TypedDict
from langgraph.config import get_stream_writer
from langgraph.graph import StateGraph, START

class State(TypedDict):
    query: str
    answer: str

def node(state: State):
    # Get the stream writer to send custom data
    writer = get_stream_writer()
    # Emit a custom key-value pair (e.g., progress update)
    writer({"custom_key": "Generating custom data inside node"})
    return {"answer": "some data"}

graph = (
    StateGraph(State)
    .add_node(node)
    .add_edge(START, "node")
    .compile()
)

inputs = {"query": "example"}

# Set stream_mode="custom" to receive the custom data in the stream
for chunk in graph.stream(inputs, stream_mode="custom", version="v2"):
    if chunk["type"] == "custom":
        print(f"Custom event: {chunk['data']['custom_key']}")

tool

from langchain.tools import tool
from langgraph.config import get_stream_writer

@tool
def query_database(query: str) -> str:
    """Query the database."""
    # Access the stream writer to send custom data
    writer = get_stream_writer()  # [!code highlight]
    # Emit a custom key-value pair (e.g., progress update)
    writer({"data": "Retrieved 0/100 records", "type": "progress"})  # [!code highlight]
    # perform query
    # Emit another custom key-value pair
    writer({"data": "Retrieved 100/100 records", "type": "progress"})
    return "some-answer"


graph = ... # define a graph that uses this tool

# Set stream_mode="custom" to receive the custom data in the stream
for chunk in graph.stream(inputs, stream_mode="custom", version="v2"):
    if chunk["type"] == "custom":
        print(f"{chunk['data']['type']}: {chunk['data']['data']}")

서브그래프 출력 (Subgraph outputs)

서브그래프의 출력을 스트리밍된 출력에 포함하려면 부모 그래프의 .stream() 메서드에서 subgraphs=True를 설정할 수 있어요. 이러면 부모 그래프와 모든 서브그래프의 출력이 스트리밍돼요.

출력은 (namespace, data) 튜플로 스트리밍돼요. namespace는 서브그래프가 호출되는 노드의 경로가 있는 튜플이에요. 예: ("parent_node:<task_id>", "child_node:<task_id>").

v2 (LangGraph 1.1 이상)

version="v2"에서는 서브그래프 이벤트가 같은 StreamPart 형식을 사용해요. ns 필드가 출처를 식별해요.

for chunk in graph.stream(
    {"foo": "foo"},
    subgraphs=True,  # [!code highlight]
    stream_mode="updates",
    version="v2", # [!code highlight]
):
    print(chunk["type"])  # "updates"
    print(chunk["ns"])    # () for root, ("node_name:<task_id>",) for subgraph
    print(chunk["data"])  # {"node_name": {"key": "value"}}

v1 (기본값)

for chunk in graph.stream(
    {"foo": "foo"},
    # Set subgraphs=True to stream outputs from subgraphs
    subgraphs=True,  # [!code highlight]
    stream_mode="updates",
):
    print(chunk)

이는 "messages"를 포함한 모든 stream_mode에 적용돼요. create_agent 같은 에이전트 빌더는 컴파일된 그래프를 반환하므로, 하나를 노드로 추가하면 서브그래프가 돼요. subgraphs=True가 없으면 부모 그래프의 stream_mode="messages"는 내부 에이전트의 LLM 호출에서 토큰 청크를 emit하지 않아요. agent.stream(...)을 직접 호출하면 emit이 되기 때문에, 이 문제는 흔히 래핑 후에만 나타나요.

from langchain.agents import create_agent
from langgraph.graph import END, START, StateGraph

graph = (
    StateGraph(State)
    .add_node("agent", create_agent(model, tools, state_schema=State))
    .add_edge(START, "agent")
    .add_edge("agent", END)
    .compile()
)

for chunk in graph.stream(
    {"messages": [{"role": "user", "content": "..."}]},
    stream_mode="messages",
    subgraphs=True,  # [!code highlight]
    version="v2",
):
    print(chunk["type"])  # "messages"
    print(chunk["ns"])    # () for root, ("agent:<task_id>",) for subgraph
    print(chunk["data"])  # (token, metadata)

확장 예시: 서브그래프에서 스트리밍

from langgraph.graph import START, StateGraph
from typing import TypedDict

# Define subgraph
class SubgraphState(TypedDict):
    foo: str  # note that this key is shared with the parent graph state
    bar: str

def subgraph_node_1(state: SubgraphState):
    return {"bar": "bar"}

def subgraph_node_2(state: SubgraphState):
    return {"foo": state["foo"] + state["bar"]}

subgraph_builder = StateGraph(SubgraphState)
subgraph_builder.add_node(subgraph_node_1)
subgraph_builder.add_node(subgraph_node_2)
subgraph_builder.add_edge(START, "subgraph_node_1")
subgraph_builder.add_edge("subgraph_node_1", "subgraph_node_2")
subgraph = subgraph_builder.compile()

# Define parent graph
class ParentState(TypedDict):
    foo: str

def node_1(state: ParentState):
    return {"foo": "hi! " + state["foo"]}

builder = StateGraph(ParentState)
builder.add_node("node_1", node_1)
builder.add_node("node_2", subgraph)
builder.add_edge(START, "node_1")
builder.add_edge("node_1", "node_2")
graph = builder.compile()

for chunk in graph.stream(
    {"foo": "foo"},
    stream_mode="updates",
    # Set subgraphs=True to stream outputs from subgraphs
    subgraphs=True,  # [!code highlight]
    version="v2",  # [!code highlight]
):
    if chunk["type"] == "updates":
        if chunk["ns"]:
            print(f"Subgraph {chunk['ns']}: {chunk['data']}")
        else:
            print(f"Root: {chunk['data']}")

Root: {'node_1': {'foo': 'hi! foo'}}
Subgraph ('node_2:dfddc4ba-c3c5-6887-5012-a243b5b377c2',): {'subgraph_node_1': {'bar': 'bar'}}
Subgraph ('node_2:dfddc4ba-c3c5-6887-5012-a243b5b377c2',): {'subgraph_node_2': {'foo': 'hi! foobar'}}
Root: {'node_2': {'foo': 'hi! foobar'}}

참고 우리는 노드 업데이트뿐 아니라, 어떤 그래프(또는 서브그래프)에서 스트리밍 중인지 알려주는 네임스페이스도 받고 있어요.

체크포인트 (Checkpoints)

그래프가 실행되는 동안 체크포인트 이벤트를 받으려면 checkpoints 스트리밍 모드를 사용해요. 각 체크포인트 이벤트는 get_state()의 출력과 같은 형식이에요. checkpointer가 필요해요.

from langgraph.checkpoint.memory import MemorySaver

graph = (
    StateGraph(State)
    .add_node(refine_topic)
    .add_node(generate_joke)
    .add_edge(START, "refine_topic")
    .add_edge("refine_topic", "generate_joke")
    .add_edge("generate_joke", END)
    .compile(checkpointer=MemorySaver())
)

config = {"configurable": {"thread_id": "1"}}

for chunk in graph.stream(
    {"topic": "ice cream"},
    config=config,
    stream_mode="checkpoints",  # [!code highlight]
    version="v2",  # [!code highlight]
):
    if chunk["type"] == "checkpoints":
        print(chunk["data"])

태스크 (Tasks)

그래프가 실행되는 동안 태스크 시작·종료 이벤트를 받으려면 tasks 스트리밍 모드를 사용해요. 태스크 이벤트는 어떤 노드가 실행 중인지, 그 결과, 오류에 대한 정보를 포함해요. checkpointer가 필요해요.

from langgraph.checkpoint.memory import MemorySaver

graph = (
    StateGraph(State)
    .add_node(refine_topic)
    .add_node(generate_joke)
    .add_edge(START, "refine_topic")
    .add_edge("refine_topic", "generate_joke")
    .add_edge("generate_joke", END)
    .compile(checkpointer=MemorySaver())
)

config = {"configurable": {"thread_id": "1"}}

for chunk in graph.stream(
    {"topic": "ice cream"},
    config=config,
    stream_mode="tasks",  # [!code highlight]
    version="v2",  # [!code highlight]
):
    if chunk["type"] == "tasks":
        print(chunk["data"])

디버그 (Debug)

그래프 실행 전반에 걸쳐 가능한 많은 정보를 스트리밍하려면 debug 스트리밍 모드를 사용해요. 스트리밍된 출력에는 노드 이름과 전체 상태가 포함돼요.

for chunk in graph.stream(
    {"topic": "ice cream"},
    stream_mode="debug",  # [!code highlight]
    version="v2",  # [!code highlight]
):
    if chunk["type"] == "debug":
        print(chunk["data"])

debug 모드는 checkpoints와 tasks 이벤트에 추가 메타데이터를 결합해요. 디버그 정보의 일부만 필요하다면 checkpoints나 tasks를 직접 사용해요.

여러 모드를 한 번에 (Multiple modes at once)

stream_mode 파라미터에 목록을 전달해 여러 모드를 한 번에 스트리밍할 수 있어요.

version="v2"에서는 모든 청크가 StreamPart 딕셔너리예요. chunk["type"]으로 모드를 구분해요.

# v2
for chunk in graph.stream(inputs, stream_mode=["updates", "custom"], version="v2"):
    if chunk["type"] == "updates":
        for node_name, state in chunk["data"].items():
            print(f"Node `{node_name}` updated: {state}")
    elif chunk["type"] == "custom":
        print(f"Custom event: {chunk['data']}")

# v1
for mode, chunk in graph.stream(inputs, stream_mode=["updates", "custom"]):
    print(chunk)

고급 (Advanced)

모든 LLM과 사용하기 (Use with any LLM)

stream_mode="custom"을 사용하면 어떤 LLM API에서든 데이터를 스트리밍할 수 있어요. 그 API가 LangChain 채팅 모델 인터페이스를 구현하지 않아도 상관없어요.

이를 통해 자체 스트리밍 인터페이스를 제공하는 raw LLM 클라이언트나 외부 서비스를 통합할 수 있어, 커스텀 설정에 매우 유연해요.

from langgraph.config import get_stream_writer

def call_arbitrary_model(state):
    """Example node that calls an arbitrary model and streams the output"""
    # Get the stream writer to send custom data
    writer = get_stream_writer()  # [!code highlight]
    # Assume you have a streaming client that yields chunks
    # Generate LLM tokens using your custom streaming client
    for chunk in your_custom_streaming_client(state["topic"]):
        # Use the writer to send custom data to the stream
        writer({"custom_llm_chunk": chunk})  # [!code highlight]
    return {"result": "completed"}

graph = (
    StateGraph(State)
    .add_node(call_arbitrary_model)
    # Add other nodes and edges as needed
    .compile()
)
# Set stream_mode="custom" to receive the custom data in the stream
for chunk in graph.stream(
    {"topic": "cats"},
    stream_mode="custom",  # [!code highlight]
    version="v2",  # [!code highlight]
):
    if chunk["type"] == "custom":
        # The chunk data will contain the custom data streamed from the llm
        print(chunk["data"])

확장 예시: 임의 채팅 모델 스트리밍

import operator
import json

from typing import TypedDict
from typing_extensions import Annotated
from langgraph.graph import StateGraph, START

from openai import AsyncOpenAI

openai_client = AsyncOpenAI()
model_name = "gpt-5.4-mini"


async def stream_tokens(model_name: str, messages: list[dict]):
    response = await openai_client.chat.completions.create(
        messages=messages, model=model_name, stream=True
    )
    role = None
    async for chunk in response:
        delta = chunk.choices[0].delta

        if delta.role is not None:
            role = delta.role

        if delta.content:
            yield {"role": role, "content": delta.content}


# this is our tool
async def get_items(place: str) -> str:
    """Use this tool to list items one might find in a place you're asked about."""
    writer = get_stream_writer()
    response = ""
    async for msg_chunk in stream_tokens(
        model_name,
        [
            {
                "role": "user",
                "content": (
                    "Can you tell me what kind of items "
                    f"i might find in the following place: '{place}'. "
                    "List at least 3 such items separating them by a comma. "
                    "And include a brief description of each item."
                ),
            }
        ],
    ):
        response += msg_chunk["content"]
        writer(msg_chunk)

    return response


class State(TypedDict):
    messages: Annotated[list[dict], operator.add]


# this is the tool-calling graph node
async def call_tool(state: State):
    ai_message = state["messages"][-1]
    tool_call = ai_message["tool_calls"][-1]

    function_name = tool_call["function"]["name"]
    if function_name != "get_items":
        raise ValueError(f"Tool {function_name} not supported")

    function_arguments = tool_call["function"]["arguments"]
    arguments = json.loads(function_arguments)

    function_response = await get_items(**arguments)
    tool_message = {
        "tool_call_id": tool_call["id"],
        "role": "tool",
        "name": function_name,
        "content": function_response,
    }
    return {"messages": [tool_message]}


graph = (
    StateGraph(State)
    .add_node(call_tool)
    .add_edge(START, "call_tool")
    .compile()
)

도구 호출을 포함한 AIMessage로 그래프를 호출해 볼게요.

inputs = {
    "messages": [
        {
            "content": None,
            "role": "assistant",
            "tool_calls": [
                {
                    "id": "1",
                    "function": {
                        "arguments": '{"place":"bedroom"}',
                        "name": "get_items",
                    },
                    "type": "function",
                }
            ],
        }
    ]
}

async for chunk in graph.astream(
    inputs,
    stream_mode="custom",
    version="v2",
):
    if chunk["type"] == "custom":
        print(chunk["data"]["content"], end="|", flush=True)

특정 채팅 모델의 스트리밍 비활성화 (Disable streaming for specific chat models)

스트리밍을 지원하는 모델과 그렇지 않은 모델을 섞어 쓰는 애플리케이션이라면, 스트리밍을 지원하지 않는 모델의 스트리밍을 명시적으로 비활성화해야 할 수 있어요.

모델을 초기화할 때 streaming=False를 설정해요.

from langchain.chat_models import init_chat_model

model = init_chat_model(
    "claude-sonnet-4-6",
    # Set streaming=False to disable streaming for the chat model
    streaming=False  # [!code highlight]
)

또는 채팅 모델 인터페이스로:

from langchain_openai import ChatOpenAI

# Set streaming=False to disable streaming for the chat model
model = ChatOpenAI(model="gpt-5.5", streaming=False)

모든 채팅 모델 통합이 streaming 파라미터를 지원하는 건 아니에요. 모델이 지원하지 않는다면 disable_streaming=True를 사용해요. 이 파라미터는 기본 클래스를 통해 모든 채팅 모델에서 사용할 수 있어요.

v2로 마이그레이션 (Migrate to v2)

이 페이지에서 사용한 v2 스트리밍 형식은 통일된 출력 형식을 제공해요. 핵심 차이점과 마이그레이션 방법을 요약하면 다음과 같아요.

Scenario	v1 (default)	v2 (version="v2")
Single stream mode	Raw data (dict)	StreamPart dict with type, ns, data
Multiple stream modes	(mode, data) tuples	Same StreamPart dict, filter on chunk["type"]
Subgraph streaming	(namespace, data) tuples	Same StreamPart dict, check chunk["ns"]
Multiple modes + subgraphs	(namespace, mode, data) triples	Same StreamPart dict
invoke() return type	Plain dict (state)	GraphOutput with .value and .interrupts
Interrupt location (stream)	__interrupt__ key in state dict	interrupts field on values stream parts
Interrupt location (invoke)	__interrupt__ key in result dict	.interrupts attribute on GraphOutput
Pydantic/dataclass output	Returns plain dict	Coerces to model/dataclass instance

v2 invoke 형식

invoke()나 ainvoke()에 version="v2"를 전달하면 .value와 .interrupts 속성을 가진 GraphOutput 객체를 반환해요.

from langgraph.types import GraphOutput

result = graph.invoke(inputs, version="v2")

assert isinstance(result, GraphOutput)
result.value       # your output — dict, Pydantic model, or dataclass
result.interrupts  # tuple[Interrupt, ...], empty if none occurred

기본 "values" 외의 stream mode에서는 invoke(..., stream_mode="updates", version="v2")가 list[tuple] 대신 list[StreamPart]를 반환해요.

GraphOutput에 대한 딕셔너리 스타일 접근(result["key"], "key" in result, result["__interrupt__"])은 하위 호환성을 위해 여전히 동작하지만 더 이상 권장되지 않고 미래 버전에서 제거될 거예요. result.value와 result.interrupts로 마이그레이션하세요.

이렇게 하면 상태와 인터럽트 메타데이터가 분리돼요. v1에서는 인터럽트가 반환된 딕셔너리의 __interrupt__ 아래에 포함돼요.

# v2 (new)
config = {"configurable": {"thread_id": "thread-1"}}
result = graph.invoke(inputs, config=config, version="v2")

if result.interrupts:
    print(result.interrupts[0].value)
    graph.invoke(Command(resume=True), config=config, version="v2")

# v1 (current default)
config = {"configurable": {"thread_id": "thread-1"}}
result = graph.invoke(inputs, config=config)

if "__interrupt__" in result:
    print(result["__interrupt__"][0].value)
    graph.invoke(Command(resume=True), config=config)

Pydantic·dataclass 상태 강제 변환 (Pydantic and dataclass state coercion)

그래프 상태가 Pydantic 모델이나 dataclass일 때 v2 values 모드는 출력을 자동으로 올바른 타입으로 강제 변환해요.

from pydantic import BaseModel
from typing import Annotated
import operator

class MyState(BaseModel):
    value: str
    items: Annotated[list[str], operator.add]

# With version="v2", chunk["data"] is a MyState instance
for chunk in graph.stream(
    {"value": "x", "items": []}, stream_mode="values", version="v2"
):
    print(type(chunk["data"]))  # <class 'MyState'>

Python < 3.11에서의 비동기 (Async with Python < 3.11)

Python < 3.11에서 asyncio task는 context 파라미터를 지원하지 않아요. 이는 LangGraph가 컨텍스트를 자동으로 전파하는 능력을 제한하고, 스트리밍 메커니즘에 두 가지 핵심 방식으로 영향을 줘요.

1. 콜백이 자동으로 전파되지 않으므로 비동기 LLM 호출(예: ainvoke())에 RunnableConfig를 명시적으로 전달해야 해요.
2. async 노드나 도구에서 get_stream_writer를 사용할 수 없어요 — writer 인자를 직접 전달해야 해요.

확장 예시: 수동 config가 있는 async LLM 호출

from typing import TypedDict
from langgraph.graph import START, StateGraph
from langchain.chat_models import init_chat_model

model = init_chat_model(model="gpt-5.4-mini")

class State(TypedDict):
    topic: str
    joke: str

# Accept config as an argument in the async node function
async def call_model(state, config):
    topic = state["topic"]
    print("Generating joke...")
    # Pass config to model.ainvoke() to ensure proper context propagation
    joke_response = await model.ainvoke(  # [!code highlight]
        [{"role": "user", "content": f"Write a joke about {topic}"}],
        config,
    )
    return {"joke": joke_response.content}

graph = (
    StateGraph(State)
    .add_node(call_model)
    .add_edge(START, "call_model")
    .compile()
)

# Set stream_mode="messages" to stream LLM tokens
async for chunk in graph.astream(
    {"topic": "ice cream"},
    stream_mode="messages",  # [!code highlight]
    version="v2",  # [!code highlight]
):
    if chunk["type"] == "messages":
        message_chunk, metadata = chunk["data"]
        if message_chunk.content:
            print(message_chunk.content, end="|", flush=True)

확장 예시: stream writer로 async 커스텀 스트리밍

from typing import TypedDict
from langgraph.types import StreamWriter

class State(TypedDict):
      topic: str
      joke: str

# Add writer as an argument in the function signature of the async node or tool
# LangGraph will automatically pass the stream writer to the function
async def generate_joke(state: State, writer: StreamWriter):  # [!code highlight]
      writer({"custom_key": "Streaming custom data while generating a joke"})
      return {"joke": f"This is a joke about {state['topic']}"}

graph = (
      StateGraph(State)
      .add_node(generate_joke)
      .add_edge(START, "generate_joke")
      .compile()
)

# Set stream_mode="custom" to receive the custom data in the stream  # [!code highlight]
async for chunk in graph.astream(
      {"topic": "ice cream"},
      stream_mode="custom",
      version="v2",
):
      if chunk["type"] == "custom":
          print(chunk["data"])

더 알아보기 (Learn more)

• 이벤트 스트리밍: v1.2의 타입 기반 프로젝션 스트리밍 API를 다뤄요.
• 서브그래프: 서브그래프 출력 스트리밍을 다뤄요.
• 인터럽트: v2에서의 인터럽트 위치를 다뤄요.