RAG Implementation

Master Retrieval-Augmented Generation (RAG) to build LLM applications that provide accurate, grounded responses using external knowledge sources.

When to Use This Skill

• Building Q&A systems over proprietary documents

• Creating chatbots with current, factual information

• Implementing semantic search with natural language queries

• Reducing hallucinations with grounded responses

• Enabling LLMs to access domain-specific knowledge

• Building documentation assistants

• Creating research tools with source citation

Core Components

1. Vector Databases

Purpose: Store and retrieve document embeddings efficiently

Options:

• Pinecone: Managed, scalable, serverless

• Weaviate: Open-source, hybrid search, GraphQL

• Milvus: High performance, on-premise

• Chroma: Lightweight, easy to use, local development

• Qdrant: Fast, filtered search, Rust-based

• pgvector: PostgreSQL extension, SQL integration

2. Embeddings

Purpose: Convert text to numerical vectors for similarity search

Models (2026):

Model Dimensions Best For

voyage-3-large 1024 Claude apps (Anthropic recommended)

voyage-code-3 1024 Code search

text-embedding-3-large 3072 OpenAI apps, high accuracy

text-embedding-3-small 1536 OpenAI apps, cost-effective

bge-large-en-v1.5 1024 Open source, local deployment

multilingual-e5-large 1024 Multi-language support

3. Retrieval Strategies

Approaches:

• Dense Retrieval: Semantic similarity via embeddings

• Sparse Retrieval: Keyword matching (BM25, TF-IDF)

• Hybrid Search: Combine dense + sparse with weighted fusion

• Multi-Query: Generate multiple query variations

• HyDE: Generate hypothetical documents for better retrieval

4. Reranking

Purpose: Improve retrieval quality by reordering results

Methods:

• Cross-Encoders: BERT-based reranking (ms-marco-MiniLM)

• Cohere Rerank: API-based reranking

• Maximal Marginal Relevance (MMR): Diversity + relevance

• LLM-based: Use LLM to score relevance

Quick Start with LangGraph

from langgraph.graph import StateGraph, START, END from langchain_anthropic import ChatAnthropic from langchain_voyageai import VoyageAIEmbeddings from langchain_pinecone import PineconeVectorStore from langchain_core.documents import Document from langchain_core.prompts import ChatPromptTemplate from langchain_text_splitters import RecursiveCharacterTextSplitter from typing import TypedDict, Annotated

class RAGState(TypedDict): question: str context: list[Document] answer: str

Initialize components

llm = ChatAnthropic(model="claude-sonnet-4-6") embeddings = VoyageAIEmbeddings(model="voyage-3-large") vectorstore = PineconeVectorStore(index_name="docs", embedding=embeddings) retriever = vectorstore.as_retriever(search_kwargs={"k": 4})

RAG prompt

rag_prompt = ChatPromptTemplate.from_template( """Answer based on the context below. If you cannot answer, say so.

Context:
{context}

Question: {question}

Answer:"""

)

async def retrieve(state: RAGState) -> RAGState: """Retrieve relevant documents.""" docs = await retriever.ainvoke(state["question"]) return {"context": docs}

async def generate(state: RAGState) -> RAGState: """Generate answer from context.""" context_text = "\n\n".join(doc.page_content for doc in state["context"]) messages = rag_prompt.format_messages( context=context_text, question=state["question"] ) response = await llm.ainvoke(messages) return {"answer": response.content}

Build RAG graph

builder = StateGraph(RAGState) builder.add_node("retrieve", retrieve) builder.add_node("generate", generate) builder.add_edge(START, "retrieve") builder.add_edge("retrieve", "generate") builder.add_edge("generate", END)

rag_chain = builder.compile()

Use

result = await rag_chain.ainvoke({"question": "What are the main features?"}) print(result["answer"])

Advanced RAG Patterns

Pattern 1: Hybrid Search with RRF

from langchain_community.retrievers import BM25Retriever from langchain.retrievers import EnsembleRetriever

Sparse retriever (BM25 for keyword matching)

bm25_retriever = BM25Retriever.from_documents(documents) bm25_retriever.k = 10

Dense retriever (embeddings for semantic search)

dense_retriever = vectorstore.as_retriever(search_kwargs={"k": 10})

Combine with Reciprocal Rank Fusion weights

ensemble_retriever = EnsembleRetriever( retrievers=[bm25_retriever, dense_retriever], weights=[0.3, 0.7] # 30% keyword, 70% semantic )

Pattern 2: Multi-Query Retrieval

from langchain.retrievers.multi_query import MultiQueryRetriever

Generate multiple query perspectives for better recall

multi_query_retriever = MultiQueryRetriever.from_llm( retriever=vectorstore.as_retriever(search_kwargs={"k": 5}), llm=llm )

Single query → multiple variations → combined results

results = await multi_query_retriever.ainvoke("What is the main topic?")

Pattern 3: Contextual Compression

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

Compressor extracts only relevant portions

compressor = LLMChainExtractor.from_llm(llm)

compression_retriever = ContextualCompressionRetriever( base_compressor=compressor, base_retriever=vectorstore.as_retriever(search_kwargs={"k": 10}) )

Returns only relevant parts of documents

compressed_docs = await compression_retriever.ainvoke("specific query")

Pattern 4: Parent Document Retriever

from langchain.retrievers import ParentDocumentRetriever from langchain.storage import InMemoryStore from langchain_text_splitters import RecursiveCharacterTextSplitter

Small chunks for precise retrieval, large chunks for context

child_splitter = RecursiveCharacterTextSplitter(chunk_size=400, chunk_overlap=50) parent_splitter = RecursiveCharacterTextSplitter(chunk_size=2000, chunk_overlap=200)

Store for parent documents

docstore = InMemoryStore()

parent_retriever = ParentDocumentRetriever( vectorstore=vectorstore, docstore=docstore, child_splitter=child_splitter, parent_splitter=parent_splitter )

Add documents (splits children, stores parents)

await parent_retriever.aadd_documents(documents)

Retrieval returns parent documents with full context

results = await parent_retriever.ainvoke("query")

Pattern 5: HyDE (Hypothetical Document Embeddings)

from langchain_core.prompts import ChatPromptTemplate

class HyDEState(TypedDict): question: str hypothetical_doc: str context: list[Document] answer: str

hyde_prompt = ChatPromptTemplate.from_template( """Write a detailed passage that would answer this question:

Question: {question}

Passage:"""

)

async def generate_hypothetical(state: HyDEState) -> HyDEState: """Generate hypothetical document for better retrieval.""" messages = hyde_prompt.format_messages(question=state["question"]) response = await llm.ainvoke(messages) return {"hypothetical_doc": response.content}

async def retrieve_with_hyde(state: HyDEState) -> HyDEState: """Retrieve using hypothetical document.""" # Use hypothetical doc for retrieval instead of original query docs = await retriever.ainvoke(state["hypothetical_doc"]) return {"context": docs}

Build HyDE RAG graph

builder = StateGraph(HyDEState) builder.add_node("hypothetical", generate_hypothetical) builder.add_node("retrieve", retrieve_with_hyde) builder.add_node("generate", generate) builder.add_edge(START, "hypothetical") builder.add_edge("hypothetical", "retrieve") builder.add_edge("retrieve", "generate") builder.add_edge("generate", END)

hyde_rag = builder.compile()

Document Chunking Strategies

Recursive Character Text Splitter

from langchain_text_splitters import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter( chunk_size=1000, chunk_overlap=200, length_function=len, separators=["\n\n", "\n", ". ", " ", ""] # Try in order )

chunks = splitter.split_documents(documents)

Token-Based Splitting

from langchain_text_splitters import TokenTextSplitter

splitter = TokenTextSplitter( chunk_size=512, chunk_overlap=50, encoding_name="cl100k_base" # OpenAI tiktoken encoding )

Semantic Chunking

from langchain_experimental.text_splitter import SemanticChunker

splitter = SemanticChunker( embeddings=embeddings, breakpoint_threshold_type="percentile", breakpoint_threshold_amount=95 )

Markdown Header Splitter

from langchain_text_splitters import MarkdownHeaderTextSplitter

headers_to_split_on = [ ("#", "Header 1"), ("##", "Header 2"), ("###", "Header 3"), ]

splitter = MarkdownHeaderTextSplitter( headers_to_split_on=headers_to_split_on, strip_headers=False )

Vector Store Configurations

Pinecone (Serverless)

from pinecone import Pinecone, ServerlessSpec from langchain_pinecone import PineconeVectorStore

Initialize Pinecone client

pc = Pinecone(api_key=os.environ["PINECONE_API_KEY"])

Create index if needed

if "my-index" not in pc.list_indexes().names(): pc.create_index( name="my-index", dimension=1024, # voyage-3-large dimensions metric="cosine", spec=ServerlessSpec(cloud="aws", region="us-east-1") )

Create vector store

index = pc.Index("my-index") vectorstore = PineconeVectorStore(index=index, embedding=embeddings)

Weaviate

import weaviate from langchain_weaviate import WeaviateVectorStore

client = weaviate.connect_to_local() # or connect_to_weaviate_cloud()

vectorstore = WeaviateVectorStore( client=client, index_name="Documents", text_key="content", embedding=embeddings )

Chroma (Local Development)

from langchain_chroma import Chroma

vectorstore = Chroma( collection_name="my_collection", embedding_function=embeddings, persist_directory="./chroma_db" )

pgvector (PostgreSQL)

from langchain_postgres.vectorstores import PGVector

connection_string = "postgresql+psycopg://user:pass@localhost:5432/vectordb"

vectorstore = PGVector( embeddings=embeddings, collection_name="documents", connection=connection_string, )

Retrieval Optimization

1. Metadata Filtering

from langchain_core.documents import Document

Add metadata during indexing

docs_with_metadata = [] for doc in documents: doc.metadata.update({ "source": doc.metadata.get("source", "unknown"), "category": determine_category(doc.page_content), "date": datetime.now().isoformat() }) docs_with_metadata.append(doc)

Filter during retrieval

results = await vectorstore.asimilarity_search( "query", filter={"category": "technical"}, k=5 )

2. Maximal Marginal Relevance (MMR)

Balance relevance with diversity

results = await vectorstore.amax_marginal_relevance_search( "query", k=5, fetch_k=20, # Fetch 20, return top 5 diverse lambda_mult=0.5 # 0=max diversity, 1=max relevance )

3. Reranking with Cross-Encoder

from sentence_transformers import CrossEncoder

reranker = CrossEncoder('cross-encoder/ms-marco-MiniLM-L-6-v2')

async def retrieve_and_rerank(query: str, k: int = 5) -> list[Document]: # Get initial results candidates = await vectorstore.asimilarity_search(query, k=20)

# Rerank
pairs = [[query, doc.page_content] for doc in candidates]
scores = reranker.predict(pairs)

# Sort by score and take top k
ranked = sorted(zip(candidates, scores), key=lambda x: x[1], reverse=True)
return [doc for doc, score in ranked[:k]]

4. Cohere Rerank

from langchain.retrievers import CohereRerank from langchain_cohere import CohereRerank

reranker = CohereRerank(model="rerank-english-v3.0", top_n=5)

Wrap retriever with reranking

reranked_retriever = ContextualCompressionRetriever( base_compressor=reranker, base_retriever=vectorstore.as_retriever(search_kwargs={"k": 20}) )

Prompt Engineering for RAG

Contextual Prompt with Citations

rag_prompt = ChatPromptTemplate.from_template( """Answer the question based on the context below. Include citations using [1], [2], etc.

If you cannot answer based on the context, say "I don't have enough information."

Context:
{context}

Question: {question}

Instructions:
1. Use only information from the context
2. Cite sources with [1], [2] format
3. If uncertain, express uncertainty

Answer (with citations):"""

)

Structured Output for RAG

from pydantic import BaseModel, Field

class RAGResponse(BaseModel): answer: str = Field(description="The answer based on context") confidence: float = Field(description="Confidence score 0-1") sources: list[str] = Field(description="Source document IDs used") reasoning: str = Field(description="Brief reasoning for the answer")

Use with structured output

structured_llm = llm.with_structured_output(RAGResponse)

Evaluation Metrics

from typing import TypedDict

class RAGEvalMetrics(TypedDict): retrieval_precision: float # Relevant docs / retrieved docs retrieval_recall: float # Retrieved relevant / total relevant answer_relevance: float # Answer addresses question faithfulness: float # Answer grounded in context context_relevance: float # Context relevant to question

async def evaluate_rag_system( rag_chain, test_cases: list[dict] ) -> RAGEvalMetrics: """Evaluate RAG system on test cases.""" metrics = {k: [] for k in RAGEvalMetrics.annotations}

for test in test_cases:
    result = await rag_chain.ainvoke({"question": test["question"]})

    # Retrieval metrics
    retrieved_ids = {doc.metadata["id"] for doc in result["context"]}
    relevant_ids = set(test["relevant_doc_ids"])

    precision = len(retrieved_ids & relevant_ids) / len(retrieved_ids)
    recall = len(retrieved_ids & relevant_ids) / len(relevant_ids)

    metrics["retrieval_precision"].append(precision)
    metrics["retrieval_recall"].append(recall)

    # Use LLM-as-judge for quality metrics
    quality = await evaluate_answer_quality(
        question=test["question"],
        answer=result["answer"],
        context=result["context"],
        expected=test.get("expected_answer")
    )
    metrics["answer_relevance"].append(quality["relevance"])
    metrics["faithfulness"].append(quality["faithfulness"])
    metrics["context_relevance"].append(quality["context_relevance"])

return {k: sum(v) / len(v) for k, v in metrics.items()}