streaming

124 notes in this subject, read out of this brain and free to use. This is page 1 of 3.

ProtocolEvent structure

Each protocol event is a ProtocolEvent envelope with fields: seq (strictly increasing within a run, use for ordering), method (channel name: messages, values, updates, custom, tools, lifecycle, etc.), params (object containing namespace—path of name:runtime_id segments from root graph, [] is root; timestamp—wall-clock milliseconds; node—graph node that emitted event when applicable; data—channel-specific payload).

stream.values for state snapshots

Use stream.values to stream full state snapshots after each step. Iterate snapshots to see state changes, then await stream.output to get the final state.

Example interleave multiple projections

Example for synchronous code: stream = graph.stream_events(input, version="v3"). Then: for name, item in stream.interleave("values", "messages", "subgraphs"): if name == "values": print(f"[state] keys={list(item)}"). elif name == "messages": print(f"[llm] node={item.node}"). elif name == "subgraphs": print(f"[subgraph] path={item.path}").

ToolCallTransformer built-in projection

LangGraph ships ToolCallTransformer as a built-in. Register it to expose stream.tool_calls on a plain StateGraph: from langgraph.prebuilt import ToolCallTransformer, then pass transformers=[ToolCallTransformer] to stream_events() or compile(). Access tool call information via stream.tool_calls with tool_name and input fields.

Stream handler transformer processing flow

The stream handler is the central dispatcher for one stream. For every protocol event it: (1) calls each registered transformer's process(event) hook in order, (2) wires named StreamChannel pushes back onto the protocol event stream, (3) stores the event in the run stream unless a transformer suppresses it, (4) calls finalize() or fail() on every transformer when the run ends.

required_stream_modes controls Pregel emission

required_stream_modes controls which Pregel stream modes the underlying graph emits during the stream. The runtime takes the union of every registered transformer's required_stream_modes and passes that union as the stream_mode argument to the graph's .stream() call. Modes that no transformer requests are never emitted. Valid values are Pregel stream modes: messages, tools, custom, values, updates, checkpoints, tasks, debug. Each transformer must declare every mode it acts on—an omitted mode is not emitted by the graph and never reaches process().

stream.stream_events() quickstart with version v3

Call graph.stream_events(input, version="v3") to create a run stream. The stream object exposes typed projections including stream.messages (for chat model output), stream.output (final state), stream.values (state snapshots), stream.subgraphs (nested graphs), stream.interrupts (human-in-the-loop payloads), and stream.extensions (custom projections). Multiple consumers can read these projections concurrently without consuming events needed by other projections.

Example resume after interrupt

Example: from langgraph.types import Command. stream = graph.stream_events(input, version="v3"). for message in stream.messages: print(message.text). if stream.interrupted: print(stream.interrupts). Then resume: stream = graph.stream_events(Command(resume={"decisions": [{"type": "approve"}]}), version="v3"). final_state = stream.output.

Streaming stack two-layer architecture

The streaming stack has two main layers: (1) Streaming emits raw graph execution events from the Pregel engine, (2) Event streaming normalizes those events, runs them through stream transformers, and exposes typed projections. The event router bridges the two layers by receiving normalized Pregel events and passing each event through registered stream transformers.

Namespace path in protocol events

The namespace field is a path from the root graph to the scope that emitted the event. The root is the empty array []. Each child execution adds one 'name:runtime_id' segment. A nested tool call inside a subgraph looks like ['researcher:6f4d', 'tools:91ac']. The name before ':' is the stable graph or node name; the suffix is a per-invocation runtime ID.

Register transformers at call time or compile time

Pass transformers at call time via stream_events(input, version='v3', transformers=[...]) for local experimentation. Compile transformers into the graph via builder.compile(transformers=[...]) when every run of that graph should produce the projection.

Resume after interrupt with Command

When a graph pauses for human input, inspect stream.interrupted and stream.interrupts, then resume by calling stream_events() again with Command. Resume requires a graph compiled with a checkpointer and a config carrying a thread ID.

StreamChannel side-channel vs main event stream

StreamChannel is the projection primitive transformers use for streaming values. It always exposes an iterable stream on stream.extensions.<name>. The constructor argument decides whether each push() also flows into the run's main event stream as a custom:<name> event. With a string name argument: pushes flow into main event stream as custom:<name> protocol events. Without a name: side-channel projection only, accessible on stream.extensions but not visible to raw event consumers. Named channel payloads must be serializable.

Concurrent consumption of multiple projections

Multiple projections can be consumed concurrently. In Python async code, use astream_events with asyncio.gather to consume multiple projections. In synchronous Python code, use stream.interleave(...) to consume multiple projections in strict arrival order. In JavaScript, use concurrent consumers with Promise.all to read multiple projections.

Lifecycle channel status events

The lifecycle channel tracks root run, subgraph, and subagent status. The data's event field is one of: started, running, completed, failed, interrupted. Beyond event, lifecycle data may include an optional graph_name, error, and cause describing why a child scope started (parent tool call, fan-out send, edge transition).

Raw Pregel event types

Raw Pregel events include: updates, values, messages, custom, checkpoints, tasks, and debug.

Event streaming recommended in-process model

Event streaming is the recommended in-process streaming model for most LangGraph application code. It returns a run stream object that can be consumed in multiple ways at the same time.

Messages channel content block lifecycle

The messages channel models output as content blocks. The data's event field is one of: message-start, content-block-start, content-block-delta, content-block-finish, message-finish. Content blocks have explicit boundaries: a block starts, emits zero or more deltas, and finishes before the next block in the same message starts. message-finish may include token usage; unrecoverable model-call failures arrive as message error events.

Stream projections table

Event streaming run stream exposes these typed projections: stream (iterate every protocol event), stream.messages (stream chat model messages and token deltas), stream.values (iterate state snapshots and await final value), stream.output (await final output), stream.subgraphs (discover and observe nested graph executions), stream.interrupts (inspect human-in-the-loop interrupt payloads), stream.interrupted (check whether run paused for human input), stream.extensions (consume custom stream transformer projections).

Example stream.subgraphs observation

Example: stream = graph.stream_events(input, version="v3"). Then: for subgraph in stream.subgraphs: print(subgraph.graph_name, subgraph.path). Then access messages: for message in subgraph.messages: print(message.text).

Quickstart example stream.messages iteration

Example: stream = graph.stream_events({"messages": [{"role": "user", "content": "What is 42 * 17?"}]}, version="v3"). Then iterate: for message in stream.messages: for token in message.text: print(token, end="", flush=True). Access final state: final_state = stream.output.

Event streaming sits above streaming layer

Event streaming sits one level above streaming, which exposes raw graph execution events through stream_mode modes such as updates, values, messages, custom, checkpoints, tasks, and debug. Use streaming when you need low-level access to those modes; use event streaming when application code benefits from typed projections.

stream.messages for token-by-token streaming

Use stream.messages for chat model output. In Python, message.text is iterable in synchronous code—iterate it for token-by-token output or call str(message.text) for complete text. message.reasoning exposes reasoning deltas and message.tool_calls exposes tool-call argument chunks. In JavaScript, message.text is both an async iterable and a promise-like value—iterate it for token-by-token output or await it for complete text.

Example consume messages channel directly

Example consuming raw content-block events instead of stream.messages projection: for event in stream: if event["method"] != "messages": continue. data = event["params"]["data"][0]. if not isinstance(data, dict): continue. if data.get("event") != "content-block-delta": continue. block = data.get("delta") or {}. if block.get("type") == "text-delta": print(block.get("text", ""), end="", flush=True). elif block.get("type") == "reasoning-delta": print(f"[thinking]{block.get('reasoning', '')}", end="", flush=True).

Protocol event channels and purposes

Protocol event channels: values (full graph state snapshots), updates (per-node state deltas), messages (content-block-centric chat model output), tools (tool call start, streamed output, finish, and error events), lifecycle (run, subgraph, and subagent status changes), checkpoints (lightweight checkpoint envelopes for branching and time travel), input (human-in-the-loop input requests and responses), tasks (Pregel task creation and result events), custom (user-defined payloads from graph code), custom:<name> (application-defined stream transformer output).

Transformers are observational not imperative

Transformers are observational and do not call back into the graph runtime. Instead, they consume events and push derived values into StreamChannel, promises, or other projection objects.

stream.subgraphs for observing nested graphs

Use stream.subgraphs to observe nested graph work without parsing namespace strings. Each subgraph exposes graph_name (the name of the compiled graph or agent) and path. A named agent dispatched from a tool surfaces under that name, and the lifecycle event that opens the scope carries a cause linking back to the dispatching tool call.

Example stream raw protocol events

Example: stream = graph.stream_events({"messages": [{"role": "user", "content": "What is 42 * 17?"}]}, version="v3"). for event in stream: namespace = event["params"]["namespace"]. print(namespace, event["method"], event["params"]["data"]).

Tools channel event types

The tools channel exposes tool execution. The data's event field is one of: tool-started, tool-output-delta, tool-finished, tool-error. Tool events are correlated by tool call ID, so a tool execution can be joined back to its originating tool-call content block on the messages channel.

Example named StreamChannel transformer

Example ToolActivityTransformer with named channel: class ToolActivityTransformer(StreamTransformer): required_stream_modes = ("tools",). def __init__(self, scope: tuple[str, ...] = ()) -> None: super().__init__(scope). self.activity = StreamChannel[ToolActivity]("tool_activity"). def init(self) -> dict: return {"tool_activity": self.activity}. def process(self, event: ProtocolEvent) -> bool: if event["method"] != "tools": return True. data = event["params"]["data"]. if isinstance(data, dict) and data.get("tool_name") and data.get("event"): status = "error" if data["event"] == "tool-error" else "started". self.activity.push({"name": data["tool_name"], "status": status}). return True.

Example unnamed StreamChannel with get_stream_writer

Example unnamed channel paired with get_stream_writer: def node(state): writer = get_stream_writer(). writer({"kind": "progress", "message": "retrieving context"}). return state. class CustomTransformer(StreamTransformer): required_stream_modes = ("custom",). def __init__(self, scope: tuple[str, ...] = ()) -> None: super().__init__(scope). self.log = StreamChannel(). def init(self) -> dict: return {"custom": self.log}. def process(self, event: ProtocolEvent) -> bool: if event["method"] == "custom": self.log.push(event["params"]["data"]). return True. Then: stream = graph.stream_events(input, version="v3", transformers=[CustomTransformer]). for item in stream.extensions["custom"]: print(item).

Example StatsTransformer final-value projection

Example final-value projection using unnamed stream: class StatsTransformer(StreamTransformer): required_stream_modes = ("messages",). def __init__(self, scope: tuple[str, ...] = ()) -> None: super().__init__(scope). self.total_tokens = 0. self.total_tokens_log = StreamChannel[int](). def init(self) -> dict: return {"total_tokens": self.total_tokens_log}. def process(self, event: ProtocolEvent) -> bool: data = event["params"]["data"]. if isinstance(data, dict): usage = data.get("usage") or {}. self.total_tokens += usage.get("output_tokens") or 0. return True. def finalize(self) -> None: self.total_tokens_log.push(self.total_tokens). self.total_tokens_log.close().

StreamTransformer interface

A transformer implements StreamTransformer with methods: init() (creates projection object—user transformer projections appear under stream.extensions), process(event: ProtocolEvent) -> bool (observes each protocol event, return false only when intentionally suppressing original event), finalize() (closes or resolves non-channel projections after successful stream), fail(err: BaseException) (propagates errors to non-channel projections).

useChannel React implementation

In React, useChannel is imported from @langchain/react and called as useChannel(stream, ["custom:redaction-stats"]), with optional target namespace parameter and options object as additional arguments.

useStream setup for custom channels

Wire up useStream as usual with apiUrl and assistantId parameters. The custom-channel selectors (useExtension, useChannel) take the same stream handle returned by useStream.

JavaScript RedactionStatsTransformer example

A JavaScript RedactionStatsTransformer example creates a StreamChannel.remote<RedactionStatsEvent>("redaction-stats"), defines init() returning {redactionStats}, and implements process(event: ProtocolEvent): boolean that calls redactInPlace(event, counts), pushes an object with kind, at, delta, counts, and total to redactionStats if delta has keys, and returns true to keep the event.

Custom stream channels overview

LangGraph agents stream more than messages and tool calls. A server-side stream transformer can inspect or rewrite the protocol as it flows to the client and publish its own structured data on a named custom channel. The frontend reads that channel with useExtension for the latest payload, and useChannel as a raw-events escape hatch.

RedactionStatsEvent type structure

The RedactionStatsEvent type has the following structure: kind (literal "update"), at (number, timestamp in milliseconds), delta (Partial<Record<PiiType, number>> where PiiType is "email" | "phone" | "ssn" | "credit_card" | "ip_address"), counts (Record<PiiType, number>), and total (number).

Custom channel use cases

Custom channels fit any server-side signal that does not map cleanly to messages, tool calls, or graph state: compliance and redaction stats (counts of scrubbed PII, blocked content, or policy hits), progress reporting (percentage complete or step labels emitted by a long-running tool), live metrics (token usage, latency, or cost accumulating during a run), sources and citations (retrieved documents pushed to a side panel as the agent grounds its answer), and domain events (any structured update your backend wants to surface without changing the message transcript).

useExtension vs useChannel comparison

useExtension returns the latest payload (T | undefined) unwrapped and typed, subscribes by channel name ("redaction-stats"), and is used when you need the current value. useChannel returns a bounded buffer of raw events (Event[]), requires manual unwrapping via event.params.data, subscribes by full channel id (["custom:redaction-stats"]), and is used when you need history, a log, or multiple channels. Both accept optional target namespace argument for scoping.

useChannel Angular implementation

In Angular, use injectChannel from @langchain/angular and call it as injectChannel(stream, ["custom:redaction-stats"]). The result is accessed as a signal via rawEvents().

useChannel Svelte implementation

In Svelte, useChannel is imported from @langchain/svelte and called as useChannel(stream, ["custom:redaction-stats"]).

useChannel Vue implementation

In Vue, useChannel is imported from @langchain/vue and called as useChannel(stream, ["custom:redaction-stats"]). The result is accessed via rawEvents.value.

useChannel options parameter

useChannel accepts an options argument to control the buffer with two options: bufferSize (default: "default") sets the maximum number of buffered events, with older events dropping once the cap is reached; replay (default: true) replays events already seen on the channel when the selector mounts instead of only live events.

useChannel raw event structure

Each entry from useChannel is a raw protocol event, so the payload sits under event.params.data. The payload must be unwrapped manually by accessing event.params?.data.

useChannel selector for raw events

useChannel is the raw-events escape hatch. It subscribes to one or more channels and returns a bounded buffer of the underlying protocol events rather than a single unwrapped value. Reach for it when you need history instead of the latest value, such as an event log or audit trail, or when you need a channel that no higher-level selector covers. Pass the full channel id ("custom:redaction-stats").

useExtension target argument for namespace scoping

useExtension accepts an optional third target argument that scopes the subscription to a namespace, the same way useMessages(stream, node) scopes messages to a discovered graph node.

useExtension return types across frameworks

The useExtension return value follows each framework's reactivity model: a plain value in React and Svelte, a Ref in Vue (accessed via latest.value), and a signal in Angular (accessed via latest()).

useExtension selector for custom channels

useExtension subscribes to a custom:<name> channel and returns the most recent payload the transformer pushed, already unwrapped and typed. Pass the bare channel name ("redaction-stats"), not the custom: prefix. It is the ergonomic choice when the UI only needs the current value, such as a live counter, progress percentage, or status badge. Return value is undefined until the first payload arrives.

Attaching transformers to agents in JavaScript

When creating an agent with createAgent(), pass transformers as a list via the streamTransformers parameter. Each transformer is a function returning a StreamTransformer instance.

Attaching transformers to agents in Python

When creating an agent with create_agent(), pass transformers as a list via the transformers parameter. Each transformer is a callable that returns a StreamTransformer instance.

StreamChannel.push() method

StreamChannel instances have a push() method that publishes a payload onto the channel. The payload can be any structured data type matching the transformer's type definition.

JavaScript StreamTransformer implementation

A JavaScript StreamTransformer is an object with init() and process() methods. init() returns an object mapping channel names to StreamChannel instances. For remote channels, use StreamChannel.remote<T>(name). The process method takes a ProtocolEvent and returns a boolean indicating whether to keep the event.

Python RedactionStatsTransformer example

A Python RedactionStatsTransformer example opens a StreamChannel named "redaction-stats" in __init__, defines init() returning {"redactionStats": self.redaction_stats}, and implements process(event: ProtocolEvent) -> bool that calls redact_in_place(event, self.counts), pushes a dict with kind, at, delta, counts, and total to self.redaction_stats if delta is non-empty, and returns True to keep the event.

Python StreamTransformer implementation

A Python StreamTransformer subclass opens a named StreamChannel in __init__, returns it from the init() method as a dictionary, and processes events in the process(process(event: ProtocolEvent) -> bool method. The init() method returns dict[str, StreamChannel], and process returns True to keep the event or False to drop it.

StreamTransformer process method behavior

The transformer's process method runs for every protocol event. It can mutate the event in place (such as scrubbing PII from messages, tools, and values data) and push side-channel updates whenever it has something to report. The method returns a boolean indicating whether to keep the event in the stream.

StreamTransformer and StreamChannel requirements

Stream transformers and StreamChannel require langgraph>=1.2 for Python and @langchain/langgraph>=1.3.1 for JavaScript.

Client-side selector frameworks supporting custom channels

The client-side selectors (useExtension, useChannel) ship with the v1 frontend SDK packages: @langchain/react, @langchain/vue, @langchain/svelte, and @langchain/angular.

Routing streaming content to nodes with useMessages

Pass each discovered subgraph snapshot to the useMessages(stream, node) selector hook to read messages scoped to that specific node. The first mounted selector opens a scoped subscription for that node namespace. When the node card unmounts, the subscription is released automatically. This avoids coupling the UI to graph state key names.

Node discovery via stream.subgraphs

Nodes are discovered automatically during graph execution via stream.subgraphs without the need for hardcoded lists. The frontend obtains a SubgraphDiscoverySnapshot for every observed step. Nodes are discovered as they run, so only nodes that are relevant to the current execution appear in the discovery map.

Give your agent this brain