rag-pipeline-gen

name	rag-pipeline-gen
type	command
description	Generate complete RAG pipeline with embeddings, vector DB, and retrieval
category	ai
version	1.0.0
author	Jeremy Longshore
shortcut	rpg
estimated_time	5-10 minutes

RAG Pipeline Generator

Generate a complete, production-ready RAG (Retrieval-Augmented Generation) pipeline with document ingestion, embedding, vector storage, retrieval, and LLM integration.

What You'll Get

When you run this command, you'll receive:

1. Document ingestion pipeline with chunking strategies
2. Embedding generation with OpenAI or open-source models
3. Vector database integration (Pinecone, Qdrant, or ChromaDB)
4. Retrieval system with reranking and hybrid search
5. LLM integration for answer generation
6. API server (FastAPI) for production deployment
7. Docker configuration for containerized deployment
8. Testing suite with evaluation metrics

Usage

/rag-pipeline-gen <vector_db> [options]

Vector Databases: pinecone, qdrant, chromadb, weaviate

Examples:

• /rpg pinecone - Generate RAG pipeline with Pinecone
• /rpg qdrant - Generate RAG pipeline with Qdrant Cloud
• /rpg chromadb - Generate RAG pipeline with ChromaDB (local development)

Generated Output

Example: Pinecone RAG Pipeline

Input:

/rpg pinecone

Output:

1. Project Structure

rag-pipeline/
├── src/
│   ├── ingestion/
│   │   ├── __init__.py
│   │   ├── document_loader.py    # Load PDFs, text, web pages
│   │   ├── chunker.py             # Text chunking strategies
│   │   └── embedder.py            # Generate embeddings
│   ├── retrieval/
│   │   ├── __init__.py
│   │   ├── vector_store.py        # Vector DB operations
│   │   ├── retriever.py           # Query and retrieval
│   │   └── reranker.py            # Reranking results
│   ├── generation/
│   │   ├── __init__.py
│   │   ├── llm_client.py          # LLM integration
│   │   └── prompt_templates.py    # Prompt engineering
│   ├── api/
│   │   ├── __init__.py
│   │   ├── main.py                # FastAPI server
│   │   └── models.py              # Pydantic models
│   └── config/
│       ├── __init__.py
│       └── settings.py            # Configuration
├── tests/
│   ├── __init__.py
│   ├── test_ingestion.py
│   ├── test_retrieval.py
│   └── test_integration.py
├── notebooks/
│   └── evaluation.ipynb           # RAG evaluation
├── docker/
│   ├── Dockerfile
│   └── docker-compose.yml
├── requirements.txt
├── .env.example
└── README.md

2. Document Loader (src/ingestion/document_loader.py)

from pathlib import Path
from typing import List, Dict
import PyPDF2
from langchain.text_splitter import RecursiveCharacterTextSplitter

class DocumentLoader:
    """Load and process documents from various sources."""

    def __init__(self, chunk_size: int = 1000, chunk_overlap: int = 200):
        self.chunker = RecursiveCharacterTextSplitter(
            chunk_size=chunk_size,
            chunk_overlap=chunk_overlap,
            separators=["\n\n", "\n", ". ", " ", ""]
        )

    def load_pdf(self, file_path: Path) -> List[Dict]:
        """Load and chunk PDF file."""
        with open(file_path, 'rb') as file:
            pdf_reader = PyPDF2.PdfReader(file)
            text = ""
            for page_num, page in enumerate(pdf_reader.pages):
                text += page.extract_text()

        # Chunk text
        chunks = self.chunker.split_text(text)

        # Create document objects
        documents = [
            {
                "text": chunk,
                "metadata": {
                    "source": (file_path),
                    : i // ((chunks) // (pdf_reader.pages) + ),
                    : i
                }
            }
             i, chunk  (chunks)
        ]

         documents

     () -> []:
        
         (file_path, , encoding=)  file:
            text = file.read()

        chunks = .chunker.split_text(text)

         [
            {
                : chunk,
                : {
                    : (file_path),
                    : i
                }
            }
             i, chunk  (chunks)
        ]

     () -> []:
        
        documents = []

         file_path  directory.glob():
             file_path.suffix == :
                documents.extend(.load_pdf(file_path))
             file_path.suffix  [, ]:
                documents.extend(.load_text(file_path))

         documents

3. Embedder (src/ingestion/embedder.py)

from typing import List
import openai
from tenacity import retry, stop_after_attempt, wait_exponential

class Embedder:
    """Generate embeddings for text chunks."""

    def __init__(
        self,
        model: str = "text-embedding-3-small",
        api_key: str = None
    ):
        self.model = model
        self.client = openai.OpenAI(api_key=api_key)

    @retry(
        stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=1, min=2, max=10)
    )
    async def embed(self, text: str) -> List[float]:
        """Generate embedding for single text."""
        response = await self.client.embeddings.create(
            model=self.model,
            input=text
        )
        return response.data[0].embedding

    async def embed_batch(
        self,
        texts: List[str],
        batch_size: int = 
    ) -> [[]]:
        
        embeddings = []

         i  (, (texts), batch_size):
            batch = texts[i:i + batch_size]

            response =  .client.embeddings.create(
                model=.model,
                =batch
            )

            batch_embeddings = [item.embedding  item  response.data]
            embeddings.extend(batch_embeddings)

         embeddings

4. Vector Store (src/retrieval/vector_store.py)

from typing import List, Dict, Optional
from pinecone import Pinecone, ServerlessSpec
import hashlib

class PineconeVectorStore:
    """Pinecone vector database operations."""

    def __init__(
        self,
        api_key: str,
        index_name: str,
        dimension: int = 1536,
        metric: str = "cosine"
    ):
        self.pc = Pinecone(api_key=api_key)
        self.index_name = index_name

        # Create index if doesn't exist
        if index_name not in self.pc.list_indexes().names():
            self.pc.create_index(
                name=index_name,
                dimension=dimension,
                metric=metric,
                spec=ServerlessSpec(cloud="aws", region="us-east-1")
            )

        self.index = self.pc.Index(index_name)

    async def upsert_documents(
        self,
        documents: List[Dict],
        embeddings: List[List[float]],
        namespace: str = "default"
    ):
        """Insert documents with embeddings into vector store."""
        vectors = []

         doc, embedding  (documents, embeddings):
            
            doc_id = ._generate_id(doc)

            vectors.append({
                : doc_id,
                : embedding,
                : {
                    : doc[],
                    **doc[]
                }
            })

        
        batch_size = 
         i  (, (vectors), batch_size):
            batch = vectors[i:i + batch_size]
            .index.upsert(vectors=batch, namespace=namespace)

         (vectors)

      () -> []:
        
        results = .index.query(
            vector=query_embedding,
            top_k=top_k,
            namespace=namespace,
            =,
            include_metadata=
        )

         [
            {
                : [],
                : [],
                : [][],
                : {
                    k: v  k, v  [].items()
                     k != 
                }
            }
               results[]
        ]

     () -> :
        
        content = 
         hashlib.md5(content.encode()).hexdigest()

      ():
        
        .index.delete(namespace=namespace, delete_all=)

5. Retriever with Reranking (src/retrieval/retriever.py)

from typing import List, Dict
import cohere
from src.ingestion.embedder import Embedder
from src.retrieval.vector_store import PineconeVectorStore

class RAGRetriever:
    """Retrieve and rerank documents for RAG."""

    def __init__(
        self,
        embedder: Embedder,
        vector_store: PineconeVectorStore,
        cohere_api_key: Optional[str] = None
    ):
        self.embedder = embedder
        self.vector_store = vector_store
        self.cohere = cohere.Client(cohere_api_key) if cohere_api_key else None

    async def retrieve(
        self,
        query: str,
        top_k: int = 5,
        initial_k: int = 20,
        rerank: bool = True,
        namespace: str = "default",
        filter: Optional[Dict] = None
    ) -> List[Dict]:
        """Retrieve relevant documents with optional reranking."""

        # 1. Embed query
        query_embedding = await self.embedder.embed(query)

        
        k = initial_k  rerank  top_k
        results =  .vector_store.search(
            query_embedding=query_embedding,
            top_k=k,
            namespace=namespace,
            =
        )

        
         rerank  .cohere  (results) > top_k:
            documents = [r[]  r  results]

            reranked = .cohere.rerank(
                query=query,
                documents=documents,
                top_n=top_k,
                model=
            )

            
            results = [results[r.index]  r  reranked.results]

         results[:top_k]

      () -> []:
        

        
        results_standard =  .retrieve(
            query=query,
            top_k=top_k,
            rerank=,
            namespace=namespace
        )

        
        query_variants =  ._generate_query_variants(query)
        results_variants = []

         variant  query_variants[:]:  
            variant_results =  .retrieve(
                query=variant,
                top_k=top_k,
                rerank=,
                namespace=namespace
            )
            results_variants.extend(variant_results)

        
        seen_ids = ()
        combined_results = []

         result  results_standard + results_variants:
             result[]   seen_ids:
                seen_ids.add(result[])
                combined_results.append(result)

        
        combined_results.sort(key= x: x[], reverse=)
         combined_results[:top_k]

      () -> []:
        
        
         [query]  

6. LLM Client (src/generation/llm_client.py)

from typing import List, Dict
from anthropic import AsyncAnthropic

class LLMClient:
    """Generate answers using LLM."""

    def __init__(self, api_key: str, model: str = "claude-3-haiku-20240307"):
        self.client = AsyncAnthropic(api_key=api_key)
        self.model = model

    async def generate_answer(
        self,
        question: str,
        context: List[Dict],
        max_tokens: int = 1024
    ) -> Dict:
        """Generate answer using retrieved context."""

        # Format context
        context_text = "\n\n".join([
            f"Source {i+1} ({ctx['metadata'].get('source', 'Unknown')}):\n{ctx['text']}"
            for i, ctx in enumerate(context)
        ])

        # Build prompt
        prompt = f"""Answer the question using ONLY the provided context. If the answer cannot be found in the context, say "I don't have enough information to answer this question."

Context:
{context_text}

Question: {question}

Answer:"""

        # Generate response
        message =  .client.messages.create(
            model=.model,
            max_tokens=max_tokens,
            messages=[{: , : prompt}]
        )

         {
            : message.content[].text,
            : [
                {
                    : ctx[].get(, ),
                    : ctx[]
                }
                 ctx  context
            ],
            : {
                : message.usage.input_tokens,
                : message.usage.output_tokens
            }
        }

7. FastAPI Server (src/api/main.py)

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import List, Optional
import asyncio

from src.ingestion.document_loader import DocumentLoader
from src.ingestion.embedder import Embedder
from src.retrieval.vector_store import PineconeVectorStore
from src.retrieval.retriever import RAGRetriever
from src.generation.llm_client import LLMClient
from src.config.settings import Settings

app = FastAPI(title="RAG API", version="1.0.0")
settings = Settings()

# Initialize components
embedder = Embedder(api_key=settings.openai_api_key)
vector_store = PineconeVectorStore(
    api_key=settings.pinecone_api_key,
    index_name=settings.pinecone_index_name
)
retriever = RAGRetriever(
    embedder=embedder,
    vector_store=vector_store,
    cohere_api_key=settings.cohere_api_key
)
llm = LLMClient(api_key=settings.anthropic_api_key)

class QueryRequest(BaseModel):
    question: str
    top_k: int = 5
    rerank: bool = True
    namespace: str = "default"

class QueryResponse(BaseModel):
    answer: str
    sources: List[Dict]
    usage: Dict

@app.post(, response_model=QueryResponse)
  ():
    
    :
        
        context =  retriever.retrieve(
            query=request.question,
            top_k=request.top_k,
            rerank=request.rerank,
            namespace=request.namespace
        )

        
        result =  llm.generate_answer(
            question=request.question,
            context=context
        )

         QueryResponse(**result)

     Exception  e:
         HTTPException(status_code=, detail=(e))

 ():
    directory: 
    namespace:  = 


  ():
    
    :
        
        loader = DocumentLoader()
        documents = loader.load_directory(Path(request.directory))

        
        texts = [doc[]  doc  documents]
        embeddings =  embedder.embed_batch(texts)

        
        num_uploaded =  vector_store.upsert_documents(
            documents=documents,
            embeddings=embeddings,
            namespace=request.namespace
        )

         {
            : ,
            : num_uploaded
        }

     Exception  e:
         HTTPException(status_code=, detail=(e))


  ():
    
     {: }

 __name__ == :
     uvicorn
    uvicorn.run(app, host=, port=)

8. Docker Configuration (docker/Dockerfile)

FROM python:3.11-slim

WORKDIR /app

# Install dependencies
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Copy application
COPY src/ ./src/

# Environment variables
ENV OPENAI_API_KEY=""
ENV PINECONE_API_KEY=""
ENV ANTHROPIC_API_KEY=""
ENV COHERE_API_KEY=""

# Expose port
EXPOSE 8000

# Run application
CMD ["uvicorn", "src.api.main:app", "--host", "0.0.0.0", "--port", "8000"]

9. Docker Compose (docker/docker-compose.yml)

version: '3.8'

services:
  rag-api:
    build:
      context: .
      dockerfile: docker/Dockerfile
    environment:
      - OPENAI_API_KEY=${OPENAI_API_KEY}
      - PINECONE_API_KEY=${PINECONE_API_KEY}
      - ANTHROPIC_API_KEY=${ANTHROPIC_API_KEY}
      - COHERE_API_KEY=${COHERE_API_KEY}
      - PINECONE_INDEX_NAME=rag-index
    ports:
      - "8000:8000"
    volumes:
      - ./data:/app/data  # Mount for document ingestion

10. Requirements (requirements.txt)

fastapi==0.109.0
uvicorn[standard]==0.27.0
anthropic==0.18.1
openai==1.12.0
pinecone-client==3.0.0
cohere==4.47
PyPDF2==3.0.1
langchain==0.1.6
pydantic==2.6.0
pydantic-settings==2.1.0
tenacity==8.2.3
python-multipart==0.0.9

11. Example Usage

import requests

# Query RAG system
response = requests.post(
    "http://localhost:8000/query",
    json={
        "question": "What is quantum computing?",
        "top_k": 5,
        "rerank": True
    }
)

result = response.json()
print(f"Answer: {result['answer']}")
print(f"Sources: {result['sources']}")
print(f"Tokens: {result['usage']}")

Response:

{
  "answer": "Quantum computing is a type of computing that uses quantum-mechanical phenomena...",
  "sources": [
    {"source": "quantum_physics.pdf", "score": 0.92},
    {"source": "computing_basics.pdf", "score": 0.87}
  ],
  "usage": {
    "input_tokens": 450,
    "output_tokens": 120
  }
}

Features Included

Production-Ready:

• Document ingestion (PDF, text, web)
• Intelligent chunking strategies
• Batch embedding generation
• Vector database integration
• Retrieval with reranking
• LLM answer generation
• FastAPI REST API
• Docker deployment
• Error handling and retries
• Source citation

Advanced Features:

• Hybrid search (multiple strategies)
• Metadata filtering
• Namespace support (multi-tenancy)
• Reranking for better relevance
• Token usage tracking
• Health check endpoints

Time Savings

Manual implementation: 16-24 hours

• Document loading and chunking
• Embedding generation
• Vector DB setup and integration
• Retrieval logic
• LLM integration
• API server
• Docker configuration
• Testing

With this command: 5-10 minutes

• Run command
• Add API keys
• Deploy to production

ROI: 96-144x time multiplier

---

Next Steps:

1. Run /rpg pinecone or /rpg qdrant or /rpg chromadb
2. Copy generated code to your project
3. Install dependencies: pip install -r requirements.txt
4. Set API keys in .env file
5. Ingest documents: POST /ingest
6. Query system: POST /query
7. Deploy: docker-compose up -d

Production checklist:

• Set up monitoring (Prometheus, Grafana)
• Implement rate limiting
• Add authentication (JWT)
• Configure logging (structured logs)
• Set up alerting (Sentry, PagerDuty)
• Run evaluation (RAGAS metrics)
• Load testing (Locust, k6)

Estimated monthly cost: $50-$200 depending on document volume and query rate.