Building a Complete RAG System

RAG System Architecture:

┌─ Document Ingestion Layer ─┐
│  Load → Clean → Chunk      │
└──────────┬──────────────────┘
           │
┌──────────▼──────────────────┐
│   Embedding Layer           │
│ tokenize → embed (batch)    │
└──────────┬──────────────────┘
           │
┌──────────▼──────────────────┐
│   Vector DB (Pinecone)      │
│ Index & Store              │
└──────────┬──────────────────┘
           │ (Query Time)
┌──────────▼──────────────────┐
│   Retrieval Layer           │
│ search → rerank → filter    │
└──────────┬──────────────────┘
           │
┌──────────▼──────────────────┐
│   Prompt Construction       │
│ format docs → build prompt  │
└──────────┬──────────────────┘
           │
┌──────────▼──────────────────┐
│   LLM Generation            │
│ call GPT-4 → get answer     │
└──────────┬──────────────────┘
           │
┌──────────▼──────────────────┐
│   Output Processing         │
│ cite sources → verify       │
└──────────────────────────────┘

from langchain_community.document_loaders import PyPDFLoader
from langchain.text_splitters import TokenTextSplitter
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_pinecone import PineconeVectorStore
from langchain.chains import RetrievalQA
from sentence_transformers import CrossEncoder

class RAGSystem:
    def __init__(self):
        self.embeddings = OpenAIEmbeddings(
            model="text-embedding-3-small"
        )
        self.llm = ChatOpenAI(model="gpt-4", temperature=0.3)
        self.vector_store = PineconeVectorStore.from_existing_index(
            index_name="my-rag",
            embedding=self.embeddings
        )
        self.reranker = CrossEncoder("cross-encoder/qnli-distilroberta-base")

    def index_documents(self, file_paths):
        """Load, chunk, and index documents."""
        docs = []
        for path in file_paths:
            loader = PyPDFLoader(path)
            docs.extend(loader.load())

        splitter = TokenTextSplitter(
            chunk_size=400,
            chunk_overlap=40,
            encoding_name="cl100k_base"
        )
        chunks = splitter.split_documents(docs)

        # Add to vector store
        self.vector_store.add_documents(chunks)
        print(f"Indexed {len(chunks)} chunks")

    def retrieve_and_rank(self, query, k=10):
        """Retrieve and rerank documents."""
        # Dense search
        dense_results = self.vector_store.similarity_search_with_score(
            query, k=k*2
        )

        # Rerank
        docs = [doc for doc, _ in dense_results]
        pairs = [[query, doc.page_content] for doc in docs]
        scores = self.reranker.predict(pairs)

        # Sort by reranker score
        ranked = sorted(zip(docs, scores), key=lambda x: -x[1])
        return [doc for doc, _ in ranked[:k]]

    def generate_answer(self, query):
        """Generate answer from retrieved docs."""
        # Retrieve
        docs = self.retrieve_and_rank(query)

        # Build prompt
        context = "\n".join([
            f"[{i}] {doc.page_content}"
            for i, doc in enumerate(docs)
        ])

        prompt = f"""Answer using ONLY these documents. Cite sources.

DOCUMENTS:
{context}

QUESTION: {query}

ANSWER:"""

        # Generate
        response = self.llm.predict(text=prompt)
        return response, docs

# Usage
rag = RAGSystem()
rag.index_documents(["tax_guide.pdf", "faq.pdf"])
answer, sources = rag.generate_answer("How do I file taxes?")

def test_rag_system():
    """Test retrieval and generation quality."""
    from ragas import evaluate
    from ragas.metrics import faithfulness, answer_relevancy

    rag = RAGSystem()

    # Test cases
    test_data = [
        {
            "query": "How do I file taxes?",
            "ground_truth": "You file taxes using Form 1040..."
        },
        {
            "query": "What deductions can I claim?",
            "ground_truth": "Common deductions include..."
        }
    ]

    results = []
    for test in test_data:
        answer, docs = rag.generate_answer(test["query"])
        results.append({
            "query": test["query"],
            "answer": answer,
            "context": "\n".join([d.page_content for d in docs]),
            "ground_truth": test["ground_truth"]
        })

    # Evaluate
    eval_result = evaluate(
        results,
        metrics=[faithfulness, answer_relevancy]
    )
    print(f"Faithfulness: {eval_result['faithfulness']}")
    print(f"Relevancy: {eval_result['answer_relevancy']}")

    # Assert minimum quality
    assert eval_result['faithfulness'] > 0.7
    assert eval_result['answer_relevancy'] > 0.75

def safe_rag_query(rag, query, max_retries=2):
    """Robust query with fallbacks."""
    try:
        answer, docs = rag.generate_answer(query)

        # Verify answer
        if not answer or len(answer) < 10:
            raise ValueError("Empty answer")

        # Check for hallucinations
        if "don't know" in answer.lower():
            return {
                "answer": answer,
                "status": "limited_info",
                "sources": docs
            }

        return {
            "answer": answer,
            "status": "success",
            "sources": docs
        }

    except Exception as e:
        print(f"Error: {e}")
        return {
            "answer": "I encountered an error. Please try again.",
            "status": "error",
            "sources": []
        }