RAG Pipeline Gen

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": str(file_path),
                    "page": i // (len(chunks) // len(pdf_reader.pages) + 1),
                    "chunk_index": i
                }
            }
            for i, chunk in enumerate(chunks)
        ]

        return documents

    def load_text(self, file_path: Path) -> List[Dict]:
        """Load and chunk text file."""
        with open(file_path, 'r', encoding='utf-8') as file:
            text = file.read()

        chunks = self.chunker.split_text(text)

        return [
            {
                "text": chunk,
                "metadata": {
                    "source": str(file_path),
                    "chunk_index": i
                }
            }
            for i, chunk in enumerate(chunks)
        ]

    def load_directory(self, directory: Path) -> List[Dict]:
        """Load all documents from directory."""
        documents = []

        for file_path in directory.glob("**/*"):
            if file_path.suffix == ".pdf":
                documents.extend(self.load_pdf(file_path))
            elif file_path.suffix in [".txt", ".md"]:
                documents.extend(self.load_text(file_path))

        return 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 = 100
    ) -> List[List[float]]:
        """Generate embeddings for batch of texts."""
        embeddings = []

        for i in range(0, len(texts), batch_size):
            batch = texts[i:i + batch_size]

            response = await self.client.embeddings.create(
                model=self.model,
                input=batch
            )

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

        return 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 = []

        for doc, embedding in zip(documents, embeddings):
            # Generate unique ID
            doc_id = self._generate_id(doc)

            vectors.append({
                "id": doc_id,
                "values": embedding,
                "metadata": {
                    "text": doc["text"],
                    **doc["metadata"]
                }
            })

        # Batch upsert
        batch_size = 100
        for i in range(0, len(vectors), batch_size):
            batch = vectors[i:i + batch_size]
            self.index.upsert(vectors=batch, namespace=namespace)

        return len(vectors)

    async def search(
        self,
        query_embedding: List[float],
        top_k: int = 5,
        namespace: str = "default",
        filter: Optional[Dict] = None
    ) -> List[Dict]:
        """Search for similar documents."""
        results = self.index.query(
            vector=query_embedding,
            top_k=top_k,
            namespace=namespace,
            filter=filter,
            include_metadata=True
        )

        return [
            {
                "id": match["id"],
                "score": match["score"],
                "text": match["metadata"]["text"],
                "metadata": {
                    k: v for k, v in match["metadata"].items()
                    if k != "text"
                }
            }
            for match in results["matches"]
        ]

    def _generate_id(self, document: Dict) -> str:
        """Generate unique ID from document content."""
        content = f"{document['text']}{document['metadata']}"
        return hashlib.md5(content.encode()).hexdigest()

    async def delete_namespace(self, namespace: str):
        """Delete all vectors in namespace."""
        self.index.delete(namespace=namespace, delete_all=True)

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)

        # 2. Vector search
        k = initial_k if rerank else top_k
        results = await self.vector_store.search(
            query_embedding=query_embedding,
            top_k=k,
            namespace=namespace,
            filter=filter
        )

        # 3. Rerank if enabled
        if rerank and self.cohere and len(results) > top_k:
            documents = [r["text"] for r in results]

            reranked = self.cohere.rerank(
                query=query,
                documents=documents,
                top_n=top_k,
                model="rerank-english-v2.0"
            )

            # Reorder results based on reranking
            results = [results[r.index] for r in reranked.results]

        return results[:top_k]

    async def hybrid_retrieve(
        self,
        query: str,
        top_k: int = 5,
        namespace: str = "default"
    ) -> List[Dict]:
        """Retrieve using multiple strategies and combine."""

        # Strategy 1: Standard retrieval
        results_standard = await self.retrieve(
            query=query,
            top_k=top_k,
            rerank=False,
            namespace=namespace
        )

        # Strategy 2: Multi-query retrieval
        query_variants = await self._generate_query_variants(query)
        results_variants = []

        for variant in query_variants[:2]:  # Use 2 variants
            variant_results = await self.retrieve(
                query=variant,
                top_k=top_k,
                rerank=False,
                namespace=namespace
            )
            results_variants.extend(variant_results)

        # Combine and deduplicate
        seen_ids = set()
        combined_results = []

        for result in results_standard + results_variants:
            if result["id"] not in seen_ids:
                seen_ids.add(result["id"])
                combined_results.append(result)

        # Sort by score and return top-k
        combined_results.sort(key=lambda x: x["score"], reverse=True)
        return combined_results[:top_k]

    async def _generate_query_variants(self, query: str) -> List[str]:
        """Generate alternative phrasings of query."""
        # Use LLM to generate variants (simplified here)
        return [query]  # Placeholder

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 = await self.client.messages.create(
            model=self.model,
            max_tokens=max_tokens,
            messages=[{"role": "user", "content": prompt}]
        )

        return {
            "answer": message.content[0].text,
            "sources": [
                {
                    "source": ctx["metadata"].get("source", "Unknown"),
                    "score": ctx["score"]
                }
                for ctx in context
            ],
            "usage": {
                "input_tokens": message.usage.input_tokens,
                "output_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("/query", response_model=QueryResponse)
async def query(request: QueryRequest):
    """Query RAG system."""
    try:
        # Retrieve relevant documents
        context = await retriever.retrieve(
            query=request.question,
            top_k=request.top_k,
            rerank=request.rerank,
            namespace=request.namespace
        )

        # Generate answer
        result = await llm.generate_answer(
            question=request.question,
            context=context
        )

        return QueryResponse(**result)

    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

class IngestRequest(BaseModel):
    directory: str
    namespace: str = "default"

@app.post("/ingest")
async def ingest(request: IngestRequest):
    """Ingest documents from directory."""
    try:
        # Load documents
        loader = DocumentLoader()
        documents = loader.load_directory(Path(request.directory))

        # Generate embeddings
        texts = [doc["text"] for doc in documents]
        embeddings = await embedder.embed_batch(texts)

        # Upload to vector store
        num_uploaded = await vector_store.upsert_documents(
            documents=documents,
            embeddings=embeddings,
            namespace=request.namespace
        )

        return {
            "status": "success",
            "documents_ingested": num_uploaded
        }

    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.get("/health")
async def health():
    """Health check endpoint."""
    return {"status": "healthy"}

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

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.