Production Deployment - Deploying AI

Let's deploy AI to production! 🚀

FastAPI + LLM Service

1from fastapi import FastAPI, HTTPException, BackgroundTasks
2from fastapi.responses import StreamingResponse
3from pydantic import BaseModel
4from typing import AsyncGenerator
5from openai import AsyncOpenAI
6import asyncio
7import uvicorn
8
9app = FastAPI(title="Safari AI API")
10client = AsyncOpenAI()
11
12class ChatRequest(BaseModel):
13    message: str
14    session_id: str = "default"
15    temperature: float = 0.7
16    stream: bool = False
17
18class ChatResponse(BaseModel):
19    response: str
20    tokens_used: int
21
22# Simple endpoint
23@app.post("/chat", response_model=ChatResponse)
24async def chat(request: ChatRequest):
25    response = await client.chat.completions.create(
26        model="gpt-5",
27        messages=[
28            {"role": "system", "content": "You are a Safari expert."},
29            {"role": "user", "content": request.message}
30        ],
31        temperature=request.temperature
32    )
33
34    return ChatResponse(
35        response=response.choices[0].message.content,
36        tokens_used=response.usage.total_tokens
37    )
38
39# Streaming endpoint
40@app.post("/chat/stream")
41async def chat_stream(request: ChatRequest):
42    async def generate() -> AsyncGenerator[str, None]:
43        stream = await client.chat.completions.create(
44            model="gpt-5",
45            messages=[{"role": "user", "content": request.message}],
46            stream=True
47        )
48
49        async for chunk in stream:
50            content = chunk.choices[0].delta.content
51            if content:
52                yield f"data: {content}\n\n"
53
54        yield "data: [DONE]\n\n"
55
56    return StreamingResponse(generate(), media_type="text/event-stream")

WebSocket Chat

1from fastapi import WebSocket, WebSocketDisconnect
2from typing import Dict, List
3import json
4
5# Connection manager
6class ConnectionManager:
7    def __init__(self):
8        self.active_connections: Dict[str, WebSocket] = {}
9
10    async def connect(self, websocket: WebSocket, client_id: str):
11        await websocket.accept()
12        self.active_connections[client_id] = websocket
13
14    def disconnect(self, client_id: str):
15        if client_id in self.active_connections:
16            del self.active_connections[client_id]
17
18    async def send_message(self, message: str, client_id: str):
19        if client_id in self.active_connections:
20            await self.active_connections[client_id].send_text(message)
21
22manager = ConnectionManager()
23
24@app.websocket("/ws/{client_id}")
25async def websocket_endpoint(websocket: WebSocket, client_id: str):
26    await manager.connect(websocket, client_id)
27
28    try:
29        while True:
30            data = await websocket.receive_text()
31            message = json.loads(data)
32
33            # Stream response
34            stream = await client.chat.completions.create(
35                model="gpt-5",
36                messages=[{"role": "user", "content": message["content"]}],
37                stream=True
38            )
39
40            async for chunk in stream:
41                content = chunk.choices[0].delta.content
42                if content:
43                    await manager.send_message(
44                        json.dumps({"type": "chunk", "content": content}),
45                        client_id
46                    )
47
48            await manager.send_message(
49                json.dumps({"type": "done"}),
50                client_id
51            )
52
53    except WebSocketDisconnect:
54        manager.disconnect(client_id)

Rate Limiting and Caching

1from fastapi import Depends
2from slowapi import Limiter
3from slowapi.util import get_remote_address
4from cachetools import TTLCache
5
6# Rate limiting
7limiter = Limiter(key_func=get_remote_address)
8
9# Cache
10response_cache = TTLCache(maxsize=1000, ttl=3600)
11
12def get_cache_key(message: str) -> str:
13    import hashlib
14    return hashlib.md5(message.encode()).hexdigest()
15
16@app.post("/chat/cached")
17@limiter.limit("10/minute")
18async def chat_cached(request: ChatRequest):
19    cache_key = get_cache_key(request.message)
20
21    # Check cache
22    if cache_key in response_cache:
23        return {"response": response_cache[cache_key], "cached": True}
24
25    # Generate response
26    response = await client.chat.completions.create(
27        model="gpt-5",
28        messages=[{"role": "user", "content": request.message}]
29    )
30
31    result = response.choices[0].message.content
32    response_cache[cache_key] = result
33
34    return {"response": result, "cached": False}

Docker Deployment

1# Dockerfile
2FROM python:3.11-slim
3
4WORKDIR /app
5
6COPY requirements.txt .
7RUN pip install --no-cache-dir -r requirements.txt
8
9COPY . .
10
11EXPOSE 8000
12
13CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

1# docker-compose.yml
2version: '3.8'
3
4services:
5  api:
6    build: .
7    ports:
8      - "8000:8000"
9    environment:
10      - OPENAI_API_KEY=${OPENAI_API_KEY}
11    restart: unless-stopped

Health Check and Monitoring

1import time
2from datetime import datetime
3
4# Health check
5@app.get("/health")
6async def health_check():
7    return {
8        "status": "healthy",
9        "timestamp": datetime.utcnow().isoformat(),
10        "version": "1.0.0"
11    }
12
13# Metrics
14from prometheus_client import Counter, Histogram
15import prometheus_client
16
17requests_total = Counter('requests_total', 'Total requests')
18request_duration = Histogram('request_duration_seconds', 'Request duration')
19
20@app.middleware("http")
21async def metrics_middleware(request, call_next):
22    requests_total.inc()
23    start = time.time()
24    response = await call_next(request)
25    request_duration.observe(time.time() - start)
26    return response
27
28@app.get("/metrics")
29async def metrics():
30    return prometheus_client.generate_latest()