스트리밍 (Streaming)

스트리밍 (Streaming)

새로 만드는 애플리케이션이라면 이벤트 스트리밍(event streaming)을 권장해요. 이건 LangGraph v1.2에서 도입된 typed-projection API로, 프로젝션별(메시지, 값, 서브그래프, 출력)로 별도의 이터레이터를 제공해서 stream_mode 청크를 분기 처리하는 대신 각각 독립적으로 소비할 수 있게 해 줍니다.

이 페이지는 LangGraph의 스트림 모드(stream-mode) API를 다룹니다. updates, values, messages, custom, checkpoints, tasks, debug 같은 스트림 모드로 그래프 실행을 노출하죠. 그래프 런타임 이벤트나 특정 스트림 모드의 출력에 직접 접근해야 할 때 이 API를 쓰면 됩니다.

시작하기 (Get started)

기본 사용법 (Basic usage)

LangGraph 그래프는 스트리밍 출력을 이터레이터로 만들어 내는 stream(동기)와 astream(비동기) 메서드를 제공합니다. 받을 데이터를 제어하려면 스트림 모드를 하나 이상 넘기면 돼요.

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!'}

전체 예시

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!'}

스트리밍 이벤트 디버깅, 토큰 단위 LLM 출력 확인, 지연 시간 모니터링은 LangSmith로 할 수 있어요. tracing quickstart를 따라 설정하면 됩니다.

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

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

stream() 또는 astream()에 version="v2"를 넘기면 통일된 출력 형식을 얻을 수 있습니다. 모든 청크는 스트림 모드, 모드 개수, 서브그래프 설정과 무관하게 일관된 모양의 StreamPart dict입니다:

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

각 스트림 모드에는 대응하는 TypedDict가 있어요 — ValuesStreamPart, UpdatesStreamPart, MessagesStreamPart, CustomStreamPart, CheckpointStreamPart, TasksStreamPart, DebugStreamPart. 이 타입들은 langgraph.types에서 import할 수 있어요. 유니온 타입 StreamPart은 part["type"]을 기준으로 하는 disjoint union이라서 에디터와 타입 체커에서 완전한 타입 내로잉(type narrowing)이 가능합니다.

v1(기본)에서는 스트리밍 옵션에 따라 출력 형식이 달라져요(단일 모드는 원시 데이터, 여러 모드는 (mode, data) 튜플, 서브그래프는 (namespace, data) 튜플). v2에서는 형식이 항상 동일합니다:

v2 (신규)

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 (현재 기본값)

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 또는 astream 메서드에 다음 스트림 모드 중 하나 이상을 리스트로 넘깁니다:

Mode	Type	Description
values	ValuesStreamPart	각 단계 후의 전체 상태(Full state).
updates	UpdatesStreamPart	각 단계 후의 상태 업데이트. 같은 단계의 여러 업데이트는 각각 따로 스트리밍됩니다.
messages	MessagesStreamPart	LLM 호출로부터의 (LLM 토큰, 메타데이터) 2-튜플.
custom	CustomStreamPart	노드에서 get_stream_writer로 내보낸 사용자 지정 데이터.
checkpoints	CheckpointStreamPart	체크포인트 이벤트(get_state()와 같은 형식). 체크포인터가 필요합니다.
tasks	TasksStreamPart	결과와 오류가 포함된 태스크 시작/종료 이벤트. 체크포인터가 필요합니다.
debug	DebugStreamPart	가능한 모든 정보 — checkpoints와 tasks를 추가 메타데이터와 함께 결합.

그래프 상태 (Graph state)

updates와 values 스트림 모드로 그래프가 실행되는 동안 상태를 스트리밍할 수 있습니다.

• 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",
    version="v2",
):
    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",
    version="v2",
):
    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에서 async 코드를 쓸 땐 올바른 스트리밍을 위해 RunnableConfig를 ainvoke()에 명시적으로 넘겨야 해요. 자세한 내용은 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(
        [
            {"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",
    version="v2",
):
    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",
    version="v2",
):
    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 모드에서 방출되지 않습니다. 이는 다음과 같은 경우에 유용합니다:

• 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")

노드로 필터링 (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",
    version="v2",
):
    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",
    version="v2",
):
    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를 사용해 스트림 라이터에 접근해 커스텀 데이터를 내보냅니다.
2. .stream() 또는 .astream() 호출 시 stream_mode="custom"으로 설정해 스트림에서 커스텀 데이터를 받습니다. 여러 모드를 조합할 수 있지만(예: ["updates", "custom"]), 최소한 하나는 "custom"이어야 합니다.

Python < 3.11 비동기에서는 get_stream_writer 사용 불가 Python < 3.11에서 실행되는 async 코드에서는 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()
    # Emit a custom key-value pair (e.g., progress update)
    writer({"data": "Retrieved 0/100 records", "type": "progress"})
    # 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,
    stream_mode="updates",
    version="v2",
):
    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,
    stream_mode="updates",
):
    print(chunk)

이 내용은 "messages"를 포함한 모든 stream_mode에 적용됩니다. create_agent 같은 에이전트 빌더는 컴파일된 그래프를 반환하므로, 그것을 노드로 추가하면 서브그래프가 됩니다. subgraphs=True 없이는 부모 그래프의 stream_mode="messages"가 내부 에이전트의 LLM 호출에서 토큰 청크를 방출하지 않습니다. agent.stream(...)을 직접 호출하면 방출되는데, 그래서 이 문제는 보통 래핑한 뒤에만 나타납니다.

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,
    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,
    version="v2",
):
    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()의 출력과 같은 형식입니다. 체크포인터가 필요합니다.

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",
    version="v2",
):
    if chunk["type"] == "checkpoints":
        print(chunk["data"])

태스크 (Tasks)

tasks 스트리밍 모드를 사용하면 그래프가 실행되는 동안 태스크 시작/종료 이벤트를 받을 수 있습니다. 태스크 이벤트에는 어떤 노드가 실행 중인지, 그 결과, 그리고 오류에 대한 정보가 포함됩니다. 체크포인터가 필요합니다.

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",
    version="v2",
):
    if chunk["type"] == "tasks":
        print(chunk["data"])

디버그 (Debug)

debug 스트리밍 모드를 사용하면 그래프 실행 전반에 걸쳐 가능한 한 많은 정보를 스트리밍할 수 있습니다. 스트리밍 출력에는 노드 이름과 전체 상태가 포함됩니다.

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

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

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

stream_mode 파라미터에 리스트를 넘기면 여러 모드를 한 번에 스트리밍할 수 있습니다. version="v2"에서는 모든 청크가 StreamPart dict입니다. 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 채팅 모델 인터페이스를 구현하지 않아도 말이죠. 덕분에 자체 스트리밍 인터페이스를 제공하는 원시 LLM 클라이언트나 외부 서비스를 통합할 수 있어, LangGraph를 커스텀 설정에 매우 유연하게 만들 수 있습니다.

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()
    # 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})
    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",
    version="v2",
):
    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로 설정하세요.

init_chat_model

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
)

Chat model interface

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 스트리밍 형식은 통일된 출력 형식을 제공합니다. 핵심 차이점과 마이그레이션 방법을 정리하면 다음과 같습니다:

시나리오	v1 (기본)	v2 (version="v2")
단일 스트림 모드	원시 데이터(dict)	type, ns, data를 가진 StreamPart dict
여러 스트림 모드	(mode, data) 튜플	동일한 StreamPart dict, chunk["type"]으로 필터
서브그래프 스트리밍	(namespace, data) 튜플	동일한 StreamPart dict, chunk["ns"] 확인
여러 모드 + 서브그래프	(namespace, mode, data) 3-튜플	동일한 StreamPart dict
invoke() 반환 타입	일반 dict(상태)	.value와 .interrupts를 가진 GraphOutput
인터럽트 위치(스트림)	상태 dict의 __interrupt__ 키	values 스트림 파트의 interrupts 필드
인터럽트 위치(invoke)	결과 dict의 __interrupt__ 키	GraphOutput의 .interrupts 속성
Pydantic/dataclass 출력	일반 dict 반환	모델/dataclass 인스턴스로 강제 변환

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"가 아닌 다른 스트림 모드에서는 invoke(..., stream_mode="updates", version="v2")가 list[tuple] 대신 list[StreamPart]를 반환합니다.

GraphOutput에 대한 dict 스타일 접근(result["key"], "key" in result, result["__interrupt__"])은 하위 호환성을 위해 여전히 동작하지만 deprecated이며 향후 버전에서 제거될 예정입니다. result.value와 result.interrupts로 마이그레이션하세요.

이렇게 하면 상태와 인터럽트 메타데이터가 분리됩니다. v1에서는 인터럽트가 반환된 dict의 __interrupt__ 아래에 포함되어 있습니다:

v2 (신규)

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 (현재 기본값)

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 모델이나 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 태스크가 context 파라미터를 지원하지 않습니다. 이는 LangGraph가 컨텍스트를 자동으로 전파하는 능력을 제한하고, LangGraph의 스트리밍 메커니즘에 두 가지 핵심 방식으로 영향을 줍니다:

1. async 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(
        [{"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",
    version="v2",
):
    if chunk["type"] == "messages":
        message_chunk, metadata = chunk["data"]
        if message_chunk.content:
            print(message_chunk.content, end="|", flush=True)

확장 예시: 스트림 라이터를 사용한 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):
      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  #
async for chunk in graph.astream(
      {"topic": "ice cream"},
      stream_mode="custom",
      version="v2",
):
      if chunk["type"] == "custom":
          print(chunk["data"])