Build a Production-Grade AI Chatbot with FastAPI
A toy chatbot is a forgivable sin when you are learning; a production chatbot needs memory, streaming, isolation between users, and resilient retries. FastAPI gives you the async foundation for all four without pulling in a heavy framework. In this tutorial you will build a chatbot service that keeps an in-memory conversation per session, streams tokens to the browser, isolates sessions so users never see each other's messages, and retries transient provider failures. By the end you will have a service you can trust under real traffic.
Designing sessions and memory
Multi-session isolation means each conversation lives in its own context bag. Pass a session ID in the request and key the history on it, so two users sharing one process never contaminate each other's memory. In a stateless deployment you would back the store with Redis, but for a single-node service a dictionary of session objects with a time-to-live is a clean start. Cap each history length so a very long conversation does not grow your prompt without bound.
sessions = {}
def get_session(sid):
if sid not in sessions:
sessions[sid] = []
return sessions[sid]
Streaming responses
Users expect words to appear as they are generated, not after an anxious pause. FastAPI supports async generators natively, so the provider's streamed tokens can flow straight to the client as a StreamingResponse. Each token arrives on the provider's async iterator, you append it to the session history in a background task or after the stream closes, and you yield the delta to the client. Because the endpoint is async and non-blocking, FastAPI keeps serving other sessions while one is streaming.
Streaming changes the perceived quality of a chatbot more than any other feature. A user who sees tokens 200ms apart experiences a responsive product; a user staring at a spinner for fifteen seconds experiences a failure.
Match the framing to your transport. If the browser consumes a plain text stream, announce the media type and keep the response streaming; if the client is a mobile app, chunk deltas into sensible frames. Whichever shape you pick, expose a heartbeat so the front end can distinguish an idle stream from a dead connection, and always render partial text in place so the user never sees a blank box.
To append the full generation after the stream completes, use a small helper that collects deltas as they are sent and pushes the assembled message into the session history once the generator finishes. That keeps the visible history accurate without duplicating tokens during the stream, and it works cleanly even if the client disconnects early, because the generator can still finalize the message in its finally block.
Keeping the event loop responsive
FastAPI shines because it never blocks a thread while waiting on I/O, but only if you keep slow work off the event loop. The provider call is async, which is perfect, but anything CPU-bound, such as a tokenizer pass or a heavy transformation, should be pushed into a thread pool with await asyncio.to_thread(...). Blocking the loop for even a few hundred milliseconds stalls every other session sharing the process, which is the fastest way to turn a small chatbot into an apparent outage.
Give each request a bounded budget, too. Set an overall timeout that covers model latency, response assembly, and streaming, and enforce it with an asyncio timeout so a misbehaving provider cannot pin your worker for minutes. When you push a token generator to the client, keep the buffer configurable; a producer that runs far ahead of a slow consumer can inflate memory and make disconnects leak work.
- Cancel tasks on disconnect. Detect that the client left and stop generating the rest of the tokens.
- Bound history size. Drop oldest turns so the prompt stays inside the model context window.
- Emit heartbeats. Send an empty token or comment periodically so proxies do not time out an idle stream.
- Log stream length. Record how many tokens actually reached the client versus were generated.
These details are invisible to a happy user, and that is exactly why they matter in production. The moment traffic spikes, a chatbot that yields control promptly and cancels cleanly stays responsive while a naive one spirals under the load.
Retries, timeouts, and circuit breaking
Providers fail. A transient network blip or a burst rate limit returns a 429 or a 5xx, and a naive service surfaces that to the user as an error. Wrap the provider call so it retries with exponential backoff and jitter on retryable status codes, and only give up after a few attempts. Add a per-request timeout so a hung upstream cannot pin your worker. When upstream keeps failing, break the circuit with a short cooldown instead of hammering the API with a retry storm.
async def call_with_retries(client, payload, max_retries=3):
for attempt in range(max_retries):
try:
return await client.post(..., json=payload)
except RetryableError as e:
await asyncio.sleep(2 ** attempt + random.random())
raise LastResortError("provider unavailable")
Do not wait for a load test to reveal weak points; think through failure modes up front. Decide what happens when the provider returns a non-retryable error and map it to a friendly user message instead of a raw stack trace. Choose how long a disconnected stream may keep generating before you cancel it, and pick a policy for what partial text remains visible when a call fails halfway through. Writing these decisions down removes the guesswork that otherwise surfaces as confusing behaviour at the exact moment you have an incident.
Shipping a small but real set of tests pays for itself quickly. Load a representative conversation, force a simulated provider failure, and assert that the retry logic, the circuit breaker, and the error handler all respond correctly. Run the same suite against the streaming route to catch encoding or framing bugs before real users do, because a chatbot that fails in production is only recoverable if your tooling already knows where to look.
Safety and hardening
A production service does not trust its own output. Enforce a hard max_tokens on every call, strip dangerous characters from user input before it reaches the model, and validate that the session ID is not attacker-controlled. Add a basic rate limiter per session or per IP so an abusive client cannot drain your quota. Finally, wrap every endpoint in a structured error handler that returns JSON the front end can render, and log the request and response for later debugging. Apply this pattern to the provider using the retry wrapper with streaming, then wire those pieces together for a chatbot that stays calm when the world around it fails.
