Streaming: Modes, Tokens, and Real-Time Progress (LangGraph in Production, Module 11)

This is Module 11 of LangGraph in Production, a free 16-module course that takes you from your first StateGraph to a deployed, durable autonomous coding agent on LangGraph v1.x. Start at Module 1 or browse the full syllabus.

---

The 18-Second Black Box

You submit the issue: “Bug: tally report --month 2026-05 returns 0.00 for all categories when the CSV has trailing spaces.”

You call .invoke(). The terminal goes silent.

Eighteen seconds later, fourteen lines appear at once. The test suite passed. A pull request draft exists. But you have no idea what happened. Did the Planner read the right file? Did the Editor for storage.py finish before report.py? Were there three failed test runs before the fourth one passed? You do not know.

In a real CLI or web UI, a user who sees nothing for 18 seconds reaches for Ctrl+C at second five. .invoke() is an opaque tunnel. Every interesting event — every token of the plan, every “editing storage.py”, every “tests: 3 failed, retrying” — is swallowed and delivered in a single indigestible batch at the end.

.astream() plus get_stream_writer changes this. They expose a window into the running graph — each node’s output as it completes, each LLM token as it generates, each custom progress event your nodes choose to emit. Forge does not change. The graph does not change. Only the way you consume the result changes.

---

In this module

You’ll learn:

• The 7 streaming modes of LangGraph v1.x and exactly what each one emits, when, and why you’d choose one over another
• How to combine multiple modes and consume the resulting (mode, chunk) tuples
• How to emit custom progress events from inside a node using get_stream_writer
• How to stream LLM tokens one by one with stream_mode="messages"
• How astream_events(version="v2") gives you named, granular events including on_tool_start and on_custom_event
• What subgraphs=True does to your chunks (preview of Module 12)

You’ll build: A stream_cli.py harness that combines updates and custom stream modes to display real-time progress from Forge — which file is being edited, when tests run — all without touching a single graph edge.

Concepts covered: CA11 — Streaming (core). This module BUILDs CA11.1–11.7 of the LangGraph theory inventory.

Prerequisites: Modules 1–10 completed. Working forge/ package (cumulative through Module 10 with the Store, RetryPolicy, CachePolicy, durability). .env with ANTHROPIC_API_KEY, or set FORGE_MODEL_PROVIDER=ollama for $0. Python 3.12, uv.

---

Where we are

M1  ✅  StateGraph, Forge, Tally — the mental model
M2  ✅  ForgeState, nodes, reducers, compile()
M3  ✅  Triage router, test-fix loop, recursion_limit
M4  ✅  Planner with create_agent, Plan structured output
M5  ✅  Fan-out with Send, parallel Editors, fan-in
M6  ✅  SqliteSaver, threads, persistence across kills
M7  ✅  interrupt() + Command(resume=), HITL plan and PR approval
M8  ✅  get_state_history, replay, fork via update_state
M9  ✅  durability=, RetryPolicy, CachePolicy, sandbox hardening
M10 ✅  the Store, namespaces, semantic recall, Planner memory-aware
M11 👉  Streaming — modes, tokens, custom events, astream_events
M12 ⬜  Subgraphs, Supervisor, multi-agent handoffs
M13–M16 ⬜

Forge entering Module 11 is a complete agent: it plans, edits files in parallel, runs tests in a fix-it loop, asks for human approval at two gates, persists state across crashes, retries flaky tests, caches expensive planning calls, and recalls relevant past fixes from long-term memory. The only thing missing is any real-time feedback. .invoke() is a black box.

After this module, you consume the same graph with .astream(). No logic changes, no new edges, no new fields in ForgeState. The streaming layer is a consumption interface, not a redesign.

Teaser: when Module 12 breaks Forge into composable subgraphs, subgraphs=True will go from a preview to essential — every nested execution visible by namespace.

---

T5 — Understanding LangGraph Streaming

.stream and .astream: Your Window Into the Graph

LangGraph’s runtime is a Pregel-style execution engine that runs graphs in supersteps — a BSP (Bulk Synchronous Parallel) model. Each superstep is a synchronization barrier: every node eligible to run in that step executes, their writes are collected, then the state is updated and the next superstep begins.

.invoke() waits for all supersteps to complete before returning a single result. .stream() and .astream() expose an iterator and async iterator respectively that yield a chunk each time an event matching the chosen stream mode occurs. Same graph. Same nodes. Same checkpointing. Different interface.

The exact v1.x signatures:

# Synchronous — fine for scripts, blocks the thread
for chunk in graph.stream(
    input,
    config,
    stream_mode="updates",   # str or list[str]
    subgraphs=False,         # True = include events from subgraphs
):
    print(chunk)

# Asynchronous — preferred in production servers
async for chunk in graph.astream(
    input,
    config,
    stream_mode="updates",
    subgraphs=False,
):
    print(chunk)

In a production server, .astream() is the right choice: a synchronous .stream() call blocks the event loop thread for the duration of the run. An async server handling multiple concurrent Forge runs needs .astream() to keep the loop free for other requests.

The 7 Stream Modes

stream_mode is the core concept. You pick what the iterator should yield. LangGraph v1.x supports exactly seven modes:

Mode	What is emitted	Frequency	Primary use
"values"	The complete ForgeState after every superstep	1 chunk per superstep	Debug / inspect full state snapshots
"updates"	The delta returned by each node ({node_name: partial_state})	1 chunk per node that ran	Show what each node changed — the workhorse for CLIs
"messages"	LLM tokens + metadata (words from the messages channel, one by one)	1 chunk per token	Display LLM generation token-by-token
"custom"	Arbitrary data emitted via get_stream_writer() from inside a node	1 chunk per writer(data) call	Business progress events (which file, which test, which step)
"debug"	Internal LangGraph runtime events (task_start, task_finish, etc.)	Many per run	Advanced runtime debugging
"checkpoints"	State snapshot at each checkpoint written	1 chunk per checkpoint	Observe persistence superstep by superstep
"tasks"	Pregel task metadata (id, node name, input/output)	1 chunk per task	Profiling and custom tracing without LangSmith

Three observations that matter in practice:

"values" gives you the complete state — that means the entire ForgeState including messages (potentially dozens of messages), edits (entire file contents), and everything else. Useful for debugging. Too heavy and too infrequent for a user-facing CLI. A typical Forge run might produce six to eight supersteps, so "values" yields six to eight large blobs.

"updates" gives you the delta — only what the node just changed. An Editor node that modifies report.py yields {"editor": {"edits": [FileEdit(...)]}}. Everything else in ForgeState is omitted. Much more useful for displaying progress. This is what stream_cli.py uses for the node-by-node activity feed.

"messages" is independent of superstep boundaries — it emits during the execution of an LLM node, before that superstep finishes. That is the only way to get individual tokens. "values" and "updates" emit after the superstep completes; by then the tokens are already merged into the final message. You cannot recover individual tokens from a "values" stream.

What a Multi-Mode Timeline Looks Like

Here is how the three most useful modes interleave on a partial Forge run (triage → planner → two parallel editors):

sequenceDiagram
    participant CLI
    participant Runtime as LangGraph Runtime
    participant Triage as triage node
    participant Planner as planner node (LLM)
    participant Ed1 as editor[storage.py]
    participant Ed2 as editor[report.py]

    Runtime->>Triage: superstep 1 — start
    Triage-->>Runtime: {route: "bug"}
    Runtime-->>CLI: updates chunk: {triage: {route: "bug"}}

    Runtime->>Planner: superstep 2 — start
    Planner-->>CLI: messages chunk: token "I"
    Planner-->>CLI: messages chunk: token " will"
    Planner-->>CLI: messages chunk: token " inspect…"
    Planner-->>Runtime: {plan: Plan(...), targets: [...]}
    Runtime-->>CLI: updates chunk: {planner: {plan: Plan(...)}}
    Runtime-->>CLI: custom chunk: {progress: "planning complete"}

    par superstep 3 — Send fan-out
        Runtime->>Ed1: editor[storage.py]
        Ed1-->>CLI: custom chunk: {progress: "editing storage.py"}
        Ed1-->>Runtime: {edits: [FileEdit(path="storage.py")]}
    and
        Runtime->>Ed2: editor[report.py]
        Ed2-->>CLI: custom chunk: {progress: "editing report.py"}
        Ed2-->>Runtime: {edits: [FileEdit(path="report.py")]}
    end
    Runtime-->>CLI: updates chunk: {editor: {edits: [...]}}

The messages chunks arrive during the planner’s superstep, token by token. The custom chunks arrive when the editor nodes call get_stream_writer(). The updates chunk for the planner arrives only when the whole superstep finishes. All three streams are happening on the same timeline.

---

Combining Modes and Consuming Tuples

When you pass a list of modes, LangGraph returns (mode, chunk) tuples at each iteration. When you pass a single string, you get bare chunks. This is a hard contract — not configurable.

# List of modes → (mode, chunk) tuples
async for mode, chunk in graph.astream(
    input,
    config,
    stream_mode=["updates", "messages", "custom"],
):
    if mode == "updates":
        node_name, delta = next(iter(chunk.items()))
        print(f"[{node_name}] state delta: {delta}")

    elif mode == "messages":
        msg_chunk, metadata = chunk   # (AIMessageChunk, dict)
        if hasattr(msg_chunk, "content") and msg_chunk.content:
            print(msg_chunk.content, end="", flush=True)

    elif mode == "custom":
        print(f"  · {chunk.get('progress', chunk)}")

A "messages" chunk is a two-element tuple: (AIMessageChunk, metadata_dict). The AIMessageChunk carries the incremental token text in .content. The metadata carries LangGraph bookkeeping including langgraph_node (which node is generating), langgraph_step (superstep number), and langgraph_run_id. You can use metadata["langgraph_node"] to filter — for example, show only the Planner’s tokens and suppress the triage node’s brief classification.

Common misconception — "values" does not give you LLM tokens. A frequent mistake is reaching for stream_mode="values" expecting to see the model “typing” its response. It does not work. "values" emits the complete state after a superstep finishes — by then the LLM call is done and its output is already merged into messages. The individual tokens existed during the LLM call, not after. Only stream_mode="messages" (or astream_events) captures them mid-generation. The rule: use "updates" to follow node-by-node progress, "messages" for LLM tokens, "custom" for your own business events. Using "values" for a user-facing CLI gives you the same 18-second silence as .invoke(), just broken into a few large dumps instead of one.

A second confusion: believing that custom events from get_stream_writer() appear everywhere. They appear in stream_mode="custom" (via .stream() / .astream()), and they appear as on_custom_event events in astream_events(version="v2"). They do not appear in astream_events(version="v1"). If your consumer uses astream_events and custom events are mysteriously missing, the version is wrong.

---

get_stream_writer: Custom Progress Events from Inside a Node

get_stream_writer is the v1.x mechanism for emitting arbitrary data from a node into the consumer’s stream. It is imported from langgraph.config — not from langgraph.types, not from langgraph.streaming:

from langgraph.config import get_stream_writer

The pattern used in Forge’s nodes.py is a thin wrapper that handles the case where the node is called outside a streaming context (like a direct unit test):

def _emit(data: dict) -> None:
    """Emit a custom progress event when streaming; a no-op otherwise."""
    try:
        get_stream_writer()(data)
    except Exception:
        pass

This is deliberately pragmatic: get_stream_writer() raises when called outside a LangGraph task context (for example, if you call the node function directly in a unit test). Wrapping it in a try/except means the same node code works both in a streaming run and in an offline test without any changes. The data you emit can be any JSON-serializable object.

Here is the actual editor node from the lab, with the _emit calls that power the CLI’s progress display:

def editor(state: dict) -> dict:
    """Map worker: edit ONE file. Receives its target via the Send payload."""
    target: FileTarget = state["target"]
    _emit({"progress": f"editing {target.path}"})       # custom event #1
    current = (Path(state["repo_path"]) / target.path).read_text()
    model = config.get_model("smart")
    resp = model.invoke([
        SystemMessage(content=EDIT_SYSTEM),
        HumanMessage(content=(
            f"Issue:\n{state['issue']}\n\nEdit file {target.path}: {target.instruction}\n"
            f"Current contents:\n{current}"
        )),
    ])
    edit = FileEdit(path=target.path, new_content=extract_code(resp.content),
                    summary=target.instruction)
    return {"edits": [edit]}   # reduced via operator.add across all editors

And the tester node:

def tester(state: ForgeState) -> dict:
    """Deterministic node: run the sandboxed pytest suite."""
    _emit({"progress": "running tests"})                 # custom event
    return {"test_result": sandbox.run_tests(state["repo_path"])}

These two lines — _emit({"progress": f"editing {target.path}"}) and _emit({"progress": "running tests"}) — are the entire Module 11 change to nodes.py. No new nodes, no new edges, no new state fields. The rest of the file is inherited byte-for-byte from Module 10.

On the consumer side, stream_cli.py dispatches on the mode:

def stream_run(issue: str, tally_src: str, thread_id: str = "cli") -> str:
    repo = make_sandbox(tally_src)
    graph = build_graph(checkpointer=checkpoint.in_memory())
    cfg = {"configurable": {"thread_id": thread_id}, "recursion_limit": 50}
    for mode, chunk in graph.stream(
        {"issue": issue, "repo_path": repo}, cfg,
        stream_mode=["updates", "custom"]
    ):
        if mode == "custom":
            print(f"  · {chunk.get('progress')}")
        else:
            print(f"[{next(iter(chunk))}]")   # mode == "updates": show node name
    return repo

The "updates" branch prints the node name from each delta (triage, planner, editor, tester, pr_approval). The "custom" branch prints the progress payload. You see exactly what the graph is doing, in order, as it happens.

For a richer CLI you could expand the custom payload with more structure — a "phase" field that lets the consumer format differently for edit start vs edit done vs test failure:

# Richer progress event in editor node (optional expansion)
writer = get_stream_writer()
writer({"phase": "edit_start", "status": f"Editing {target.path}"})
# ... do the edit ...
writer({"phase": "edit_done", "status": f"Done: {target.path}", "summary": edit.summary})

# Consumer side — format by phase
elif mode == "custom":
    evt = chunk
    phase = evt.get("phase", "")
    if phase == "edit_start":
        print(f"  ✏  {evt['status']}")
    elif phase == "edit_done":
        print(f"  ✓  {evt['status']} — {evt.get('summary', '')}")
    elif phase == "test_done":
        icon = "✅" if evt.get("passed") else "❌"
        print(f"  {icon} {evt['status']}")

You can also declare StreamWriter as a parameter type hint if you prefer explicit injection over the context-lookup approach:

from langgraph.config import StreamWriter

def tester_node(state: ForgeState, *, writer: StreamWriter) -> dict:
    writer({"phase": "test_start", "status": "Running pytest…"})
    result = sandbox.run_tests(state["repo_path"])
    writer({
        "phase": "test_done",
        "status": f"Tests: {result.total} total",
        "passed": result.passed,
        "failures": result.failures,
    })
    return {"test_result": result}

LangGraph injects the writer automatically when the node signature contains a writer: StreamWriter parameter. The two approaches — get_stream_writer() inside the function body vs writer: StreamWriter in the signature — are functionally equivalent. The get_stream_writer() pattern works in both sync nodes and async coroutines. Design guideline: custom events are the business data channel. Internal LangGraph events belong in "debug" and "updates". Your application’s progress data — which file Forge is editing, what the test output says, whether the plan was approved — belongs in "custom".

---

Token Streaming and astream_events(version="v2")

Token streaming with "messages" mode

stream_mode="messages" is the dedicated mode for LLM token streaming. You combine it with "updates" to get both node-level progress and token-level detail:

async for mode, chunk in graph.astream(
    {"issue": issue_text},
    config,
    stream_mode=["updates", "messages"],
):
    if mode == "messages":
        msg_chunk, metadata = chunk
        # metadata["langgraph_node"] tells you which node is generating
        node = metadata.get("langgraph_node", "?")
        if hasattr(msg_chunk, "content") and msg_chunk.content:
            print(f"[{node}] {msg_chunk.content}", end="", flush=True)
    elif mode == "updates":
        node_name = next(iter(chunk))
        print(f"\n[{node_name}] done")

The metadata["langgraph_node"] field is your filter handle. The Planner generates a detailed plan. The triage node generates a single word. You probably want to show the Planner’s tokens to the user and suppress the triage node’s output. Check node == "planner" before printing.

The AIMessageChunk object has a .content attribute that holds the token text. It also has a .content_blocks attribute for structured content blocks if the model returns structured output. For plain text token display, .content is what you want.

astream_events(version="v2")

astream_events is the more granular alternative. Instead of (mode, chunk) tuples, you receive named event dictionaries with a richer schema. The critical argument is version="v2" — without it, custom events are silently dropped:

async for event in graph.astream_events(
    {"issue": issue_text},
    config,
    version="v2",          # required for on_custom_event
):
    kind = event["event"]
    name = event.get("name", "")
    data = event.get("data", {})

    if kind == "on_chat_model_stream":
        chunk = data.get("chunk")
        if chunk and chunk.content:
            print(chunk.content, end="", flush=True)

    elif kind == "on_custom_event":
        # writer(data) calls from get_stream_writer — v2 only
        print(f"[progress] {data}")

    elif kind == "on_tool_start":
        print(f"\n→ {name}({data.get('input', {})})")

    elif kind == "on_tool_end":
        print(f"← result: {str(data.get('output', ''))[:80]}")

The key events from astream_events(version="v2"):

Event	When	data contents
on_chain_start	Start of a node or subgraph	{"input": ...}
on_chain_end	End of a node or subgraph	{"output": ...}
on_chat_model_start	Start of an LLM call	{"messages": [...]}
on_chat_model_stream	LLM token	{"chunk": AIMessageChunk}
on_chat_model_end	End of an LLM call	{"output": AIMessage}
on_tool_start	Start of a @tool call	{"input": {...}}
on_tool_end	End of a @tool call	{"output": ToolMessage}
on_custom_event	writer(data) called — v2 only	data as passed

When should you use .astream() vs astream_events()? Use .astream() with combined modes for the vast majority of cases: CLIs, web UIs, anything that needs node progress and LLM tokens. Use astream_events() when you need the granularity of individual tool calls — for custom observability, debugging the Planner’s read_file / search_code calls, or building a trace without LangSmith. The on_tool_start / on_tool_end events tell you exactly which tools the Planner invoked and with what arguments — that detail is not available in "updates" mode alone.

---

Subgraph Streaming: subgraphs=True (Preview)

subgraphs=True is a parameter on .stream() / .astream() that changes the structure of every chunk. Without it, events from subgraphs (if any exist) are aggregated into the parent graph’s chunks. With it, each node in each subgraph emits its own chunk carrying a namespace tuple identifying its position in the graph hierarchy:

async for chunk in graph.astream(
    input, config,
    stream_mode="updates",
    subgraphs=True,
):
    # chunk = (namespace_tuple, delta)
    # namespace_tuple = () for the root graph
    # namespace_tuple = ("planner:1a2b3c",) for a subgraph named "planner"
    namespace, delta = chunk
    if namespace:
        print(f"[subgraph {namespace[-1]}] {delta}")
    else:
        print(f"[root] {delta}")

On the current Forge graph (through Module 10), there are no subgraphs. Every node lives in the root graph. So namespace is always () and subgraphs=True has no visible effect — you can add the parameter right now and nothing breaks. The chunks simply become two-element tuples ((), delta) where the first element is always the empty tuple.

This becomes essential in Module 12, where the Planner, Coder, and Reviewer become separate compiled subgraphs connected through a Supervisor. At that point, events from inside the Planner subgraph arrive with namespace = ("planner:abc123",) and you can filter the stream by subgraph. One parameter, added now, unlocks full visibility into a hierarchical multi-agent graph.

Note the connection to the checkpoint_ns field you saw in Module 6: subgraphs have their own checkpoint namespaces ("planner:abc123") that appear both in the StateSnapshot objects from get_state_history and in the streaming chunks from subgraphs=True. The framework is consistent — the same namespace system governs both persistence and streaming.

---

T6 — Hands-On Lab: Streaming Forge in Real Time

Objective: Add live progress to Forge by instrumenting three nodes with get_stream_writer and consuming the stream in a new stream_cli.py. The graph, its edges, and ForgeState do not change.

Observable result (offline, no API key required):

Run the unit tests to verify the streaming plumbing:

$ uv run pytest tests/test_module_11.py -v
test_stream_mode_updates_lists_nodes        PASSED
test_stream_mode_custom_captures_writer_events  PASSED
test_stream_mode_values_is_full_state       PASSED
test_combined_modes_yield_mode_tagged_tuples  PASSED

With a live API key (FORGE_LIVE_TESTS=1), stream_cli.py produces:

$ uv run python stream_cli.py "report crashes on empty category"
[intake]
[triage]
[recall]
[planner]
[plan_approval]
  · editing report.py
[apply]
  · running tests
[pr_approval]

Step 1 — Copy the Module 10 state

Start from the complete module-10/ directory. Every file in forge/ is inherited byte-for-byte. The only files that change in this module are forge/nodes.py (modified) and stream_cli.py (new).

Verify the inherited ForgeState has exactly 11 fields and no review field — Module 12 adds that:

class ForgeState(TypedDict, total=False):
    issue: str                                           # M2
    messages: Annotated[list[AnyMessage], add_messages]  # M2
    repo_path: str                                       # M2
    route: str                                           # M3
    attempts: int                                        # M3
    test_result: TestResult | None                       # M3
    plan: Plan | None                                    # M4
    targets: list[FileTarget]                            # M5
    edits: Annotated[list[FileEdit], operator.add]       # M5
    plan_approved: bool                                  # M7
    pull_request: PullRequest | None                     # M7
    # review: Review | None                             # M12 — absent here

Step 2 — Add get_stream_writer to forge/nodes.py

This is the only logical change to the graph internals. Import get_stream_writer from langgraph.config and add a thin _emit helper at the top of nodes.py:

from langgraph.config import get_stream_writer

def _emit(data: dict) -> None:
    """Emit a custom progress event when streaming; a no-op otherwise."""
    try:
        get_stream_writer()(data)
    except Exception:
        pass

The try/except is intentional here: get_stream_writer() raises a LookupError when called outside a LangGraph task execution context — for example, when a unit test calls the node function directly instead of through .stream(). Wrapping it means your nodes stay testable offline without any mocking.

Add _emit calls in three nodes:

In editor (the map worker that edits one file):

def editor(state: dict) -> dict:
    target: FileTarget = state["target"]
    _emit({"progress": f"editing {target.path}"})   # ← add this
    # ... rest of the node unchanged ...

In tester (the deterministic pytest runner):

def tester(state: ForgeState) -> dict:
    _emit({"progress": "running tests"})             # ← add this
    return {"test_result": sandbox.run_tests(state["repo_path"])}

That is it. Two _emit calls added to the existing node bodies. The Planner’s tokens arrive naturally via stream_mode="messages" — no change to the planner node is needed.

Step 3 — Write stream_cli.py

The new artefact for this module. It uses the synchronous .stream() (fine for a CLI script) with combined modes:

"""Stream a Forge run with live progress (Module 11)."""
from __future__ import annotations

import sys

from forge import checkpoint
from forge.graph import build_graph
from forge.sandbox import make_sandbox


def stream_run(issue: str, tally_src: str, thread_id: str = "cli") -> str:
    repo = make_sandbox(tally_src)
    graph = build_graph(checkpointer=checkpoint.in_memory())
    cfg = {"configurable": {"thread_id": thread_id}, "recursion_limit": 50}
    for mode, chunk in graph.stream(
        {"issue": issue, "repo_path": repo}, cfg,
        stream_mode=["updates", "custom"]
    ):
        if mode == "custom":
            print(f"  · {chunk.get('progress')}")
        else:
            print(f"[{next(iter(chunk))}]")
    return repo


def main(argv: list[str] | None = None) -> int:
    issue = (argv or sys.argv[1:] or ["report crashes on empty category"])[0]
    stream_run(issue, "tally")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

The ["updates", "custom"] list makes every iteration yield a (mode, chunk) tuple. The "updates" branch extracts the node name from the delta dict with next(iter(chunk)). The "custom" branch prints the progress payload. The whole script is 30 lines.

Step 4 — Verify the streaming contract with unit tests

The test_module_11.py tests verify the streaming plumbing against a tiny two-node graph that avoids any LLM calls:

def test_stream_mode_custom_captures_writer_events():
    custom = list(_graph().stream({"log": []}, stream_mode="custom"))
    assert {"progress": "in a"} in custom and {"progress": "in b"} in custom

def test_combined_modes_yield_mode_tagged_tuples():
    items = list(_graph().stream({"log": []}, stream_mode=["updates", "custom"]))
    assert all(isinstance(i, tuple) and len(i) == 2 for i in items)
    assert {"updates", "custom"} <= {i[0] for i in items}

And test_pipeline.py verifies that the real Forge graph emits custom events during a complete run:

def test_streaming_emits_live_progress(monkeypatch, make_routed_model):
    # ... setup: fake models, sandboxed repo, HITL plan approval ...
    progress = []
    for mode, chunk in graph.stream(
        Command(resume={"approved": True}), cfg,
        stream_mode=["updates", "custom"]
    ):
        if mode == "custom":
            progress.append(chunk.get("progress"))
    assert any("editing report.py" in (p or "") for p in progress)
    assert any("running tests" in (p or "") for p in progress)

This test runs fully offline using RoutedFakeModel — no API key required. The pytest of Tally runs for real inside the sandbox (it is deterministic and requires no LLM).

Step 5 — Try it yourself

(a) Compare "values" vs "updates". Add a mode switch to stream_cli.py and run with stream_mode="values". Observe the difference: each chunk is the full ForgeState dictionary (large), arrives once per superstep (infrequent), and contains no individual LLM tokens. Switch back to "updates" and the chunks shrink to deltas.

(b) Add subgraphs=True. Change the .stream() call to subgraphs=True and unpack accordingly:

for namespace, (mode, chunk) in graph.stream(
    ..., stream_mode=["updates", "custom"], subgraphs=True
):
    print(f"ns={namespace} mode={mode}")

Every chunk will have namespace = () because Forge is still a flat graph with no subgraphs. Nothing breaks. The namespace plumbing is already in place. Module 12 is when it becomes meaningful.

(c) Switch to astream_events. Write a small async script that calls graph.astream_events(input, config, version="v2") and filters for on_tool_start events. You will see every read_file and list_files call the Planner makes, with the exact file path argument. That visibility is not available in .stream() without custom events on the tool nodes themselves.

---

T7 — In Production

Backpressure and buffer saturation. A consumer that processes chunks slowly — a WebSocket UI that serializes JSON and sends over a slow connection — can fall behind. LangGraph’s internal buffer is finite. In production, avoid calling writer(data) at very high frequency. A natural rate is one or two custom events per major operation (edit start, edit done, test start, test result). Never emit a custom event per LLM token — that role belongs to stream_mode="messages".

Prefer "updates" over "values" in production logs. The "values" mode dumps the complete state at each superstep, which can include the full content of every edited file in edits[*].new_content. Log that to a file or a structured logging system and you will fill disk quickly on large repos. "updates" logs only the delta, and "custom" logs only your intentional progress events. Both are safe to log continuously.

Streaming costs the same in tokens as .invoke(). The LLM sees the same prompt, generates the same tokens, and pays the same API cost whether you use .invoke() or .astream(). The streaming interface only changes when your code receives the output — the model generates the same way. There is no streaming surcharge.

"debug" and "checkpoints" modes in production. These are for diagnostic runs, not for continuous operation. "debug" is verbose enough to slow a tight event loop. "checkpoints" emits on every checkpoint write, which in a run with durability="sync" (introduced in Module 9) is on every superstep. Reserve both for troubleshooting sessions.

LangSmith is complementary, not an alternative. LangSmith (Module 15) traces every LLM call, tool call, and node execution automatically — whether or not you use streaming. astream_events and LangSmith serve different purposes: one gives your application live data to display, the other gives you a persistent trace for debugging and evaluation. In production you want both. One sentence of teaser for Module 14: LangGraph Studio, launched with langgraph dev, also visualizes the stream in real time — that is Module 14’s territory.

Subgraphs in production streaming. When Module 12 introduces the Planner, Coder, and Reviewer as compiled subgraphs, subgraphs=True is what makes each subgraph’s internal nodes visible to the top-level consumer. A Supervisor routing between a Planner subgraph and a Coder subgraph would otherwise appear as a single opaque chunk per subgraph invocation.

---

T8 — Mastery Corner

What to really understand here

CA11.1 — .stream / .astream are the same graph. Switching from .invoke() to .astream() does not change what the graph computes. It changes when your code receives the intermediate results. The Pregel superstep model is the same; .astream() just exposes the barrier boundaries.

CA11.2 — The 7 modes have distinct contracts. "values" is complete state after each superstep. "updates" is the node’s returned delta. "messages" is LLM tokens during generation. "custom" is whatever you emit with get_stream_writer(). "debug", "checkpoints", "tasks" are for observability and profiling. They cannot substitute for each other.

CA11.3 — List of modes → tuples. Passing a list turns every iteration into a (mode, chunk) tuple. Passing a string gives bare chunks. This is the single most common confusion when combining modes.

CA11.4 — astream_events requires version="v2" for custom events. Without version="v2", on_custom_event events are silently dropped. No error. No warning. They just do not appear.

CA11.5 — Token streaming is independent of superstep boundaries. "messages" yields during execution, not after. "values" yields after. These are fundamentally different timings.

CA11.6 — get_stream_writer is context-bound. The writer is bound to the current LangGraph task context. Calling it outside that context (direct function call in a test) raises. Design your nodes to handle this gracefully — the _emit wrapper pattern is the right approach.

CA11.7 — subgraphs=True changes chunk structure. Every chunk becomes (namespace, original_chunk). On a flat graph, namespace = () always. On a hierarchical graph, namespace identifies the subgraph.

Quiz

Q1. You want to display the Planner’s reasoning tokens to the user as they are generated, word by word. Which stream_mode do you use?

• A. "values" — it emits the state after each superstep, and messages are in the state
• B. "updates" — it emits the planner’s delta when it finishes, including the new message
• C. "messages" — it emits LLM tokens during generation, before the superstep completes
• D. "debug" — it emits internal task events that include model output

Q2. Your Forge consumer calls graph.astream_events(input, config, version="v2") and filters for on_custom_event events. You verify that get_stream_writer() is called in the editor node, but no on_custom_event events appear. What is the most likely cause?

• A. get_stream_writer must be imported from langgraph.types, not langgraph.config
• B. The editor node must be declared as async def for custom events to appear
• C. Custom events require version="v2" in astream_events — which you have — but you also need to pass stream_mode="custom" as an additional argument
• D. The _emit wrapper swallows the exception when get_stream_writer() is called outside the streaming context — but your consumer is inside astream_events, so the issue must be elsewhere; re-read the question: all conditions look correct

Q3. You call graph.stream(input, config, stream_mode=["updates", "messages"]). What is the structure of each item you receive from the iterator?

• A. A dict with keys "updates" and "messages" containing both payloads at once
• B. A (mode, chunk) tuple where mode is either "updates" or "messages" and chunk is the corresponding payload
• C. A bare chunk, same as calling with stream_mode="updates", with an additional .mode attribute
• D. An alternating sequence: one updates chunk then one messages chunk per superstep, always in that order

Q4. You add subgraphs=True to a .stream() call on the current Forge graph (Module 11, no subgraphs defined). What happens to the chunks?

• A. An exception is raised because subgraphs=True requires at least one subgraph to be present
• B. The chunks are identical to subgraphs=False — the parameter has no effect on a flat graph
• C. Each chunk becomes a (namespace, original_chunk) tuple with namespace = () for all root-graph events
• D. The chunks disappear — events from the root graph are filtered out when subgraphs=True

Q5. A developer calls get_stream_writer() inside a node function body, then calls the node function directly in a unit test (not through .stream()). What happens and what is the fix?

• A. It works normally — get_stream_writer() falls back to a no-op writer when not in a streaming context
• B. It raises a LookupError (or similar context error) because the writer is bound to a LangGraph task context; fix by wrapping the call in a try/except or by using the writer: StreamWriter injection pattern and passing a mock writer in the test
• C. It raises a RuntimeError because streaming must be enabled at graph compile time with streaming=True
• D. It silently returns None and calling None(data) raises TypeError; fix by checking if writer is not None

---

Answers

A1 — C. "messages" is the only mode that emits individual LLM tokens during generation. "values" and "updates" both emit after the superstep finishes, at which point the full message exists but the individual tokens are gone. "debug" emits internal task metadata, not model output.

A2 — D (trick question — read carefully). The stated conditions are self-consistent. If you actually have version="v2" and get_stream_writer() is correctly called, custom events should appear. In practice, the most common cause of missing on_custom_event events is using version="v1" (silently dropped). But the question describes a scenario where version="v2" is confirmed. If this happens in a real debugging session, the next step is verifying that _emit’s try/except is not masking an import error in the node. The canonical diagnostic: call the node inside a .stream() run with stream_mode="custom" and confirm events appear there first.

A3 — B. When stream_mode is a list, LangGraph wraps every yielded item as a (mode, chunk) tuple. This is unconditional — not configurable. When stream_mode is a single string, items are bare chunks.

A4 — C. subgraphs=True always changes the chunk structure to (namespace, original_chunk). On a flat graph with no subgraphs, namespace is always the empty tuple (). No exception, no filtering. The parameter is forward-compatible — you can add it now and the Module 12 subgraphs will automatically stream with non-empty namespaces.

A5 — B. get_stream_writer() raises when called outside a LangGraph task execution context (typically a LookupError). The _emit wrapper with try/except is the standard pattern for making nodes testable without mocking. Alternatively, declare writer: StreamWriter in the function signature — LangGraph injects the real writer during execution, and your test can inject a mock or a list-appending lambda.

Traps to avoid

Trap 1 — Using "values" mode to get LLM tokens. This is the most common streaming mistake. "values" emits the complete state after each superstep completes. By the time it emits, the LLM call is over and all tokens are merged into the final message. You get a large blob, not a stream. The 18-second silence in the introduction is exactly this mistake. Use "messages" for tokens.

Trap 2 — Forgetting version="v2" in astream_events. Custom events from get_stream_writer() appear as on_custom_event only in astream_events(version="v2"). With version="v1", they are silently discarded. No error. No warning. This is a silent data loss bug. Always write version="v2" explicitly. If you are calling astream_events and your custom events are nowhere to be found, check the version first.

Trap 3 — Calling get_stream_writer() once per LLM token. This inverts the role of the "custom" and "messages" modes. Token-level data belongs in "messages" — the LangGraph runtime handles that automatically for any node that calls an LLM. The "custom" channel is for business-level events: “editing storage.py”, “tests: 3 failed”, “plan approved”. If you hook into the model’s streaming callback and emit a custom event per token, you flood the "custom" channel with noise and lose the meaningful business signal. Reserve custom events for coarse-grained progress milestones.

---

Key Takeaways

• .stream() / .astream() expose the same LangGraph graph as an iterator, yielding chunks as supersteps complete. .invoke() is just .stream() that waits for all chunks and returns the last one. Same graph, different interface.
• LangGraph v1.x has 7 stream modes: "values" (full state per superstep), "updates" (node delta per node), "messages" (LLM tokens during generation), "custom" (business data via get_stream_writer), "debug" (internal runtime events), "checkpoints" (state at each checkpoint write), "tasks" (Pregel task metadata). Pass a list to combine modes — you get (mode, chunk) tuples.
• get_stream_writer() (imported from langgraph.config, not langgraph.types) emits arbitrary JSON-serializable data from inside a node into the "custom" stream. It is context-bound to the LangGraph task — wrap calls in try/except if you want nodes to remain callable outside .stream().
• stream_mode="messages" is the only mode that yields individual LLM tokens. It emits during the LLM call, before the superstep barrier. All other modes emit after the barrier.
• astream_events(version="v2") gives you named events (on_chat_model_stream, on_tool_start, on_tool_end, on_custom_event). Custom events only appear in version="v2" — the version argument is not optional.
• subgraphs=True changes every chunk to a (namespace, original_chunk) tuple. On a flat graph, namespace = () always. In a hierarchical graph (Module 12), the namespace identifies which subgraph the event came from.
• The streaming layer is a consumption interface, not a graph redesign. Forge gains full real-time visibility in Module 11 with two _emit calls in nodes.py and a 30-line stream_cli.py. Zero changes to the graph, edges, or ForgeState.

---

What’s Next

This is Module 11 of LangGraph in Production, a free 16-module course that takes you from your first StateGraph to a deployed, durable autonomous coding agent on LangGraph v1.x. Start at Module 1 or browse the full syllabus.

Want the full LangGraph mastery question bank (interview-style) and the next module in your inbox? Subscribe here — it’s free, like everything in this series.

Forge’s graph is now fully transparent in real time — but it is still one flat structure. Module 12 breaks it into composable subgraphs (Planner / Coder / Reviewer) and adds a hand-built Supervisor that routes between specialist agents via Command handoffs. That is when subgraphs=True stops being a preview and becomes essential — each subgraph’s internal nodes streaming with their own namespace.

---

Links: Module 10 — Long-Term Memory and the Store | Module 12 — Subgraphs and Multi-Agent Systems | Course index | Lab code (module-11)

---

References

1. LangGraph How-to: Streaming — https://langchain-ai.github.io/langgraph/how-tos/streaming/ — comprehensive reference for all stream modes and the subgraphs=True parameter. As of June 2026.

2. LangGraph API reference: langgraph.config — get_stream_writer, StreamWriter — https://langchain-ai.github.io/langgraph/reference/config/ — confirms the canonical import path from langgraph.config import get_stream_writer. As of June 2026.

3. LangChain / LangGraph astream_events guide — https://python.langchain.com/docs/how_to/streaming/#using-stream-events — covers version="v2" requirement and the on_custom_event type. As of June 2026.

4. LangGraph How-to: Stream from subgraphs — https://langchain-ai.github.io/langgraph/how-tos/streaming-subgraphs/ — details on subgraphs=True and namespace tuples. As of June 2026.

5. Lab code — stream_cli.py and forge/nodes.py for Module 11: https://github.com/dupuis1212/langgraph-prod-labs/tree/main/module-11. Tested against langgraph==1.2.5, langchain==1.3.9, langchain-core==1.4.7, langchain-anthropic==1.4.6, Python 3.12. As of June 2026.