Step-by-step: build a retrieval-augmented generation pipeline in FastAPI.
Retrieval-augmented generation (RAG) lets your AI API answer questions using your own documents instead of relying solely on the LLM's training data. This guide walks through building a complete RAG pipeline: document ingestion, embedding generation, vector storage with pgvector, and context-injected LLM answers.
Enable pgvector in PostgreSQL
Install the pgvector extension in your PostgreSQL database and create a table with a vector column for storing embeddings.
-- Enable the extension
CREATE EXTENSION IF NOT EXISTS vector;
-- Create embeddings table
CREATE TABLE document_chunks (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
collection VARCHAR(100) NOT NULL,
content TEXT NOT NULL,
embedding vector(1536), -- OpenAI ada-002 dimension
metadata JSONB DEFAULT '{}',
created_at TIMESTAMPTZ DEFAULT NOW()
);
CREATE INDEX ON document_chunks USING ivfflat (embedding vector_cosine_ops) WITH (lists = 100);Build the document ingestion pipeline
Parse documents into text, split into overlapping chunks, and store them for embedding. Support PDF, Markdown, and plain text formats.
def chunk_text(text: str, chunk_size: int = 512, overlap: int = 64) -> list[str]:
words = text.split()
chunks = []
for i in range(0, len(words), chunk_size - overlap):
chunk = " ".join(words[i:i + chunk_size])
if chunk:
chunks.append(chunk)
return chunks
async def ingest_document(file_path: str, collection: str, db: AsyncSession):
text = extract_text(file_path) # PDF/DOCX/MD parser
chunks = chunk_text(text)
embeddings = await generate_embeddings(chunks)
for chunk, embedding in zip(chunks, embeddings):
db.add(DocumentChunk(collection=collection, content=chunk, embedding=embedding))
await db.commit()Generate embeddings via OpenAI
Call OpenAI's embedding API to convert text chunks into vector representations. Batch chunks for efficiency and track embedding token costs.
from openai import AsyncOpenAI
client = AsyncOpenAI()
async def generate_embeddings(texts: list[str]) -> list[list[float]]:
response = await client.embeddings.create(
model="text-embedding-ada-002",
input=texts,
)
return [item.embedding for item in response.data]Implement similarity search
Query pgvector for the most similar document chunks to a user's question. Use cosine distance for relevance ranking and return top-k results.
async def search_similar(query: str, collection: str, top_k: int = 5, db: AsyncSession) -> list[str]:
query_embedding = (await generate_embeddings([query]))[0]
result = await db.execute(
text("""
SELECT content, 1 - (embedding <=> :embedding) AS similarity
FROM document_chunks
WHERE collection = :collection
ORDER BY embedding <=> :embedding
LIMIT :top_k
"""),
{"embedding": str(query_embedding), "collection": collection, "top_k": top_k},
)
return [row.content for row in result.fetchall()]Generate answers with context injection
Combine retrieved document chunks with the user's question in an LLM prompt. The LLM answers based on your documents, not just its training data.
async def rag_answer(question: str, collection: str, db: AsyncSession) -> str:
context_chunks = await search_similar(question, collection, db=db)
context = "\n\n".join(context_chunks)
response = await client.chat.completions.create(
model="gpt-4o",
messages=[
{"role": "system", "content": f"Answer based on this context:\n\n{context}"},
{"role": "user", "content": question},
],
)
return response.choices[0].message.contentSkip the setup, start building
Everything in this guide is already implemented and configured in FastAPI AI Kit. Clone the repo and start building your product features immediately.
Ready to ship your AI backend this weekend?
Join developers who skipped weeks of boilerplate and went straight to building.