# RAG Patterns

> When to activate: RAG, retrieval augmented generation, LangChain, LlamaIndex, vector stores, embeddings, retrieval, chunking, reranking, hybrid search, pgvector, Chroma, RAG evaluation, contextual compression

- Skill: `mattakushi432/rag-patterns` (Agent Skill)
- Install (CLI): `npx skillmds@latest add mattakushi432/rag-patterns`
- Raw SKILL.md: https://api.skillmd.com/api/skills/mattakushi432/rag-patterns/raw
- Safety review: pending
- Works with: Claude Code, Claude.ai, OpenAI Codex
- Category: AI & ML
- Author: Mattakushi432 (https://skillmd.com/u/mattakushi432)
- Updated: 2026-09-17
- Page: https://skillmd.com/skills/mattakushi432/rag-patterns

---

# RAG (Retrieval-Augmented Generation) Patterns

## Pipeline Architecture

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

## Document Processing

```python
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

```python
# 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)

```python
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)

```python
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)

```python
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

```python
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)

```python
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)

```python
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

```python
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)

```python
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.*