Vector Databases and LangChain: Building Production RAG Systems

The RAG Revolution

Retrieval-Augmented Generation (RAG) combines the power of large language models with the precision of vector search to create AI systems that can reason over your specific data. This approach has revolutionized how we build AI applications that need to be both knowledgeable and accurate.

At ScriptLabs Studios, we've implemented RAG systems for enterprises ranging from legal research platforms to technical documentation systems, consistently achieving 40% better accuracy than standalone LLMs.

Understanding Vector Databases

Vector databases are specialized systems designed to store, index, and search high-dimensional vectors efficiently. Unlike traditional databases that work with structured data, vector databases excel at finding semantic similarity in unstructured content.

Key Vector Database Features

• High-dimensional indexing - Efficient search through 1000+ dimensional vectors
• Similarity search - Find semantically similar content
• Metadata filtering - Combine vector search with traditional filters
• Real-time updates - Dynamic content indexing
• Scalability - Handle millions of vectors with sub-second query times

Vector Database Comparison

Pinecone: Managed Vector Database

Pinecone offers a fully managed solution with excellent performance:

import pinecone
from langchain.vectorstores import Pinecone
from langchain.embeddings import OpenAIEmbeddings

# Initialize Pinecone
pinecone.init(
    api_key="your-api-key",
    environment="your-env"
)

# Create index
index_name = "document-search"
if index_name not in pinecone.list_indexes():
    pinecone.create_index(
        name=index_name,
        dimension=1536,  # OpenAI ada-002 dimensions
        metric="cosine",
        pods=1,
        pod_type="p1.x1"
    )

# Initialize vector store
embeddings = OpenAIEmbeddings()
vectorstore = Pinecone.from_texts(
    texts=documents,
    embedding=embeddings,
    index_name=index_name
)

Weaviate: Open Source with GraphQL

import weaviate
from langchain.vectorstores import Weaviate

client = weaviate.Client(
    url="http://localhost:8080",
    auth_client_secret=weaviate.AuthApiKey(api_key="your-key")
)

# Define schema
class_obj = {
    "class": "Document",
    "vectorizer": "text2vec-openai",
    "moduleConfig": {
        "text2vec-openai": {
            "model": "ada",
            "modelVersion": "002",
            "type": "text"
        }
    },
    "properties": [
        {
            "dataType": ["text"],
            "name": "content",
        },
        {
            "dataType": ["text"],
            "name": "source",
        }
    ]
}

client.schema.create_class(class_obj)

# Create vector store
vectorstore = Weaviate(
    client=client,
    index_name="Document",
    text_key="content"
)

ChromaDB: Simple and Effective

import chromadb
from langchain.vectorstores import Chroma

# Initialize ChromaDB client
client = chromadb.PersistentClient(path="./chroma_db")

# Create vector store
vectorstore = Chroma(
    client=client,
    collection_name="documents",
    embedding_function=OpenAIEmbeddings(),
    persist_directory="./chroma_db"
)

Building Production RAG Systems

Document Processing Pipeline

from langchain.document_loaders import PyPDFLoader, TextLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.embeddings import OpenAIEmbeddings
import hashlib
import asyncio

class DocumentProcessor:
    def __init__(self, vectorstore, chunk_size=1000, chunk_overlap=200):
        self.vectorstore = vectorstore
        self.text_splitter = RecursiveCharacterTextSplitter(
            chunk_size=chunk_size,
            chunk_overlap=chunk_overlap,
            separators=["

", "
", " ", ""]
        )
        self.embeddings = OpenAIEmbeddings()
    
    async def process_document(self, file_path: str, metadata: dict = None):
        """Process a single document with metadata tracking"""
        # Load document
        if file_path.endswith('.pdf'):
            loader = PyPDFLoader(file_path)
        else:
            loader = TextLoader(file_path)
        
        documents = loader.load()
        
        # Split into chunks
        chunks = self.text_splitter.split_documents(documents)
        
        # Add metadata and generate IDs
        for i, chunk in enumerate(chunks):
            chunk.metadata.update(metadata or {})
            chunk.metadata['source'] = file_path
            chunk.metadata['chunk_index'] = i
            chunk.metadata['chunk_id'] = self._generate_chunk_id(
                chunk.page_content, file_path, i
            )
        
        # Store in vector database
        await self.vectorstore.aadd_documents(chunks)
        
        return len(chunks)
    
    def _generate_chunk_id(self, content: str, source: str, index: int) -> str:
        """Generate deterministic chunk ID for deduplication"""
        content_hash = hashlib.md5(
            f"{content}{source}{index}".encode()
        ).hexdigest()
        return f"{source}_{index}_{content_hash[:8]}"
    
    async def batch_process(self, file_paths: list, batch_size: int = 10):
        """Process multiple documents in batches"""
        results = []
        for i in range(0, len(file_paths), batch_size):
            batch = file_paths[i:i + batch_size]
            tasks = [
                self.process_document(path) 
                for path in batch
            ]
            batch_results = await asyncio.gather(*tasks)
            results.extend(batch_results)
            
        return results

Advanced Chunking Strategies

from langchain.text_splitter import TextSplitter
import re

class SemanticChunker(TextSplitter):
    """Custom chunker that respects semantic boundaries"""
    
    def __init__(self, chunk_size=1000, chunk_overlap=100):
        self.chunk_size = chunk_size
        self.chunk_overlap = chunk_overlap
    
    def split_text(self, text: str) -> list[str]:
        # Split on semantic boundaries
        sections = self._split_on_sections(text)
        chunks = []
        
        for section in sections:
            if len(section) <= self.chunk_size:
                chunks.append(section)
            else:
                # Further split large sections
                sub_chunks = self._split_large_section(section)
                chunks.extend(sub_chunks)
        
        return self._add_overlap(chunks)
    
    def _split_on_sections(self, text: str) -> list[str]:
        # Split on headings, paragraphs, and natural breaks
        patterns = [
            r'

',  # Paragraph breaks
            r'
(?=[A-Z][^.]*:)',  # Section headers
            r'
(?=d+.)',  # Numbered lists
            r'
(?=-)',  # Bullet points
        ]
        
        sections = [text]
        for pattern in patterns:
            new_sections = []
            for section in sections:
                new_sections.extend(re.split(pattern, section))
            sections = [s.strip() for s in new_sections if s.strip()]
        
        return sections
    
    def _split_large_section(self, section: str) -> list[str]:
        # Split large sections while preserving sentence boundaries
        sentences = re.split(r'(?<=[.!?])s+', section)
        chunks = []
        current_chunk = ""
        
        for sentence in sentences:
            if len(current_chunk + sentence) <= self.chunk_size:
                current_chunk += sentence + " "
            else:
                if current_chunk:
                    chunks.append(current_chunk.strip())
                current_chunk = sentence + " "
        
        if current_chunk:
            chunks.append(current_chunk.strip())
        
        return chunks
    
    def _add_overlap(self, chunks: list[str]) -> list[str]:
        if self.chunk_overlap == 0 or len(chunks) <= 1:
            return chunks
        
        overlapped_chunks = [chunks[0]]
        
        for i in range(1, len(chunks)):
            # Add overlap from previous chunk
            prev_chunk = chunks[i-1]
            current_chunk = chunks[i]
            
            overlap_text = prev_chunk[-self.chunk_overlap:]
            overlapped_chunk = overlap_text + " " + current_chunk
            overlapped_chunks.append(overlapped_chunk)
        
        return overlapped_chunks

Query Processing and Retrieval

Hybrid Search Implementation

from langchain.retrievers import BM25Retriever, EnsembleRetriever
from rank_bm25 import BM25Okapi
import numpy as np

class HybridRetriever:
    def __init__(self, vectorstore, documents, weights=[0.7, 0.3]):
        self.vectorstore = vectorstore
        self.vector_weight, self.bm25_weight = weights
        
        # Initialize BM25
        tokenized_docs = [doc.page_content.split() for doc in documents]
        self.bm25 = BM25Okapi(tokenized_docs)
        self.documents = documents
    
    async def retrieve(self, query: str, k: int = 10) -> list:
        # Vector search
        vector_results = await self.vectorstore.asimilarity_search_with_score(
            query, k=k*2
        )
        
        # BM25 search
        bm25_scores = self.bm25.get_scores(query.split())
        bm25_results = [
            (self.documents[i], score) 
            for i, score in enumerate(bm25_scores)
        ]
        bm25_results.sort(key=lambda x: x[1], reverse=True)
        bm25_results = bm25_results[:k*2]
        
        # Combine and rerank
        return self._combine_results(vector_results, bm25_results, k)
    
    def _combine_results(self, vector_results, bm25_results, k):
        # Normalize scores
        vector_scores = [score for _, score in vector_results]
        bm25_scores = [score for _, score in bm25_results]
        
        if vector_scores:
            max_vector = max(vector_scores)
            min_vector = min(vector_scores)
            vector_norm = [(score - min_vector) / (max_vector - min_vector) 
                          for score in vector_scores]
        
        if bm25_scores:
            max_bm25 = max(bm25_scores)
            min_bm25 = min(bm25_scores)
            bm25_norm = [(score - min_bm25) / (max_bm25 - min_bm25) 
                        for score in bm25_scores]
        
        # Combine scores
        combined_results = {}
        
        for i, (doc, _) in enumerate(vector_results):
            doc_id = doc.metadata.get('chunk_id', str(hash(doc.page_content)))
            combined_results[doc_id] = {
                'document': doc,
                'score': self.vector_weight * vector_norm[i]
            }
        
        for i, (doc, _) in enumerate(bm25_results):
            doc_id = doc.metadata.get('chunk_id', str(hash(doc.page_content)))
            if doc_id in combined_results:
                combined_results[doc_id]['score'] += self.bm25_weight * bm25_norm[i]
            else:
                combined_results[doc_id] = {
                    'document': doc,
                    'score': self.bm25_weight * bm25_norm[i]
                }
        
        # Sort by combined score
        sorted_results = sorted(
            combined_results.values(),
            key=lambda x: x['score'],
            reverse=True
        )
        
        return [result['document'] for result in sorted_results[:k]]

Query Expansion and Rewriting

from langchain.llms import OpenAI
from langchain.chains import LLMChain
from langchain.prompts import PromptTemplate

class QueryProcessor:
    def __init__(self):
        self.llm = OpenAI(temperature=0.1)
        self.expansion_chain = self._create_expansion_chain()
        self.rewrite_chain = self._create_rewrite_chain()
    
    def _create_expansion_chain(self):
        template = """
        Given the user question below, generate 3 alternative phrasings that might help find relevant information.
        Focus on:
        1. Different terminology that means the same thing
        2. More specific technical terms
        3. Broader conceptual framings
        
        Original question: {question}
        
        Alternative phrasings:
        1.
        2. 
        3.
        """
        
        prompt = PromptTemplate(template=template, input_variables=["question"])
        return LLMChain(llm=self.llm, prompt=prompt)
    
    def _create_rewrite_chain(self):
        template = """
        Rewrite the following question to be more specific and searchable.
        If the question is vague, make it more concrete.
        If it uses colloquial language, make it more technical.
        
        Original: {question}
        Rewritten:
        """
        
        prompt = PromptTemplate(template=template, input_variables=["question"])
        return LLMChain(llm=self.llm, prompt=prompt)
    
    async def process_query(self, question: str) -> dict:
        # Generate query variants
        expansion_result = await self.expansion_chain.arun(question=question)
        rewritten = await self.rewrite_chain.arun(question=question)
        
        alternatives = self._parse_alternatives(expansion_result)
        
        return {
            'original': question,
            'rewritten': rewritten,
            'alternatives': alternatives,
            'all_queries': [question, rewritten] + alternatives
        }
    
    def _parse_alternatives(self, expansion_result: str) -> list:
        lines = expansion_result.strip().split('
')
        alternatives = []
        
        for line in lines:
            line = line.strip()
            if line and (line.startswith('1.') or line.startswith('2.') or line.startswith('3.')):
                alternative = line[2:].strip()
                if alternative:
                    alternatives.append(alternative)
        
        return alternatives

RAG Chain Implementation

Production RAG System

from langchain.chains import ConversationalRetrievalChain
from langchain.memory import ConversationBufferMemory
from langchain.prompts import PromptTemplate
import logging

class ProductionRAGSystem:
    def __init__(self, vectorstore, llm, retriever_k=10):
        self.vectorstore = vectorstore
        self.llm = llm
        self.retriever_k = retriever_k
        self.retriever = self._create_retriever()
        self.memory = ConversationBufferMemory(
            memory_key="chat_history",
            return_messages=True,
            output_key="answer"
        )
        self.chain = self._create_chain()
        self.logger = logging.getLogger(__name__)
    
    def _create_retriever(self):
        return self.vectorstore.as_retriever(
            search_type="similarity_score_threshold",
            search_kwargs={
                "k": self.retriever_k,
                "score_threshold": 0.7
            }
        )
    
    def _create_chain(self):
        template = """You are a helpful AI assistant. Use the following context to answer the user's question.
        
        Context: {context}
        
        Chat History: {chat_history}
        
        Question: {question}
        
        Instructions:
        - Answer based primarily on the provided context
        - If the context doesn't contain enough information, say so clearly
        - Cite specific sources when possible using [Source: filename] format
        - Be concise but comprehensive
        - If asked about something not in the context, explain what information is available
        
        Answer:"""
        
        prompt = PromptTemplate(
            template=template,
            input_variables=["context", "chat_history", "question"]
        )
        
        return ConversationalRetrievalChain.from_llm(
            llm=self.llm,
            retriever=self.retriever,
            memory=self.memory,
            combine_docs_chain_kwargs={"prompt": prompt},
            return_source_documents=True,
            verbose=True
        )
    
    async def query(self, question: str, session_id: str = None) -> dict:
        """Process a query and return structured response"""
        try:
            # Get response from chain
            response = await self.chain.acall({"question": question})
            
            # Process sources
            sources = self._process_sources(response["source_documents"])
            
            result = {
                "answer": response["answer"],
                "sources": sources,
                "confidence": self._calculate_confidence(
                    response["source_documents"]
                ),
                "session_id": session_id
            }
            
            # Log query
            self.logger.info(f"Query processed: {question[:100]}...")
            
            return result
            
        except Exception as e:
            self.logger.error(f"Error processing query: {str(e)}")
            return {
                "answer": "I encountered an error processing your question. Please try again.",
                "sources": [],
                "confidence": 0.0,
                "error": str(e)
            }
    
    def _process_sources(self, source_documents) -> list:
        """Extract and format source information"""
        sources = []
        seen_sources = set()
        
        for doc in source_documents:
            source_id = doc.metadata.get('source', 'Unknown')
            chunk_index = doc.metadata.get('chunk_index', 0)
            
            if source_id not in seen_sources:
                sources.append({
                    'source': source_id,
                    'content_preview': doc.page_content[:200] + "...",
                    'metadata': doc.metadata,
                    'relevance_score': getattr(doc, 'score', 0.0)
                })
                seen_sources.add(source_id)
        
        return sources
    
    def _calculate_confidence(self, source_documents) -> float:
        """Calculate confidence based on source document scores"""
        if not source_documents:
            return 0.0
        
        scores = [getattr(doc, 'score', 0.0) for doc in source_documents]
        return min(1.0, sum(scores) / len(scores))
    
    def clear_memory(self, session_id: str = None):
        """Clear conversation memory"""
        self.memory.clear()
    
    def get_conversation_history(self, session_id: str = None) -> list:
        """Get conversation history for session"""
        return self.memory.chat_memory.messages

Monitoring and Optimization

Performance Monitoring

import time
import asyncio
from collections import defaultdict
import json

class RAGMonitor:
    def __init__(self):
        self.metrics = defaultdict(list)
        self.query_logs = []
    
    async def monitor_query(self, rag_system, question: str, session_id: str = None):
        """Monitor a query with detailed metrics"""
        start_time = time.time()
        
        # Monitor retrieval
        retrieval_start = time.time()
        retrieved_docs = await rag_system.vectorstore.asimilarity_search_with_score(
            question, k=rag_system.retriever_k
        )
        retrieval_time = time.time() - retrieval_start
        
        # Monitor full query
        result = await rag_system.query(question, session_id)
        total_time = time.time() - start_time
        
        # Record metrics
        metrics = {
            'timestamp': time.time(),
            'question': question,
            'retrieval_time': retrieval_time,
            'total_time': total_time,
            'num_sources': len(result.get('sources', [])),
            'confidence': result.get('confidence', 0.0),
            'answer_length': len(result.get('answer', '')),
            'session_id': session_id
        }
        
        self.query_logs.append(metrics)
        
        # Update running metrics
        self.metrics['retrieval_times'].append(retrieval_time)
        self.metrics['total_times'].append(total_time)
        self.metrics['confidences'].append(result.get('confidence', 0.0))
        
        return result
    
    def get_performance_summary(self) -> dict:
        """Get performance summary statistics"""
        if not self.query_logs:
            return {}
        
        retrieval_times = [log['retrieval_time'] for log in self.query_logs]
        total_times = [log['total_time'] for log in self.query_logs]
        confidences = [log['confidence'] for log in self.query_logs]
        
        return {
            'total_queries': len(self.query_logs),
            'avg_retrieval_time': sum(retrieval_times) / len(retrieval_times),
            'avg_total_time': sum(total_times) / len(total_times),
            'avg_confidence': sum(confidences) / len(confidences),
            'p95_retrieval_time': sorted(retrieval_times)[int(len(retrieval_times) * 0.95)],
            'p95_total_time': sorted(total_times)[int(len(total_times) * 0.95)],
        }
    
    def export_logs(self, filename: str):
        """Export query logs for analysis"""
        with open(filename, 'w') as f:
            json.dump(self.query_logs, f, indent=2)

Advanced RAG Patterns

Multi-Modal RAG

from langchain.embeddings import OpenAIEmbeddings
import base64

class MultiModalRAG(ProductionRAGSystem):
    def __init__(self, vectorstore, llm, image_vectorstore=None):
        super().__init__(vectorstore, llm)
        self.image_vectorstore = image_vectorstore
        self.image_embeddings = OpenAIEmbeddings(model="clip")
    
    async def process_multimodal_query(self, question: str, image_path: str = None):
        """Process query that may include image context"""
        text_results = await super().query(question)
        
        if image_path and self.image_vectorstore:
            # Process image
            image_results = await self._search_images(image_path)
            
            # Combine results
            combined_context = self._combine_text_image_context(
                text_results['sources'],
                image_results
            )
            
            # Generate enhanced response
            enhanced_response = await self._generate_multimodal_response(
                question, combined_context, image_path
            )
            
            return {
                **text_results,
                'image_sources': image_results,
                'multimodal_answer': enhanced_response
            }
        
        return text_results
    
    async def _search_images(self, image_path: str):
        """Search for similar images in vector database"""
        with open(image_path, 'rb') as f:
            image_data = base64.b64encode(f.read()).decode()
        
        # Generate image embeddings (simplified)
        results = await self.image_vectorstore.asimilarity_search(
            image_data, k=5
        )
        
        return results

Best Practices and Recommendations

Production Deployment Checklist

• Data Quality
  • Clean and preprocess documents before indexing
  • Remove duplicate content to avoid repetitive responses
  • Implement proper metadata structure for filtering
• Performance Optimization
  • Use appropriate chunk sizes (500-1500 characters typically work best)
  • Implement caching for frequently asked questions
  • Monitor query latency and optimize retrieval parameters
• Security and Privacy
  • Implement proper access controls for sensitive documents
  • Use metadata filtering to enforce permissions
  • Consider data residency requirements for vector storage
• Monitoring and Maintenance
  • Track answer quality through user feedback
  • Monitor for concept drift in your document collection
  • Implement automated reindexing for updated documents

Future Directions

Emerging Trends in RAG

• Graph-Enhanced RAG - Combining knowledge graphs with vector search
• Agents with RAG - Multi-step reasoning with retrieval
• Real-time RAG - Streaming updates to knowledge bases
• Federated RAG - Searching across multiple vector databases

Conclusion

Vector databases and RAG systems represent a fundamental shift in how we build AI applications that need to be both accurate and current. By combining the semantic understanding of embeddings with the precision of retrieval, we can create systems that provide relevant, factual responses grounded in your specific data.

The key to successful RAG implementation lies in understanding your data, choosing the right vector database for your needs, and implementing robust processing pipelines that can scale with your requirements.

As this technology continues to evolve, we're seeing exciting developments in multimodal RAG, real-time updates, and integration with other AI techniques. The future of knowledge-based AI systems is being built on these foundations today.

Ready to build a production RAG system for your organization? Let's discuss how vector databases and LangChain can transform your knowledge management and AI capabilities.