07 Oct LangGraph Streaming: Stream Responses and Events

So far, invoke() made you wait until the whole graph finished. Streaming lets your app show results as they happen: progress updates, partial state, or the model’s answer word by word. It makes apps feel much faster.

The method: stream()

You have used stream() since Chapter 4. The stream_mode argument decides what you receive:

• updates: only what each node changed. Best for progress and debugging.
• values: the full state after each step. Best for showing the whole picture.
• messages: the model’s output token by token. Best for a chat typing effect.
• custom: data you send yourself. Best for progress bars and status text.

Setup: a small graph to compare modes

We will reuse the text analyzer from Chapter 4:

from typing import TypedDict
from langgraph.graph import StateGraph, START, END

class State(TypedDict):
    text: str
    word_count: int
    level: str

def count_words(state: State):
    return {"word_count": len(state["text"].split())}

def label_length(state: State):
    return {"level": "short" if state["word_count"] < 10 else "long"}

builder = StateGraph(State)
builder.add_node("count_words", count_words)
builder.add_node("label_length", label_length)
builder.add_edge(START, "count_words")
builder.add_edge("count_words", "label_length")
builder.add_edge("label_length", END)
graph = builder.compile()

inputs = {"text": "LangGraph helps you build AI apps as graphs"}

Mode 1: updates

for chunk in graph.stream(inputs, stream_mode="updates"):
    print(chunk)

Output:

{'count_words': {'word_count': 8}}
{'label_length': {'level': 'short'}}

Each item names the node and shows only its changes.

Mode 2: values

for chunk in graph.stream(inputs, stream_mode="values"):
    print(chunk)

Output:

{'text': 'LangGraph helps you build AI apps as graphs'}
{'text': 'LangGraph helps you build AI apps as graphs', 'word_count': 8}
{'text': 'LangGraph helps you build AI apps as graphs', 'word_count': 8, 'level': 'short'}

Here, each item is the complete state after a step. The first item is the starting input, and the last one equals the invoke() result.

Mode 3: messages (token streaming)

This mode gives the chat “typing” effect. It streams the model’s output as it is generated, even though the LLM call sits inside a node.

from dotenv import load_dotenv
from langchain_anthropic import ChatAnthropic
from langgraph.graph import StateGraph, MessagesState, START, END

load_dotenv()
llm = ChatAnthropic(model="claude-sonnet-5-5")

def chatbot(state: MessagesState):
    return {"messages": [llm.invoke(state["messages"])]}

builder = StateGraph(MessagesState)
builder.add_node("chatbot", chatbot)
builder.add_edge(START, "chatbot")
builder.add_edge("chatbot", END)
chat = builder.compile()

def as_text(content):
    # Some providers return a string, others a list of content blocks
    if isinstance(content, str):
        return content
    return "".join(b.get("text", "") for b in content if isinstance(b, dict))

for chunk, metadata in chat.stream(
    {"messages": [("human", "Explain a Python list in two sentences.")]},
    stream_mode="messages",
):
    print(as_text(chunk.content), end="", flush=True)
print()

Sample output (appears word by word in your terminal):

A Python list is an ordered, changeable collection of items written inside square brackets. You can store any mix of data types in it and access items by their position.

How it works:

• In messages mode, each item is a tuple: a message chunk and a metadata dictionary
• chunk.content holds a small piece of text, and end=”” with flush=True prints the pieces on one line as they arrive
• as_text() protects you from provider differences in the shape of content
• The metadata tells you where the chunk came from. For example, metadata[“langgraph_node”] is the node name (“chatbot”), so if a graph has several LLM nodes, you can show only the ones you want:

if metadata["langgraph_node"] == "chatbot":
    print(as_text(chunk.content), end="", flush=True)

Even though the node calls llm.invoke() (not a streaming call), LangGraph still delivers tokens as they arrive.

Mode 4: custom (your own progress messages)

Sometimes you want to report progress from inside a node, like “Step 2 of 4” during a long task. Use get_stream_writer():

import time
from typing import TypedDict
from langgraph.config import get_stream_writer
from langgraph.graph import StateGraph, START, END

class State(TypedDict):
    result: str

def long_task(state: State):
    writer = get_stream_writer()
    for i in range(1, 4):
        time.sleep(0.5)                       # pretend to work
        writer({"progress": f"Step {i} of 3"})
    return {"result": "done"}

builder = StateGraph(State)
builder.add_node("long_task", long_task)
builder.add_edge(START, "long_task")
builder.add_edge("long_task", END)
graph = builder.compile()

for chunk in graph.stream({"result": ""}, stream_mode="custom"):
    print(chunk)

Output (one line every half second):

{'progress': 'Step 1 of 3'}
{'progress': 'Step 2 of 3'}
{'progress': 'Step 3 of 3'}

Whatever you pass to writer() is sent straight to whoever is streaming. Use it for progress bars, status text, or partial results.

Combining modes

Pass a list to receive several kinds of output at once. Each item then arrives as a (mode, data) tuple:

for mode, data in graph.stream(inputs, stream_mode=["updates", "values"]):
    print(mode, "->", data)

This is useful in real apps, where you might show tokens with messages and progress with custom in the same run.

Streaming with agents

Everything works the same on the agent from Chapters 11 and 12. With stream_mode=”updates”, you would see the agent node, then tools, then agent again. With messages, you would see the final answer stream in, and the tool-call chunks appear as well. Filter by langgraph_node or by chunk type if you only want the final text.

Async streaming

Web servers (like FastAPI) use async code. The async version is astream():

import asyncio

async def main():
    async for chunk in graph.astream(inputs, stream_mode="updates"):
        print(chunk)

asyncio.run(main())

It takes the same stream_mode options. Chapter 23 uses it to stream to a browser.

Which mode should you pick?

• Debug which node did what: updates
• Show the full state changing: values
• Build a chat with a typing effect: messages
• Show a progress bar or status: custom
• Several of the above: a list of modes

Common mistakes

• Expecting stream() to return one result: it returns an iterator, so loop over it
• Forgetting flush=True when printing tokens: text may appear all at once at the end
• Not unpacking messages items: each one is a (chunk, metadata) tuple
• Using get_stream_writer() outside a node: it only works while the graph is running
• Reading chunk.content as always a string: some providers return blocks, so use a helper like as_text()
• Using stream() in async code: use astream() there
• Forgetting the custom mode when using writer(): without stream_mode=”custom” (or including it in a list), you won’t receive the data

Summary

• stream() gives results as the graph runs, instead of waiting for the end
• Four main modes: updates, values, messages, and custom
• messages mode streams model tokens for a chat typing effect, and metadata tells you which node produced them
• get_stream_writer() lets a node send its own progress messages
• Pass a list of modes to combine them, and use astream() for async apps

Practice

1. Run the analyzer in all three non-LLM modes and describe the difference in one sentence each
2. Add a time.sleep(1) to count_words and watch updates arrive one at a time
3. Stream your Chapter 9 chatbot and print only chunks from the chatbot node
4. Add a get_stream_writer() progress message to a node and stream with stream_mode=[“custom”, “updates”]

---

If you liked the tutorial, spread the word and share the link and our website, Studyopedia, with others.

---

For Videos, Join Our YouTube Channel: Join Now

---

Read More:

• Generative AI Tutorial
• AI Ethics
• Machine Learning Tutorial
• Deep Learning Tutorial
• Ollama Tutorial
• Retrieval Augmented Generation (RAG) Tutorial
• ChatGPT Tutorial
• Microsoft Copilot Tutorial