Streaming Tokens in LangChain

LangChain streaming lets you show tokens to a user as the model generates them instead of making them wait for the full response. If you have ever watched a chat UI print words one at a time, that is streaming, and LangChain gives you three different APIs to do it: .stream(), .astream(), and .astream_events(). Picking the wrong one is the most common reason streaming "doesn't work" in a LangChain app, so this article walks through what each one actually does, where the chunks come from, and how to wire streaming into a real backend.

Why streaming matters more than it looks

A non-streamed call to an LLM blocks until the whole response is generated. For short answers that is fine. For anything longer than a couple of sentences, the user stares at a spinner for several seconds, which feels broken even when the backend is healthy. Streaming turns that dead time into visible progress: tokens appear as soon as the model produces them, time-to-first-token becomes the metric that matters, and the perceived latency drops even though total generation time is unchanged.

There's a second, less obvious reason to care: streaming forces you to confront how your chain is actually structured. A chain that streams cleanly is usually a chain built the LangChain Expression Language (LCEL) way, with runnables piped together. A chain that silently swallows streaming (common with older LLMChain-style code, or anything that buffers output inside a custom function) is a chain worth refactoring anyway.

The three streaming APIs, and when to use each

`.stream()` for synchronous, single-chain output

Every LangChain "Runnable" (LLMs, chat models, prompts, output parsers, retrievers, and full chains built by piping them together with |) implements .stream(). It returns a generator that yields chunks as they're produced.

from langchain_anthropic import ChatAnthropic

model = ChatAnthropic(model="claude-sonnet-4-5")

for chunk in model.stream("Explain what a hash map is in two sentences."):
    print(chunk.content, end="", flush=True)

Each chunk here is an AIMessageChunk, not a plain string. That distinction matters because AIMessageChunk objects support the + operator, so you can accumulate them into a full message:

full = None
for chunk in model.stream("List three sorting algorithms."):
    full = chunk if full is None else full + chunk
    print(chunk.content, end="", flush=True)

print()
print(full.content)   # the complete accumulated text
print(full.response_metadata)  # populated once the stream is done

.stream() works for a single chat model call and for LCEL chains built from streamable components. It does not work for arbitrary Python functions dropped into a chain with RunnableLambda unless that function itself is a generator.

`.astream()` for async backends

If your application is async (FastAPI, an async worker, anything built on asyncio), use .astream() instead. It has the identical shape but yields inside an async for loop:

async def generate(prompt: str):
    async for chunk in model.astream(prompt):
        yield chunk.content

Mixing sync .stream() into an async web handler will block the event loop on network I/O, which defeats the purpose of running an async server in the first place. If your server is async, keep the whole path async, including the streaming call.

`.astream_events()` for chains with multiple steps

This is the one people reach for too late. .stream() and .astream() only stream the *final* output of a chain. If your chain is a pipeline, prompt -> retriever -> model -> output parser, and you want visibility into intermediate steps (which documents got retrieved, when the model started generating, when a tool was called), you need .astream_events().

async def generate_with_events(prompt: str):
    async for event in chain.astream_events(prompt, version="v2"):
        kind = event["event"]

        if kind == "on_chat_model_stream":
            chunk = event["data"]["chunk"]
            if chunk.content:
                yield chunk.content

        elif kind == "on_retriever_end":
            docs = event["data"]["output"]
            print(f"retrieved {len(docs)} docs")

        elif kind == "on_tool_start":
            print(f"calling tool: {event['name']}")

Always pass version="v2" explicitly. The event schema changed between v1 and v2, and code that omits the version argument is fragile against library upgrades. astream_events gives you a flat stream of events tagged with a name, event type (on_chain_start, on_llm_stream, on_tool_end, and so on), run_id, and data payload. It is more verbose than .stream() but it's the only API that lets you distinguish "the model is now generating" from "a retriever just ran" from "a tool call just finished."

Streaming through a full LCEL chain

Streaming composes automatically through LCEL as long as every link in the chain supports it. A typical retrieval-augmented generation (RAG) chain:

from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough

prompt = ChatPromptTemplate.from_template(
    "Answer the question using only this context:\n{context}\n\nQuestion: {question}"
)

def format_docs(docs):
    return "\n\n".join(d.page_content for d in docs)

chain = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | prompt
    | model
    | StrOutputParser()
)

for chunk in chain.stream("What does the retry policy say about backoff?"):
    print(chunk, end="", flush=True)

Notice StrOutputParser() at the end. It converts AIMessageChunk objects into plain string chunks as they pass through, so chain.stream() yields strings directly instead of message objects. That's a deliberate design choice in LangChain: output parsers are streaming-aware and transform chunk-by-chunk rather than waiting for the full message.

The retriever step (retriever | format_docs) does *not* stream, because retrieval is a single blocking call that returns a list of documents. That's expected. Streaming only kicks in once you hit the model call. If you watch the chain run with .astream_events(), you'll see the retriever's on_retriever_start/on_retriever_end fire once each, followed by a burst of on_chat_model_stream events as the model output comes in.

A common trap: `RunnableLambda` breaks streaming

If you wrap a plain Python function in your chain, streaming silently stops working past that point unless the function is itself a generator:

from langchain_core.runnables import RunnableLambda

def clean_output(text: str) -> str:
    return text.strip().replace("  ", " ")

# This BREAKS streaming: the lambda buffers the whole chain output
# before running, because it's not a generator.
chain = prompt | model | StrOutputParser() | RunnableLambda(clean_output)

Because clean_output takes a full string and returns a full string, LangChain has no choice but to materialize the entire upstream output before calling it. If you need to keep streaming, either move the transformation to something that operates on strings safely per-chunk (rare, since most text transforms need the full string), or accept that this stage of the chain is the last streamable point and do post-processing after .stream() has finished yielding.

A cleaner pattern is to stream the raw model output to the user for perceived latency, and run any transformation as a separate, non-streamed step server-side (e.g., logging, moderation checks) after the full text has been assembled.

Streaming from a FastAPI endpoint with Server-Sent Events

The most common real-world target for LangChain streaming is a web API. Server-Sent Events (SSE) are the simplest transport: one-directional, works over plain HTTP, and every browser has a built-in EventSource client.

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import json

app = FastAPI()

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

@app.post("/chat")
async def chat(prompt: str):
    return StreamingResponse(sse_stream(prompt), media_type="text/event-stream")

A few details that matter in production:

• Set media_type="text/event-stream" explicitly, and if you're behind a reverse proxy like Nginx, disable buffering for the route (proxy_buffering off or the equivalent) or chunks will queue up server-side and arrive all at once anyway.
• Always send a terminal sentinel like [DONE] so the client knows when to stop listening, since SSE doesn't have a built-in "stream ended cleanly" signal separate from the connection closing.
• Wrap the generator body in a try/except and yield an error event on failure. Without this, an exception mid-stream just closes the connection and the client has no idea why.

async def sse_stream(prompt: str):
    try:
        async for event in chain.astream_events(prompt, version="v2"):
            if event["event"] == "on_chat_model_stream":
                content = event["data"]["chunk"].content
                if content:
                    yield f"data: {json.dumps({'token': content})}\n\n"
    except Exception as exc:
        yield f"data: {json.dumps({'error': str(exc)})}\n\n"
    finally:
        yield "data: [DONE]\n\n"

On the frontend, EventSource (or a small fetch + ReadableStream reader if you need custom headers, since EventSource doesn't support them) consumes this directly. If you're using WebSockets instead of SSE because you also need client-to-server messages mid-stream (for interrupting generation, for example), the same async for event in chain.astream_events(...) loop works, you just push each chunk over websocket.send_text() instead of formatting it as an SSE line.

Streaming intermediate steps in agents

Agents that call tools are where .astream_events() earns its complexity. A tool-calling agent has several phases that a naive .stream() call collapses into an opaque wait: reasoning, tool call, tool result, more reasoning, final answer. If you only stream the final output, the user sees nothing during tool execution, which for a slow tool (a database query, a web search) can be the majority of the latency.

async def stream_agent(agent_executor, input_text: str):
    async for event in agent_executor.astream_events(
        {"input": input_text}, version="v2"
    ):
        kind = event["event"]

        if kind == "on_chat_model_stream":
            content = event["data"]["chunk"].content
            if content:
                yield {"type": "token", "content": content}

        elif kind == "on_tool_start":
            yield {"type": "tool_start", "tool": event["name"], "input": event["data"].get("input")}

        elif kind == "on_tool_end":
            yield {"type": "tool_end", "tool": event["name"], "output": str(event["data"].get("output"))}

This lets the frontend render "Searching the web..." while a tool runs, then switch to token-by-token rendering once the model resumes generating the final answer. That's the difference between an agent that feels responsive and one that feels stuck.

Callbacks: the older, lower-level mechanism

Before .astream_events() existed, streaming in LangChain leaned heavily on callback handlers, and you'll still see this pattern in older codebases and in some framework integrations that haven't been updated.

from langchain_core.callbacks import BaseCallbackHandler

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

model = ChatAnthropic(model="claude-sonnet-4-5", streaming=True, callbacks=[TokenPrinter()])
model.invoke("Summarize the CAP theorem in one paragraph.")

A few things to know if you inherit code like this:

• You still have to set streaming=True (or use .stream()/.astream(), which sets it implicitly) or on_llm_new_token never fires, because the model just returns the complete response in one shot.
• Callback handlers run synchronously by default. If you need an async callback for a web server, subclass AsyncCallbackHandler and implement async def on_llm_new_token instead, otherwise you'll block the event loop from inside the handler.
• astream_events() is generally the better choice for new code because it gives you a single unified stream you can iterate with async for, rather than a handler object whose methods fire as side effects. Reach for callbacks specifically when you need to hook into a third-party integration that only exposes the callback interface, or when you want file-level logging that runs independent of any particular stream consumer.

Debugging streaming that isn't streaming

When .stream() returns everything at once instead of chunk by chunk, the cause is almost always one of these:

• A non-streaming model or provider setting. Some providers or specific model configurations don't support token-level streaming, or need streaming=True set explicitly at the client level.
• A blocking step in the chain. Anything that has to see the whole output before it can run (a RunnableLambda over a full string, a Pydantic output parser that needs complete valid JSON, a summarization step) will buffer everything upstream of it.
• Reverse proxy buffering. The application code is streaming correctly, but Nginx, a load balancer, or a CDN is buffering the response body and flushing it in one chunk. Check with curl -N directly against the app server, bypassing the proxy, to isolate this.
• Structured output parsers. .with_structured_output() and JSON-mode parsers generally cannot stream partial results token-by-token, because partial JSON isn't valid JSON. Some parsers support streaming partial, progressively-more-complete dict shapes, but this is a different (and much more limited) kind of streaming than raw text tokens. If you need structured output and streaming both, stream the raw text and parse it into structure only once the stream completes.

FAQ

Does `.stream()` work with every LangChain model integration? No. Streaming support depends on the underlying provider and integration package actually implementing token-level streaming. Chat model integrations from major providers generally support it; some smaller or local-model integrations only support call-level (non-streaming) invocation, or provider a "streaming" API that just chunks up an already-complete response. Check the integration's documentation before assuming streaming works, and test with a real for chunk in model.stream(...) call rather than trusting the class signature.

What's the difference between `AIMessageChunk` and a plain string? AIMessageChunk is LangChain's streaming message type. It carries .content (the text delta), plus metadata fields that only populate as the stream progresses (things like usage_metadata or response_metadata, which are often empty until the final chunk). Chunks support + so you can reduce a stream into one complete message. If you only need text, pipe the chain through StrOutputParser() so .stream() yields plain strings instead of chunk objects.

Can I cancel a stream partway through? Yes, at the transport level: closing the generator (breaking out of the for loop) stops LangChain from pulling further chunks, and closing the underlying HTTP response to the model provider cancels the upstream request too, in most integrations. If you're serving over SSE or WebSockets, detecting client disconnect and breaking your server-side loop is the right way to stop generation early and avoid paying for tokens nobody will see.

Should I use `.stream()` or `.astream_events()` by default? Use .stream()/.astream() for a single model call or a simple chain where you only care about the final text output. Reach for .astream_events() as soon as you have multiple steps (retrieval, tool calls, multi-model chains) and want visibility into what's happening at each stage, or when you're building a UI that needs to distinguish "retrieving" from "generating" from "calling a tool."

Why does my agent's streamed output include tool call syntax instead of clean text? That usually means you're streaming the raw on_llm_stream events from a model configured to emit tool calls, rather than filtering for the final answer generation phase. Check the event's tags or name field, agent frameworks typically tag intermediate reasoning steps differently from the final response, and filter your astream_events() loop to only forward tokens from the step you actually want the user to see.

Does streaming cost more or less than a non-streamed call? Streaming doesn't change how many tokens are generated or billed, it changes how they're delivered to the client. Cost is a function of input and output token counts, not delivery mechanism, so switching between .invoke() and .stream() for the same prompt and response has no cost difference on its own.