We use cookies to enhance your experience on the site

Production Deployment - Deploying AI

The code from the previous lesson runs in your terminal and nobody but you can reach it. Deployment turns a field notebook into a radio station: what matters is not only that it transmits, but how much it costs to stay on air and whether anyone notices when it goes quiet.

The /chat endpoint - the simplest way to expose a model

FastAPI describes an endpoint with the

@app.post

decorator, and the shape of the incoming and outgoing data with a Pydantic

BaseModel

class. Thanks to that you never check by hand whether the client actually sent a

message

field - the library rejects a malformed request before your function even runs. This is exactly the shape of your first task in this lesson: create a FastAPI endpoint with LLM integration, a

/chat

endpoint that accepts a message and returns an AI response.

1from fastapi import FastAPI
2from pydantic import BaseModel
3from openai import AsyncOpenAI
4
5app = FastAPI()
6client = AsyncOpenAI()
7
8class ChatRequest(BaseModel):
9    message: str
10
11class ChatResponse(BaseModel):
12    response: str
13
14@app.post("/chat", response_model=ChatResponse)
15async def chat(request: ChatRequest):
16    completion = await client.chat.completions.create(
17        model="gpt-5",
18        messages=[{"role": "user", "content": request.message}]
19    )
20    return ChatResponse(response=completion.choices[0].message.content)

ChatRequest

describes what comes in,

ChatResponse

what goes out, and

response_model

enforces that the reply really has that shape. The

AsyncOpenAI

class lets you

await

the call, so while one request waits on the model the server keeps serving other clients. The

return

statement hands back the whole answer at once, which means the user stares at an empty screen for a dozen seconds.

Streaming: StreamingResponse and an async generator

A regular return cannot be talked into streaming. That is not a setting you forgot, it follows from what

return

means: it ends the function and hands over a value, so the value has to be complete already. Using temporary files does not rescue it either, because the write happens only once the full text has arrived, so the wait is exactly as long. And it is certainly not impossible - you just need a function that hands back text in pieces, plus the

StreamingResponse

class to wrap it in an HTTP response.

1from fastapi.responses import StreamingResponse
2
3@app.post("/chat/stream")
4async def chat_stream(request: ChatRequest):
5    async def generate():
6        stream = await client.chat.completions.create(
7            model="gpt-5",
8            messages=[{"role": "user", "content": request.message}],
9            stream=True
10        )
11        async for chunk in stream:
12            content = chunk.choices[0].delta.content
13            if content:
14                yield f"data: {content}\n\n"
15        yield "data: [DONE]\n\n"
16
17    return StreamingResponse(generate(), media_type="text/event-stream")

The whole difference sits in the word

yield

. The

generate

function never returns a result and never finishes - it hands over a fragment and suspends until someone asks for the next one. A function like that is an async generator, and

StreamingResponse

pushes its fragments to the client as they appear. You implement AI response streaming in FastAPI using StreamingResponse with an async generator - there is no second way that works.

Two silent failures are worth memorising, @name, because neither one raises an error. Without the

if content:

guard the literal text

None

lands in the middle of the stream, since the first and last chunks carry an empty

delta.content

. And without the closing

yield

carrying

[DONE]

, the client keeps spinning its loading indicator long after the server has finished working.

WebSocket - when one direction is not enough

Streaming solves half the problem: the server can talk for longer, but still only in reply to a question. HTTP works as request and response, and the server has no way to speak first - yet in a chatbot you want to push a notification, or the result of a background job that nobody happens to be asking about.

WebSocket keeps one connection open in which both sides send messages at any moment - that is bidirectional real-time communication, and it is the one real reason to reach for it. It is not faster than HTTP: it rides on the same TCP. It is not more secure either -

ws://

travels in plain text, and

wss://

uses the very same TLS as

https

. And it certainly does not mean it doesn't require a server: it needs one more than HTTP does, because that server holds the connection open for the entire conversation.

1from fastapi import WebSocket, WebSocketDisconnect
2
3active_connections: dict[str, WebSocket] = {}
4
5@app.websocket("/ws/{client_id}")
6async def websocket_endpoint(websocket: WebSocket, client_id: str):
7    await websocket.accept()
8    active_connections[client_id] = websocket
9    try:
10        while True:
11            question = await websocket.receive_text()
12            answer = await ask_model(question)
13            await websocket.send_text(answer)
14    except WebSocketDisconnect:
15        del active_connections[client_id]

The

ask_model

function is the model call from the first example, and the

while True

loop handles message after message without tearing the connection down. The

active_connections

dictionary keeps every live socket under a client id, which is precisely what lets you send someone a message they never asked for. Skipping the delete on
WebSocketDisconnect
is a memory leak: the server runs for weeks while the dictionary swells with dead entries.

Caching and rate limits - keeping the bill from growing forever

Every model call is a separate charge for tokens, so when a hundred people ask the same thing you pay a hundred times for one identical answer. Caching reduces API costs and speeds up responses, because a remembered answer comes back in milliseconds without touching the network at all. Two small libraries are enough here:

cachetools

for the memory,

slowapi

for the ceiling on traffic.

1from cachetools import TTLCache
2from slowapi import Limiter
3from slowapi.util import get_remote_address
4
5limiter = Limiter(key_func=get_remote_address)
6response_cache = TTLCache(maxsize=1000, ttl=3600)
7
8@app.post("/chat/cached")
9@limiter.limit("10/minute")
10async def chat_cached(request: ChatRequest):
11    key = request.message.strip().lower()
12    if key in response_cache:
13        return {"response": response_cache[key], "cached": True}
14
15    response_cache[key] = await ask_model(request.message)
16    return {"response": response_cache[key], "cached": False}

TTLCache

drops entries by itself after an hour (

ttl=3600

), so answers never go stale for good, and the key is the question text, which means two identical questions land on the same entry. So it is simply untrue that there are no benefits here: the cache cuts the bill and it cuts the wait.

It is worth knowing what a cache does not do. It does not improve response quality - it returns the same text that was produced earlier, so a weak answer comes back just as weak, only faster. It is not required by law either; how much you spend is your own decision. The

@limiter.limit("10/minute")

decorator from

slowapi

caps requests coming from a single address, so one client cannot burn the whole budget in a quarter of an hour.

Before you put this in front of the world

The three things that matter most before deploying an LLM to production are testing, monitoring, and securing API keys. The key has to come from an environment variable, never from the code - hardcoded, it lands in the repository, and from there on somebody else's bill. Monitoring starts with a

/health

endpoint, which is how your hosting learns that the process is still alive.

1import logging
2import os
3
4logging.basicConfig(level=logging.INFO)
5logger = logging.getLogger("safari-ai")
6
7client = AsyncOpenAI(api_key=os.environ["OPENAI_API_KEY"])
8
9@app.get("/health")
10async def health_check():
11    logger.info("health ok")
12    return {"status": "healthy"}

Reading

os.environ["OPENAI_API_KEY"]

has one extra benefit: when the variable is missing the application refuses to start, instead of accepting traffic and only failing on the first real question. It is worth wrapping the model call in

try

as well, so an outage at the provider comes back as an

HTTPException

while

logger.exception

writes the trace into the logs.

Three temptations to resist. Using the latest model without testing looks harmless, because a model name is just a string - except the new model formats its answers differently and quietly breaks the parsing on your side. Disabling logging for speed saves a fraction of a millisecond and takes away the only record of what happened overnight. Exposing the API without rate limits ends in an invoice: the key is yours, so other people's requests are billed to you.

Summary

You implement AI response streaming in FastAPI using StreamingResponse with an async generator, meaning a function that uses
```
yield
```
.
A regular return gives you no streaming, because it hands over a complete value and ends the function. Using temporary files does not help either, and the whole thing is definitely possible.
You pick WebSocket because it enables bidirectional real-time communication. It is not faster than HTTP, it is not more secure, and it very much does require a server holding the connection open.
The
```
/chat
```
endpoint is
```
@app.post
```
plus Pydantic models:
```
ChatRequest
```
on the way in,
```
ChatResponse
```
on the way out through
```
response_model
```
.
Before deploying an LLM to production, the priorities are testing, monitoring, and securing API keys in environment variables. Using the latest model without testing, disabling logging for speed, and exposing the API without rate limits are the three easiest ways to break production.
Caching reduces API costs and speeds up responses. It does not improve response quality, it is not required by law, and it is not the case that there are no benefits.

The next lesson covers the ReAct pattern, in which an agent alternates between reasoning and acting, repeating the Thought - Action - Observation cycle. For now, remember this: deployment is not the last commit, it is the first day when strangers spend your key on your mistakes.

Go to CodeWorlds→