FastAPI + OpenAI: Building a Production AI Chatbot

Calling client.chat.completions.create() is the easy 10% of building a chatbot backend. The other 90% — conversation state, streaming to the client, rate limiting abusive users, and controlling cost per session — is what separates a demo from something you can put in front of paying customers.

Conversation state: the part tutorials skip

An LLM API call is stateless. Every request needs the full conversation history sent as context, which means your backend owns session state, not the model.

# app/chat/session_store.py
class SessionStore:
    async def get(self, session_id: str) -> list[dict]:
        row = await db.get(ChatSession, session_id)
        return row.messages if row else []

    async def append(self, session_id: str, message: dict):
        session = await db.get_or_create(ChatSession, session_id)
        session.messages.append(message)
        await db.commit()

Two things go wrong here in practice: unbounded history growth (every message ever sent gets re-sent as context, blowing up token costs) and no truncation strategy. Cap history at a token budget, not a message count — a summarization step for older turns is worth the complexity once conversations get long.

Streaming without blocking your event loop

FastAPI's StreamingResponse combined with OpenAI's streaming API gives you token-by-token output over SSE:

@router.post("/v1/chat/stream")
async def chat_stream(body: ChatRequest, key: APIKey = Depends(get_api_key)):
    history = await session_store.get(body.session_id)

    async def event_stream():
        total_tokens = 0
        stream = await openai_client.chat.completions.create(
            model="gpt-4o",
            messages=[*history, {"role": "user", "content": body.message}],
            stream=True,
        )
        async for chunk in stream:
            delta = chunk.choices[0].delta.content or ""
            total_tokens += 1  # approximate; reconcile with usage on stream end
            yield f"data: {delta}\n\n"
        await meter.record(key.id, total_tokens)
        await session_store.append(body.session_id, {"role": "user", "content": body.message})

    return StreamingResponse(event_stream(), media_type="text/event-stream")

The token count from streaming chunks is approximate — OpenAI's streaming API doesn't return exact usage until the final chunk in newer API versions. Reconcile with the real usage object when it's available rather than trusting a running chunk count for billing.

Rate limiting without punishing legitimate users

Per-key rate limits need two tiers: a per-minute burst limit (stops abuse and runaway client bugs) and a per-day cap (controls your actual OpenAI bill exposure per customer).

@router.post("/v1/chat")
@require_api_key(tier=["free", "pro"])
@rate_limit(per_minute=10, per_day=200)
async def chat(body: ChatRequest, key: APIKey = Depends(get_api_key)):
    ...

Free tiers should have a meaningfully lower daily cap than paid tiers — not just to encourage upgrades, but because free-tier abuse (scraping, automation) is the most common source of unplanned OpenAI cost.

Controlling cost per session

Two levers matter most: model selection and context length. Route simple queries to a cheaper model and reserve GPT-4o for requests that actually need it.

def select_model(message: str, history_length: int) -> str:
    if history_length < 3 and len(message) < 200:
        return "gpt-4o-mini"  # cheaper, faster for simple exchanges
    return "gpt-4o"

This single routing decision often cuts LLM spend by 40-60% on chatbot workloads, since the majority of turns in most conversations are short and simple.

Handling provider errors gracefully

OpenAI's API has rate limits and occasional outages. A chatbot that returns a raw 500 on a transient OpenAI error looks broken to the end user, even though the failure is one API call away.

from openai import RateLimitError, APIError

async def chat_with_retry(messages: list[dict], max_retries: int = 3):
    for attempt in range(max_retries):
        try:
            return await openai_client.chat.completions.create(model="gpt-4o", messages=messages)
        except RateLimitError:
            await asyncio.sleep(2 ** attempt)
        except APIError:
            if attempt == max_retries - 1:
                raise
            await asyncio.sleep(1)

What FastAPI AI Kit includes

The kit's chatbot-relevant pieces ship pre-built: session persistence with conversation history, an SSE streaming endpoint, per-key rate limiting with configurable tiers, and automatic retry with backoff on the OpenAI provider. Token tracking feeds directly into Stripe metering if you're billing per conversation or per token.