RAG in Production: Seven Pitfalls Nobody Tells You About

# ❌ Defaulting to ada-002
from openai import OpenAI
response = client.embeddings.create(
    model="text-embedding-ada-002",
    input="Your text"
)

# ✅ Consider alternatives based on your use case
# Chinese embedding: multilingual-e5-large, BGE
# Code retrieval: codellama embedding, GTE-code

# Fixed-size chunking
texts = text_splitter.split_text(document)

from langchain.text_splitter import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,      # Target chunk size
    chunk_overlap=50,    # Overlap between chunks, preserve context
    separators=["。\n", "，\n", "\n", " "]  # Split by sentence/paragraph
)

def semantic_chunk(text):
    """
    Chunk by semantic unit: each chunk is a complete Q&A or topic paragraph
    """
    sections = []
    
    for section in text.split("\n## "):  # Split by Markdown headers
        if len(section) < 100:
            continue  # Skip too-short header lines
        
        # If chunk too large, split by sub-headers
        if len(section) > 1000:
            subsections = section.split("\n### ")
            for sub in subsections:
                if len(sub) > 50:
                    sections.append(sub.strip())
        else:
            sections.append(section.strip())
    
    return sections

# This is insufficient
results = vector_store.similarity_search(query, k=5)

from langchain.retrievers import EnsembleRetriever

# BM25 keyword retrieval
bm25_retriever = BM25Retriever.from_texts(chunks)

# Vector retrieval
vector_retriever = vector_store.as_retriever(search_kwargs={"k": 5})

# Hybrid retrieval
ensemble_retriever = EnsembleRetriever(
    retrievers=[bm25_retriever, vector_retriever],
    weights=[0.3, 0.7]  # Low keyword weight, high vector weight
)

def rewrite_query(query, llm):
    """
    Use LLM to rewrite user question into a form better suited for retrieval
    """
    prompt = f"""Rewrite the following natural language question into a form better suited for knowledge base retrieval.
    Keep the original meaning, but use more formal language closer to documentation phrasing.
    
    Original: {query}
    Rewritten:"""
    
    response = llm.complete(prompt)
    return response.text.strip()

# User asks: service won't start
original = "service won't start"
rewritten = rewrite_query(original, llm)  # Might be "service startup failure troubleshooting"

# Use rewritten query for retrieval
results = ensemble_retriever.get_relevant_documents(rewritten)

def rerank_and_truncate(query, docs, llm, max_tokens=3000):
    """
    Rerank with LLM, then truncate to max_tokens
    """
    # Retrieve more candidates first
    candidates = vector_store.similarity_search(query, k=10)
    
    # Rerank with LLM (simple approach: score each)
    scored = []
    for doc in candidates:
        score_prompt = f"""Rate the relevance of this document chunk to answering the question.
        Question: {query}
        Document: {doc.page_content}
        Relevance: 1-5 score"""
        
        score = int(llm.complete(score_prompt).text.strip())
        scored.append((score, doc))
    
    # Take top ones
    top_docs = sorted(scored, key=lambda x: x[0], reverse=True)[:5]
    
    # Truncate to max_tokens
    total_tokens = 0
    selected = []
    for score, doc in top_docs:
        doc_tokens = len(doc.page_content) // 4  # Rough estimate
        if total_tokens + doc_tokens <= max_tokens:
            selected.append(doc)
            total_tokens += doc_tokens
    
    return selected

def summarize_chunks(chunks, llm, max_per_chunk=200):
    """
    Summarize each chunk first, then put in context
    """
    summarized = []
    for chunk in chunks:
        if len(chunk.page_content) > max_per_chunk * 4:  # Exceeds ~200 tokens
            summary = llm.complete(
                f"Summarize the following, keep key information, no more than {max_per_chunk} characters:\n\n{chunk.page_content}"
            )
            summarized.append(summary.text.strip())
        else:
            summarized.append(chunk.page_content)
    
    return summarized

# Periodically rebuild embedding
import schedule

def rebuild_index_if_needed():
    """
    Check document modification time, rebuild embedding if changed
    """
    docs_version = get_docs_version()  # From DB or file hash
    
    if docs_version != cached_version:
        print("Documents updated, rebuilding index...")
        
        # Incremental update: only process changed documents
        changed_docs = get_changed_documents(cached_version, docs_version)
        
        for doc in changed_docs:
            # Delete old embedding
            vector_store.delete_by_doc_id(doc.id)
            
            # Rebuild new embedding
            new_embedding = embed_text(doc.content)
            vector_store.add_texts([doc.content], ids=[doc.id])
        
        update_cached_version(docs_version)

# Check daily
schedule.every().day.at("02:00").do(rebuild_index_if_needed)

def evaluate_rag_system(test_questions, ground_truth, rag_pipeline):
    """
    Basic evaluation: retrieval Recall and Answer accuracy
    """
    results = {
        'retrieval_recall': [],
        'answer_relevance': []
    }
    
    for question, expected_docs in zip(test_questions, ground_truth):
        # Retrieval evaluation
        retrieved = rag_pipeline.retrieve(question)
        retrieved_ids = {doc.id for doc in retrieved}
        expected_ids = {doc.id for doc in expected_docs}
        
        recall = len(retrieved_ids & expected_ids) / len(expected_ids)
        results['retrieval_recall'].append(recall)
        
        # Answer quality evaluation (LLM-scored)
        answer = rag_pipeline.answer(question)
        relevance_score = llm.evaluate(
            f"Question: {question}\nAnswer: {answer}\nScore 1-5, 5 is best"
        )
        results['answer_relevance'].append(int(relevance_score))
    
    return {
        'avg_recall': sum(results['retrieval_recall']) / len(results['retrieval_recall']),
        'avg_relevance': sum(results['answer_relevance']) / len(results['answer_relevance'])
    }