Building a langchain fastapi streaming chatbot ui means pushing model tokens to the browser the moment they are generated, not after the full response completes. This tutorial gives you a working end-to-end pattern using LangChain’s async callback handlers, FastAPI’s StreamingResponse, and a few lines of JavaScript. You will be able to type a message and watch the assistant reply word by word.
Step 1: Scaffold the project
Start with a clean virtual environment and install the minimal dependency set. We use langchain-openai because it exposes ChatOpenAI with native streaming support, and FastAPI for the server.
mkdir stream-chat && cd stream-chat
python -m venv .venv && source .venv/bin/activate
pip install "fastapi" "uvicorn[standard]" "langchain-openai" "langchain" "python-dotenv"Create a .env file with your API key. If you point ChatOpenAI at the n4n.ai OpenAI-compatible endpoint, you get access to 240+ models with automatic fallback when a provider is rate-limited or degraded, but any OpenAI-compatible backend works.
OPENAI_API_KEY=sk-...
# Optional: OPENAI_BASE_URL=https://api.n4n.ai/v1Step 2: Configure the streaming LLM and a callback handler
LangChain emits tokens through callback handlers. For a langchain fastapi streaming chatbot ui, the cleanest bridge between the model and the web layer is an asyncio.Queue fed by a custom AsyncCallbackHandler.
import asyncio
from langchain_openai import ChatOpenAI
from langchain.callbacks.base import AsyncCallbackHandler
from langchain_core.messages import HumanMessage
class QueueCallbackHandler(AsyncCallbackHandler):
def __init__(self, queue: asyncio.Queue):
self.queue = queue
async def on_llm_new_token(self, token: str, **kwargs):
# Called synchronously as the model streams
await self.queue.put(token)
async def on_llm_end(self, *args, **kwargs):
# Sentinel value signals the generator to close
await self.queue.put(None)The handler does one job: move tokens from LangChain’s event loop into a queue the HTTP handler owns.
Step 3: Run the chain with the queue
We wrap the model invocation in a coroutine that attaches the handler and sends a single user message. Keep the chain simple; you can swap in LCEL later.
async def stream_chat(message: str, queue: asyncio.Queue):
handler = QueueCallbackHandler(queue)
llm = ChatOpenAI(
model="gpt-3.5-turbo",
streaming=True,
temperature=0.7,
callbacks=[handler],
)
# ainvoke triggers the callback stream
await llm.ainvoke([HumanMessage(content=message)])Note that streaming=True is mandatory. Without it, on_llm_new_token never fires. The langchain fastapi streaming chatbot ui pattern breaks if you forget this flag.
Step 4: Expose a Server-Sent Events endpoint
FastAPI’s StreamingResponse accepts an async generator. We yield SSE-formatted lines (data: ...\n\n) so the browser can parse them without custom framing.
from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
app = FastAPI()
@app.post("/chat")
async def chat(request: Request):
data = await request.json()
message = data.get("message", "")
queue: asyncio.Queue = asyncio.Queue()
async def event_generator():
task = asyncio.create_task(stream_chat(message, queue))
try:
while True:
if await request.is_disconnected():
task.cancel()
break
token = await queue.get()
if token is None:
yield "data: [DONE]\n\n"
break
yield f"data: {token}\n\n"
except asyncio.CancelledError:
pass
finally:
await task
return StreamingResponse(event_generator(), media_type="text/event-stream")The request.is_disconnected() check prevents wasted inference if the user closes the tab. Cancelling the task propagates to LangChain’s async run loop.
Step 5: Build the frontend
A vanilla JS page is enough to prove the pipeline. Save this as static/index.html and mount it with StaticFiles.
<!doctype html>
<html lang="en">
<head><meta charset="utf-8"><title>Chat</title></head>
<body>
<textarea id="input" rows="3" cols="60" placeholder="Ask something..."></textarea><br>
<button onclick="send()">Send</button>
<div id="output" style="white-space: pre-wrap; margin-top: 1rem;"></div>
<script>
async function send() {
const msg = document.getElementById('input').value;
const out = document.getElementById('output');
out.textContent = '';
const res = await fetch('/chat', {
method: 'POST',
headers: {'Content-Type': 'application/json'},
body: JSON.stringify({message: msg})
});
const reader = res.body.getReader();
const decoder = new TextDecoder();
while (true) {
const {value, done} = await reader.read();
if (done) break;
const chunk = decoder.decode(value);
chunk.split('\n\n').forEach(line => {
if (line.startsWith('data: ')) {
const data = line.slice(6);
if (data === '[DONE]') return;
out.textContent += data;
}
});
}
}
</script>
</body>
</html>Mount static files in main.py:
from fastapi.staticfiles import StaticFiles
app.mount("/", StaticFiles(directory="static", html=True), name="static")Step 6: Run and verify success
Launch the server:
uvicorn main:app --reload --port 8000Open http://localhost:8000/. Type “Explain Python’s GIL in one sentence” and click Send. You should see the response append character-by-character. Verification checklist:
- Browser Network tab shows a single POST with
text/event-streamresponses arriving in multiple chunks. - No full delay before first paint; first token appears within a second.
- Closing the tab mid-stream does not raise unhandled exceptions in the server log (the cancel path works).
Why callbacks instead of direct astream
LangChain’s model.astream() returns an async iterator of tokens and is simpler for single-shot calls. The callback approach shines when you compose multiple chains, retrievers, or agents and still need a single token firehose. For a langchain fastapi streaming chatbot ui built on LCEL, you can later replace stream_chat with chain.astream() and push into the same queue, keeping the SSE layer untouched.
Handling backpressure and cleanup
asyncio.Queue is unbounded by default. A slow client can accumulate tokens in memory if the network stalls. In production, use asyncio.Queue(maxsize=1024) and handle queue.full() by dropping or coalescing. Always await the background task in finally to avoid “Task was destroyed but it is pending” warnings.
Production notes
- Run behind a proxy that buffers SSE correctly (nginx needs
proxy_buffering off;). - Use
gunicorn -k uvicorn.workers.UvicornWorkerfor multiple workers; each streaming connection pins a worker, so size accordingly. - If you need per-token metering or routing directives, an OpenAI-compatible gateway that honors cache-control hints can be dropped in by changing
base_urlonly.
The langchain fastapi streaming chatbot ui you now have is minimal but structurally correct: tokens flow from LangChain callbacks to an asyncio queue, through a FastAPI generator, and into a browser via SSE. Swap the model, add retrieval, or harden the frontend without touching the streaming core.