Streaming with LangChain LCEL

LangChain LCEL streaming lets you get tokens back from a chain as they're generated instead of waiting for the whole response, and it works by every Runnable in your pipeline implementing a shared stream and astream interface that LCEL composes automatically when you pipe components together with the | operator. The catch is that streaming only flows end to end if every step in the chain actually supports it, one non-streaming component in the middle (a RunnableLambda that does return result instead of yielding, or an output parser that buffers) and your whole chain quietly falls back to collect-then-emit. This article walks through how LCEL streaming actually works, the three APIs you'll use (stream, astream, astream_events), and how to hook it into a FastAPI server with Server-Sent Events.

Why LCEL streaming is different from calling `.stream()` on a model

If you've only ever streamed a raw model call, streaming feels trivial: call model.stream(prompt), iterate the chunks, done. LCEL streaming is a different problem because you're not streaming one call, you're streaming a graph of calls, some of which might not even be doing text generation (a retriever, a Python function, a parser).

LCEL solves this by giving every Runnable the same four methods:

• invoke / ainvoke: run once, return the full result
• batch / abatch: run over a list of inputs
• stream / astream: run once, yield partial results as they're produced
• astream_events: run once, yield a structured event log of everything that happened inside, including nested steps

When you build a chain with prompt | model | parser, LCEL wires each component's stream output into the next component's stream input. If model supports streaming (most chat model integrations do) and parser supports streaming (the built-in StrOutputParser does), then calling .stream() on the whole chain gives you tokens as the model produces them, with each token passed through the parser one at a time.

Here's the baseline that actually streams:

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

prompt = ChatPromptTemplate.from_template("Write a short poem about {topic}")
model = ChatOpenAI(model="gpt-4o-mini", temperature=0.7)
parser = StrOutputParser()

chain = prompt | model | parser

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

Run this and you'll see text appear word by word in your terminal, not all at once after a pause. That's the whole point: the user perceives lower latency because the first token shows up in a few hundred milliseconds instead of waiting for the full generation.

The silent fallback trap

The dangerous part of LCEL streaming is that a broken chain doesn't throw an error, it just stops streaming and buffers everything, then dumps it at the end. You'll see one big chunk instead of a sequence of small ones, and unless you're specifically watching for chunk count or timing, it looks like it "worked."

This happens whenever a step in the chain can only produce output after consuming its entire input. The most common offenders:

• A RunnableLambda wrapping a function that does return some_value instead of being a generator
• Any step that needs to see the full text before it can act, like a JSON parser that calls json.loads() on the complete string
• Custom output parsers that don't implement parse_result in a streaming-friendly way
• Some retrieval or reranking steps, which are inherently batch operations by nature (you can't stream a "top 5 documents" result meaningfully)

Here's a broken example:

from langchain_core.runnables import RunnableLambda

def add_signature(text: str) -> str:
    return text + "\n\n-- Generated by AI"

chain = prompt | model | parser | RunnableLambda(add_signature)

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

add_signature needs the full string to concatenate a suffix, so LCEL can't stream through it, everything upstream gets buffered until add_signature runs once on the complete text, and you get a single chunk at the end. This is often fine (you may not care about streaming past that point), but if you expected token-by-token output through the whole chain, this is the bug to look for.

If you genuinely need to transform streamed chunks and preserve streaming, write a generator-based transform instead:

from langchain_core.runnables import RunnableGenerator

def add_signature_stream(chunks):
    for chunk in chunks:
        yield chunk
    yield "\n\n-- Generated by AI"

chain = prompt | model | parser | RunnableGenerator(add_signature_stream)

RunnableGenerator wraps a function that takes an iterator of upstream chunks and yields its own chunks, so LCEL keeps streaming through it instead of collapsing to batch mode.

Async streaming with `astream`

Most production code that streams to a browser or API client runs inside an async web framework, so you'll use astream instead of stream. The interface is identical, just async:

import asyncio

async def main():
    chain = prompt | model | parser
    async for chunk in chain.astream({"topic": "databases"}):
        print(chunk, end="", flush=True)

asyncio.run(main())

astream is not just a wrapper around stream run in a thread, LangChain integrations implement native async streaming against the provider's async HTTP client, so you get real concurrency, meaning you can serve multiple streaming requests from a single event loop without blocking each other.

`astream_events`: streaming structured events, not just text

stream and astream give you the final output type of the chain (usually a string), chunk by chunk. That's fine for a simple prompt-to-text pipeline, but real chains have multiple steps, retrieval, tool calls, sub-chains, and you often want to know what's happening inside, not just the final text. That's what astream_events is for.

astream_events yields a stream of event dictionaries, each with a event type (on_chat_model_stream, on_retriever_end, on_chain_start, on_tool_start, and so on), a name, and a data payload. This is the API you want when building a UI that shows "Searching documents..." followed by "Generating answer..." followed by the streamed tokens.

async def main():
    chain = prompt | model | parser

    async for event in chain.astream_events({"topic": "databases"}, 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())

Always pass version="v2" explicitly. The event schema changed between v1 and v2, and relying on the default means your code behaves differently depending on which LangChain version happens to be installed.

For a retrieval-augmented chain, astream_events is what lets you show the retrieval step happening before the generation step, without writing your own event bus:

from langchain_core.runnables import RunnablePassthrough

rag_chain = (
    {"context": retriever, "question": RunnablePassthrough()}
    | prompt
    | model
    | parser
)

async for event in rag_chain.astream_events({"question": "How does indexing work?"}, version="v2"):
    if event["event"] == "on_retriever_end":
        docs = event["data"]["output"]
        print(f"[retrieved {len(docs)} documents]")
    elif event["event"] == "on_chat_model_stream":
        chunk = event["data"]["chunk"].content
        if chunk:
            print(chunk, end="", flush=True)

Streaming with tool calls and agents

Tool-calling adds a wrinkle: the model doesn't stream plain text for a tool call, it streams partial JSON arguments that only become valid once the call is complete. If you're building an agent loop, don't try to render tool-call arguments token by token in the UI, buffer them until on_tool_start or the equivalent completion event fires, and only stream the final natural-language response tokens live.

async for event in agent_chain.astream_events({"input": "What's the weather in Mumbai?"}, version="v2"):
    kind = event["event"]
    if kind == "on_tool_start":
        print(f"\n[calling tool: {event['name']} with {event['data'].get('input')}]")
    elif kind == "on_tool_end":
        print(f"[tool result: {event['data'].get('output')}]")
    elif kind == "on_chat_model_stream":
        chunk = event["data"]["chunk"].content
        if chunk:
            print(chunk, end="", flush=True)

This gives you a clean separation: structured tool events for your "thinking" UI, streamed text tokens for the final answer.

Wiring LCEL streaming into FastAPI with Server-Sent Events

The most common place this all lands in production is a FastAPI backend serving a chat UI over Server-Sent Events (SSE). SSE is simpler than WebSockets for one-directional streaming and works with plain fetch + EventSource on the frontend.

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

app = FastAPI()

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

async def event_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")
async def chat(question: str):
    return StreamingResponse(
        event_generator(question),
        media_type="text/event-stream",
    )

A few details that matter here:

• Set media_type="text/event-stream", not application/json. Browsers rely on this to keep the connection open and parse it as an event stream.
• Each SSE message needs to end with a double newline (\n\n). Skip this and clients will buffer messages together incorrectly.
• Send an explicit terminator ([DONE] or similar) so the frontend knows when to close the EventSource connection, since SSE itself doesn't have a built-in "stream ended cleanly" signal that all clients handle the same way.
• If you're behind a reverse proxy (nginx, Caddy), disable response buffering for this route, otherwise the proxy buffers the whole response before forwarding it and you lose the streaming benefit entirely.

On the frontend, consuming this is a short fetch loop against the ReadableStream:

async function streamChat(question) {
  const response = await fetch(`/chat?question=${encodeURIComponent(question)}`);
  const reader = response.body.getReader();
  const decoder = new TextDecoder();

  while (true) {
    const { done, value } = await reader.read();
    if (done) break;
    const text = decoder.decode(value);
    for (const line of text.split("\n\n")) {
      if (line.startsWith("data: ")) {
        const payload = line.slice(6);
        if (payload === "[DONE]") return;
        document.getElementById("output").textContent += payload;
      }
    }
  }
}

If you want structured events (tool calls, retrieval steps, tokens) instead of plain text, swap chain.astream for chain.astream_events(..., version="v2") in the generator and serialize each event as JSON before sending it over SSE:

import json

async def event_generator(question: str):
    async for event in chain.astream_events({"question": question}, version="v2"):
        if event["event"] == "on_chat_model_stream":
            content = event["data"]["chunk"].content
            if content:
                yield f"data: {json.dumps({'type': 'token', 'content': content})}\n\n"
    yield f"data: {json.dumps({'type': 'done'})}\n\n"

Debugging a chain that isn't streaming

When your chain returns everything in one chunk and you're not sure why, don't guess, isolate the problem systematically:

1. Call .stream() on each component individually, starting from the model alone (model.stream(messages)), then add one piped step at a time ((prompt | model).stream(...), then (prompt | model | parser).stream(...)) until the chunk count drops to one. That's your culprit.
2. Check if the offending component is a RunnableLambda around a plain function. If so, either accept the buffering (fine for post-processing that must see the whole text) or rewrite it as a RunnableGenerator.
3. Confirm the underlying model integration actually supports streaming. Not every provider integration streams by default, some require streaming=True in the constructor, and a handful of hosted models don't support streaming at all for certain response modes (like forced JSON output in some configurations).
4. Check for a callback handler that's buffering output for logging or tracing purposes, this is a common cause of "streaming works locally but not in production" once observability tooling gets added.

FAQ

Does `chain.stream()` and `chain.astream()` give exactly the same chunks? They should produce equivalent content, one is the sync version and one the async version of the same underlying execution. Use astream in any async context (FastAPI, async agent loops) and stream in plain scripts, don't mix a sync stream() call inside an async def route, it will block the event loop.

Why does my chain stream fine with `invoke` replaced by `stream`, but a `RunnableParallel` step breaks it? RunnableParallel runs its branches concurrently and only emits once all branches have something to emit for that step. If one branch is a retriever (batch by nature) and another is a model (streaming), the parallel step's overall streaming behavior is limited by the slowest, non-streaming branch. This is expected: you'll typically see the retrieval branch resolve fully, then the model branch stream normally after.

Can I stream from `chain.batch()` for multiple inputs at once? Not directly, batch and stream are separate execution modes. If you need to stream several independent generations concurrently, run multiple astream calls concurrently with asyncio.gather or asyncio.as_completed, and route each one to its own output channel (a separate SSE connection per request, or a multiplexed WebSocket with an ID per message).

What's the difference between `astream_events` v1 and v2? v2 standardized event names and payload shapes across all runnable types and is the version actively maintained going forward. Always pin version="v2" explicitly in your code, LangChain will otherwise warn you (or in newer releases, default to v2 outright), and explicit versioning avoids surprises across upgrades.

Do output parsers always support streaming? No. StrOutputParser streams cleanly since it's a pass-through on string chunks. Structured output parsers that need the complete response to validate against a schema (like a Pydantic-based parser) generally cannot stream partial, valid objects, they'll buffer until the full response arrives, then parse once. If you need partial structured output while streaming, look at parsers explicitly built for it that emit progressively-filled partial objects rather than a JSON schema validator that only accepts complete JSON.

Should I use `stream` or `astream_events` for a simple single-model chatbot? Use astream for a simple, single-purpose chatbot chain where you only care about the final text tokens, it's simpler and has less overhead. Reach for astream_events once your chain has multiple meaningful steps (retrieval, tool calls, sub-chains) that you want to expose in the UI as separate stages, since the event-level granularity is what makes that possible without hand-rolling your own instrumentation.