Building a Production RAG System from Scratch

The Problem: Enterprise Knowledge Management

Our company had 100K+ documents scattered across wikis, Confluence, Google Docs, and Slack. Finding information took hours. We needed a system that could:

• Answer questions across all documents
• Maintain context and accuracy
• Scale to millions of documents
• Cost-effective (API budget: $500/month)

Solution: EnterpriseRAG

A production-grade RAG (Retrieval-Augmented Generation) system using:

• LangChain for orchestration
• Claude API for generation
• Qdrant for vector storage
• FastAPI for serving
• Redis for caching

---

Architecture Overview

┌──────────────┐
│  Document    │
│  Ingestion   │
└──────┬───────┘
       │
       ▼
┌──────────────┐     ┌──────────────┐
│  Chunking &  │────▶│   Embedding  │
│  Processing  │     │   Model      │
└──────────────┘     └──────┬───────┘
                            │
                            ▼
                     ┌──────────────┐
                     │   Vector DB  │
                     │   (Qdrant)   │
                     └──────┬───────┘
                            │
       ┌────────────────────┴────────────────────┐
       │                                         │
       ▼                                         ▼
┌──────────────┐                         ┌──────────────┐
│   Query      │                         │   Context    │
│   Processing │                         │   Retrieval  │
└──────┬───────┘                         └──────┬───────┘
       │                                        │
       └────────────────┬───────────────────────┘
                        ▼
                 ┌──────────────┐
                 │  Claude API  │
                 │  Generation  │
                 └──────┬───────┘
                        │
                        ▼
                 ┌──────────────┐
                 │   Response   │
                 │   + Sources  │
                 └──────────────┘

---

Implementation

1. Document Ingestion Pipeline

import asyncio
from langchain.document_loaders import (
    TextLoader,
    PDFLoader,
    ConfluenceLoader,
    SlackLoader
)
from langchain.text_splitter import RecursiveCharacterTextSplitter

class DocumentPipeline:
    def __init__(self):
        self.splitter = RecursiveCharacterTextSplitter(
            chunk_size=1000,
            chunk_overlap=200,
            separators=["\n\n", "\n", " ", ""]
        )
    
    async def ingest_documents(self, source_type, source_config):
        """Ingest documents from various sources"""
        
        # Load documents based on source
        if source_type == 'confluence':
            loader = ConfluenceLoader(**source_config)
        elif source_type == 'slack':
            loader = SlackLoader(**source_config)
        elif source_type == 'pdf':
            loader = PDFLoader(**source_config)
        else:
            loader = TextLoader(**source_config)
        
        # Load and split
        documents = await loader.aload()
        chunks = self.splitter.split_documents(documents)
        
        print(f"📄 Loaded {len(documents)} documents")
        print(f"📦 Split into {len(chunks)} chunks")
        
        return chunks
    
    async def process_batch(self, chunks, batch_size=100):
        """Process chunks in batches for efficiency"""
        batches = [chunks[i:i+batch_size] 
                  for i in range(0, len(chunks), batch_size)]
        
        tasks = [self.embed_batch(batch) for batch in batches]
        results = await asyncio.gather(*tasks)
        
        return [item for sublist in results for item in sublist]

2. Embedding Strategy

from langchain.embeddings import OpenAIEmbeddings
from qdrant_client import QdrantClient
from qdrant_client.models import Distance, VectorParams, PointStruct
import hashlib

class EmbeddingSystem:
    def __init__(self):
        self.embeddings = OpenAIEmbeddings(
            model="text-embedding-ada-002"
        )
        self.client = QdrantClient(host="localhost", port=6333)
        self.collection_name = "enterprise_docs"
        
        # Create collection if not exists
        self.setup_collection()
    
    def setup_collection(self):
        """Initialize vector collection"""
        try:
            self.client.create_collection(
                collection_name=self.collection_name,
                vectors_config=VectorParams(
                    size=1536,  # OpenAI ada-002 dimension
                    distance=Distance.COSINE
                )
            )
            print("✅ Vector collection created")
        except Exception as e:
            print(f"⚠️  Collection exists: {e}")
    
    async def embed_batch(self, chunks):
        """Embed a batch of chunks with deduplication"""
        points = []
        
        for i, chunk in enumerate(chunks):
            # Generate unique ID based on content
            chunk_id = hashlib.md5(
                chunk.page_content.encode()
            ).hexdigest()
            
            # Generate embedding
            vector = await self.embeddings.aembed_query(
                chunk.page_content
            )
            
            # Create point
            points.append(PointStruct(
                id=chunk_id,
                vector=vector,
                payload={
                    "text": chunk.page_content,
                    "metadata": chunk.metadata,
                    "source": chunk.metadata.get("source", "unknown")
                }
            ))
        
        # Upsert to Qdrant
        self.client.upsert(
            collection_name=self.collection_name,
            points=points
        )
        
        return points

3. Smart Retrieval

from langchain.vectorstores import Qdrant
import anthropic

class SmartRetriever:
    def __init__(self):
        self.vectorstore = Qdrant(
            client=QdrantClient(host="localhost", port=6333),
            collection_name="enterprise_docs",
            embeddings=OpenAIEmbeddings()
        )
        self.claude = anthropic.Anthropic()
    
    async def retrieve_context(self, query, k=5):
        """Retrieve relevant context with hybrid search"""
        
        # 1. Semantic search
        semantic_results = await self.vectorstore.asimilarity_search(
            query, k=k
        )
        
        # 2. Keyword boost (for technical terms)
        if any(term in query.lower() for term in ['api', 'error', 'bug']):
            keyword_results = await self.vectorstore.asimilarity_search(
                query, k=k, filter={"category": "technical"}
            )
            semantic_results.extend(keyword_results)
        
        # 3. Deduplicate and rank
        unique_docs = self.deduplicate(semantic_results)
        
        # 4. Re-rank with Claude (optional but powerful)
        reranked = await self.rerank_with_claude(query, unique_docs)
        
        return reranked[:k]
    
    async def rerank_with_claude(self, query, documents):
        """Use Claude to re-rank documents by relevance"""
        
        # Prepare ranking prompt
        docs_text = "\n\n".join([
            f"[{i}] {doc.page_content[:200]}..."
            for i, doc in enumerate(documents)
        ])
        
        message = self.claude.messages.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=1024,
            messages=[{
                "role": "user",
                "content": f"""Rank these documents by relevance to: "{query}"
                
Documents:
{docs_text}

Return ONLY the indices in order of relevance (e.g., "2,0,4,1,3")"""
            }]
        )
        
        # Parse ranking
        indices = [int(i.strip()) for i in message.content[0].text.split(',')]
        return [documents[i] for i in indices if i < len(documents)]
    
    def deduplicate(self, documents):
        """Remove duplicate documents"""
        seen = set()
        unique = []
        
        for doc in documents:
            content_hash = hashlib.md5(doc.page_content.encode()).hexdigest()
            if content_hash not in seen:
                seen.add(content_hash)
                unique.append(doc)
        
        return unique

4. Generation with Claude

class RAGGenerator:
    def __init__(self):
        self.claude = anthropic.Anthropic()
        self.retriever = SmartRetriever()
    
    async def answer_question(self, question, include_sources=True):
        """Generate answer with retrieved context"""
        
        # 1. Retrieve relevant context
        print(f"🔍 Searching for: {question}")
        context_docs = await self.retriever.retrieve_context(question, k=5)
        
        # 2. Build context string
        context = "\n\n".join([
            f"Source {i+1} ({doc.metadata.get('source', 'unknown')}):\n{doc.page_content}"
            for i, doc in enumerate(context_docs)
        ])
        
        # 3. Generate answer with Claude
        print(f"💭 Generating answer...")
        
        message = self.claude.messages.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=2048,
            messages=[{
                "role": "user",
                "content": f"""You are a helpful assistant answering questions about our company's documentation.

Context from our knowledge base:
{context}

Question: {question}

Instructions:
- Answer based ONLY on the provided context
- If the answer isn't in the context, say so
- Cite sources when making specific claims
- Be concise but complete

Answer:"""
            }]
        )
        
        answer = message.content[0].text
        
        # 4. Add source citations
        if include_sources:
            sources = [
                {
                    "title": doc.metadata.get("title", "Untitled"),
                    "source": doc.metadata.get("source", "unknown"),
                    "url": doc.metadata.get("url", None)
                }
                for doc in context_docs
            ]
            
            return {
                "answer": answer,
                "sources": sources,
                "confidence": self.calculate_confidence(context_docs, answer)
            }
        
        return {"answer": answer}
    
    def calculate_confidence(self, docs, answer):
        """Calculate confidence score based on context quality"""
        
        # Simple heuristic: more sources + longer context = higher confidence
        num_sources = len(docs)
        avg_length = sum(len(doc.page_content) for doc in docs) / len(docs)
        
        confidence = min(
            (num_sources / 5) * 0.5 +  # Max 50% from number of sources
            (min(avg_length, 1000) / 1000) * 0.5,  # Max 50% from context length
            1.0
        )
        
        return round(confidence * 100)

5. Caching for Performance

import redis
import json

class RAGCache:
    def __init__(self):
        self.redis = redis.Redis(host='localhost', port=6379, db=0)
        self.ttl = 3600  # 1 hour
    
    def get_cached_answer(self, question):
        """Get cached answer if exists"""
        key = f"rag:{hashlib.md5(question.encode()).hexdigest()}"
        cached = self.redis.get(key)
        
        if cached:
            print("📦 Cache hit!")
            return json.loads(cached)
        
        return None
    
    def cache_answer(self, question, answer):
        """Cache answer for future requests"""
        key = f"rag:{hashlib.md5(question.encode()).hexdigest()}"
        self.redis.setex(
            key,
            self.ttl,
            json.dumps(answer)
        )

---

Results After 3 Months

Performance Metrics

Metric	Value
Documents Indexed	100,000+
Average Query Time	1.2s
Accuracy	92% (measured vs human answers)
API Cost	$300/month (40% savings via caching)
User Satisfaction	4.8/5
Questions/Day	1,500+

Real Impact

Before RAG:

• Average search time: 15 minutes
• “I don’t know” responses: 40%
• Employee frustration: High

After RAG:

• Average answer time: <2 seconds
• Accurate answers: 92%
• Employee productivity: +25%
• Support tickets: -60%

---

Lessons Learned

1. Chunking Strategy Matters

What Worked:

• 1000 chars with 200 overlap
• Semantic boundaries (paragraphs, sections)
• Metadata preservation

What Didn’t:

• Fixed 512 token chunks (lost context)
• No overlap (missed connections)
• Ignoring document structure

2. Hybrid Search is Better

Combining semantic + keyword search improved accuracy from 85% → 92%.

3. Re-ranking is Worth It

Claude re-ranking added 50ms latency but improved relevance by 15%.

4. Caching Saves Money

40% of queries are duplicates. Redis caching cut API costs by 40%.

5. Confidence Scores Build Trust

Users trust answers with confidence scores + source citations more.

---

Production Tips

1. Monitor Everything

from prometheus_client import Counter, Histogram

query_counter = Counter('rag_queries_total', 'Total queries')
query_duration = Histogram('rag_query_duration_seconds', 'Query duration')
accuracy_gauge = Gauge('rag_accuracy', 'Answer accuracy')

@query_duration.time()
async def answer_with_metrics(question):
    query_counter.inc()
    result = await answer_question(question)
    # Track accuracy, latency, cache hits, etc.
    return result

2. Handle Failures Gracefully

async def answer_with_fallback(question):
    try:
        return await answer_question(question)
    except Exception as e:
        logger.error(f"RAG failed: {e}")
        # Fallback to simple semantic search
        return await fallback_search(question)

3. Cost Optimization

• Cache aggressively
• Batch embeddings
• Use cheaper models for re-ranking
• Compress context intelligently

---

Code Repository

Full implementation: github.com/vaibhav7k/enterprise-rag

Includes:

• Complete source code
• Docker setup
• Monitoring dashboards
• Example queries
• Performance benchmarks

---

Conclusion

Building a production RAG system taught me:

• Context quality > Model quality
• Caching is critical for cost
• Hybrid search beats pure semantic
• Re-ranking significantly improves relevance
• Monitoring is essential for debugging

The system now handles 1,500+ queries daily with 92% accuracy, saving employees ~4 hours/week in document search time.

Tech Stack: Python, LangChain, Claude API, Qdrant, FastAPI, Redis, Docker

---

Have questions about RAG systems? Email me or @vaibhav7k