RAG Retrieval-Augmented Generation Complete Guide [2026] - The Key Technology for Smarter AI

prompt_template = """
Answer the user's question based on the following reference information. If the reference information doesn't contain relevant content, please state that clearly.

Reference Information:
{context}

User Question: {question}

Answer:
"""

import numpy as np

def cosine_similarity(vec1, vec2):
    """Cosine similarity: most commonly used method"""
    return np.dot(vec1, vec2) / (np.linalg.norm(vec1) * np.linalg.norm(vec2))

def euclidean_distance(vec1, vec2):
    """Euclidean distance: smaller distance means more similar"""
    return np.linalg.norm(vec1 - vec2)

def dot_product(vec1, vec2):
    """Dot product: equivalent to cosine similarity when vectors are normalized"""
    return np.dot(vec1, vec2)

from langchain.text_splitter import RecursiveCharacterTextSplitter

def split_documents(documents, chunk_size=500, chunk_overlap=50):
    """
    Document chunking strategy
    - chunk_size: size of each chunk, recommended 300-1000 characters
    - chunk_overlap: overlap portion to maintain context continuity
    """
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=chunk_size,
        chunk_overlap=chunk_overlap,
        separators=["\n\n", "\n", ".", "!", "?", ";", " "]
    )
    return splitter.split_documents(documents)

from langchain.embeddings import OpenAIEmbeddings, HuggingFaceEmbeddings

openai_embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

local_embeddings = HuggingFaceEmbeddings(
    model_name="BAAI/bge-base-en-v1.5",
    model_kwargs={'device': 'cuda'}
)

from langchain.vectorstores import Chroma

vectorstore = Chroma.from_documents(
    documents=chunks,
    embedding=embeddings,
    persist_directory="./chroma_db"
)

from langchain.chains import RetrievalQA
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4-turbo", temperature=0)

qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",
    retriever=vectorstore.as_retriever(search_kwargs={"k": 5}),
    return_source_documents=True
)

result = qa_chain.invoke({"query": "What is RAG technology?"})
print(result["result"])

import os
from typing import List, Dict, Any
from langchain.document_loaders import DirectoryLoader, TextLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores import Chroma
from langchain_openai import ChatOpenAI
from langchain.prompts import PromptTemplate
from langchain.chains import RetrievalQA

class RAGSystem:
    """Complete RAG system implementation"""
    
    def __init__(
        self,
        embedding_model: str = "text-embedding-3-small",
        llm_model: str = "gpt-4-turbo",
        persist_directory: str = "./rag_db"
    ):
        self.embeddings = OpenAIEmbeddings(model=embedding_model)
        self.llm = ChatOpenAI(model=llm_model, temperature=0)
        self.persist_directory = persist_directory
        self.vectorstore = None
        self.qa_chain = None
        
    def load_documents(self, directory: str, glob: str = "**/*.txt") -> List:
        """Load documents"""
        loader = DirectoryLoader(
            directory,
            glob=glob,
            loader_cls=TextLoader,
            loader_kwargs={'encoding': 'utf-8'}
        )
        return loader.load()
    
    def process_documents(
        self,
        documents: List,
        chunk_size: int = 500,
        chunk_overlap: int = 50
    ) -> List:
        """Document chunking processing"""
        splitter = RecursiveCharacterTextSplitter(
            chunk_size=chunk_size,
            chunk_overlap=chunk_overlap,
            separators=["\n\n", "\n", ".", "!", "?", " "]
        )
        return splitter.split_documents(documents)
    
    def build_vectorstore(self, chunks: List) -> None:
        """Build vector database"""
        self.vectorstore = Chroma.from_documents(
            documents=chunks,
            embedding=self.embeddings,
            persist_directory=self.persist_directory
        )
        
    def setup_qa_chain(self, k: int = 5) -> None:
        """Set up QA chain"""
        prompt_template = """You are a professional AI assistant. Please answer the user's question based on the following reference information.
        
If the reference information doesn't contain relevant content, please clearly state "Unable to answer this question based on available materials."
Please cite specific sources when answering.

Reference Information:
{context}

User Question: {question}

Answer:"""
        
        prompt = PromptTemplate(
            template=prompt_template,
            input_variables=["context", "question"]
        )
        
        self.qa_chain = RetrievalQA.from_chain_type(
            llm=self.llm,
            chain_type="stuff",
            retriever=self.vectorstore.as_retriever(
                search_kwargs={"k": k}
            ),
            return_source_documents=True,
            chain_type_kwargs={"prompt": prompt}
        )
    
    def query(self, question: str) -> Dict[str, Any]:
        """Execute query"""
        if not self.qa_chain:
            raise ValueError("Please call setup_qa_chain() to initialize the QA chain first")
        
        result = self.qa_chain.invoke({"query": question})
        return {
            "answer": result["result"],
            "sources": [
                {
                    "content": doc.page_content[:200] + "...",
                    "metadata": doc.metadata
                }
                for doc in result["source_documents"]
            ]
        }

if __name__ == "__main__":
    rag = RAGSystem()
    
    docs = rag.load_documents("./knowledge_base")
    chunks = rag.process_documents(docs)
    rag.build_vectorstore(chunks)
    rag.setup_qa_chain()
    
    result = rag.query("What are the main advantages of RAG technology?")
    print(f"Answer: {result['answer']}")
    print(f"\nCitation Sources:")
    for i, source in enumerate(result['sources'], 1):
        print(f"{i}. {source['content']}")

from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import LLMChainExtractor

class AdvancedRAGSystem(RAGSystem):
    """Advanced RAG system: supports multiple retrieval strategies"""
    
    def setup_hybrid_retriever(self, k: int = 5):
        """Hybrid retrieval: combines semantic search and keyword search"""
        from langchain.retrievers import BM25Retriever, EnsembleRetriever
        
        bm25_retriever = BM25Retriever.from_documents(self.chunks)
        bm25_retriever.k = k
        
        dense_retriever = self.vectorstore.as_retriever(
            search_kwargs={"k": k}
        )
        
        self.retriever = EnsembleRetriever(
            retrievers=[bm25_retriever, dense_retriever],
            weights=[0.3, 0.7]
        )
    
    def setup_reranking(self, k: int = 5):
        """Reranking: use LLM to rerank retrieval results"""
        base_retriever = self.vectorstore.as_retriever(
            search_kwargs={"k": k * 2}
        )
        
        compressor = LLMChainExtractor.from_llm(self.llm)
        
        self.retriever = ContextualCompressionRetriever(
            base_compressor=compressor,
            base_retriever=base_retriever
        )
    
    def setup_multi_query(self):
        """Multi-query retrieval: generate multiple query variants to improve recall"""
        from langchain.retrievers.multi_query import MultiQueryRetriever
        
        self.retriever = MultiQueryRetriever.from_llm(
            retriever=self.vectorstore.as_retriever(),
            llm=self.llm
        )

optimized_prompt = """
You are an expert assistant in the {domain} field.

## Task
Answer user questions based on the provided reference materials.

## Rules
1. Only use information from the reference materials
2. If materials are insufficient, clearly state so
3. Cite specific sources to increase credibility
4. Use clear, professional language

## Reference Materials
{context}

## User Question
{question}

## Answer
"""