RAG Architecture Patterns: The Complete Enterprise Implementation Guide

# Parent-child chunking (retrieves small chunks, returns parent context):
from langchain.text_splitter import RecursiveCharacterTextSplitter

# Small child chunks for precision retrieval:
child_splitter = RecursiveCharacterTextSplitter(chunk_size=200, chunk_overlap=20)

# Larger parent chunks for context-rich generation:
parent_splitter = RecursiveCharacterTextSplitter(chunk_size=1500, chunk_overlap=100)

def ingest_with_parent_child(document: str, doc_id: str):
    parent_chunks = parent_splitter.split_text(document)
    
    for i, parent in enumerate(parent_chunks):
        parent_id = f"{doc_id}_parent_{i}"
        parent_store.set(parent_id, parent)  # Store full parent in regular DB
        
        # Create child chunks — each points to its parent:
        children = child_splitter.split_text(parent)
        for j, child in enumerate(children):
            vector_store.add(
                text=child,
                metadata={"parent_id": parent_id, "doc_id": doc_id}
            )

# On retrieval: search child chunks, return parent context:
def retrieve(query: str, k: int = 5) -> list[str]:
    child_results = vector_store.similarity_search(query, k=k*3)  # over-fetch
    parent_ids = {r.metadata["parent_id"] for r in child_results}
    # Return full parent texts (much richer context than child chunks):
    return [parent_store.get(pid) for pid in list(parent_ids)[:k]]

# Naive: misses "JWT" if user asks about "JSON Web Tokens" exact term
results = vector_store.similarity_search("JWT authentication", k=5)

# Hybrid: combines semantic + keyword (BM25) search:
from langchain.retrievers import EnsembleRetriever

vector_retriever = vector_store.as_retriever(search_kwargs={"k": 5})
bm25_retriever = BM25Retriever.from_documents(docs, k=5)

# Ensemble with Reciprocal Rank Fusion:
ensemble = EnsembleRetriever(
    retrievers=[vector_retriever, bm25_retriever],
    weights=[0.6, 0.4]  # 60% semantic, 40% keyword
)
# Now "JWT" is matched by BM25 even if embeddings don't capture acronyms

def rewrite_query(user_query: str) -> list[str]:
    """Generate multiple search queries to improve recall."""
    response = llm.invoke(f"""
    Generate 3 different search queries to find information relevant to:
    "{user_query}"
    
    Return only the 3 queries, one per line.
    """)
    queries = response.content.strip().split('\n')
    return queries[:3]

# Search with all rewritten queries, merge results:
all_results = []
for query in rewrite_query(user_query) + [user_query]:
    all_results.extend(vector_store.similarity_search(query, k=3))
# Deduplicate and re-rank

def hyde_retrieve(query: str) -> list[str]:
    # Generate a hypothetical ideal answer:
    hypothetical_answer = llm.invoke(
        f"Write a detailed answer to: {query}\n"
        "Base it on knowledge you have. Do not say 'I don't know'."
    ).content
    
    # Embed the hypothetical answer (not the query):
    hypothesis_embedding = embed(hypothetical_answer)
    
    # Search vector store using the hypothetical answer's embedding:
    # This finds documents similar to the "ideal answer" rather than the question
    return vector_store.similarity_search_by_vector(hypothesis_embedding, k=5)

from sentence_transformers import CrossEncoder

reranker = CrossEncoder('cross-encoder/ms-marco-MiniLM-L-6-v2')

def retrieve_and_rerank(query: str, k: int = 5) -> list[str]:
    # Step 1: Retrieve 20 candidates (over-fetch):
    candidates = vector_store.similarity_search(query, k=20)
    
    # Step 2: Re-rank using cross-encoder (considers query+document jointly):
    pairs = [[query, doc.page_content] for doc in candidates]
    scores = reranker.predict(pairs)
    
    # Step 3: Return top-k after re-ranking:
    ranked = sorted(zip(candidates, scores), key=lambda x: x[1], reverse=True)
    return [doc.page_content for doc, _ in ranked[:k]]

from ragas import evaluate
from ragas.metrics import faithfulness, answer_relevancy, context_precision, context_recall

# Build evaluation dataset:
eval_data = {
    "question": ["What is our refund policy?", "How do I reset my password?"],
    "answer": [generated_answers[0], generated_answers[1]],
    "contexts": [retrieved_contexts[0], retrieved_contexts[1]],
    "ground_truth": ["Refunds within 30 days...", "Click 'Forgot Password'..."],
}

results = evaluate(eval_data, metrics=[
    faithfulness,        # Is the answer grounded in the retrieved context?
    answer_relevancy,    # Is the answer relevant to the question?
    context_precision,   # Is the retrieved context useful?
    context_recall,      # Was the relevant context actually retrieved?
])
# Target: faithfulness > 0.9, answer_relevancy > 0.85, context_recall > 0.8