langgraph-streaming

name	langgraph-streaming
description	Stream LangGraph graph execution using stream modes (updates, values, messages, custom, checkpoints, tasks, debug). Use when implementing real-time output, token-by-token LLM streaming, custom progress events, subgraph streaming, or migrating to the v2 streaming format. Covers StreamPart format, get_stream_writer, filtering by node/tag, and async patterns.

Documentation Index

Fetch the complete documentation index at: https://docs.langchain.com/llms.txt Use this file to discover all available pages before exploring further.

---

LangGraph Streaming

Always use version="v2" — unified StreamPart format regardless of mode count or subgraphs.

for chunk in graph.stream(inputs, stream_mode=["updates", "custom"], version="v2"):
    if chunk["type"] == "updates": ...
    elif chunk["type"] == "custom": ...

Every chunk shape: {"type": str, "ns": tuple, "data": ...}

Stream Modes

Mode	What you get	Requires checkpointer
updates	State changes returned by each node	No
values	Full state snapshot after each step	No
messages	(token_chunk, metadata) from LLM calls	No
custom	Arbitrary data via get_stream_writer()	No
checkpoints	Checkpoint events (same as get_state())	Yes
tasks	Task start/finish with results and errors	Yes
debug	Everything — combines checkpoints + tasks	Yes

updates — node state changes

for chunk in graph.stream(inputs, stream_mode="updates", version="v2"):
    if chunk["type"] == "updates":
        for node_name, state in chunk["data"].items():
            print(f"{node_name}: {state}")

values — full state after each step

for chunk in graph.stream(inputs, stream_mode="values", version="v2"):
    if chunk["type"] == "values":
        print(chunk["data"])   # complete state dict / model

messages — LLM token streaming

Works even when using .invoke() inside nodes (not just .stream()).

for chunk in graph.stream(inputs, stream_mode="messages", version="v2"):
    if chunk["type"] == "messages":
        msg, metadata = chunk["data"]
        if msg.content:
            print(msg.content, end="", flush=True)

Filter by node:

if metadata["langgraph_node"] == "my_node":
    print(msg.content, end="")

Filter by LLM tag:

model = init_chat_model("gpt-5.4-mini", tags=["my-llm"])
# ...
if metadata["tags"] == ["my-llm"]:
    print(msg.content, end="")

Suppress a model from the stream (nostream tag — still runs, tokens just not emitted):

internal_model = init_chat_model("gpt-5.4-mini").with_config({"tags": ["nostream"]})

custom — arbitrary data from inside nodes/tools

from langgraph.config import get_stream_writer

def my_node(state):
    writer = get_stream_writer()
    writer({"status": "step 1 done", "progress": 50})
    return {...}

for chunk in graph.stream(inputs, stream_mode="custom", version="v2"):
    if chunk["type"] == "custom":
        print(chunk["data"])

Works in tools too — call get_stream_writer() inside the @tool function.

For non-LangChain LLMs, use custom mode to stream raw tokens:

def call_raw_llm(state):
    writer = get_stream_writer()
    for token in your_streaming_client(state["prompt"]):
        writer({"token": token})

Multiple modes at once

for chunk in graph.stream(inputs, stream_mode=["updates", "custom"], version="v2"):
    if chunk["type"] == "updates": ...
    elif chunk["type"] == "custom": ...

Subgraph streaming

for chunk in graph.stream(inputs, subgraphs=True, stream_mode="updates", version="v2"):
    print(chunk["ns"])    # () = root; ("node_name:<task_id>",) = subgraph
    print(chunk["data"])

---

v2 invoke format

from langgraph.types import GraphOutput

result = graph.invoke(inputs, version="v2")
result.value       # state dict, Pydantic model, or dataclass
result.interrupts  # tuple[Interrupt, ...] — empty if no interrupt occurred

Check for interrupts:

if result.interrupts:
    graph.invoke(Command(resume=True), config=config, version="v2")

Pydantic/dataclass states are automatically coerced — chunk["data"] is a model instance, not a plain dict.

---

Async

astream is the async equivalent of stream. All modes and options are identical.

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

Python < 3.11: context vars don't propagate automatically in async tasks.

• Pass config explicitly to model.ainvoke(..., config) for messages mode to work
• Use writer: StreamWriter as a node argument instead of get_stream_writer() for custom mode:

from langgraph.types import StreamWriter

async def my_node(state, writer: StreamWriter):
    writer({"progress": 50})
    ...

---

Key Rules

1. Always use version="v2" — v1 format changes shape based on mode count and subgraphs flag; v2 is always {type, ns, data}
2. stream_mode accepts a list for multiple concurrent modes; filter by chunk["type"]
3. messages mode captures .invoke() calls too — no need to switch to .stream() inside nodes
4. custom mode requires stream_mode="custom" (or in a list) — emitting via get_stream_writer() silently drops if custom isn't in the mode list
5. checkpoints and tasks require a checkpointer on the compiled graph
6. nostream tag suppresses tokens without affecting execution
7. get_stream_writer() doesn't work in async on Python < 3.11 — use writer: StreamWriter parameter instead

Related Resources

• Event streaming (typed projections, LangGraph v1.2+): /oss/python/langgraph/event-streaming
• Subgraphs: /oss/python/langgraph/use-subgraphs
• Interrupts: /oss/python/langgraph/interrupts
• LangChain agent streaming: /oss/python/langchain/streaming