RAG Patterns

RAG (Retrieval-Augmented Generation) Patterns

Pipeline Architecture

Documents → Chunking → Embedding → Vector Store
                                       ↓
Query → Embedding → Retrieval (top-k) → Reranking → LLM → Response

Document Processing

from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.document_loaders import PyPDFLoader, DirectoryLoader

# Load
loader = DirectoryLoader("./docs", glob="**/*.pdf", loader_cls=PyPDFLoader)
documents = loader.load()

# Chunk — overlap prevents context loss at boundaries
splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200,
    separators=["\n\n", "\n", ". ", " ", ""],  # try larger separators first
    length_function=len,
)
chunks = splitter.split_documents(documents)

# Preserve metadata for filtering/citation
for chunk in chunks:
    chunk.metadata.update({
        "source": chunk.metadata.get("source", "unknown"),
        "page": chunk.metadata.get("page", 0),
    })

Chunking Strategies

# Semantic chunking (split at meaning boundaries)
from langchain_experimental.text_splitter import SemanticChunker
from langchain_openai import OpenAIEmbeddings
semantic_splitter = SemanticChunker(
    OpenAIEmbeddings(),
    breakpoint_threshold_type="percentile",
    breakpoint_threshold_amount=95,
)

# Markdown-aware (preserves headers as context)
from langchain.text_splitter import MarkdownHeaderTextSplitter
md_splitter = MarkdownHeaderTextSplitter(
    headers_to_split_on=[("#", "h1"), ("##", "h2"), ("###", "h3")]
)

# Parent-child chunking (index small, retrieve large)
from langchain.retrievers import ParentDocumentRetriever
from langchain.storage import InMemoryStore

parent_splitter = RecursiveCharacterTextSplitter(chunk_size=2000)
child_splitter = RecursiveCharacterTextSplitter(chunk_size=256)
store = InMemoryStore()
retriever = ParentDocumentRetriever(
    vectorstore=vectorstore, docstore=store,
    child_splitter=child_splitter, parent_splitter=parent_splitter,
)

Vector Stores

pgvector (Postgres-native, good when you already run Postgres)

from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import PGVector

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

vectorstore = PGVector(
    connection_string=settings.database_url,
    embedding_function=embeddings,
    collection_name="documents",
    pre_delete_collection=False,
)
vectorstore.add_documents(chunks)

results = vectorstore.similarity_search_with_score(
    query="What is the refund policy?",
    k=5,
    filter={"source": "policy.pdf"},  # metadata filtering
)

Chroma (embedded, good for local dev / small deployments)

from langchain_community.vectorstores import Chroma
from langchain_community.embeddings import HuggingFaceEmbeddings

embeddings = HuggingFaceEmbeddings(model_name="BAAI/bge-small-en-v1.5")
vectorstore = Chroma.from_documents(chunks, embeddings, persist_directory="./chroma_db")
retriever = vectorstore.as_retriever(search_kwargs={"k": 5})

Hybrid Search (Dense + Sparse)

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

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

bm25_retriever = BM25Retriever.from_documents(chunks)
bm25_retriever.k = 5

ensemble_retriever = EnsembleRetriever(
    retrievers=[bm25_retriever, dense_retriever],
    weights=[0.4, 0.6],  # weight towards semantic
)

Reranking

from sentence_transformers import CrossEncoder

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

def rerank(query: str, documents: list, top_k: int = 3) -> list:
    pairs = [(query, doc.page_content) for doc in documents]
    scores = reranker.predict(pairs)
    ranked = sorted(zip(scores, documents), reverse=True)
    return [doc for _, doc in ranked[:top_k]]

initial_results = retriever.invoke(query)           # get 10 candidates
reranked = rerank(query, initial_results, top_k=3)   # keep top 3

RAG Chain — Two Idioms

Manual LCEL Chain (fine-grained control over the prompt)

from langchain_anthropic import ChatAnthropic
from langchain.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough

template = """Answer based only on the provided context.
If the context doesn't contain the answer, say "I don't have that information."

Context:
{context}

Question: {question}"""

prompt = ChatPromptTemplate.from_template(template)
model = ChatAnthropic(model="claude-sonnet-4-6")

def format_docs(docs) -> str:
    return "\n\n---\n\n".join([
        f"[Source: {doc.metadata.get('source', 'unknown')}]\n{doc.page_content}"
        for doc in docs
    ])

rag_chain = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | prompt
    | model
    | StrOutputParser()
)

answer = rag_chain.invoke("What is the return policy?")

RetrievalQA Wrapper (faster to stand up, less prompt control)

from langchain.chains import RetrievalQA

qa_chain = RetrievalQA.from_chain_type(
    llm=model, retriever=retriever, return_source_documents=True
)
result = qa_chain.invoke({"query": "What is the retry policy?"})

Advanced: Contextual Compression

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

compressor = LLMChainExtractor.from_llm(model)
compression_retriever = ContextualCompressionRetriever(
    base_compressor=compressor,
    base_retriever=dense_retriever,
)
# Extracts only relevant sentences from each retrieved chunk
compressed_docs = compression_retriever.invoke("What is the SLA?")

Evaluation (RAGAS)

from ragas import evaluate
from ragas.metrics import faithfulness, answer_relevancy, context_precision, context_recall

dataset = {
    "question": ["What is the retry policy?"],
    "answer": [generated_answer],
    "contexts": [[ctx.page_content for ctx in retrieved_docs]],
    "ground_truth": ["The retry policy uses exponential backoff..."],
}

results = evaluate(
    dataset=dataset,
    metrics=[faithfulness, answer_relevancy, context_precision, context_recall],
)
# faithfulness: is the answer grounded in context?
# answer_relevancy: does the answer address the question?
# context_precision / context_recall: are retrieved chunks relevant and complete?

Checklist

• Chunk size and overlap tuned for the document type (prose vs. code vs. tables)
• Metadata (source, page) preserved through chunking for citation
• Reranking applied when retrieval k is much larger than what's fed to the LLM
• Prompt explicitly instructs the model to say "I don't know" when context is insufficient
• RAGAS or an equivalent eval run before shipping a prompt/retrieval change

See Also

• skills/ml-ai/pytorch-patterns.md
• skills/ml-ai/llm-integration.md
• skills/ml-ai/vector-db-patterns.md

Merged from the former skills/python-ecosystem/rag-patterns.md and skills/ml-ai/rag-patterns.md, which had diverged into two different filename-colliding skills. RAG is ML/AI-specific, so this canonical version now lives under ml-ai/ only.