LangChain

LangChain Streaming: Returning Tokens as They're Generated

Pramod Dutta · Jun 27, 2026 · 14 min read

Why Your Chatbot Feels Slow Even When The Model Is Fast

You've built a chain, wired it to GPT-4 or Claude, and it works. Then you deploy it and someone tries it live. They type a question, hit enter, and then... nothing. Three seconds. Five seconds. The cursor blinks. Finally the entire answer appears at once, dumped on screen like a wall of text nobody asked to read all at once.

This is the single most common reason AI product demos feel unpolished. It's not the model's fault — LLMs generate text one token at a time, and that generation is happening the whole time you're waiting. The problem is that your code is waiting for the *entire* response to finish before showing the user anything. Every ChatGPT-style interface you've ever used that feels "instant" is not actually faster at generating text — it's just showing you tokens as they arrive instead of making you wait for the full buffer.

LangChain gives you this exact capability through streaming, and the good news is it's not an advanced, bolt-on feature you have to fight for. Streaming is built into the Runnable interface that every LangChain component implements — LLMs, chat models, chains, agents, even retrievers. If your chain is composed with LangChain Expression Language (LCEL), you already have .stream() and .astream() available on it, often with zero code changes to the chain itself.

In this article we'll go from the simplest possible streaming call to production patterns: streaming through multi-step chains, handling structured output while still streaming, async streaming for concurrent requests, and wiring token-by-token output into a FastAPI backend with Server-Sent Events. By the end you'll know exactly which method to reach for and why your current implementation might be silently buffering the whole response anyway.

The Core Idea: Tokens Are Chunks, Not Strings

Before touching code, it helps to be precise about what "streaming" actually returns. When you call .invoke() on a chat model, you get back a single AIMessage object containing the full response. When you call .stream() instead, you get back an iterator of `AIMessageChunk` objects — small pieces of the message that arrive incrementally and are designed to be concatenated together.

This distinction matters because chunks aren't just substrings. They support the + operator so you can accumulate them properly, and they carry partial metadata (like token usage that only becomes accurate on the final chunk). Here's the simplest possible example:

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o", temperature=0)

for chunk in llm.stream("Explain what a closure is in JavaScript, in 3 sentences."):
    print(chunk.content, end="", flush=True)

Run this and you'll see text appear word by word (sometimes a few characters at a time) instead of waiting for the full answer. Compare that to:

response = llm.invoke("Explain what a closure is in JavaScript, in 3 sentences.")
print(response.content)

Same prompt, same model, but .invoke() blocks until generation is complete. The wall-clock time to see the *first* character is dramatically different even though the wall-clock time to see the *last* character is roughly the same. That first-token latency is what your users actually perceive as speed.

Streaming a Full Chain, Not Just the Model

Real applications rarely call the model directly — you're almost always composing a prompt template, the model, and an output parser into a chain. The good news is that LCEL chains propagate streaming through every step, as long as every step in the chain supports it.

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI

prompt = ChatPromptTemplate.from_template(
    "You are a patient tutor. Explain {topic} to a beginner in simple terms."
)
llm = ChatOpenAI(model="gpt-4o", temperature=0.3)
parser = StrOutputParser()

chain = prompt | llm | parser

for chunk in chain.stream({"topic": "recursion"}):
    print(chunk, end="", flush=True)

Notice that parser here is StrOutputParser, which unwraps AIMessageChunk objects into plain string chunks as they pass through. This is important: not every output parser streams cleanly. A parser that needs the *entire* output before it can produce a result — like one that parses a complete JSON blob — will necessarily buffer everything internally and only yield at the end, even though you called .stream(). LangChain will not error out on this; it will just silently behave like .invoke() from the caller's perspective. If you add a step to your chain and streaming suddenly "stops working," that step is almost always the culprit.

A quick way to sanity-check which step broke streaming is to bisect the chain — stream the prompt-and-model portion alone, then add the parser back:

partial_chain = prompt | llm
for chunk in partial_chain.stream({"topic": "recursion"}):
    print(chunk.content, end="", flush=True)

If this streams token by token but the full chain with the parser doesn't, you've isolated the problem to the parser.

Async Streaming With `.astream()`

Synchronous .stream() is fine for a script or a notebook, but production backends are almost always async — you're serving many concurrent users, and you don't want one slow LLM call blocking your event loop. LangChain mirrors every sync streaming method with an async equivalent using .astream():

import asyncio
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o", temperature=0)

async def main():
    async for chunk in llm.astream("Give me 3 tips for writing clean Python code."):
        print(chunk.content, end="", flush=True)

asyncio.run(main())

The pattern is identical to the sync version — for becomes async for, .stream() becomes .astream() — but the underlying behavior is what matters: .astream() yields control back to the event loop between chunks, so other coroutines (other users' requests, other background tasks) can make progress while this one is waiting on network I/O from the model provider.

This also composes with full chains exactly like before:

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI

prompt = ChatPromptTemplate.from_template("Summarize this in one paragraph:\n\n{text}")
llm = ChatOpenAI(model="gpt-4o", temperature=0)
chain = prompt | llm | StrOutputParser()

async def summarize(text: str):
    async for chunk in chain.astream({"text": text}):
        print(chunk, end="", flush=True)

A common mistake here is mixing sync and async carelessly — calling .stream() inside an async def function works because it's just a regular generator, but it will block the event loop for the duration of each chunk's network wait. If your app is async end-to-end (FastAPI, for example), use .astream() consistently so you actually get the concurrency benefit you're paying the async complexity tax for.

Streaming Events, Not Just Text: `.astream_events()`

.stream() and .astream() give you the final output chunks of whatever is at the *end* of your chain. But what if you want visibility into what's happening *inside* the chain — which tool an agent decided to call, what a retriever returned, when a particular sub-chain started and finished? For that, LangChain exposes .astream_events(), which yields a stream of structured event dictionaries describing every step of execution, not just the final tokens.

import asyncio
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI

prompt = ChatPromptTemplate.from_template("Write a haiku about {subject}.")
llm = ChatOpenAI(model="gpt-4o", temperature=0.7)
chain = prompt | llm | StrOutputParser()

async def main():
    async for event in chain.astream_events({"subject": "autumn rain"}, version="v2"):
        kind = event["event"]
        if kind == "on_chat_model_stream":
            content = event["data"]["chunk"].content
            if content:
                print(content, end="", flush=True)
        elif kind == "on_chain_start":
            print(f"\n[started: {event['name']}]")
        elif kind == "on_chain_end":
            print(f"\n[finished: {event['name']}]")

asyncio.run(main())

This is the tool you want when you're streaming an agent's behavior to a frontend and want to show the user "Searching the web..." or "Calling calculator tool..." status updates in addition to the final generated text. Plain .astream() on an agent will often only give you the final answer chunks (or, depending on the agent type, intermediate messages that are hard to disambiguate). .astream_events() gives you the event type explicitly, so your frontend logic becomes a simple dispatch on event["event"] rather than guessing at message shapes.

One gotcha worth flagging: the version="v2" argument is not optional decoration — event schemas changed between v1 and v2, and omitting it (or using an old LangChain version that defaults to v1) will give you a deprecation warning and a different event shape. Pin it explicitly so your event-handling code doesn't break silently on a library upgrade.

Streaming Structured Output (Without Losing the Streaming)

A frequent point of confusion: if you use with_structured_output() to force a model to return a Pydantic model or JSON schema, can you still stream it? Partially, yes — LangChain supports streaming *partial* structured objects as they're being built, which is genuinely useful for progressively rendering a form or a structured card in the UI as fields fill in.

from pydantic import BaseModel, Field
from langchain_openai import ChatOpenAI

class MovieReview(BaseModel):
    title: str = Field(description="Movie title")
    rating: int = Field(description="Rating out of 10")
    summary: str = Field(description="One paragraph summary")

llm = ChatOpenAI(model="gpt-4o", temperature=0)
structured_llm = llm.with_structured_output(MovieReview)

for chunk in structured_llm.stream("Review the movie Inception."):
    print(chunk)

Depending on the underlying method LangChain uses (function calling vs JSON mode), what you get back per chunk may be a progressively more complete dictionary or partial Pydantic object rather than clean incremental deltas — the API surface is consistent, but under the hood the model provider is still deciding how granularly it emits function-call arguments. Treat each chunk as "the best current guess at the full structured object," and simply re-render your UI with the latest chunk each time rather than trying to diff between them.

If you need character-by-character streaming of a *specific field* (say, a long summary field inside a structured object), that's a case where plain string streaming through a prompt asking for a fenced text block, followed by a manual parse, is often more predictable than structured-output streaming — worth knowing before you spend an afternoon fighting partial JSON.

Wiring Streaming Into a FastAPI Backend

None of this matters to your end users unless the tokens make it across the network to a browser. The standard pattern is Server-Sent Events (SSE): the backend keeps an HTTP connection open and pushes chunks as they're generated, and the frontend uses EventSource or a fetch with a readable stream to render them as they arrive.

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI

app = FastAPI()

prompt = ChatPromptTemplate.from_template("Answer clearly and concisely: {question}")
llm = ChatOpenAI(model="gpt-4o", temperature=0)
chain = prompt | llm | StrOutputParser()

async def token_generator(question: str):
    async for chunk in chain.astream({"question": question}):
        yield f"data: {chunk}\n\n"
    yield "data: [DONE]\n\n"

@app.get("/chat/stream")
async def chat_stream(question: str):
    return StreamingResponse(
        token_generator(question),
        media_type="text/event-stream",
    )

A few details matter here that are easy to get wrong on the first pass:

The media_type must be text/event-stream for browsers to treat it as SSE and process it incrementally rather than waiting for the connection to close.
Each SSE message needs to be formatted as data: <payload>\n\n — the double newline is the field separator SSE clients expect, not a stylistic choice.
Sending an explicit [DONE] sentinel lets the frontend know to close its EventSource connection and stop the "thinking" indicator, rather than relying on the HTTP connection closing (which can be ambiguous with certain proxies and load balancers).
If you deploy behind Nginx or a reverse proxy, response buffering is often enabled by default and will silently un-stream your beautifully streamed backend. You'll need proxy_buffering off (or the equivalent on your platform) or you'll be right back to waiting for the full response before anything appears.

On the frontend side, a minimal vanilla JS consumer looks like this:

const evtSource = new EventSource(`/chat/stream?question=${encodeURIComponent(userInput)}`);

evtSource.onmessage = (event) => {
  if (event.data === "[DONE]") {
    evtSource.close();
    return;
  }
  responseDiv.textContent += event.data;
};

That's the entire client-side logic required to get a token-by-token typing effect — no special library needed for the basic case, though most production chat UIs eventually reach for something like the Vercel AI SDK to handle reconnection, backpressure, and markdown rendering as tokens arrive.

Callbacks: The Older, Lower-Level Streaming Mechanism

Before LCEL's .stream() existed, LangChain streaming was done through callback handlers — subclassing BaseCallbackHandler and overriding on_llm_new_token. You'll still see this pattern in older tutorials and some codebases, and it's worth recognizing even if you don't reach for it by default anymore:

from langchain_core.callbacks import BaseCallbackHandler
from langchain_openai import ChatOpenAI

class TokenPrinter(BaseCallbackHandler):
    def on_llm_new_token(self, token: str, **kwargs) -> None:
        print(token, end="", flush=True)

llm = ChatOpenAI(model="gpt-4o", streaming=True, callbacks=[TokenPrinter()])
llm.invoke("Tell me a short joke about compilers.")

Two things to note if you encounter this pattern in the wild. First, streaming=True has to be set explicitly on the model constructor for callbacks to fire per-token — .stream() and .astream() handle this for you automatically, which is one reason they're the recommended default now. Second, callbacks are push-based (the framework calls your handler) rather than pull-based (you iterate a generator), which makes them awkward to compose with async for loops and modern async frameworks. Unless you're maintaining legacy code or need callback-specific features like tracing hooks into LangSmith, prefer .astream() for new code — it's simpler to reason about and easier to test.

Debugging: Why Your Stream Isn't Actually Streaming

If you've wired up .stream() correctly and you're still seeing the full response appear all at once, the cause is almost always one of these:

An output parser or chain step that buffers. As mentioned earlier, any component that needs the complete output before producing a result (certain JSON parsers, some retrieval steps, .with_structured_output() in some modes) will collapse streaming into a single final chunk.
`temperature=0` and prompt caching confusing you into thinking nothing changed. This isn't a streaming bug, just worth ruling out separately when comparing runs.
A proxy or reverse proxy buffering the HTTP response, discussed above — the backend is streaming correctly, the network layer isn't.
Printing without `flush=True` in a script. Python's stdout is line-buffered or block-buffered depending on context; without explicit flushing, chunks can appear to arrive in batches even though they were yielded individually.
Using `.invoke()` by habit somewhere in a wrapper function that later gets called from streaming-looking code — a surprisingly common copy-paste mistake once a codebase has both patterns present.

A fast way to confirm whether the *model call itself* is streaming, independent of your application code, is to time the first chunk versus the last chunk:

import time
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o", temperature=0)

start = time.time()
first_chunk_time = None
for chunk in llm.stream("Write a 200 word story about a lighthouse keeper."):
    if first_chunk_time is None:
        first_chunk_time = time.time() - start
    print(chunk.content, end="", flush=True)

total_time = time.time() - start
print(f"\n\nFirst token: {first_chunk_time:.2f}s | Total: {total_time:.2f}s")

If first_chunk_time is close to total_time, something upstream is buffering — the model itself is not the bottleneck. If first_chunk_time is a small fraction of total_time, streaming is working correctly at the model layer and any perceived lag downstream is in your application or network code.

Wrapping Up

Streaming isn't a cosmetic nicety bolted onto LangChain — it's a first-class part of the Runnable interface, which means .stream() and .astream() work uniformly across LLMs, chains, and (with .astream_events()) even the internal steps of agents. The mental model to keep is simple: you're working with chunks that concatenate into a full response, not a fundamentally different generation process. Every step in your chain has to support incremental output for that benefit to reach the user, and the moment one step needs the full input before it can act, streaming quietly degrades back into blocking behavior — which is the single most common bug reported in production LangChain apps that "used to stream."

Get comfortable with the sync-to-async migration (.stream() to .astream()), understand when you need event-level visibility versus plain token output, and treat your reverse proxy configuration as part of the streaming pipeline, not an afterthought. Once those pieces are in place, the "instant" feel of modern AI products isn't a mystery — it's just tokens being shown to the user the moment they exist instead of being held back until the very end.

If you want to go deeper — building streaming agents with tool-call visibility, handling backpressure under real production load, and wiring all of this into a polished chat UI end-to-end — that's exactly what we cover hands-on in the LangChain Tutorial 2026 course here on teachyou.ai.