Agentic AI using LangGraph

Chapter: Streaming in LangGraph

In LLM-based applications, presenting long responses all at once introduces significant user latency, bad UX, and unnecessary token consumption. Streaming addresses this issue by delivering output tokens incrementally as they are generated by the model, creating a dynamic "typewriter" effect. This chapter covers the foundational mechanics of streaming, Python generator objects, LangGraph's .stream() methods, and front-end integration using Streamlit's st.write_stream.

1. Context & The Problem Statement: The "Wall of Text" Lag

When building conversational AI interfaces without streaming, the application follows a synchronous request-response cycle:

The Naive Approach Problems

1. High Perceived Latency (Time-To-First-Token / TTFT Delay): For large text generation tasks (e.g., requesting a 500-word blog or complex code output), the user is left staring at an unresponsive screen for 5 to 10 seconds. Non-technical users often perceive this delay as an application freeze or crash, causing high user drop-off rates.
2. Poor Readability: Receiving 500–1,000 words in a single instantaneous layout shift forces the user to digest a massive "wall of text" abruptly, making content scanning and comprehension difficult.
3. Inefficient Token Spending: If an LLM generates an unhelpful or off-topic answer, the user cannot halt the response mid-flight in a non-streamed setup. Every token generated by the provider is charged, leading to wasted API credits.

2. Core Mental Model: What is Streaming?

Definition: In LLM systems, streaming is the mechanism where the model emits individual generated tokens (word fragments/characters) over an open connection as soon as they are computed, rather than buffering the entire output until completion.

Real-Life Analogy

Imagine placing an order at a multi-course restaurant:

• Without Streaming (Buffered): The kitchen prepares your appetizer, main course, side dishes, and dessert, holding all of them until every item is ready. You sit empty-handed at your table for two hours before receiving all dishes at once.
• With Streaming: As soon as the appetizer is prepared, the waiter brings it to your table immediately while the kitchen continues cooking the main course. You begin eating right away, experiencing zero unproductive waiting time.

3. Key Advantages of Token Streaming

Implementing streaming transforms the user experience across six distinct operational dimensions:

4. Technical Foundation: Python Generators vs. Standard Functions

To stream tokens in Python without loading everything into memory at once, LangGraph relies on Generators and the yield keyword rather than standard return functions.

Functions vs. Generators

• Standard Function (return): Executes completely, computes all data, stores it in memory, and returns the final value to the caller, terminating execution.
• Generator Function (yield): Computes one item at a time, yields execution control (and the computed value) back to the caller, pauses its state, and resumes execution when the next item is requested.

# Standard Return Pattern (Buffered)
def generate_response_buffered():
    tokens = ["What ", "is ", "the ", "recipe ", "for ", "pasta?"]
    return "".join(tokens)  # Evaluates completely before returning

# Generator Yield Pattern (Streamed)
def generate_response_streamed():
    tokens = ["What ", "is ", "the ", "recipe ", "for ", "pasta?"]
    for token in tokens:
        yield token  # Returns control on each emitted token on-the-fly

Mathematical Time to First Token (TTFT) Comparison

Let the generation of a token $t_i$ take duration $d(t_i)$, and total tokens generated be $N$.

• Time to First Token (Buffered):

• Time to First Token (Streamed):

Since $d(t_1) \ll \sum_{i=1}^{N} d(t_i)$, the perceived initial latency is reduced by an order of magnitude.

5. LangGraph Execution Paradigm: .invoke() vs. .stream()

In LangGraph, compiled graphs provide two primary execution methods:

1. The Naive Approach: .invoke()

Returns the final updated state dictionary after all node executions in the graph finish.

# Synchronous, non-streaming graph invocation
final_state = chatbot.invoke(
    {"messages": [HumanMessage(content="Write a 500-word blog on cricket.")]},
    config={"configurable": {"thread_id": "1"}}
)
print(final_state["messages"][-1].content)

2. The Streaming Approach: .stream()

Returns an iterable Python Generator emitting graph artifacts during execution.

# Streamed graph invocation returning a generator object
stream_generator = chatbot.stream(
    {"messages": [HumanMessage(content="Write a 500-word blog on cricket.")]},
    config={"configurable": {"thread_id": "1"}},
    stream_mode="messages"
)

6. Understanding LangGraph Stream Modes

LangGraph provides multiple streaming modes via the stream_mode parameter to support different application needs:

Note: For building streaming chatbots, stream_mode="messages" is required to stream individual LLM tokens.

7. Backend Implementation Walkthrough

Here is how to modify a pure Python/LangGraph backend script to process streamed message chunks:

from langchain_core.messages import HumanMessage
from my_graph_module import chatbot  # Pre-compiled state graph

# 1. Define input parameters
initial_state = {"messages": [HumanMessage(content="What is the recipe to make pasta?")]}
config = {"configurable": {"thread_id": "session_001"}}

# 2. Invoke graph in streamed mode using stream_mode="messages"
stream_generator = chatbot.stream(
    initial_state,
    config=config,
    stream_mode="messages"
)

# 3. Iterate over generator yielding (message_chunk, metadata)
for message_chunk, metadata in stream_generator:
    # Filter and print content chunks as they arrive
    if message_chunk.content:
        print(message_chunk.content, end="", flush=True)

How the Generator Iteration Works

8. Frontend Integration with Streamlit (st.write_stream)

Streamlit natively simplifies generator streaming through its utility function: st.write_stream().

st.write_stream() accepts a Python generator or iterator, executes the typewriter UI effect automatically, and returns the combined full final string output once generation ends.

Updated Front-End Application (frontend.py)

import streamlit as st
from langchain_core.messages import HumanMessage
from backend import chatbot  # Compiled LangGraph instance

st.title("Streamed Agentic Chatbot")

# Initialize session state message history
if "messages" not in st.messages:
    st.session_state["messages"] = []

# Display prior chat history
for msg in st.session_state["messages"]:
    with st.chat_message(msg["role"]):
        st.markdown(msg["content"])

# Process new user input
if user_input := st.chat_input("Type your message here..."):
    # Render user prompt
    st.session_state["messages"].append({"role": "user", "content": user_input})
    with st.chat_message("user"):
        st.markdown(user_input)

    # Render AI assistant response via streaming
    with st.chat_message("assistant"):
        # Define thread configuration
        config = {"configurable": {"thread_id": "default_user"}}
        
        # Generator expression yielding content tokens from stream chunks
        generator = (
            message_chunk.content 
            for message_chunk, metadata in chatbot.stream(
                {"messages": [HumanMessage(content=user_input)]},
                config=config,
                stream_mode="messages"
            )
            if message_chunk.content
        )
        
        # st.write_stream handles UI animation and returns full string when finished
        ai_message_text = st.write_stream(generator)
    
    # Store complete AI response in session memory
    st.session_state["messages"].append({"role": "assistant", "content": ai_message_text})

9. Common Implementation Pitfalls & Debugging Workflows

The Hardcoded Input Bug

A common bug when moving from .invoke() to .stream() is hardcoding the initial prompt inside the stream configuration:

# ❌ BUG: Hardcoded payload inside stream call
generator = (
    message_chunk.content 
    for message_chunk, metadata in chatbot.stream(
        {"messages": [HumanMessage(content="What is the recipe to make pasta?")]}, # Hardcoded!
        config=config,
        stream_mode="messages"
    )
)

Symptom

Regardless of what prompt the user types into st.chat_input(), the application consistently streams the pasta recipe response.

Fix

Ensure the payload dynamically maps to the variable containing user_input:

# ✅ CORRECT: Pass user_input dynamically
{"messages": [HumanMessage(content=user_input)]}

10. Complete End-to-End System Workflow

The following execution map displays the path from the user's interface input to backend graph iteration and back to the front-end rendering engine:

Summary and Key Takeaways

• The Problem: Buffering large LLM responses creates high latency ($5\text{s} - 10\text{s}$ blank screens), poor text readability, higher user drop-off, and uncancelable token spending.
• The Solution: Streaming returns tokens incrementally as they are calculated by the LLM, giving users immediate feedback via a typewriter effect.
• Core Python Primitive: Streaming relies on Generators (yield) rather than traditional functions (return) to process data incrementally on the fly.
• LangGraph Integration: Replace graph.invoke() with graph.stream() using stream_mode="messages" to receive token chunks.
• Streamlit Integration: Use st.write_stream() to handle front-end token rendering automatically while returning the final string to persist in message state.
• Key Benefit for Agents: Beyond chat text, streaming enables step-by-step UI updates during multi-step tool calls and agentic reasoning workflows.