Rag Pipeline Gen

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
…

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:
…

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

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
…

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
…

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."""
…

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(
…