Production RAG Systems: Architecture & Best Practices [2025]

# Production Data Pipeline (Python + Celery)
from celery import Celery
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
import logging

app = Celery('rag_pipeline', broker='redis://localhost:6379')
embeddings = OpenAIEmbeddings()
logger = logging.getLogger(__name__)

@app.task(bind=True, max_retries=3)
def process_document(self, doc_id: str, source_url: str):
    """Process a single document through the pipeline"""
    try:
        # 1. Extract
        logger.info(f"Extracting document {doc_id}")
        content = extract_content(source_url)
        
        # 2. Transform
        logger.info(f"Chunking document {doc_id}")
        text_splitter = RecursiveCharacterTextSplitter(
            chunk_size=1000,
            chunk_overlap=200,
            separators=["\n\n", "\n", ". ", " "]
        )
        chunks = text_splitter.split_text(content)
        
        # 3. Generate embeddings
        logger.info(f"Generating embeddings for {len(chunks)} chunks")
        vectors = embeddings.embed_documents(chunks)
        
        # 4. Load to vector store
        logger.info(f"Storing vectors for {doc_id}")
        vector_store.upsert(
            ids=[f"{doc_id}_{i}" for i in range(len(chunks))],
            vectors=vectors,
            metadata=[{
                "doc_id": doc_id,
                "chunk_index": i,
                "source": source_url,
                "text": chunk
            } for i, chunk in enumerate(chunks)]
        )
        
        # 5. Update metadata DB
        metadata_db.update_document_status(doc_id, "processed")
        logger.info(f"Successfully processed {doc_id}")
        
    except Exception as e:
        logger.error(f"Error processing {doc_id}: {e}")
        # Retry with exponential backoff
        raise self.retry(exc=e, countdown=2 ** self.request.retries)

# Batch processing for efficiency
@app.task
def process_document_batch(doc_ids: list):
    """Process multiple documents in parallel"""
    for doc_id in doc_ids:
        process_document.delay(doc_id)

# Cost Tracking Implementation
from dataclasses import dataclass
from datetime import datetime
import tiktoken

@dataclass
class CostTracker:
    """Track costs for RAG operations"""
    
    # Pricing (as of 2025)
    GPT4_INPUT_COST = 0.03 / 1000  # per token
    GPT4_OUTPUT_COST = 0.06 / 1000
    GPT35_INPUT_COST = 0.0015 / 1000
    GPT35_OUTPUT_COST = 0.002 / 1000
    EMBEDDING_COST = 0.0001 / 1000
    
    def __init__(self):
        self.encoding = tiktoken.encoding_for_model("gpt-4")
        self.daily_costs = {}
    
    def track_llm_call(
        self,
        model: str,
        prompt: str,
        response: str,
        user_id: str
    ):
        """Track cost of LLM call"""
        input_tokens = len(self.encoding.encode(prompt))
        output_tokens = len(self.encoding.encode(response))
        
        if "gpt-4" in model:
            cost = (input_tokens * self.GPT4_INPUT_COST + 
                   output_tokens * self.GPT4_OUTPUT_COST)
        else:
            cost = (input_tokens * self.GPT35_INPUT_COST + 
                   output_tokens * self.GPT35_OUTPUT_COST)
        
        self._record_cost("llm", cost, user_id)
        
        return {
            "input_tokens": input_tokens,
            "output_tokens": output_tokens,
            "cost": cost
        }
    
    def track_embedding_call(self, texts: list, user_id: str):
        """Track cost of embedding generation"""
        total_tokens = sum(
            len(self.encoding.encode(text)) for text in texts
        )
        cost = total_tokens * self.EMBEDDING_COST
        
        self._record_cost("embedding", cost, user_id)
        
        return {"tokens": total_tokens, "cost": cost}
    
    def _record_cost(self, operation: str, cost: float, user_id: str):
        """Record cost in database"""
        date = datetime.utcnow().date()
        
        cost_db.insert({
            "date": date,
            "operation": operation,
            "cost": cost,
            "user_id": user_id,
            "timestamp": datetime.utcnow()
        })
        
        # Update daily total
        if date not in self.daily_costs:
            self.daily_costs[date] = 0
        self.daily_costs[date] += cost
        
        # Alert if daily cost exceeds threshold
        if self.daily_costs[date] > 100:  # $100/day
            alert_admin(f"Daily cost exceeded: $" + "{self.daily_costs[date]:.2f}")
    
    def get_cost_report(self, start_date, end_date):
        """Generate cost report"""
        costs = cost_db.query(
            start_date=start_date,
            end_date=end_date
        )
        
        report = {
            "total_cost": sum(c["cost"] for c in costs),
            "by_operation": {},
            "by_user": {},
            "by_day": {}
        }
        
        for cost in costs:
            # By operation
            op = cost["operation"]
            report["by_operation"][op] = report["by_operation"].get(op, 0) + cost["cost"]
            
            # By user
            user = cost["user_id"]
            report["by_user"][user] = report["by_user"].get(user, 0) + cost["cost"]
            
            # By day
            day = cost["date"]
            report["by_day"][day] = report["by_day"].get(day, 0) + cost["cost"]
        
        return report

# Usage
cost_tracker = CostTracker()

async def query_with_cost_tracking(query: str, user_id: str):
    """Query with cost tracking"""
    
    # Track embedding cost
    embedding_cost = cost_tracker.track_embedding_call([query], user_id)
    
    # Process query
    response = await rag_system.query(query)
    
    # Track LLM cost
    llm_cost = cost_tracker.track_llm_call(
        model="gpt-4",
        prompt=response["prompt"],
        response=response["answer"],
        user_id=user_id
    )
    
    # Add cost info to response
    response["cost"] = {
        "embedding": embedding_cost["cost"],
        "llm": llm_cost["cost"],
        "total": embedding_cost["cost"] + llm_cost["cost"]
    }
    
    return response