This langchain stream tokens tutorial shows how to pull individual tokens from a LangChain LLM call the moment they are generated, instead of waiting for the full completion. We’ll use LangChain’s callback handlers and the native stream/astream APIs against an OpenAI-compatible chat model, then push those tokens into a minimal FastAPI service.
Step 1: Install the required packages
Streaming matters for perceived latency: a user sees progress instead of a spinner. Set up a clean environment and install the current LangChain packages. The langchain-openai package provides the ChatOpenAI integration; langchain-core holds the callback and message primitives.
python -m venv venv
source venv/bin/activate
pip install langchain-openai langchain-core fastapi uvicornTarget Python 3.10 or newer. LangChain’s async streaming relies on asyncio improvements that are stable in that range. If you later add a vector store or parser, install those separately to avoid bloating the base image.
Step 2: Configure the chat model for streaming
Instantiate ChatOpenAI with streaming=True. Without that flag, the client buffers the response and you will not receive incremental tokens. If you route through n4n.ai, point base_url at its OpenAI-compatible endpoint; it handles provider fallback and meters per-token usage.
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(
model="gpt-4o-mini",
temperature=0.7,
streaming=True,
base_url="https://api.n4n.ai/v1", # optional gateway
api_key="your-api-key",
)The model string must match what your endpoint serves. When using a gateway, the same ChatOpenAI class works unchanged as long as the route speaks the OpenAI chat protocol. Provider cache-control hints are forwarded automatically by compliant gateways, so set extra_body only if you need explicit TTLs.
Step 3: Capture tokens with a callback handler
The simplest way to observe tokens is a custom BaseCallbackHandler. Override on_llm_new_token; LangChain calls it for every token the model emits.
from langchain_core.callbacks import BaseCallbackHandler
class TokenPrinter(BaseCallbackHandler):
def on_llm_new_token(self, token: str, **kwargs):
# kwargs carries run_id, chunk, parent_run_id, etc.
print(token, end="", flush=True)
handler = TokenPrinter()Pass the handler via the config dict. The call still returns the full AIMessage, but tokens print live:
response = llm.invoke(
"Explain Rust's ownership model in three bullets",
config={"callbacks": [handler]},
)
print() # newline after streamThis pattern fits synchronous contexts or when you want side-effect logging without changing your control flow. Be aware that callbacks fire for every LLM inside a chain, so filter by kwargs.get("run_id") if you embed this in a larger graph.
Step 4: Stream asynchronously with astream
For modern async apps, use astream. It returns an async iterator of AIMessageChunk objects. Each chunk’s content is a token or small token group. In a langchain stream tokens tutorial, this is the primitive you’ll build endpoints around.
import asyncio
async def stream_tokens():
async for chunk in llm.astream("Write a haiku about distributed systems"):
if chunk.content:
print(chunk.content, end="", flush=True)
print()
asyncio.run(stream_tokens())astream avoids thread pooling and gives you direct iteration. Tokens are not words—they may split mid-word depending on the tokenizer—but for display purposes that is invisible.
Step 5: Expose tokens over HTTP with FastAPI
Wrap astream in a StreamingResponse. Yield raw strings; FastAPI flushes them as they are produced.
from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
app = FastAPI()
@app.get("/stream")
async def stream_query(q: str, request: Request):
async def event_gen():
try:
async for chunk in llm.astream(q):
if await request.is_disconnected():
break
if chunk.content:
yield chunk.content
except asyncio.CancelledError:
pass
return StreamingResponse(event_gen(), media_type="text/plain")Run it:
uvicorn main:app --port 8000 --reloadFor browser or client streaming, add --no-access-log in high traffic to cut log noise, but keep it during development. The media_type="text/plain" keeps proxies from buffering; use text/event-stream only if you adopt SSE framing.
Step 6: Verify token streaming works
Start the script from Step 4 or the server from Step 5. You should see text appear incrementally, not all at once after a pause.
For the HTTP route, use curl with -N to disable buffering:
curl -N "http://localhost:8000/stream?q=Tell%20me%20a%20short%20joke"Expected behavior: the joke prints token-by-token in your terminal. If you get the full response after a delay, confirm streaming=True is set and that your gateway isn’t forcing buffered mode.
To assert programmatically, wrap the client in a small timer:
import time, asyncio
from langchain_openai import ChatOpenAI
async def timed_stream():
llm = ChatOpenAI(model="gpt-4o-mini", streaming=True, api_key="key")
start = time.time()
async for chunk in llm.astream("Count to five slowly"):
if chunk.content:
print(f"{time.time()-start:.2f}s: {chunk.content}", end="", flush=True)
asyncio.run(timed_stream())If timestamps spread across seconds, streaming is live.
Step 7: Stream through chains and agents
A raw model is easy; real apps use prompts and parsers. LangChain 0.2+ offers astream_events to tap tokens inside a chain. This langchain stream tokens tutorial approach works for RunnableSequence and most LCEL chains.
from langchain_core.prompts import ChatPromptTemplate
prompt = ChatPromptTemplate.from_messages([("system", "Be terse."), ("user", "{input}")])
chain = prompt | llm
async def stream_chain(q: str):
async for event in chain.astream_events({"input": q}, version="v2"):
if event["event"] == "on_llm_stream":
token = event["data"]["chunk"].content
if token:
print(token, end="", flush=True)For agents, stream intermediate steps separately; token streaming only applies to the underlying LLM calls. Do not assume astream_events captures tool outputs as tokens—those are separate event types.
Step 8: Handle client disconnects and backpressure
When a browser closes, your async generator should stop pulling from the model. FastAPI cancels the task; we already added is_disconnected() checks and CancelledError handling in Step 5. Backpressure is minimal for text tokens, but if you batch or transform, yield control with await asyncio.sleep(0) occasionally.
If you build a queue between the LLM and a WebSocket, cap its size. An unbounded asyncio.Queue will happily eat memory if the client is slow.
Step 9: Production notes
- Keep
streaming=Truein the model constructor; toggling per call is error-prone. - If you use a gateway that honors client routing directives, set
modelper request viawith_configrather than rebuilding the client. - Token callbacks fire for every LLM in a chain; use
run_idfiltering if you only care about the final answer. - For per-token cost tracking, sum
len(chunk.content)or use your gateway’s usage metering—n4n.ai returns per-token usage in headers if you ask for it.
That covers a complete langchain stream tokens tutorial: from package install to verified HTTP streaming, with callbacks and async generators.
Step 10: Synchronous streaming with stream()
If you’re writing a CLI and don’t want asyncio, use the blocking stream() iterator.
for chunk in llm.stream("Summarize the OSI model in one line"):
if chunk.content:
print(chunk.content, end="", flush=True)It uses a background thread under the hood. The langchain stream tokens tutorial wouldn’t be complete without noting this for scripts that run in a single-threaded REPL. Token boundaries remain identical to the async path.
Adapt the FastAPI skeleton to WebSockets if you need bidirectional traffic, but the token source stays the same astream loop. Once you have tokens flowing, add framing, retry on 429, and you’re shipping.