rag-pipelines

name	rag-pipelines
description	RAG pipeline design — chunking, embeddings, retrieval strategies, evaluation, and demo patterns.
context	fork
globs	[]
alwaysApply	false

RAG Pipelines

Retrieval-Augmented Generation pipeline design: chunking, embeddings, vector stores, retrieval strategies, and evaluation.

---

Pipeline Architecture

Documents → Chunk → Embed → Store → Retrieve → Generate

---

Chunking Strategies

Strategy	Best For	Chunk Size
Fixed-size	General docs	500-1000 tokens
Semantic	Long-form content	By paragraph/section
Code-aware	Source code	By function/class
Recursive	Mixed content	1000 → 500 → 200 tokens

from langchain.text_splitter import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200,
    separators=["\n\n", "\n", ". ", " ", ""]
)

---

Embedding Models

Model	Dimensions	Speed	Quality
text-embedding-3-small	1536	Fast	Good
text-embedding-3-large	3072	Medium	Best
bge-large-en	1024	Fast	Good
e5-large-v2	1024	Fast	Good

---

Vector Stores

Store	Use Case	Scaling
Pinecone	Production, managed	Auto-scales
Weaviate	Production, self-hosted	Horizontal
pgvector	PostgreSQL shops	Vertical
Chroma	Prototyping, local	Single-node

from langchain.vectorstores import Chroma

vector_store = Chroma.from_documents(
    documents=chunks,
    embedding=embeddings,
    persist_directory="./chroma_db"
)

---

Retrieval Strategies

Similarity Search

retriever = vector_store.as_retriever(search_type="similarity", k=4)

MMR (Diversity-focused)

retriever = vector_store.as_retriever(
    search_type="mmr",
    search_kwargs={"k": 4, "lambda_mult": 0.7}
)

Hybrid (BM25 + Semantic)

from langchain.retrievers import EnsembleRetriever

retriever = EnsembleRetriever(
    retrievers=[bm25_retriever, vector_retriever],
    weights=[0.3, 0.7]
)

Self-Querying

from langchain.retrievers.self_query.base import SelfQueryRetriever

retriever = SelfQueryRetriever.from_llm(
    llm, vector_store, document_contents, metadata_field_info
)

---

Evaluation (RAGAS)

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

result = evaluate(
    dataset=eval_dataset,
    metrics=[faithfulness, answer_relevance, context_precision]
)

Metric	Target
Faithfulness	≥0.8
Answer Relevance	≥0.8
Context Precision	≥0.7

---

Demo Patterns

Gradio

import gradio as gr

def answer(question):
    return qa_chain.run(question)

gr.Interface(fn=answer, inputs="text", outputs="text").launch()

Streamlit

import streamlit as st

question = st.text_input("Ask a question")
if question:
    st.write(qa_chain.run(question))

---

Best Practices

• Chunk overlap prevents context loss at boundaries
• Metadata filtering improves retrieval precision
• Hybrid retrieval balances recall and precision
• Evaluate before deploying — use RAGAS metrics
• Cache frequent queries to reduce latency