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()}