A model call is just another dependency. It is slower than your database, less reliable than your database, priced per token, and it can return something syntactically perfect and factually wrong. Every design decision in this post follows from those four properties.
I have shipped full stack Python AI projects for teams in the US, UK and UAE: CV screening, multi-tenant WhatsApp SaaS, natural-language lead generation, clinical decision support, computer vision APIs. The components barely change between them. What changes is the plumbing around the request path once a language model is inside it, and that is where most builds go wrong.
What stack do full stack Python AI projects actually run on?
A deliberately unremarkable one. FastAPI for the API, Pydantic for every boundary, Celery with Redis for long work, Postgres with pgvector for rows and embeddings, React with TypeScript on the front end, Docker for build parity, and CI that runs more than unit tests. The interesting engineering is not in choosing these; it is in deciding where the model call sits relative to them.
The reference stack, and what each layer has to do differently once AI is involved:
| Layer | Choice | Why this one | What AI changes about it |
|---|---|---|---|
| API | FastAPI | Async-native, typed handlers, OpenAPI for free | Must support streaming responses and per-route timeout budgets |
| Contracts | Pydantic v2 | One validation layer for HTTP, queue payloads and config | Now validates model output too, not just inbound requests |
| Background work | Celery + Redis | Retries, time limits, scheduled jobs, visible queues | Most model work belongs here rather than in a handler |
| Data and retrieval | Postgres + pgvector | One store, one backup, one transaction boundary | Embeddings live in the same row as the record they describe |
| Front end | React + TypeScript | Types generated from the OpenAPI schema | Must render partial, streaming, cancellable state |
| Runtime | Docker | Same image locally, in CI and in production | Pin provider SDK versions as tightly as any other dependency |
| CI | Tests, lint, types, evals | Catches regressions before users do | Adds an eval job with its own token budget |
"The model is not your architecture. It is one slow, expensive, non-deterministic dependency that your architecture has to survive."
What changes when AI enters the request path?
Four assumptions break at once: latency moves from milliseconds to seconds, failure becomes partial rather than binary, output is non-deterministic so tests cannot assert equality, and every call carries a price. A CRUD app quietly depends on all four being false.
The four shifts, and what each one forces you to build:
- Latency: a single completion can take longer than a typical gateway timeout, so you need streaming or a queue, not a faster server.
- Partial failure: a stream can die halfway through a valid answer, so you need resumability or a clear discard-and-retry rule, plus idempotency on anything the answer triggers.
- Non-determinism: two identical requests give different text, so assertions move from equality to schema validity, invariants and scored evaluation.
- Cost: throughput now has a marginal price, so rate limits and caps become product features rather than infrastructure settings.
- Trust boundary: model output is untrusted input. Anything it produces that reaches SQL, a shell, an HTTP call or a customer needs validating first.
If the model is also choosing its own steps rather than answering once, the operational surface grows again, and I cover that separately in building production-ready AI agents with Python. This post assumes the simpler and far more common case: an application that makes model calls as part of otherwise normal features.
How do you handle timeouts on slow model calls?
Pick one of three shapes per endpoint: answer inline under a hard timeout, stream tokens so the connection stays alive, or hand the job to Celery and return a job ID. My rule is that anything that can plausibly exceed thirty seconds goes to the queue, no exceptions.
How I decide, per endpoint:
- 1Measure p95 latency with your real prompts and your real context sizes, not a hello-world call. Retrieval and long system prompts dominate.
- 2Under a few seconds with small output: answer inline, with an explicit SDK timeout set below your gateway timeout.
- 3Slow but user-facing: stream. The user sees progress in under a second even when the full answer takes twenty.
- 4Slow and not interactive, such as bulk enrichment, document ingestion or scoring: Celery task, job row in Postgres, webhook or poll when done.
- 5Set timeouts as a descending chain: provider SDK timeout below Celery soft time limit, below Celery hard time limit, below whatever the client is willing to wait.
Celery's task documentation says task functions should ideally be idempotent, and presents acks_late as an option for tasks that are, because a terminated worker process can cause the same task to run again. With model calls that matter twice over: a retried task is a second bill and a second side effect. I key every AI task on a deterministic idempotency key derived from the input, and store the result against that key before doing anything visible with it.
Why does Pydantic matter more once a model generates your data?
Because the model has become an input source you do not control, producing data you intend to store. Pydantic is where you catch it. The same discipline you already apply to request bodies applies to completions, and the failure mode without it is a plausible-looking record with a hallucinated enum value.
import json
from typing import Literal
from pydantic import BaseModel, Field, ValidationError
class LeadScore(BaseModel):
tier: Literal['hot', 'warm', 'cold']
score: int = Field(ge=0, le=100)
reason: str = Field(max_length=280)
evidence_ids: list[str] = Field(default_factory=list, max_length=5)
def score_lead(client, payload: dict, attempts: int = 2) -> LeadScore:
schema = LeadScore.model_json_schema()
messages = [
{'role': 'system', 'content': 'Score the lead. Return JSON matching the schema.'},
{'role': 'user', 'content': json.dumps(payload)},
]
for attempt in range(attempts):
raw = call_model(client, messages, schema=schema, timeout=20)
try:
return LeadScore.model_validate_json(raw)
except ValidationError as exc:
messages.append({'role': 'assistant', 'content': raw})
messages.append({
'role': 'user',
'content': 'That failed validation: ' + exc.json() + '. Return corrected JSON only.',
})
raise ValueError('model could not produce a valid LeadScore')Three things are doing the work there. Literal instead of str means an invented tier fails loudly rather than landing in the database. The validation error is fed back to the model as text, which repairs most single-field mistakes on the second attempt. And the retry count is finite, so a confused model becomes an exception instead of a loop.
Provider-side constraints help too. OpenAI documents structured outputs that constrain generation to a JSON Schema you supply, which removes most shape errors before they reach you. Treat that as a first line of defence and keep the Pydantic validation anyway, because schema conformance is not the same as semantic correctness. Practical patterns for wiring that into a business system are in connecting the OpenAI API to your CRM and on the OpenAI integration page.
Do you need a vector database, or is Postgres with pgvector enough?
For nearly every build I have shipped, Postgres with pgvector is enough. One store, one backup, one migration path, and metadata filters live in the same query as the similarity search, which is exactly what multi-tenant applications need.
What pgvector gives you, per its documentation:
- Two index types: HNSW, with better query performance but slower builds and higher memory, and IVFFlat, with faster builds and lower memory but lower query performance.
- Distance operators for L2, cosine, inner product, L1 and binary Hamming or Jaccard, so you match whatever your embedding model was trained for.
- Ordinary SQL around it, so tenant_id, document version and updated_at filters compose with the vector search rather than fighting it.
- Your existing transactions, backups, replicas and permissions, instead of a second system with its own operational story.
Move to a dedicated vector service when you genuinely outgrow it: very large corpora where index build time dominates, or sustained query volumes where you want that workload isolated from your transactional database. Below that, a separate vector store is one more thing to keep in sync and one more place for stale documents to hide. The harder problem is almost always the ingestion pipeline, which is why chunking, deduplication and re-embedding on change belong in a proper data engineering design rather than in a helper function. The Airflow pipeline behind my analytics build exists for exactly that reason.
How do you stream tokens to a React and TypeScript front end?
Server-sent events over a FastAPI StreamingResponse, read on the client with fetch and a ReadableStream reader. WebSockets are more machinery than a one-directional token stream needs, and SSE survives ordinary HTTP infrastructure better.
import json
from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
app = FastAPI()
async def token_stream(request: Request, prompt: str):
usage = {}
async for chunk in call_model_stream(prompt):
if await request.is_disconnected():
break
if chunk.delta:
yield 'data: ' + json.dumps({'delta': chunk.delta}) + '\n\n'
if chunk.usage:
usage = chunk.usage
yield 'event: done\ndata: ' + json.dumps({'usage': usage}) + '\n\n'
@app.post('/api/answer')
async def answer(request: Request, prompt: str):
return StreamingResponse(
token_stream(request, prompt),
media_type='text/event-stream',
headers={'Cache-Control': 'no-cache', 'X-Accel-Buffering': 'no'},
)The details that break streaming in production:
- Proxy buffering. Nginx and several hosted proxies buffer responses by default and your stream arrives as one lump; X-Accel-Buffering set to no is the usual fix.
- Client disconnect. FastAPI's documentation warns that an async generator only cancels at an await point, so check for disconnects inside the loop or you will keep paying for tokens nobody reads.
- A terminal event. Send a final event carrying token usage, a message ID and any citations, so the front end can persist a complete record rather than reassembling one from deltas.
- Idle heartbeats. A comment line every fifteen seconds keeps intermediaries from closing a slow stream.
- Front-end state. Model the UI as idle, streaming, complete or failed. Half-rendered markdown and an abort button are part of the feature, not polish.
How do you keep AI costs from surprising you?
Cost equals tokens per call multiplied by calls per user action multiplied by volume. Log token usage per request from the first commit, attach it to a tenant and a feature, and enforce caps in code rather than relying on a billing alert that arrives after the damage.
Where spend actually comes from, and the lever that moves it:
| Driver | What controls it | Highest-leverage fix |
|---|---|---|
| Input tokens | System prompt plus retrieved context plus history | Retrieve fewer, shorter chunks; trim history aggressively |
| Output tokens | Verbosity and max token limits | Cap max output; request structured JSON instead of prose |
| Calls per action | Retries, re-ranking, validation repairs, agent steps | Cache by input hash; cap retries at two |
| Embeddings | Corpus size and how often you re-embed | Store a content hash and re-embed only on change |
| Model tier | Which model each route uses | Route easy cases to a smaller model, escalate on low confidence |
Providers price per million input and output tokens, usually with cached input billed lower than fresh input, and those rates change often enough that any figure in a blog post is stale. Check the official pricing page for the model you have chosen, then multiply by token counts you measured in your own logs. That measured number, not the headline rate, is what you build a margin on.
What do evals and CI look like for full stack Python AI projects?
Three tiers. Fast unit tests with the provider mocked, a small contract test that hits the real API, and an eval suite of real cases scored against known-good outcomes that runs nightly or on prompt changes. Without the third tier, a prompt edit is an unreviewed production change.
The pipeline I set up on full stack Python AI projects:
- 1Unit and integration tests with the model client mocked, so the suite stays fast and free and runs on every push.
- 2Schema tests asserting that every Pydantic output model rejects the malformed shapes you have actually seen in logs.
- 3One live contract test per provider with a tiny prompt, catching SDK upgrades and auth breakage without a meaningful bill.
- 4An eval job over twenty to fifty real historical cases, scoring field-level accuracy deterministically and reserving an LLM judge for open-ended text only.
- 5A cost assertion in the eval job: fail the build if median tokens per case rises beyond an agreed threshold.
- 6Version the prompt in the repository alongside code, so a regression can be bisected like any other change.
Everything ships in Docker with the provider SDK version pinned. Model deprecations and SDK changes are a real source of breakage, and a pinned image plus a nightly contract test is how you find out on your schedule rather than during a customer demo.
What would I cut from a first build?
Most of it. The first version of nearly every AI feature I have shipped was one FastAPI endpoint, one Pydantic model, one Postgres table and a Celery task, with no framework in between.
Defer until something concrete forces the change:
- Orchestration frameworks, until you have a multi-step flow that plain functions genuinely cannot express.
- A dedicated vector database, until pgvector shows measured pain rather than theoretical limits.
- Multi-provider abstraction layers, until you have a real second provider and a real reason.
- Fine-tuning, until prompt and retrieval work have both plateaued against your eval set.
- Streaming, on anything that is not user-facing. A job ID and a webhook are simpler and more reliable.
The pattern held on AgentFlow, a visual AI agent builder, where the visual layer was the last thing built rather than the first, and on a YOLO-based detection API, where the queue and the result store mattered more than the model wrapper. Good full stack Python AI projects look like good Python projects with one awkward dependency handled carefully. If you want that shape built properly the first time, that is what my full-stack Python development work is for.
Key takeaways
- The reference stack for full stack Python AI projects is FastAPI, Pydantic, Celery, Postgres with pgvector, React with TypeScript, Docker and CI.
- Any model call that can exceed roughly thirty seconds belongs in a Celery task with an idempotency key, not in a request handler.
- Model output is untrusted input; validate it with Pydantic and feed validation errors back for a bounded number of repair attempts.
- Postgres with pgvector removes an entire service from most retrieval designs and keeps tenant filters in the same query as the similarity search.
- Stream user-facing answers with server-sent events, and disable proxy buffering or the stream arrives as a single block.
- CI needs an eval tier over real historical cases, plus a token-budget assertion, or prompt edits ship unreviewed.
Frequently asked questions
- What is the best Python stack for AI projects in 2026?
- FastAPI, Pydantic v2, Celery with Redis, Postgres with the pgvector extension, and a React plus TypeScript front end, all in Docker. It is deliberately conventional. The parts that need thought are streaming, background execution, output validation and cost logging, none of which a framework choice solves for you.
- Should AI calls go in a FastAPI request handler or a Celery task?
- Inline in the handler only when p95 latency is a few seconds and the output is small. Stream when it is slow but user-facing. Otherwise use Celery, return a job ID, and notify by webhook or polling. Celery's docs also recommend idempotent tasks before enabling acks_late.
- Is pgvector good enough for production RAG, or do I need a vector database?
- pgvector is good enough for most production retrieval. It supports HNSW and IVFFlat indexes and multiple distance operators, and it lets you filter by tenant, version and date in the same SQL query. Move to a dedicated service only when index build time or query volume becomes a measured problem.
- How much does it cost to run an AI feature in a Python app?
- Cost equals tokens per call times calls per user action times volume. Providers bill per million input and output tokens, with cached input usually cheaper, and rates change frequently. Check the official pricing page, then multiply by token counts measured from your own request logs rather than estimates.
- How do you test non-deterministic AI output in CI?
- Replace equality assertions with three tiers: mocked unit tests for logic, schema tests proving your Pydantic models reject malformed output, and an eval suite over real historical cases scored on field-level accuracy. Add a token-budget assertion so a prompt change that doubles cost fails the build.
- What is the difference between streaming and background jobs for LLM responses?
- Streaming keeps one HTTP connection open and pushes tokens as they generate, so the user sees progress immediately. A background job returns a job ID and completes asynchronously. Stream when a human is waiting on the text; queue when the work is bulk, long, or nobody is watching the screen.
- Can I add AI to an existing Python web application without rewriting it?
- Yes, in most cases. Add a Pydantic model for the output, put the call in a Celery task or a streaming endpoint, store results in your existing Postgres database, and log token usage per request. The invasive parts are timeout budgets and cost caps, not the framework you already run.
- Do I need LangChain or a similar framework to build AI features?
- Not for a first build. A provider SDK, Pydantic and a Celery task cover single-call features, which is most of them. Frameworks earn their place once you need durable multi-step orchestration, retries across steps and shared state. Adopt one when you feel the specific problem it solves.
Sources
Written by Syed Husnain Haider Bukhari
AI engineer, data scientist, and founder of Revolutionary Technologies LLC. Ships production AI agents, automations, and data platforms for teams in the US, UK, and UAE — including AgentFlow, AI Walay, and ProLeads.
Get in touch →Related pages
Want this built instead of researched?
Book a 30-minute scoping call. You get a one-page plan and a fixed-scope quote within 48 hours.
Start a conversation