Step-by-Step Guide to Building and Deploying a Production RAG Model: From Document Chunking to Cloud API

Building a toy RAG prototype in a Jupyter notebook takes 20 lines of code. However, engineering a production-grade RAG system capable of sub-second response times, multi-tenant security isolation, zero hallucination rates, and scalable cloud deployment requires a disciplined end-to-end software architecture.

In this step-by-step engineering handbook, we will design, code, containerize, and deploy a robust RAG service from absolute scratch using Python, LangChain, ChromaDB, Sentence-Transformers, FastAPI, and Docker.

account_tree The 3-Tier Production RAG Architecture

Before writing any code, it is essential to visualize the complete data pipeline. A resilient production RAG system separates into three autonomous stages:

terminal Step 1: Setting Up the Python Engineering Environment

Create an isolated Python 3.11+ environment and install the verified core dependencies. We will use LangChain for modular orchestration, ChromaDB as an embedded vector database, sentence-transformers for lightweight high-speed local embeddings, and FastAPI for our microservice layer.

# Create and activate a clean virtual environment
python -m venv venv
source venv/bin/activate  # On Windows: .\venv\Scripts\activate

# Install essential RAG packages
pip install --upgrade pip
pip install \
    langchain==0.2.14 \
    langchain-community==0.2.12 \
    chromadb==0.5.5 \
    sentence-transformers==3.0.1 \
    pypdf==4.3.1 \
    fastapi==0.112.1 \
    uvicorn[standard]==0.30.6 \
    pydantic==2.8.2 \
    python-multipart==0.0.9

Organize your production project structure to maintain strict separation of concerns:

rag-production-service/
├── app/
│   ├── __init__.py
│   ├── main.py              # FastAPI endpoints and lifecycle events
│   ├── config.py            # Environment settings and hyperparameters
│   ├── ingestion.py         # Document parsing and chunking logic
│   ├── vectorstore.py       # ChromaDB initialization and indexing
│   ├── retriever.py         # Semantic search and re-ranking pipeline
│   └── generator.py         # Prompt engineering and LLM inference
├── data/
│   └── sample_docs/         # Source enterprise documents (PDF/MD/TXT)
├── vector_db_data/          # Persistent ChromaDB storage directory
├── Dockerfile               # Production multi-stage container build
├── docker-compose.yml       # Containerized service orchestration
├── requirements.txt         # Pinned production dependencies
└── .env                     # Sensitive API keys and environment variables

content_cut Step 2: Document Ingestion & Context-Aware Semantic Chunking

The single most common reason production RAG pipelines fail is poor chunking. If chunks are too small (e.g., 100 characters), semantic context is fractured. If chunks are too large (e.g., 3,000 characters), embedding vectors become diluted with background noise, causing retrieval accuracy to plummet.

We employ a RecursiveCharacterTextSplitter configured with a chunk size of 750 tokens and an overlap of 100 tokens, respecting natural paragraph and sentence boundaries while injecting document metadata into every fragment.

# app/ingestion.py
import os
from typing import List
from langchain_community.document_loaders import PyPDFLoader, TextLoader, DirectoryLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_core.documents import Document

def load_documents_from_directory(dir_path: str) -> List[Document]:
    """
    Ingests all PDF, Markdown, and Text files from a target directory
    while automatically attaching file metadata.
    """
    documents = []
    if not os.path.exists(dir_path):
        os.makedirs(dir_path, exist_ok=True)
        return documents

    for filename in os.listdir(dir_path):
        filepath = os.path.join(dir_path, filename)
        if filename.endswith(".pdf"):
            loader = PyPDFLoader(filepath)
            documents.extend(loader.load())
        elif filename.endswith((".txt", ".md")):
            loader = TextLoader(filepath, encoding="utf-8")
            documents.extend(loader.load())
            
    print(f"[Ingestion] Loaded {len(documents)} raw document pages/files.")
    return documents

def chunk_documents(documents: List[Document], chunk_size: int = 750, chunk_overlap: int = 100) -> List[Document]:
    """
    Recursively splits raw text into semantically cohesive passages.
    Preserves document source paths and line markers in metadata.
    """
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=chunk_size,
        chunk_overlap=chunk_overlap,
        separators=["\n\n", "\n", " ", ""],
        length_function=len
    )
    
    chunked_docs = splitter.split_documents(documents)
    
    # Enrich metadata with unique chunk identifiers
    for idx, doc in enumerate(chunked_docs):
        doc.metadata["chunk_id"] = f"{doc.metadata.get('source', 'unknown')}_part_{idx}"
        
    print(f"[Chunking] Generated {len(chunked_docs)} semantic passages.")
    return chunked_docs

database Step 3: Dense Vector Embeddings & Persistent ChromaDB Indexing

Next, we transform textual passages into dense numerical vectors that capture semantic meaning. We use the open-source all-MiniLM-L6-v2 embedding model from Sentence-Transformers. It generates 384-dimensional dense vectors with remarkable cosine separation, sub-15ms inference latency on standard CPUs, and zero third-party API costs.

# app/vectorstore.py
import os
from typing import List
import chromadb
from chromadb.config import Settings
from langchain_community.embeddings import HuggingFaceEmbeddings
from langchain_community.vectorstores import Chroma
from langchain_core.documents import Document

CHROMA_PERSIST_DIR = os.getenv("CHROMA_PERSIST_DIR", "./vector_db_data")
COLLECTION_NAME = "enterprise_knowledge_base"

def get_embedding_function():
    """
    Instantiates an ultra-fast, local embedding model.
    Runs efficiently on standard server CPUs without requiring dedicated GPUs.
    """
    return HuggingFaceEmbeddings(
        model_name="sentence-transformers/all-MiniLM-L6-v2",
        model_kwargs={"device": "cpu"},
        encode_kwargs={"normalize_embeddings": True}
    )

def initialize_vector_store() -> Chroma:
    """
    Connects to the persistent ChromaDB database on disk.
    """
    os.makedirs(CHROMA_PERSIST_DIR, exist_ok=True)
    embeddings = get_embedding_function()
    
    vectorstore = Chroma(
        collection_name=COLLECTION_NAME,
        embedding_function=embeddings,
        persist_directory=CHROMA_PERSIST_DIR
    )
    return vectorstore

def index_chunks(vectorstore: Chroma, chunks: List[Document]) -> int:
    """
    Upserts chunked passages into ChromaDB with HNSW vector indexing.
    """
    if not chunks:
        return 0
    
    ids = [chunk.metadata["chunk_id"] for chunk in chunks]
    vectorstore.add_documents(documents=chunks, ids=ids)
    print(f"[VectorStore] Successfully indexed {len(chunks)} chunks into ChromaDB.")
    return len(chunks)

manage_search Step 4: Hybrid Semantic Retrieval & Cross-Encoder Re-Ranking

Standard vector search retrieves the top-K nearest neighbors based purely on cosine distance. However, cosine similarity can return passages that share general topical keywords but lack the exact factual answer.

In production, we implement a two-stage retrieval pipeline:

1. Coarse Retrieval: ChromaDB retrieves the top 10 candidate chunks via approximate nearest neighbor (ANN) search.
2. Cross-Encoder Re-Ranking: A secondary cross-encoder model jointly encodes the query and each candidate passage, scoring their factual relevance to select the top 3 highest-quality contexts.

# app/retriever.py
from typing import List, Tuple
from langchain_core.documents import Document
from langchain_community.vectorstores import Chroma
from sentence_transformers import CrossEncoder

class AdvancedRAGRetriever:
    def __init__(self, vectorstore: Chroma):
        self.vectorstore = vectorstore
        # Load high-accuracy cross-encoder for semantic reranking
        self.reranker = CrossEncoder("cross-encoder/ms-marco-MiniLM-L-6-v2")

    def retrieve(self, query: str, top_k: int = 3, initial_candidates: int = 10) -> List[Tuple[Document, float]]:
        """
        Retrieves top candidate chunks and reranks them using joint cross-attention.
        """
        # 1. First-Stage Retrieval: Vector Cosine Similarity
        candidate_docs = self.vectorstore.similarity_search(query, k=initial_candidates)
        
        if not candidate_docs:
            return []

        # 2. Second-Stage Scoring: Cross-Encoder Evaluation
        query_passage_pairs = [[query, doc.page_content] for doc in candidate_docs]
        rerank_scores = self.reranker.predict(query_passage_pairs)

        # 3. Pair documents with their reranker score and sort descending
        scored_docs = list(zip(candidate_docs, rerank_scores))
        scored_docs.sort(key=lambda x: x[1], reverse=True)

        # 4. Return top-K most relevant contexts
        top_results = scored_docs[:top_k]
        return top_results

psychology Step 5: Generator Orchestration & Hallucination Guardrails

The prompt template is the firewall of your generative system. If a prompt is ambiguous, the LLM will fall back on its pre-trained assumptions and invent facts. We design an immutable system prompt enforcing three strict business rules:

• Grounding: Answer only using the explicitly provided reference context.
• Fallback: If the retrieved passages do not contain the answer, state clearly: "The provided enterprise documentation does not contain sufficient information to answer this question."
• Attribution: Explicitly cite the document source and section for every claim.

# app/generator.py
import os
from typing import List, Tuple
from langchain_core.documents import Document

def format_context_with_citations(scored_docs: List[Tuple[Document, float]]) -> str:
    """
    Structures retrieved passages into formatted markdown blocks with source tags.
    """
    context_blocks = []
    for doc, score in scored_docs:
        source_name = os.path.basename(doc.metadata.get("source", "Unknown Document"))
        chunk_id = doc.metadata.get("chunk_id", "N/A")
        context_blocks.append(
            f"--- SOURCE: {source_name} | CHUNK: {chunk_id} (Relevance Score: {score:.2f}) ---\n"
            f"{doc.page_content.strip()}\n"
        )
    return "\n".join(context_blocks)

def build_grounded_prompt(query: str, context_text: str) -> str:
    """
    Assembles the final, hardened system prompt.
    """
    prompt = f"""You are JoelX Orbit's Enterprise AI Knowledge Assistant.
Answer the user's question with utmost technical accuracy, strictly adhering to the provided documentation below.

BUSINESS CONSTRAINTS:
1. ONLY utilize facts mentioned in the reference context.
2. If the context does not contain the answer, say: 'I cannot answer this question based on the verified documentation provided.' Do NOT extrapolate or fabricate facts.
3. Include brackets referencing the source document (e.g. [source: handbook.pdf]) for any key technical specification you mention.

=== VERIFIED REFERENCE CONTEXT ===
{context_text}
==================================

USER QUESTION:
{query}

FINAL DETAILED ANSWER:"""
    return prompt

api Step 6: Building the Production REST API with FastAPI

We encapsulate the entire RAG pipeline inside a high-throughput, asynchronous FastAPI microservice with Pydantic request validation, error boundaries, and health check probes.

# app/main.py
from fastapi import FastAPI, HTTPException, UploadFile, File, BackgroundTasks
from pydantic import BaseModel, Field
from typing import List, Optional
import os
import shutil

from app.vectorstore import initialize_vector_store, index_chunks
from app.ingestion import load_documents_from_directory, chunk_documents
from app.retriever import AdvancedRAGRetriever
from app.generator import format_context_with_citations, build_grounded_prompt

app = FastAPI(
    title="JoelX Orbit Enterprise RAG API",
    description="Production-grade Retrieval-Augmented Generation Microservice",
    version="1.0.0"
)

# Global runtime state
vectorstore = initialize_vector_store()
retriever = AdvancedRAGRetriever(vectorstore)

class QueryRequest(BaseModel):
    query: str = Field(..., min_length=3, max_length=500, example="What are the deployment steps for our API?")
    top_k: Optional[int] = Field(default=3, ge=1, le=10)

class SourceCitation(BaseModel):
    source: str
    relevance_score: float
    snippet: str

class QueryResponse(BaseModel):
    query: str
    answer: str
    citations: List[SourceCitation]
    latency_ms: float

@app.get("/health", tags=["Monitoring"])
async def health_check():
    """Kubernetes / Cloud Readiness & Liveness Probe."""
    return {"status": "healthy", "service": "rag-engine", "collection": "enterprise_knowledge_base"}

@app.post("/api/v1/query", response_model=QueryResponse, tags=["Inference"])
async def query_rag(request: QueryRequest):
    """
    Executes hybrid retrieval, reranking, and generation on the knowledge base.
    """
    import time
    start_time = time.time()
    
    try:
        # Retrieve and rerank top contexts
        top_docs = retriever.retrieve(request.query, top_k=request.top_k)
        
        if not top_docs:
            return QueryResponse(
                query=request.query,
                answer="No relevant documentation found in the knowledge base.",
                citations=[],
                latency_ms=round((time.time() - start_time) * 1000, 2)
            )

        context_str = format_context_with_citations(top_docs)
        prompt = build_grounded_prompt(request.query, context_str)
        
        # Here we mock or execute your preferred LLM call (e.g. Gemini, OpenAI, or local Ollama)
        simulated_llm_response = (
            f"Based on your enterprise documents, the system operates on a multi-stage architecture. "
            f"Refer to the cited sources below for comprehensive parameters."
        )

        citations = [
            SourceCitation(
                source=os.path.basename(doc.metadata.get("source", "doc")),
                relevance_score=round(score, 3),
                snippet=doc.page_content[:150] + "..."
            )
            for doc, score in top_docs
        ]

        total_time = round((time.time() - start_time) * 1000, 2)
        return QueryResponse(
            query=request.query,
            answer=simulated_llm_response,
            citations=citations,
            latency_ms=total_time
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"RAG Inference Error: {str(e)}")

@app.post("/api/v1/ingest", tags=["Data Pipeline"])
async def upload_and_ingest(file: UploadFile = File(...)):
    """
    Dynamically uploads a new document, parses, chunks, and indexes it into ChromaDB.
    """
    upload_dir = "./data/sample_docs"
    os.makedirs(upload_dir, exist_ok=True)
    file_location = os.path.join(upload_dir, file.filename)
    
    with open(file_location, "wb+") as file_object:
        shutil.copyfileobj(file.file, file_object)

    raw_docs = load_documents_from_directory(upload_dir)
    chunks = chunk_documents(raw_docs)
    indexed_count = index_chunks(vectorstore, chunks)

    return {
        "status": "success",
        "filename": file.filename,
        "indexed_chunks": indexed_count
    }

deployed_code Step 7: Production Containerization with Docker & Cloud Deployment

To ensure your RAG microservice runs identically in local staging, AWS ECS, Google Cloud Run, or on a dedicated VPS, we package it inside an optimized, multi-stage Dockerfile that runs under an unprivileged user for security compliance.

# Dockerfile - Multi-Stage Production Build
FROM python:3.11-slim AS builder

WORKDIR /app

# Install build tools for native compilation
RUN apt-get update && apt-get install -y --no-install-recommends \
    build-essential \
    curl \
    && rm -rf /var/lib/apt/lists/*

COPY requirements.txt .
RUN pip install --no-cache-dir --user -r requirements.txt

# Final Runtime Stage
FROM python:3.11-slim AS runner

WORKDIR /app

# Create unprivileged application user
RUN groupadd -g 1001 appgroup && \
    useradd -u 1001 -g appgroup -s /bin/bash appuser

# Copy installed Python packages from builder
COPY --from=builder /root/.local /home/appuser/.local
ENV PATH=/home/appuser/.local/bin:$PATH

# Copy application code
COPY --chown=appuser:appgroup . /app

# Ensure storage directories exist with proper write permissions
RUN mkdir -p /app/vector_db_data /app/data/sample_docs && \
    chown -R appuser:appgroup /app

USER appuser

EXPOSE 8000

HEALTHCHECK --interval=30s --timeout=5s --start-period=10s --retries=3 \
  CMD curl -f http://localhost:8000/health || exit 1

CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "2"]

Next, define a docker-compose.yml file to manage data volumes, environment variables, and automated service restarts:

version: '3.8'

services:
  rag-engine:
    build: .
    container_name: joelx_rag_engine
    restart: unless-stopped
    ports:
      - "8000:8000"
    environment:
      - CHROMA_PERSIST_DIR=/app/vector_db_data
      - PYTHONUNBUFFERED=1
    volumes:
      # Persistent vector database storage on host disk
      - ./vector_db_data:/app/vector_db_data
      - ./data/sample_docs:/app/data/sample_docs
    deploy:
      resources:
        limits:
          memory: 4G
          cpus: '2.0'

# 1. Build and launch container in detached mode
docker-compose up -d --build

# 2. Verify health and query API
curl http://localhost:8000/health
curl -X POST http://localhost:8000/api/v1/query \
     -H "Content-Type: application/json" \
     -d '{"query": "How do we initialize the vector store?", "top_k": 3}'

verified_user Step 8: Continuous RAG Evaluation (RAGAS) & Security Hardening

Deploying the code is only half the battle. In production, models degrade as documentation evolves. To prevent regression, integrate automated evaluation using the RAGAS (Retrieval Augmented Generation Assessment) framework: