streaming

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

Scoped messages vs. stream.values for node content

Use useMessages(stream, node) to render node-scoped streaming and final messages for display in node cards. Use stream.values only when you intentionally need to read a whole-graph state field using the actual state key name. Scoped messages are tied to the producing node and support parallel graph paths without guessing from message order.

Handling incomplete markdown in streaming content

Streaming content may include partial tokens or markdown that has not been fully formed yet. If you render markdown in node cards, ensure your renderer handles incomplete syntax gracefully, such as unclosed bold markers like '**'.

useStream setup for graph execution

Wire up useStream with apiUrl and assistantId parameters. Access stream.subgraphs.values() to retrieve discovered nodes for the current run, and pass this to UI components for rendering node cards and progress indicators.

Functional API streaming execution example

To stream events from an entrypoint workflow in Python: ```python config = { "configurable": { "thread_id": "some_thread_id" } } stream = my_workflow.stream_events(some_input, config, version="v3") for message in stream.messages: for token in message.text: print(token, end="", flush=True) ``` For async streaming: ```python stream = await my_workflow.astream_events(some_input, config, version="v3") async for message in stream.messages: async for token in message.text: print(token, end="", flush=True) ```

Interrupt payloads surface differently in Python and JavaScript APIs

In Python with event streaming (graph.stream_events(..., version="v3")), interrupt values appear on stream.interrupts and stream.interrupted is True when the run pauses. With default invoke() API, interrupts surface under result["__interrupt__"]. In JavaScript, values are returned under the __interrupt__ field via result.__interrupt__.

Stream with human-in-the-loop using stream_events()

When building interactive agents with human-in-the-loop workflows, use graph.stream_events(..., version="v3") in a loop. The stream object provides: stream.messages (chat-model output as content blocks, iterate message.text for token deltas), stream.values (full state snapshots after each step), stream.interrupted / stream.interrupts (check if graph paused and read payloads), and stream.subgraphs[*].messages (for nested subgraph message chunks). Resume by calling stream_events again with Command(resume=...) and repeat until stream.interrupted is false.

Stream output format v2 structure

With version='v2' passed to stream() or astream(), every chunk is a StreamPart dict with consistent shape: {"type": "values" | "updates" | "messages" | "custom" | "checkpoints" | "tasks" | "debug", "ns": (), "data": ...}. The ns field is a namespace tuple populated for subgraph events (empty tuple for root graph). The data payload type varies by stream mode.

Stream modes available

LangGraph supports seven stream modes: (1) 'values' - full state snapshot after each step; (2) 'updates' - only changed keys from each node; (3) 'messages' - (message_chunk, metadata) tuples from LLM calls; (4) 'custom' - arbitrary data from get_stream_writer(); (5) 'checkpoints' - checkpoint events matching get_state() format, requires checkpointer; (6) 'tasks' - task start/finish events with results and errors, requires checkpointer; (7) 'debug' - combines checkpoints and tasks with extra metadata.

Type narrowing with stream mode

When streaming with version='v2', each stream mode has a corresponding TypedDict: ValuesStreamPart, UpdatesStreamPart, MessagesStreamPart, CustomStreamPart, CheckpointStreamPart, TasksStreamPart, DebugStreamPart. The union type StreamPart is a disjoint union on part['type'], enabling full type narrowing in editors and type checkers.

Messages stream mode output format

The 'messages' stream mode outputs tuples of (message_chunk, metadata) where message_chunk is the token or message segment from the LLM and metadata is a dictionary containing details about the graph node and LLM invocation, including tags and langgraph_node fields.

Filtering LLM tokens by tags in messages stream

When streaming in 'messages' mode, you can filter streamed tokens by checking the 'tags' field in the metadata. LLM models can be tagged at initialization (e.g., init_chat_model(model='gpt-5.4-mini', tags=['joke'])). In the stream handler, check if metadata['tags'] matches the desired tags to filter output.

Filtering LLM tokens by node in messages stream

To stream tokens only from specific nodes in 'messages' mode, filter by the 'langgraph_node' field in the streamed metadata. Check if metadata['langgraph_node'] equals the target node name to include only tokens from that node.

Using nostream tag to exclude LLM output

Apply the 'nostream' tag to LLM invocations to exclude their tokens from the 'messages' stream mode entirely. Invocations tagged with 'nostream' still run and produce output, but their tokens are not emitted. Use this when you need LLM output for internal processing but do not want to stream it to the client, or when streaming the same content through a different channel.

Custom data streaming with get_stream_writer

To emit custom user-defined data from a node or tool, call get_stream_writer() to access the stream writer and emit data. Then set stream_mode='custom' when calling stream() or astream(). The emitted data can be any dict structure and will be accessible via chunk['data'] when stream_mode='custom'.

Custom stream writer limitation in async Python < 3.11

In async code running on Python < 3.11, get_stream_writer() will not work. Instead, add a 'writer' parameter to your node or tool function and pass it manually from the config object.

Streaming subgraph outputs

To include outputs from subgraphs in streamed outputs, set subgraphs=True in the stream() method. Outputs are streamed as normal StreamPart chunks with version='v2', where the 'ns' field identifies the source: empty tuple for root, tuple like ('node_name:<task_id>',) for subgraphs.

Subgraph message streaming requires subgraphs=True

When a compiled graph (like one created with create_agent) is added as a node (creating a subgraph), streaming in 'messages' mode on the parent graph will not emit token chunks from the inner graph's LLM calls unless subgraphs=True is set. Invoking the inner graph directly will emit tokens.

Multiple stream modes at once

Pass a list of stream modes as the stream_mode parameter to stream multiple modes simultaneously. With version='v2', each chunk is a StreamPart dict with a 'type' field indicating which mode produced it. Use chunk['type'] to distinguish between modes and route to appropriate handlers.

Checkpoints stream mode requirements

The 'checkpoints' stream mode requires a checkpointer to be configured when compiling the graph. It streams checkpoint events in the same format as get_state() output after each step.

Tasks stream mode requirements

The 'tasks' stream mode requires a checkpointer to be configured when compiling the graph. It streams task start/finish events containing information about which node is running, its results, and any errors.

Debug stream mode combines checkpoints and tasks

The 'debug' stream mode combines 'checkpoints' and 'tasks' events with additional metadata, streaming as much information as possible throughout graph execution. Use 'checkpoints' or 'tasks' directly if you only need a subset of debug information.

Streaming from arbitrary LLMs with custom mode

Use stream_mode='custom' to stream data from any LLM API, even if it does not implement the LangChain chat model interface. Call your custom streaming client inside a node, use get_stream_writer() to emit chunks as custom data, and consume them by listening to 'custom' stream mode.

Version v2 streaming format advantages

Version v2 provides a unified output format regardless of stream mode, number of modes, or subgraph settings. With v1 (default), format changes based on options: single mode returns raw data, multiple modes return (mode, data) tuples, subgraphs return (namespace, data) tuples. v2 always uses consistent StreamPart structure.

Stream method parameters

The stream() and astream() methods accept: stream_mode (string or list of strings for modes), version (string, 'v2' recommended for unified format), subgraphs (boolean to include subgraph outputs), and config (dict with configurable thread_id for checkpointing).

Async with Python < 3.11 requires explicit config

When using Python < 3.11 with async code, you must explicitly pass RunnableConfig to ainvoke() to enable proper streaming. Upgrade to Python 3.11+ to avoid this requirement.

Tool progress stream mode (JavaScript only)

The 'tools' stream mode (available in JavaScript) emits tool lifecycle events: on_tool_start (invocation begins), on_tool_event (intermediate data from async generator tools), on_tool_end (tool returns final result), on_tool_error (tool throws error). Each event includes name, input/output/error/data, and toolCallId.

Async generator tools for streaming progress

In JavaScript, define tools as async generators (async function*) to emit on_tool_event events. Each yield sends intermediate data to the stream; the return value becomes the tool's final result.

useStream React hook toolProgress array

The useStream hook from @langchain/langgraph-sdk/react exposes a toolProgress array (when 'tools' is in streamMode) containing ToolProgress objects with fields: name (tool name), state ('starting'|'running'|'completed'|'error'), toolCallId, input, data (most recent yielded data), result (final output on completion), error (on failure).

Basic streaming example with updates mode

Example shows streaming with version='v2': 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"]}')

Full v2 streaming example with custom writer

Example creates a StateGraph with a node that uses get_stream_writer() to emit {'status': 'thinking of a joke...'}, then streams with stream_mode=['updates', 'custom'] and version='v2'. The output shows both custom status messages and node updates.

Graph state streaming updates vs values

Example defines State with topic and joke fields. refine_topic node appends ' and cats' to topic. generate_joke creates joke about the topic. Streaming with stream_mode='updates' shows only node outputs. Streaming with stream_mode='values' shows full state after each step, including intermediate states with empty joke field.

LLM token streaming example

Example calls init_chat_model(model='gpt-5.4-mini') and invokes it with a user prompt. Messages stream tokens using stream_mode='messages' and version='v2'. Output is accessed via chunk['data'] which is (message_chunk, metadata) tuple, allowing token-by-token output.

Filtering messages by tags full example

Example creates two tagged models: joke_model = init_chat_model(model='gpt-5.4-mini', tags=['joke']) and poem_model with tags=['poem']. In async astream with stream_mode='messages', check if metadata['tags'] == ['joke'] to filter only joke tokens. Python < 3.11 requires passing config explicitly to ainvoke.

Filtering messages by node full example

Example defines write_joke and write_poem nodes that call model.invoke(). Both run concurrently from START. In stream with stream_mode='messages' and version='v2', check metadata['langgraph_node'] == 'write_poem' to filter only poem tokens.

Custom data streaming example

Example node calls get_stream_writer() to emit {'custom_key': 'Generating custom data inside node'}. Graph streams with stream_mode='custom' and version='v2'. Output shows chunk['type'] == 'custom' with chunk['data']['custom_key'] containing the message.

Tool custom data example

Example @tool query_database uses get_stream_writer() to emit progress updates: {'data': 'Retrieved 0/100 records', 'type': 'progress'}. Graph streams with stream_mode='custom' and version='v2'. Output shows progress type and data fields.

Subgraph streaming example with version='v2'

Parent graph has node_1 and node_2 (subgraph). With stream_mode='updates', subgraphs=True, version='v2', chunks show chunk['ns'] as () for root or ('node_2:<task_id>',) for subgraph. Root updates show node_1 and node_2 updates; subgraph chunks show subgraph_node_1 and subgraph_node_2 updates.

Checkpoints streaming example

Example compiles graph with checkpointer=MemorySaver(). Streams with stream_mode='checkpoints', version='v2', config={'configurable': {'thread_id': '1'}}. Output shows checkpoint data in chunk['data'].

Tasks streaming example

Example compiles graph with checkpointer=MemorySaver(). Streams with stream_mode='tasks', version='v2', config={'configurable': {'thread_id': '1'}}. Output shows task event data in chunk['data'].

Arbitrary LLM streaming example

Example uses AsyncOpenAI client to stream tokens. get_items async tool calls stream_tokens() and emits each chunk via get_stream_writer(). Graph node processes tool calls. Stream with stream_mode='custom', version='v2' outputs each custom chunk. Shows how to integrate non-LangChain LLMs.

Disable streaming for models that don't support it

Set streaming=False when initializing a model to disable streaming. Use init_chat_model("claude-sonnet-4-6", streaming=False) or ChatOpenAI(model="gpt-5.5", streaming=False) in Python. In TypeScript, use new ChatOpenAI({ model: "gpt-5.5", streaming: false }). Not all chat model integrations support the streaming parameter; if your model doesn't support it, use disable_streaming=True in Python or disableStreaming: true in TypeScript instead, as these parameters are available on all chat models via the base class.

v2 streaming format unified output

The v2 streaming format provides a unified output format across all scenarios. Single stream mode returns StreamPart dict with type, ns, data fields. Multiple stream modes return the same StreamPart dict, filtered on chunk["type"]. Subgraph streaming returns the same StreamPart dict, checking chunk["ns"]. Multiple modes with subgraphs all use the same StreamPart dict.

v2 invoke return type

When you pass version="v2" to invoke() or ainvoke(), it returns a GraphOutput object with .value and .interrupts attributes. result.value contains your output as a dict, Pydantic model, or dataclass. result.interrupts is a tuple of Interrupt objects, empty if none occurred.

GraphOutput backwards compatibility

Dict-style access on GraphOutput (result["key"], "key" in result, result["__interrupt__"]) still works for backwards compatibility but is deprecated and will be removed in a future version. Migrate to result.value and result.interrupts instead.

v1 vs v2 invoke return type comparison

v1 returns a plain dict (state) with interrupts embedded under __interrupt__ key. v2 returns GraphOutput with .value and .interrupts attributes, separating state from interrupt metadata. To check for interrupts in v2: if result.interrupts: print(result.interrupts[0].value). To check in v1: if "__interrupt__" in result: print(result["__interrupt__"][0].value).

Pydantic and dataclass state coercion in v2

When your graph state is a Pydantic model or dataclass, v2 values mode automatically coerces output to the correct type. For example, with stream_mode="values" and version="v2", chunk["data"] returns a MyState instance (the Pydantic model or dataclass class) rather than a plain dict.

v1 vs v2 streaming mode differences

v1 single stream mode returns raw data (dict), v2 returns StreamPart dict with type, ns, data. v1 multiple stream modes return (mode, data) tuples, v2 returns same StreamPart dict filtered on chunk["type"]. v1 subgraph streaming returns (namespace, data) tuples, v2 returns same StreamPart dict checking chunk["ns"]. v1 multiple modes with subgraphs return (namespace, mode, data) triples, v2 returns same StreamPart dict.

Async context propagation in Python < 3.11

In Python versions < 3.11, asyncio tasks do not support the context parameter, limiting LangGraph's ability to automatically propagate context. This affects streaming in two ways: (1) You must explicitly pass RunnableConfig into async LLM calls (e.g., ainvoke()), as callbacks are not automatically propagated. (2) You cannot use get_stream_writer in async nodes or tools—you must pass a writer argument directly.

Async LLM call with manual config propagation

In async node functions for Python < 3.11, accept config as an argument and pass it to model.ainvoke() to ensure proper context propagation. Example: async def call_model(state, config): joke_response = await model.ainvoke([{"role": "user", "content": f"Write a joke about {state['topic']}"}], config)

Stream LLM tokens with messages stream_mode

To stream LLM tokens in async context, use stream_mode="messages" with version="v2". Check for chunk["type"] == "messages" to find message chunks. Access message content via message_chunk, metadata = chunk["data"], then check if message_chunk.content exists.

StreamWriter in async nodes

In async nodes or tools, add writer as an argument in the function signature. LangGraph will automatically pass the StreamWriter to the function. Use writer({"custom_key": "value"}) to stream custom data. Set stream_mode="custom" when calling astream() to receive the custom data.

Custom streaming with stream writer example

Example showing async custom streaming: 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']}"}. Call with graph.astream({"topic": "ice cream"}, stream_mode="custom", version="v2") and check chunk["type"] == "custom" to access chunk["data"].

Stream mode with tool calls

When invoking a graph with tool calls using streamMode="custom", the stream returns custom chunk data. Example: graph.stream(inputs, { streamMode: "custom" }) yields chunks where you can access chunk.content.

Streaming with Functional API

The Functional API uses the same streaming mechanism as the Graph API. Use get_stream_writer() (Python) or pass a stream writer parameter (JavaScript) to emit custom data. Use stream_events() to process streamed output and iterate over (mode, chunk) pairs.

Async with Python < 3.11 using StreamWriter

If using Python < 3.11 with async code, get_stream_writer() will not work. Instead, use StreamWriter class directly as a function parameter: async def main(inputs: dict, writer: StreamWriter) -> int

Example: streaming with custom data emission

Use get_stream_writer() in entrypoint to obtain writer instance. Call writer.write() to emit custom data before and after computation. Use stream_events() with version='v3', then iterate over stream.values to receive value snapshots.

Custom stream events from subgraphs streaming mode support

Enabled streaming for subgraph custom events by updating TeeStream to handle event types separately, allowing custom events emitted from subgraphs to be streamed.

Remote graphs now use LangGraphJS native v3 stream for v2 event-streaming

Fixed protocol v2 event streaming against JS sidecar (remote) graphs by serving them through LangGraphJS's native v3 stream instead of the legacy reconstruction path. This resolved tool calls not rendering, headless interrupts never executing or resuming, and 400: tool_use ids must be unique errors on the final message after a resume.

Custom stream events from subgraphs forwarded to client

Fixed custom stream events emitted from subgraphs not being forwarded to the client when using stream_mode=['custom'] with stream_subgraphs=True on JS deployments.

join_stream with stream_mode filter fixed

Fixed a bug where calling join_stream with a stream_mode filter could cause non-message events from subgraphs to be incorrectly filtered from the results.

Give your agent this brain