rag-pipeline

RAG Pipeline

Build production-ready Retrieval-Augmented Generation pipelines with hybrid search, re-ranking, and quality evaluation.

When to Use

• Building a semantic search system over documents
• Answering questions from a knowledge base (PDFs, Markdown, code)
• Creating a retrieval layer for an AI agent
• Indexing project documentation, skills, or bundles into a vector store
• Improving an existing RAG pipeline's accuracy or performance

Available Operations

1. Ingest documents (load, split, enrich with metadata)
2. Generate embeddings and index into pgvector
3. Configure hybrid retrieval (semantic + keyword BM25)
4. Add re-ranking for precision
5. Build a query chain with LLM
6. Evaluate retrieval quality with golden datasets

Multi-Step Workflow

Step 1: Set Up Environment

Install required dependencies and verify database connectivity.

pip install langchain langchain-openai langchain-postgres langchain-community langchain-cohere pgvector rank-bm25

Verify pgvector is available:

psql $DATABASE_URL -c "CREATE EXTENSION IF NOT EXISTS vector;"

Step 2: Ingest Documents

Load documents from the target directory and split into chunks with appropriate overlap.

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

# Load documents by type
loader = DirectoryLoader(
    "./documents/",
    glob="**/*.md",
    loader_cls=UnstructuredMarkdownLoader
)
docs = loader.load()

# Split with Markdown-aware separators
splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200,
    separators=["\n## ", "\n### ", "\n\n", "\n", ". ", " "]
)
chunks = splitter.split_documents(docs)

Step 3: Enrich Chunks with Metadata

Every chunk must carry metadata for filtering and traceability.

from datetime import datetime

for chunk in chunks:
    chunk.metadata.update({
        "source": chunk.metadata.get("source", "unknown"),
        "doc_type": classify_document(chunk),  # skill, agent_md, prd, code
        "language": detect_language(chunk),
        "created_at": datetime.now().isoformat(),
    })