useStream initialization parameters
useStream accepts an object with apiUrl specifying the backend server URL (for example, http://localhost:2024) and assistantId specifying the graph name from langgraph.json to identify which agent to connect to.
LangChain · Agents · all subjects
agents/streaming
116 notes in this subject, read out of this brain and free to use. This is page 2 of 2.
LangChain SDK v1 frontend packages
LangChain frontend SDKs are built for agent applications, not just token-streaming chatbots. The current documentation uses v1 frontend SDK packages. Migration guides are available for earlier versions of React, Vue, Svelte, and Angular.
Use cases for reasoning tokens
Reasoning tokens enable transparency (show users the model's reasoning to build trust), debugging (inspect the model's thought process to identify where it goes wrong), educational tools (teach students problem-solving by revealing how AI approaches questions), decision support (let domain experts validate the reasoning behind recommendations), and quality assurance (audit reasoning chains for compliance in regulated industries).
useStream hook for reasoning-capable agents
The useStream hook connects to reasoning-capable agents and iterates stream.messages in chat UI. Branch on HumanMessage.isInstance and AIMessage.isInstance, then pass each assistant message to a component that reads contentBlocks and separates reasoning from text. Set isStreaming on the last message while stream.isLoading is true so thinking blocks update as tokens arrive.
useStream React implementation for reasoning tokens
React implementation using useStream with reasoning-capable agent: import { useStream } from "@langchain/react"; import { AIMessage, HumanMessage } from "langchain"; function Chat() { const stream = useStream<typeof myAgent>({ apiUrl: "http://localhost:2024", assistantId: "reasoning", }); return ( <div className="messages"> {stream.messages.map((msg, i) => { if (HumanMessage.isInstance(msg)) { return <HumanBubble key={i} text={msg.text} />; } if (AIMessage.isInstance(msg)) { return ( <AIResponse key={i} message={msg} isStreaming={stream.isLoading && i === stream.messages.length - 1} /> ); } return null; })} </div> ); }
ThinkingBubble component for reasoning tokens
A ThinkingBubble component presents reasoning tokens in a visually distinct, collapsible container. It accepts reasoning (string) and isStreaming (boolean) props. The component shows a preview of the first 120 characters when collapsed, displays a character count, shows a spinner when streaming, and includes a chevron toggle to expand/collapse the full reasoning. When expanded, it displays the complete reasoning in a pre-formatted text container.
ThinkingBubble React implementation
React ThinkingBubble component implementation: import { useState } from "react"; function ThinkingBubble({ reasoning, isStreaming, }: { reasoning: string; isStreaming: boolean; }) { const [isExpanded, setIsExpanded] = useState(false); const charCount = reasoning.length; const previewLength = 120; const preview = reasoning.length > previewLength ? reasoning.slice(0, previewLength) + "..." : reasoning; return ( <div className="thinking-bubble"> <button className="thinking-header" onClick={() => setIsExpanded(!isExpanded)} > <span className="thinking-icon"> {isStreaming ? ( <span className="thinking-spinner" /> ) : ( "💭" )} </span> <span className="thinking-label"> {isStreaming ? "Thinking..." : `Thought process (${charCount} chars)`} </span> <span className={`chevron ${isExpanded ? "expanded" : ""}`}>▶</span> </button> {isExpanded && ( <div className="thinking-content"> <pre>{reasoning}</pre> </div> )} {!isExpanded && !isStreaming && ( <div className="thinking-preview">{preview}</div> )} </div> ); }
AIResponse component combining thinking and text
An AIResponse component combines ThinkingBubble and text rendering. It extracts reasoning blocks and text blocks from the message's contentBlocks, determines if currently in reasoning phase (isStreaming and no text yet) or text phase (isStreaming and text present), and renders accordingly. It shows a cursor blink animation during text streaming.
AIResponse React implementation
React AIResponse component combining thinking and text: function AIResponse({ message, isStreaming, }: { message: AIMessage; isStreaming: boolean; }) { const reasoningBlocks = message.contentBlocks .filter((b) => b.type === "reasoning") .map((b) => b.reasoning) .join(""); const textBlocks = message.contentBlocks .filter((b) => b.type === "text") .map((b) => b.text) .join(""); const hasReasoning = reasoningBlocks.length > 0; const hasText = textBlocks.length > 0; const isReasoningPhase = isStreaming && !hasText; const isTextPhase = isStreaming && hasText; return ( <div className="ai-response"> {hasReasoning && ( <ThinkingBubble reasoning={reasoningBlocks} isStreaming={isReasoningPhase} /> )} {hasText && ( <div className="ai-text-bubble"> <p>{textBlocks}</p> {isTextPhase && <span className="cursor-blink">▊</span>} </div> )} </div> ); }
Filtering empty reasoning blocks
Some models produce empty reasoning blocks as placeholders. Filter these out using: const meaningfulReasoning = message.contentBlocks.filter((b) => b.type === "reasoning" && b.reasoning.trim().length > 0);
Handling multiple reasoning-text cycles
A single message can alternate between reasoning and text blocks. To preserve interleaving, iterate contentBlocks in order rather than grouping by type. Use: message.contentBlocks.forEach((block) => { if (block.type === "reasoning") { /* Render ThinkingBubble */ } else if (block.type === "text") { /* Render text paragraph */ } });
Best practices for reasoning token UI
Best practices for displaying reasoning tokens: (1) Default to collapsed - show reasoning on demand, not by default; (2) Show character count - gives users a quick sense of how much thinking went into the response; (3) Differentiate visually - use distinct colors, borders, or backgrounds so reasoning is never confused with the actual answer; (4) Animate transitions - smooth expand/collapse animations improve perceived quality; (5) Consider accessibility - use proper ARIA attributes (aria-expanded, aria-controls) on the toggle button; (6) Truncate in previews - show a short preview when collapsed so users can decide whether to expand.
Messages without reasoning blocks
Not every AI message will contain reasoning blocks. When contentBlocks has only text blocks, render a standard message bubble without the ThinkingBubble. Always check hasReasoning before rendering a ThinkingBubble component.
Extracting reasoning and text blocks from AIMessage
To extract reasoning and text blocks from an AIMessage, filter the contentBlocks array by type. Example code: function extractBlocks(msg: AIMessage) { const reasoningBlocks = msg.contentBlocks.filter((b) => b.type === "reasoning").map((b) => b.reasoning); const textBlocks = msg.contentBlocks.filter((b) => b.type === "text").map((b) => b.text); return { reasoning: reasoningBlocks.join(""), text: textBlocks.join("") }; }
Reasoning tokens not available on all models
Not all models produce reasoning tokens. This pattern applies specifically to models that support extended thinking or chain-of-thought output. Standard chat models return only text blocks.
Reasoning tokens structure in AIMessage
Models with reasoning capabilities like OpenAI's GPT-5 and Anthropic's Claude with extended thinking produce structured content blocks within an AIMessage. The contentBlocks property contains two distinct types of blocks: reasoning blocks (the model's internal chain-of-thought, problem decomposition, and step-by-step analysis) and text blocks (the final polished response). Reasoning blocks have type 'reasoning' with a reasoning property, and text blocks have type 'text' with a text property. Example: { type: "reasoning", reasoning: "Let me think about this step by step..." } and { type: "text", text: "The answer is 42." }
contentBlocks property on AIMessage
The contentBlocks property on an AIMessage contains all content blocks in the order they were generated. These blocks can be filtered by type to separate reasoning from text. A single message may contain multiple reasoning blocks if the model pauses its reasoning, produces partial text, then reasons further.
HITL streaming with stream_events
Stream real-time updates while the agent runs and handles interrupts using stream_events(). Use stream.messages to stream LLM tokens and stream.values to check agent state snapshots. Use version='v3' for streaming mode. Check stream.interrupted to determine if the run paused for human input, and resume with stream_events(Command(resume={...}), config, version='v3').
HITL streaming JavaScript example
Example JavaScript code for streaming with HITL: import { Command } from "@langchain/langgraph"; const config = { configurable: { thread_id: "some_id" } }; const stream = await agent.streamEvents( { messages: [{ role: "user", content: "Delete old records from the database" }] }, { ...config, version: "v3" } ); for await (const message of stream.messages) { for await (const token of message.text) { process.stdout.write(token); } } if (stream.interrupted) { console.log(`\n\nInterrupt: ${JSON.stringify(stream.interrupts)}`); } const resumeStream = await agent.streamEvents( new Command({ resume: { decisions: [{ type: "approve" }] } }), { ...config, version: "v3" } ); for await (const message of resumeStream.messages) { for await (const token of message.text) { process.stdout.write(token); } }
HITL streaming Python example
Example Python code for streaming with HITL: from langgraph.types import Command config = {"configurable": {"thread_id": "some_id"}} stream = agent.stream_events( {"messages": [{"role": "user", "content": "Delete old records from the database"}]}, config=config, version="v3", ) for message in stream.messages: for token in message.text: print(token, end="", flush=True) if stream.interrupted: print(f"\n\nInterrupt: {stream.interrupts}") stream = agent.stream_events( Command(resume={"decisions": [{"type": "approve"}]}), config=config, version="v3", ) for message in stream.messages: for token in message.text: print(token, end="", flush=True)
Streaming tool calls and responses code example
```python from langchain.agents import create_agent from langchain.messages import AIMessage, AIMessageChunk, ToolMessage def get_weather(city: str) -> str: return f"It's always sunny in {city}!" agent = create_agent("openai:gpt-5.5", tools=[get_weather]) def _render_message_chunk(token: AIMessageChunk) -> None: if token.text: print(token.text, end="|") if token.tool_call_chunks: print(token.tool_call_chunks) def _render_completed_message(message) -> None: if isinstance(message, AIMessage) and message.tool_calls: print(f"Tool calls: {message.tool_calls}") if isinstance(message, ToolMessage): print(f"Tool response: {message.content_blocks}") input_message = {"role": "user", "content": "What is the weather in Boston?"} for chunk in agent.stream( {"messages": [input_message]}, stream_mode=["messages", "updates"], version="v2", ): if chunk["type"] == "messages": token, metadata = chunk["data"] if isinstance(token, AIMessageChunk): _render_message_chunk(token) elif chunk["type"] == "updates": for source, update in chunk["data"].items(): if source in ("model", "tools"): _render_completed_message(update["messages"][-1]) ``` This example streams both partial tool call chunks and completed parsed tool calls with responses.
Streaming with human-in-the-loop interrupts code example
```python from langchain.agents import create_agent from langchain.agents.middleware import HumanInTheLoopMiddleware from langchain.messages import AIMessageChunk from langgraph.checkpoint.memory import InMemorySaver from langgraph.types import Command def get_weather(city: str) -> str: return f"It's always sunny in {city}!" checkpointer = InMemorySaver() agent = create_agent( "openai:gpt-5.5", tools=[get_weather], middleware=[ HumanInTheLoopMiddleware(interrupt_on={"get_weather": True}), ], checkpointer=checkpointer, ) input_message = {"role": "user", "content": "Can you look up the weather in Boston and San Francisco?"} config = {"configurable": {"thread_id": "some_id"}} interrupts = [] for chunk in agent.stream( {"messages": [input_message]}, config=config, stream_mode=["messages", "updates"], version="v2", ): if chunk["type"] == "updates": for source, update in chunk["data"].items(): if source == "__interrupt__": interrupts.extend(update) # Build decisions for each interrupt decisions = {} for interrupt in interrupts: decisions[interrupt.id] = {"decisions": [{"type": "approve"}] * len(interrupt.value["action_requests"])} # Resume with decisions for chunk in agent.stream( Command(resume=decisions), config=config, stream_mode=["messages", "updates"], version="v2", ): # Process resumed stream pass ``` This example shows collecting interrupts during streaming and resuming execution with approval decisions.
Streaming from sub-agents with name parameter code example
```python from langchain.agents import create_agent from langchain.chat_models import init_chat_model from langchain.messages import AIMessageChunk def get_weather(city: str) -> str: return f"It's always sunny in {city}!" weather_model = init_chat_model("openai:gpt-5.5") weather_agent = create_agent( model=weather_model, tools=[get_weather], name="weather_agent", ) def call_weather_agent(query: str) -> str: result = weather_agent.invoke({ "messages": [{"role": "user", "content": query}] }) return result["messages"][-1].text supervisor_model = init_chat_model("openai:gpt-5.5") agent = create_agent( model=supervisor_model, tools=[call_weather_agent], name="supervisor", ) input_message = {"role": "user", "content": "What is the weather in Boston?"} current_agent = None for chunk in agent.stream( {"messages": [input_message]}, stream_mode=["messages", "updates"], subgraphs=True, version="v2", ): if chunk["type"] == "messages": token, metadata = chunk["data"] if agent_name := metadata.get("lc_agent_name"): if agent_name != current_agent: print(f"🤖 {agent_name}: ") current_agent = agent_name if isinstance(token, AIMessageChunk): if token.text: print(token.text, end="|") ``` This example shows tracking which agent is emitting tokens using lc_agent_name metadata with subgraphs=True.
Streaming LLM tokens code example
```python from langchain.agents import create_agent def get_weather(city: str) -> str: return f"It's always sunny in {city}!" agent = create_agent( model="gpt-5-nano", tools=[get_weather], ) for chunk in agent.stream( {"messages": [{"role": "user", "content": "What is the weather in SF?"}]}, stream_mode="messages", version="v2", ): if chunk["type"] == "messages": token, metadata = chunk["data"] print(f"node: {metadata['langgraph_node']}") print(f"content: {token.content_blocks}") ``` This example streams LLM tokens as they are generated, printing the node name and content blocks for each token.
Stream modes: updates, messages, custom
LangChain supports three stream modes for agents: (1) 'updates' streams state updates after each agent step, emitting separate events if multiple nodes run in the same step; (2) 'messages' streams tuples of (token, metadata) from graph nodes where an LLM is invoked; (3) 'custom' streams custom data from inside graph nodes using the stream writer.
Agent progress streaming with stream_mode='updates'
To stream agent progress, use the stream() or astream() methods with stream_mode='updates'. This emits an event after every agent step. For an agent that calls a tool once, you will see updates for: LLM node (AIMessage with tool call requests), Tool node (ToolMessage with execution result), LLM node (final AI response).
Persist conversation history with thread_id and checkpointer
Pass a thread_id via config (Python) or configurable (JavaScript) so conversation is checkpointed and follow-up turns can resume the same history. thread_id is independent of stream_mode. Persisting conversation history requires the agent to be configured with a checkpointer. On LangSmith deployments a checkpointer is provisioned automatically. Locally, pass one explicitly, for example create_agent(..., checkpointer=InMemorySaver()).
Stream LLM tokens with stream_mode='messages'
To stream tokens as they are produced by the LLM, use stream_mode='messages'. This streams tuples of (token, metadata) from any graph nodes where an LLM is invoked. The output includes incremental message chunks generated by all LLM calls in the agent.
Custom stream updates with get_stream_writer()
To stream updates from tools as they are executed, use get_stream_writer() (Python) or config.writer (JavaScript). In Python, call writer = get_stream_writer() inside your tool to emit arbitrary data. In JavaScript, access config.writer as a parameter in your tool function. Use stream_mode='custom' when streaming to receive these updates.
Stream multiple modes simultaneously
Pass stream_mode as a list to specify multiple streaming modes, for example stream_mode=['updates', 'custom']. Each streamed chunk is a StreamPart dict with 'type', 'ns', and 'data' keys. Use chunk['type'] to determine the stream mode and chunk['data'] to access the payload.
Streaming reasoning tokens with standard content blocks
Some models perform internal reasoning before producing a final answer. Stream reasoning tokens by filtering standard content blocks for the type 'reasoning' when using stream_mode='messages'. LangChain normalizes provider-specific formats (Anthropic thinking blocks, OpenAI reasoning summaries, etc.) into a standard 'reasoning' content block type via the content_blocks property. Reasoning output must be enabled on the model.
Stream tool calls and parsed tool calls
Use stream_mode='messages' to stream partial JSON as tool calls are generated. To access completed, parsed tool calls: (1) if those messages are tracked in state (as in the model node of create_agent), use stream_mode=['messages', 'updates'] to access completed messages through state updates; (2) if those messages are not tracked in state, use custom updates or aggregate chunks during the streaming loop.
Aggregate message chunks to access completed messages
If completed messages are not reflected in state updates, you can aggregate message chunks in the streaming loop by adding chunks together: full_message = token if full_message is None else full_message + token. Check token.chunk_position == 'last' to detect when a full message is complete and access full_message.tool_calls for parsed tool calls.
Stream with human-in-the-loop interrupts
To handle human-in-the-loop interrupts during streaming: (1) configure the agent with HumanInTheLoopMiddleware and a checkpointer; (2) collect interrupts generated during stream_mode='updates' by checking if source == '__interrupt__'; (3) collect Interrupt objects and call _render_interrupt(interrupt) to display approval requests; (4) respond by passing a Command(resume=decisions) back into the streaming loop with decisions matching the order of collected interrupts.
Streaming from sub-agents with lc_agent_name metadata
When there are multiple LLMs in an agent, pass a name parameter to each agent when creating it (e.g., name='weather_agent'). This name is available in metadata via the lc_agent_name key when streaming in 'messages' mode. Set subgraphs=True when creating the stream to emit messages from inner agents. The name attached to an agent is also attached to any AIMessages generated by that agent.
Disable streaming for specific models
Set streaming=False (Python) or streaming: false (JavaScript) when initializing a model to disable streaming of individual tokens. This is useful for multi-agent systems to control which agents stream output, mixing models with different streaming support, or deploying to LangSmith to prevent certain model outputs from being streamed to the client. Not all chat model integrations support the streaming parameter; if unsupported, use disable_streaming=True (Python) or disableStreaming: true (JavaScript) instead.
v2 streaming format with unified output
Pass version='v2' to stream() or astream() to get a unified output format (requires LangGraph >= 1.1). Every chunk is a StreamPart dict with 'type', 'ns', and 'data' keys—the same shape regardless of stream mode or number of modes. This eliminates the need to unpack (mode, data) tuples like in v1. The v2 format also improves invoke()—it returns a GraphOutput object with .value (state) and .interrupts (tuple of Interrupt objects) attributes.
Event streaming API in LangChain v1.3
For new applications, event streaming (the typed-projection API introduced in LangChain v1.3) is recommended over the stream_mode branching approach. Event streaming gives you separate iterators per projection (messages, values, tool calls, subgraphs) so you can consume them independently instead of branching on stream_mode chunks.
Streaming tool calls with guardrail middleware example
Completed messages within middleware (like guardrails) can be streamed by using get_stream_writer() to emit the completed message to the stream. This allows access to completed messages during streaming even when they are not in state updates. The example shows a safety_guardrail using model-based evaluation with stream_writer(result) to emit the evaluation AIMessage.
Streaming custom updates code example
```python from langchain.agents import create_agent from langgraph.config import get_stream_writer def get_weather(city: str) -> str: writer = get_stream_writer() writer(f"Looking up data for city: {city}") writer(f"Acquired data for city: {city}") return f"It's always sunny in {city}!" agent = create_agent( model="claude-sonnet-4-6", tools=[get_weather], ) for chunk in agent.stream( {"messages": [{"role": "user", "content": "What is the weather in SF?"}]}, stream_mode="custom", version="v2", ): if chunk["type"] == "custom": print(chunk["data"]) ``` This example emits custom string updates during tool execution using get_stream_writer().
Streaming multiple modes code example
```python from langchain.agents import create_agent from langgraph.config import get_stream_writer def get_weather(city: str) -> str: writer = get_stream_writer() writer(f"Looking up data for city: {city}") writer(f"Acquired data for city: {city}") return f"It's always sunny in {city}!" agent = create_agent(model="gpt-5-nano", tools=[get_weather]) for chunk in agent.stream( {"messages": [{"role": "user", "content": "What is the weather in SF?"}]}, stream_mode=["updates", "custom"], version="v2", ): print(f"stream_mode: {chunk['type']}") print(f"content: {chunk['data']}") ``` This example streams both state updates and custom events, using chunk['type'] to distinguish between them.
Tool stream writer for real-time updates
Use runtime.stream_writer to emit custom updates during tool execution. This is useful for providing progress feedback to users during long-running operations. The tool must be invoked within a LangGraph execution context.
Voice agent streaming pipeline pattern
The demo implements a streaming pipeline where each stage processes data asynchronously. Each stage processes events independently and concurrently: audio transcription begins as soon as audio arrives, the agent starts reasoning as soon as a transcript is available, and speech synthesis begins as soon as agent text is generated. This enables sub-700ms latency to support natural conversation.
Producer-consumer pattern in STT stage
The STT stage uses a producer-consumer pattern where audio chunks are sent to the STT service concurrently with receiving transcript events. This allows transcription to begin before all audio has arrived.
Python STT stream implementation with AssemblyAI
The stt_stream async generator function transforms audio bytes into VoiceAgentEvent objects using a producer-consumer pattern. The producer task sends audio chunks to AssemblyAI and signals completion by calling await stt.close(). The consumer task receives and yields transcription events. Both run concurrently with asyncio.create_task() and cleanup is handled in the finally block.
TypeScript STT stream implementation with AssemblyAI
The sttStream async generator function transforms audio (Uint8Array) into VoiceAgentEvent objects using a producer-consumer pattern. The producer sends audio chunks to AssemblyAI in a background async IIFE. The consumer receives transcription events. Both run concurrently and the function yields events from passthrough iterator.
LangChain agent stream implementation with memory
The agent_stream async generator function processes STT output events through a LangChain agent. A unique thread_id (created with uuid7()) is generated for conversation memory. The function passes through all upstream events unchanged. When an stt_output event arrives, it sends a HumanMessage to the agent using astream_events with version="v3" and thread_id in configurable dict. Agent response tokens are yielded as AgentChunkEvent objects.
Agent streaming with stream_events v3
Agent responses are streamed using agent.astream_events() with version="v3". The method takes input dict with messages list, configurable dict with thread_id, and version="v3" parameter. Streaming yields message objects with text attribute that can be iterated over token by token.
Producer-consumer pattern in TTS stage
The TTS stage uses concurrent processing by merging two async streams: upstream processing (passes through all events and sends agent text chunks to TTS provider) and audio reception (receives synthesized audio chunks from TTS provider). Both streams run concurrently.
Python TTS stream implementation with Cartesia
The tts_stream async generator function merges two concurrent async streams. The process_upstream() coroutine iterates over event_stream, yields all events, and sends agent_chunk text to TTS via await tts.send_text(). The tts.receive_events() coroutine yields audio chunks. Both streams run concurrently via merge_async_iters() and cleanup is handled in finally block.
TypeScript TTS stream implementation with Cartesia
The ttsStream async generator function uses producer-consumer pattern. The producer reads events from eventStream, pushes them to passthrough iterator, and sends agent_chunk text to TTS. The consumer receives audio from TTS and pushes events. Both run as background async IIFEs, and the function yields from passthrough iterator.
Streaming TTS behavior
Some TTS providers like Cartesia begin synthesizing audio as soon as they receive text, enabling audio playback to start before the agent finishes generating its complete response. This reduces overall latency.
Complete voice pipeline with RunnableGenerator
The complete voice pipeline chains three stages using Python's RunnableGenerator: RunnableGenerator(stt_stream) for audio to STT events, piped to RunnableGenerator(agent_stream) for STT events to agent events, piped to RunnableGenerator(tts_stream) for agent events to TTS audio. The pipeline.atransform() method processes the WebSocket audio stream and TTS audio chunks are sent back to the client.
Voice pipeline composition in TypeScript
Voice pipeline stages are chained by passing output of one async generator as input to the next: transcriptEventStream = sttStream(inputStream), agentEventStream = agentStream(transcriptEventStream), outputEventStream = ttsStream(agentEventStream). Iterate over outputEventStream and send tts_chunk audio events to client.
TypeScript agent stream implementation
In TypeScript, voice agents use createAgent() with model, tools array, checkpointer (new MemorySaver()), and systemPrompt. Tools are created with tool() function from @langchain/core/tools with name, description, and schema. The agentStream() async generator creates unique threadId, passes through upstream events, and calls agent.streamEvents() with messages list and configurable.thread_id when stt_output event arrives.
Stream events from supervisor with interleave
Supervisor agents stream events using stream_events method with version='v3'. The stream can be consumed using interleave method to separate different event types. For example, stream.interleave('messages', 'tool_calls') yields tuples of (kind, item) where kind is 'messages' or 'tool_calls'. Messages have .text attribute that can be printed token-by-token. Tool calls have .tool_name and .input attributes.