Building a RAG API with FastAPI and pgvector

A fastapi rag api pgvector stack gives you a self-hosted retrieval-augmented generation backend without dragging in a separate vector database appliance. This tutorial builds a minimal but production-shaped service: ingest documents into Postgres, embed them with an OpenAI-compatible endpoint, and serve grounded answers over HTTP.

Prerequisites

• Python 3.11+ and pip
• Postgres 15+ with the pgvector extension installed (CREATE EXTENSION vector;)
• An OpenAI-compatible API key (or a gateway key — see note below)
• psql access to your database

Install the Python deps:

pip install fastapi uvicorn asyncpg openai pydantic numpy

If you point the embedding and chat client at n4n.ai, one OpenAI-compatible endpoint covers 240+ models and automatically falls back when a provider is rate-limited, so you avoid writing retry logic yourself.

Database schema

Connect to your DB and enable the extension. Then create a table that stores raw text and a 1536-dim embedding (matching text-embedding-3-small).

CREATE EXTENSION IF NOT EXISTS vector;

CREATE TABLE documents (
    id SERIAL PRIMARY KEY,
    content TEXT NOT NULL,
    embedding vector(1536)
);

-- Optional: speed up cosine search once you have data
-- CREATE INDEX ON documents USING ivfflat (embedding vector_cosine_ops) WITH (lists = 100);

Embedding client

Use the async OpenAI client so it doesn’t block the event loop. Set base_url to any compatible gateway.

from openai import AsyncOpenAI

# Swap base_url for your provider or gateway
client = AsyncOpenAI(
    base_url="https://api.openai.com/v1",
    api_key="sk-your-key"
)

EMBED_MODEL = "text-embedding-3-small"

async def embed(text: str) -> list[float]:
    resp = await client.embeddings.create(model=EMBED_MODEL, input=text)
    return resp.data[0].embedding

FastAPI app skeleton

We’ll use asyncpg for a connection pool. The app exposes two endpoints: /ingest and /query.

import asyncpg
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

app = FastAPI()
pool = None

@app.on_event("startup")
async def startup():
    global pool
    pool = await asyncpg.create_pool(
        dsn="postgresql://user:pass@localhost:5432/rag"
    )

Ingest endpoint

Chunking is deliberately naive (fixed size). For real corpora, use sentence-aware splitting.

class IngestRequest(BaseModel):
    text: str
    chunk_size: int = 500

@app.post("/ingest")
async def ingest(req: IngestRequest):
    chunks = [
        req.text[i:i+req.chunk_size]
        for i in range(0, len(req.text), req.chunk_size)
    ]
    async with pool.acquire() as conn:
        for chunk in chunks:
            vec = await embed(chunk)
            await conn.execute(
                "INSERT INTO documents (content, embedding) VALUES ($1, $2)",
                chunk,
                vec
            )
    return {"inserted": len(chunks)}

Test it with a short payload:

curl -X POST http://localhost:8000/ingest \
  -H "Content-Type: application/json" \
  -d '{"text":"Postgres is a relational database. pgvector adds vector similarity search."}'

Expected response:

{"inserted": 1}

Query endpoint

Retrieve top-3 by cosine distance, then call a chat model with the context stuffed into the system prompt.

class QueryRequest(BaseModel):
    question: str
    top_k: int = 3

@app.post("/query")
async def query(req: QueryRequest):
    qvec = await embed(req.question)
    async with pool.acquire() as conn:
        rows = await conn.fetch(
            """
            SELECT content, 1 - (embedding <=> $1) AS score
            FROM documents
            ORDER BY embedding <=> $1
            LIMIT $2
            """,
            qvec,
            req.top_k
        )
    context = "\n---\n".join(r["content"] for r in rows)
    if not context:
        raise HTTPException(status_code=404, detail="No documents indexed")

    resp = await client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": f"Answer from context:\n{context}"},
            {"role": "user", "content": req.question}
        ]
    )
    return {
        "answer": resp.choices[0].message.content,
        "sources": [r["content"] for r in rows]
    }

Run the service:

uvicorn main:app --port 8000 --reload

Ask a question:

curl -X POST http://localhost:8000/query \
  -H "Content-Type: application/json" \
  -d '{"question":"What does pgvector add to Postgres?"}'

Expected shape (text may vary):

{
  "answer": "pgvector adds vector similarity search to Postgres.",
  "sources": ["Postgres is a relational database. pgvector adds vector similarity search."]
}

Production notes

The <=> operator computes cosine distance; 1 - distance is your similarity score. For datasets past ~100k rows, build an ivfflat index and tune lists to rows / 1000. Without an index, Postgres does a sequential scan on every query — fine for a demo, unacceptable at scale.

Don’t embed queries and documents with different models. Dimension mismatch throws at insert time, but a silent model swap breaks retrieval silently. Pin the embedding model in code and in your migration scripts.

For multi-tenant data, add a tenant_id column and filter in the ORDER BY query. pgvector doesn’t change Postgres’ row security model — use it.

Streaming the chat response is straightforward with FastAPI’s StreamingResponse and the OpenAI client’s stream=True. Do that before shipping to users; token-by-token output beats a 2-second blank wait.

Where the gateway helps

If you run this against a single provider, a 429 burns the whole request. An OpenAI-compatible gateway that honors client routing directives and forwards provider cache-control hints lets you set base_url once and get fallback for free. That’s the only place I’d add abstraction — the rest of the fastapi rag api pgvector code should stay boring and explicit.