Enterprise-Grade RAG Chatbots for Financial Services

# Confer's RAG Pipeline Implementation

from typing import List, Dict, Any
import numpy as np
from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores import Pinecone
from langchain.retrievers import BM25Retriever, EnsembleRetriever
from langchain.retrievers.document_compressors import LLMChainExtractor
from langchain.chains import RetrievalQA
from langchain.llms import OpenAI

class FinancialRAGPipeline:
    """Enterprise-grade RAG pipeline optimized for financial services."""
    
    def __init__(
        self,
        vector_db_name: str,
        embedding_model: str = "text-embedding-ada-002",
        llm_model: str = "gpt-4",
        top_k: int = 5,
        reranker_threshold: float = 0.7
    ):
        # Initialize embedding model
        self.embeddings = OpenAIEmbeddings(model=embedding_model)
        
        # Initialize vector store
        self.vector_store = Pinecone.from_existing_index(
            index_name=vector_db_name,
            embedding=self.embeddings,
            namespace="financial_documents"
        )
        
        # Initialize retrievers
        self.dense_retriever = self.vector_store.as_retriever(
            search_type="similarity",
            search_kwargs={"k": top_k * 2}
        )
        
        self.keyword_retriever = BM25Retriever.from_documents(
            self.vector_store.similarity_search(
                "", k=1000
            )
        )
        
        # Create ensemble retriever
        self.ensemble_retriever = EnsembleRetriever(
            retrievers=[self.dense_retriever, self.keyword_retriever],
            weights=[0.7, 0.3]
        )
        
        # Document compressor for filtering irrelevant content
        self.compressor = LLMChainExtractor.from_llm(OpenAI(temperature=0))
        
        # LLM for response generation
        self.llm = OpenAI(model_name=llm_model, temperature=0.2)
        
        # Reranker threshold
        self.reranker_threshold = reranker_threshold
        
    def query(self, user_query: str, user_context: Dict[str, Any] = None) -> Dict[str, Any]:
        """Process user query and generate contextual response."""
        # Step 1: Retrieve relevant documents
        raw_docs = self.ensemble_retriever.get_relevant_documents(user_query)
        
        # Step 2: Compress and filter documents
        filtered_docs = self.compressor.compress_documents(raw_docs, user_query)
        
        # Step 3: Apply reranking if needed
        if len(filtered_docs) > 3:
            filtered_docs = self._rerank_documents(filtered_docs, user_query)
        
        # Step 4: Integrate user context if available
        prompt = self._build_prompt(user_query, filtered_docs, user_context)
        
        # Step 5: Generate response
        response = self.llm.predict(prompt)
        
        return {
            "query": user_query,
            "response": response,
            "sources": [doc.metadata for doc in filtered_docs[:3]],
            "confidence": self._calculate_confidence(filtered_docs, user_query)
        }
        
    def _rerank_documents(self, documents, query):
        # Implementation of cross-encoder reranking
        # ...
        
    def _build_prompt(self, query, documents, user_context):
        # Construct prompt with retrieved context and user information
        # ...
        
    def _calculate_confidence(self, documents, query):
        # Calculate confidence score based on document relevance
        # ...