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

Vector Store with pgvector

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,
)

# Index documents
vectorstore.add_documents(chunks)

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

Reranking

from sentence_transformers import CrossEncoder

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

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

RAG Chain

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: list[Document]) -> str:
    return "\n\n---\n\n".join([
        f"[Source: {doc.metadata.get('source', 'unknown')}]\n{doc.page_content}"
        for doc in docs
    ])

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

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

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

Evaluation

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

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