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 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.

1# ===========================================
2# Safari Lab: FastAPI + LLM Integration
3# ===========================================
4# Build a REST API that connects to an LLM
5# to serve safari wildlife information.
6
7from fastapi import FastAPI, HTTPException
8from pydantic import BaseModel
9from openai import OpenAI
10
11app = FastAPI(title="Safari AI Guide API")
12client = OpenAI(api_key="your-api-key-here")
13
14# --- Request/Response models ---
15class AnimalQuery(BaseModel):
16    animal: str
17    question: str
18    max_tokens: int = 150
19
20class GuideResponse(BaseModel):
21    animal: str
22    answer: str
23    tokens_used: int
24
25# --- Safari guide endpoint ---
26@app.post("/ask-guide", response_model=GuideResponse)
27async def ask_safari_guide(query: AnimalQuery):
28    """Ask the safari guide AI about any animal."""
29    try:
30        response = client.chat.completions.create(
31            model="gpt-4",
32            messages=[
33                {
34                    "role": "system",
35                    "content": "You are Darwin, an expert safari guide. "
36                               "Give concise, factual answers about wildlife."
37                },
38                {
39                    "role": "user",
40                    "content": f"About the {query.animal}: {query.question}"
41                }
42            ],
43            max_tokens=query.max_tokens,
44            temperature=0.7
45        )
46
47        return GuideResponse(
48            animal=query.animal,
49            answer=response.choices[0].message.content,
50            tokens_used=response.usage.total_tokens
51        )
52    except Exception as e:
53        raise HTTPException(status_code=500, detail=str(e))
54
55# --- Health check endpoint ---
56@app.get("/health")
57async def health_check():
58    return {"status": "active", "service": "Safari AI Guide"}
59
60# --- List available species ---
61@app.get("/species")
62async def list_species():
63    return {
64        "species": [
65            "Lion", "Elephant", "Giraffe", "Zebra",
66            "Cheetah", "Hippo", "Rhino", "Leopard"
67        ]
68    }
69
70# TODO: Add an endpoint for batch animal queries
71# class BatchQuery(BaseModel):
72#     animals: list[str]
73#     question: str
74#
75# @app.post("/ask-guide/batch")
76# async def batch_query(query: BatchQuery):
77#     results = []
78#     for animal in query.animals:
79#         single_query = AnimalQuery(animal=animal, question=query.question)
80#         result = await ask_safari_guide(single_query)
81#         results.append(result)
82#     return {"results": results}
83
84# TODO: Add a streaming endpoint
85# from fastapi.responses import StreamingResponse
86#
87# @app.post("/ask-guide/stream")
88# async def stream_guide(query: AnimalQuery):
89#     async def generate():
90#         stream = client.chat.completions.create(
91#             model="gpt-4",
92#             messages=[{"role": "user", "content": query.question}],
93#             stream=True
94#         )
95#         for chunk in stream:
96#             if chunk.choices[0].delta.content:
97#                 yield chunk.choices[0].delta.content
98#     return StreamingResponse(generate(), media_type="text/plain")
99
100print("=== Safari AI Guide API ===")
101print("Endpoints:")
102print("  POST /ask-guide - Ask about an animal")
103print("  GET  /health    - Health check")
104print("  GET  /species   - List species")
105print("\nRun with: uvicorn main:app --reload")
106