How Streaming Makes AI Responses Visible in Real Time with LangGraph

Why Streaming Is Needed

Traditional LLM calls wait several seconds before returning a complete reply, causing poor user experience. Streaming output displays each generated token immediately, reducing perceived latency from about 3 seconds to 0.5 seconds, creating a typewriter effect and making the model’s reasoning visible.

Streaming Modes in LangGraph

values

– emits the full state after each step; useful for debugging and full‑state inspection. updates – emits node‑level output only; focuses on the result of individual nodes. messages – emits LLM token‑level increments; enables a typewriter‑style front‑end display. debug – emits debugging events such as execution order. custom – emits user‑defined data pushed from inside a node via write().

Multiple modes can be combined, e.g. stream_mode=["values", "updates"].

Minimal Code Example

from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver

class ChatState(TypedDict):
    messages: list

def chatbot(state: ChatState) -> dict:
    return {"messages": [{"role": "assistant", "content": "你好！"}]}

builder = StateGraph(ChatState)
builder.add_node("chatbot", chatbot)
builder.add_edge(START, "chatbot")
builder.add_edge("chatbot", END)
# Streaming itself does not depend on a checkpointer; a checkpointer is needed only for multi‑turn or resumable chats
app = builder.compile(checkpointer=InMemorySaver())

Mode Comparison

Mode 1: values (full state)

config = {"configurable": {"thread_id": "stream-demo"}}
for chunk in app.stream({"messages": [{"role": "user", "content": "你好"}]}, config, stream_mode="values"):
    print(chunk)

Output:

{'messages': [{'role': 'user', 'content': '你好'}]}
{'messages': [{'role': 'assistant', 'content': '你好！'}]}

Mode 2: updates (node updates)

for chunk in app.stream(input_data, config, stream_mode="updates"):
    print(f"节点: {list(chunk.keys())}")

Output:

节点: ['chatbot']

Mode 3: Combined output

for chunk in app.stream(input_data, config, stream_mode=["values", "updates"]):
    print(chunk)

Asynchronous Streaming

async for chunk in app.astream(input_data, config, stream_mode="updates"):
    print(chunk)

FastAPI integration:

from fastapi import FastAPI
from fastapi.responses import StreamingResponse

app = FastAPI()

@app.get("/chat/stream")
async def stream_chat(message: str):
    async def event_stream():
        async for chunk in app.astream({"messages": [{"role": "user", "content": message}]},
                                   {"configurable": {"thread_id": "chat-1"}},
                                   stream_mode="updates"):
            yield f"data: {chunk}

"
    return StreamingResponse(event_stream(), media_type="text/event-stream")

Streaming Chatbot Flow

┌─────────────────────────────────────┐
│           Streaming Chat Flow       │
│  ┌─────────┐   ┌─────────┐          │
│  │  User   │──▶│ Router  │          │
│  └─────────┘   └────┬────┘          │
│                │                 │
│          ┌─────▼─────┐           │
│          │ Chat Node │           │
│          │ (stream) │           │
│          └─────┬─────┘           │
│                │                 │
│                ▼                 │
│          Return to User           │
└─────────────────────────────────────┘

Key Points Recap

Stream modes: values, updates, messages, debug, custom. values streams the complete state at each step. updates streams node‑level output. messages streams LLM token increments for a typewriter effect. astream provides asynchronous streaming.

Related Links

Official documentation: https://langchain-ai.github.io/langgraph/concepts/streaming/

GitHub repository: https://github.com/langchain-ai/langgraph