configuration, and fault tolerance.

Step 1: Environment Initialization

Install the required dependencies. The stack relies on lightweight, well-maintained libraries optimized for local and containerized deployments.

pip install anthropic chromadb sentence-transformers pypdf2 python-dotenv

Step 2: Document Ingestion & Semantic Chunking

Arbitrary text splitting destroys semantic continuity. The ingestion module must parse documents, extract raw text, and split content using sentence-aware boundaries with configurable overlap. Overlap prevents context loss at chunk edges, which is critical for maintaining coherence during retrieval.

import re
from pathlib import Path
from typing import List, Dict
import PyPDF2

class DocumentIngestor:
    def __init__(self, chunk_size: int = 400, overlap: int = 50):
        self.chunk_size = chunk_size
        self.overlap = overlap

    def extract_text(self, file_path: Path) -> str:
        if file_path.suffix.lower() == ".pdf":
            with open(file_path, "rb") as f:
                reader = PyPDF2.PdfReader(f)
                return "\n".join(page.extract_text() or "" for page in reader.pages)
        return file_path.read_text(encoding="utf-8")

    def split_into_segments(self, raw_text: str) -> List[str]:
        sentences = re.split(r'(?<=[.!?]) +', raw_text)
        segments, current = [], []
        current_len = 0

        for sentence in sentences:
            word_count = len(sentence.split())
            if current_len + word_count > self.chunk_size and current:
                segments.append(" ".join(current))
                overlap_words = current[-self.overlap:] if len(current) > self.overlap else current
                current = overlap_words
                current_len = len(" ".join(current).split())
            current.append(sentence)
            current_len += word_count

        if current:
            segments.append(" ".join(current))
        return segments

    def process_files(self, paths: List[Path]) -> List[Dict]:
        results = []
        for p in paths:
            if not p.exists():
                continue
            raw = self.extract_text(p)
            chunks = self.split_into_segments(raw)
            for idx, chunk in enumerate(chunks):
                results.append({
                    "id": f"{p.stem}_seg_{idx}",
                    "content": chunk,
                    "metadata": {"source_file": p.name, "total_chunks": len(chunks)}
                })
        return results

Architecture Rationale: The DocumentIngestor class encapsulates parsing and splitting logic. Sentence-aware splitting preserves grammatical boundaries, reducing semantic fragmentation. Overlap is calculated dynamically based on word count rather than fixed character offsets, ensuring consistent context preservation across varying document densities. Metadata propagation guarantees traceability during retrieval.

Step 3: Vector Index Construction

ChromaDB provides a persistent, embeddable vector store that eliminates the need for external database infrastructure during development and small-scale production. The embedding function is injected directly into the collection, ensuring consistent vectorization across indexing and querying phases.

import chromadb
from chromadb.utils import embedding_functions
from typing import List, Dict

class VectorRepository:
    def __init__(self, storage_path: str = "./vector_store"):
        self.client = chromadb.PersistentClient(path=storage_path)
        self.embedder = embedding_functions.SentenceTransformerEmbeddingFunction(
            model_name="all-MiniLM-L6-v2"
        )

    def upsert_collection(self, collection_name: str, items: List[Dict]) -> None:
        collection = self.client.get_or_create_collection(
            name=collection_name,
            embedding_function=self.embedder,
            metadata={"hnsw:space": "cosine"}
        )
        ids = [item["id"] for item in items]
        documents = [item["content"] for item in items]
        metadatas = [item["metadata"] for item in items]

        collection.upsert(ids=ids, documents=documents, metadatas=metadatas)
        print(f"Indexed {len(ids)} segments into '{collection_name}'")

Architecture Rationale: upsert replaces add to prevent duplicate insertion errors during incremental updates. The all-MiniLM-L6-v2 model is selected for its balance of inference speed and semantic accuracy on general-domain text. Cosine distance is explicitly configured as the similarity metric, which aligns with standard embedding normalization practices. Persistent storage ensures embeddings survive application restarts, eliminating redundant computation.

Step 4: Retrieval & Generation Orchestration

The orchestration layer handles query embedding, similarity search, context assembly, and LLM invocation. Prompt construction must enforce strict grounding to prevent model drift.

import anthropic
import os
from typing import List

class RAGOrchestrator:
    def __init__(self, vector_repo: VectorRepository, collection_name: str):
        self.repo = vector_repo
        self.collection_name = collection_name
        self.client = anthropic.Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))

    def retrieve_context(self, query: str, top_k: int = 4) -> List[str]:
        collection = self.repo.client.get_collection(
            name=self.collection_name,
            embedding_function=self.repo.embedder
        )
        results = collection.query(query_texts=[query], n_results=top_k)
        return results["documents"][0]

    def generate_response(self, query: str, context_segments: List[str]) -> str:
        formatted_context = "\n\n---\n\n".join(context_segments)
        prompt = (
            "You are a technical assistant. Answer the user's question using ONLY the provided context. "
            "If the context does not contain sufficient information, state that explicitly. "
            "Do not invent facts or reference external knowledge.\n\n"
            f"Context:\n{formatted_context}\n\n"
            f"Question: {query}"
        )
        response = self.client.messages.create(
            model="claude-sonnet-4-5",
            max_tokens=1024,
            messages=[{"role": "user", "content": prompt}]
        )
        return response.content[0].text

    def process_query(self, query: str) -> str:
        context = self.retrieve_context(query)
        return self.generate_response(query, context)

Architecture Rationale: The orchestrator separates retrieval from generation, enabling independent scaling and testing. The prompt template enforces strict grounding constraints, which significantly reduces hallucination rates. claude-sonnet-4-5 is invoked with a controlled token limit to manage cost and output length. Environment variable injection for the API key follows security best practices, preventing credential leakage in version control.

Pitfall Guide

Production RAG systems fail when engineering teams optimize for prototype speed rather than retrieval reliability. The following pitfalls represent the most common failure modes observed in enterprise deployments.

1. Context Window Saturation

Explanation: Retrieving too many chunks or feeding oversized segments into the prompt exhausts the model's context window, causing truncation or degraded attention distribution. Fix: Implement token-aware chunking and dynamic context truncation. Count tokens before prompt assembly and cap retrieval at a safe threshold (typically 4–6 segments for standard prompts).

2. Semantic Drift from Arbitrary Splitting

Explanation: Splitting text at fixed character or word boundaries severs sentences mid-thought, creating fragments that lack semantic completeness. Fix: Use sentence-aware or paragraph-aware splitting with configurable overlap. Preserve grammatical boundaries to ensure each chunk carries independent meaning.

3. Embedding Model Mismatch

Explanation: General-purpose embedding models underperform on highly specialized domains (e.g., legal contracts, medical records, internal codebases). Fix: Evaluate domain-specific embeddings or fine-tune a lightweight model on representative queries and documents. Validate retrieval accuracy using a labeled test set before production deployment.

4. Metadata Blindness

Explanation: Losing source attribution during chunking prevents filtering, auditing, and user trust verification. Fix: Propagate metadata through every pipeline stage. Store source file, section headers, and chunk indices. Enable metadata-aware queries to restrict search scope dynamically.

5. Synchronous Blocking in Web Services

Explanation: Tying request threads to LLM API calls causes thread pool exhaustion under concurrent load, degrading system responsiveness. Fix: Implement asynchronous I/O or background task queues (e.g., Celery, ARQ). Decouple retrieval and generation into non-blocking workflows with proper timeout handling.

6. Ignoring Retrieval Quality Validation

Explanation: Assuming top-k similarity search is sufficient leads to poor answers when semantic distance does not correlate with factual relevance. Fix: Introduce a cross-encoder reranker in production pipelines. Rerankers score query-chunk pairs directly, significantly improving precision at the cost of additional compute.

7. Hardcoded Credentials and Configuration

Explanation: Embedding API keys, file paths, and model names directly in source code creates security vulnerabilities and deployment friction. Fix: Externalize all configuration using environment variables, .env files, or secret managers. Validate required variables at startup and fail fast if missing.

Production Bundle

Action Checklist

• Verify dependency versions and lockfile integrity before deployment
• Configure environment variables for API keys and storage paths
• Test chunking logic against representative document samples
• Validate retrieval accuracy using a curated query-test set
• Implement token counting and context truncation safeguards
• Add structured logging for retrieval latency and generation success rates
• Containerize the application with volume mounts for persistent vector storage
• Set up monitoring alerts for API rate limits and embedding failures

Decision Matrix

Scenario	Recommended Approach	Why	Cost Impact
Internal knowledge base (<10k docs)	ChromaDB + all-MiniLM-L6-v2	Low operational overhead, fast local indexing	Minimal (compute only)
Customer support with high concurrency	Async orchestration + Redis cache for embeddings	Prevents thread exhaustion, reduces redundant API calls	Moderate (infrastructure + caching)
Legal/medical compliance	Cross-encoder reranker + strict metadata filtering	Ensures factual precision and auditability	Higher (reranker compute + validation)
Real-time analytics dashboard	Streaming retrieval + incremental upserts	Maintains freshness without full re-indexing	Low-Moderate (network + storage I/O)

Configuration Template

# rag_config.yaml
storage:
  vector_path: "./data/vector_store"
  collection_name: "enterprise_docs"

ingestion:
  chunk_size: 400
  overlap: 50
  supported_formats: [".pdf", ".txt", ".md"]

retrieval:
  top_k: 4
  similarity_metric: "cosine"

generation:
  model: "claude-sonnet-4-5"
  max_tokens: 1024
  temperature: 0.1
  grounding_enforcement: true

security:
  api_key_env: "ANTHROPIC_API_KEY"
  log_level: "INFO"

Quick Start Guide

1. Initialize the environment: Create a virtual environment, install dependencies, and set ANTHROPIC_API_KEY in your shell or .env file.
2. Prepare sample documents: Place PDF or text files in a ./docs directory. Ensure they contain domain-specific content relevant to your use case.
3. Run the indexing pipeline: Execute the DocumentIngestor and VectorRepository modules to parse, chunk, and embed your files. Verify the vector store directory is populated.
4. Test retrieval and generation: Instantiate RAGOrchestrator, submit a test query, and validate that the response references only the provided context. Adjust top_k or chunk size if answers appear fragmented or overly verbose.
5. Deploy: Wrap the orchestrator in a lightweight web framework (FastAPI/Flask), containerize with Docker, and mount the vector storage volume for persistence across restarts.