RAG Architecture Patterns: Beyond Basic Vector Search

Retrieval-Augmented Generation (RAG) systems usually start with plain vector similarity search. That baseline breaks down on three predictable fronts: exact keyword matches, questions whose answer spans several documents, and retrieved context that nobody checked for relevance.

The default worth reaching for first is hybrid search combined with parent-child chunking. Reranking, self-corrective retrieval (CRAG), and knowledge graphs (GraphRAG) sit above it as escalations, added when measurement shows the default falls short. The patterns below map onto AWS Bedrock Knowledge Bases and OpenSearch, and RAGAS metrics decide whether an escalation paid for itself.

The Problem with Basic RAG

Vector similarity alone leaves three specific gaps in production.

Missing Exact Matches

Vector embeddings excel at capturing semantic meaning but struggle with precise matches. When users search for “AWS CDK”, “GAN architecture”, or specific product codes, pure semantic search often misses these exact terms. The embedding model treats “GAN” (Generative Adversarial Network) as semantically similar to general “neural network” content, diluting precision.

# Basic RAG implementation - common starting point
from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings

vectorstore = Chroma.from_documents(
    documents=docs,
    embedding=OpenAIEmbeddings(model="text-embedding-3-small")
)

# Vector similarity search
query = "What is GAN architecture?"
results = vectorstore.similarity_search(query, k=5)

# Problem: May miss documents with exact "GAN" term
# Returns semantically similar "neural network" docs instead

Multi-Hop Reasoning Failures

Basic RAG retrieves documents based on single-step similarity. Complex queries requiring connections across multiple documents fail systematically:

• “Which AWS service launched in 2020 has the lowest cold start time?”
• “What are the security implications of using serverless databases with Lambda?”

These questions need information synthesis from disparate sources, something single-step retrieval cannot handle.

No Quality Verification

Standard RAG pipelines pass retrieved documents directly to the LLM without verifying relevance. Irrelevant context causes hallucinations and degraded answer quality. This becomes particularly problematic when retrieval returns marginally related documents that the LLM treats as authoritative.

Hybrid Search: Combining Semantic and Keyword Retrieval

The first practical upgrade for RAG systems combines dense vector search with sparse keyword matching. This hybrid approach addresses the exact match problem while maintaining semantic understanding.

Implementation Strategy

Hybrid search runs two parallel retrievals:

1. Dense retrieval: Vector similarity (semantic understanding)
2. Sparse retrieval: BM25 keyword matching (exact term matching)

Results merge using Reciprocal Rank Fusion (RRF):

RRF(d) = Σ 1/(k + rank(d))

Where k is a constant (typically 60) and rank(d) is the document’s position in each result set.

Working Implementation

from langchain.retrievers import EnsembleRetriever
from langchain_community.vectorstores import Chroma
from langchain_community.retrievers import BM25Retriever
from langchain_openai import OpenAIEmbeddings

# Dense vector retriever
vectorstore = Chroma.from_documents(
    documents=docs,
    embedding=OpenAIEmbeddings(model="text-embedding-3-small")
)
vector_retriever = vectorstore.as_retriever(
    search_kwargs={"k": 10}
)

# Sparse keyword retriever
bm25_retriever = BM25Retriever.from_documents(docs)
bm25_retriever.k = 10

# Hybrid ensemble with RRF
ensemble_retriever = EnsembleRetriever(
    retrievers=[vector_retriever, bm25_retriever],
    weights=[0.5, 0.5]  # Equal weighting
)

# Retrieve with both methods
results = ensemble_retriever.invoke(
    "What is GAN architecture?"
)

Performance Characteristics

What the second retriever changes:

• Named entities land reliably, because BM25 matches the literal string that the embedding model blurs into its neighbourhood
• Abbreviations behave the same way (GAN, RAG, AWS, CDK)
• Latency grows by one retrieval pass, which runs in parallel with the vector query
• Recall improves without costing precision, since the two result sets are merged rather than intersected

The alpha parameter in weighted fusion controls the balance:

# Alternative: Weighted fusion instead of RRF
def weighted_fusion(vector_results, bm25_results, alpha=0.5):
    """
    alpha = 0.0: Pure keyword search
    alpha = 0.5: Equal weight
    alpha = 1.0: Pure semantic search
    """
    fused_scores = {}

    for doc in vector_results:
        fused_scores[doc.id] = alpha * doc.score

    for doc in bm25_results:
        if doc.id in fused_scores:
            fused_scores[doc.id] += (1 - alpha) * doc.score
        else:
            fused_scores[doc.id] = (1 - alpha) * doc.score

    return sorted(
        fused_scores.items(),
        key=lambda x: x[1],
        reverse=True
    )

Multi-Stage Reranking: Precision After Recall

Hybrid search improves recall, but production systems often need even higher precision. Multi-stage reranking addresses this through a two-phase approach.

Architecture Pattern

1. Stage 1: High-recall retrieval - Cast a wide net (k=50-100)
2. Stage 2: Cross-encoder reranking - Precision scoring on candidates
3. Stage 3: Top-k selection - Final set for LLM (k=5-10)

This pattern separates retrieval (fast, broad) from relevance scoring (slower, precise).

Cross-Encoder vs Bi-Encoder

Understanding the difference matters for implementation:

• Bi-encoder (traditional embeddings): Encodes query and documents separately, compares vectors
• Cross-encoder: Feeds query + document pairs into BERT-based model for direct relevance scores

Cross-encoders produce more accurate relevance scores but don’t scale to large collections (must score each candidate individually). This makes them perfect for the reranking stage.

Implementation

from sentence_transformers import CrossEncoder
import numpy as np

# Stage 1: High-recall retrieval
initial_results = vectorstore.similarity_search(
    query,
    k=50  # Cast wide net
)

# Stage 2: Cross-encoder reranking
reranker = CrossEncoder('cross-encoder/ms-marco-MiniLM-L-6-v2')

# Create query-document pairs
query_doc_pairs = [
    (query, doc.page_content)
    for doc in initial_results
]

# Score all pairs
scores = reranker.predict(query_doc_pairs)

# Stage 3: Sort by score and take top-k
reranked_indices = np.argsort(scores)[::-1][:10]
final_docs = [initial_results[i] for i in reranked_indices]
final_scores = [scores[i] for i in reranked_indices]

# Use final_docs for LLM generation
print(f"Top result score: {final_scores[0]:.3f}")

What Reranking Buys

Reranking changes the order of the result set, not its size. Because the cross-encoder reads the query and the document together, it separates two chunks that a bi-encoder maps to nearly the same vector. The gain shows up in ranking metrics such as MRR@5 and in precision on domain-specific queries, where the top three slots decide the answer the model writes.

The cost is one model inference per candidate. With 50 candidates that is 50 forward passes added to every query, which rules the pattern out under a sub-second latency budget.

When to Use Reranking

Implement reranking when:

• Accuracy requirements exceed 85% precision
• Complex technical queries with nuanced relevance
• Legal, medical, financial domains with high accuracy stakes
• Computational resources allow for cross-encoder inference

Skip reranking when:

• Sub-500ms latency requirements
• Simple FAQ systems
• Limited computational budget
• Basic semantic matching suffices

Chunking Strategies: Context Preservation

How you split documents into chunks significantly impacts retrieval quality. Poorly chunked content can destroy otherwise solid RAG implementations.

Strategy Comparison

Fixed-Size Chunking (Baseline)

from langchain.text_splitter import CharacterTextSplitter

splitter = CharacterTextSplitter(
    chunk_size=512,
    chunk_overlap=50
)
chunks = splitter.split_documents(documents)

Problems:

• Breaks sentences arbitrarily
• Splits code blocks mid-function
• No respect for logical boundaries

Semantic Chunking

from langchain_experimental.text_splitter import SemanticChunker
from langchain_openai import OpenAIEmbeddings

splitter = SemanticChunker(
    embeddings=OpenAIEmbeddings(model="text-embedding-3-small"),
    breakpoint_threshold_type="percentile"
)
chunks = splitter.split_documents(documents)

Benefits:

• Preserves topic coherence
• Natural section boundaries
• Higher preprocessing cost

Parent-Child Hierarchical Chunking

This approach searches on small chunks for precision but returns larger parent chunks for context. It works well as a default strategy for technical documentation.

from langchain.retrievers import ParentDocumentRetriever
from langchain.storage import InMemoryStore
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings

# Parent splitter (large chunks for context)
parent_splitter = RecursiveCharacterTextSplitter(
    chunk_size=2000,
    chunk_overlap=200
)

# Child splitter (small chunks for precision)
child_splitter = RecursiveCharacterTextSplitter(
    chunk_size=400,
    chunk_overlap=50
)

# Store parent docs separately
docstore = InMemoryStore()

# Vector store indexes child chunks
vectorstore = Chroma(
    embedding_function=OpenAIEmbeddings(model="text-embedding-3-small")
)

# Retriever configuration
retriever = ParentDocumentRetriever(
    vectorstore=vectorstore,
    docstore=docstore,
    child_splitter=child_splitter,
    parent_splitter=parent_splitter,
)

# Add documents
retriever.add_documents(documents)

# Retrieval searches child chunks but returns parents
results = retriever.invoke(
    "How do I optimize Lambda cold starts?"
)
# Results contain full parent context

Performance Impact

Parent-child retrieval trades storage for coherence. The corpus lives twice: parents in a document store, children in the vector index. Query-time cost stays close to plain vector search, because the extra work is a key lookup in the docstore after the vector hit.

The model then reads the section a fragment belongs to, in place of a 400-token slice that starts mid-sentence and ends mid-table.

Best Practices

Rules that hold across content types:

1. Terminate at natural boundaries: End chunks at sentence or paragraph breaks
2. Add metadata: Include document title, section headers in chunk metadata
3. Overlap strategically: 10-20% overlap prevents information loss at boundaries
4. Match strategy to content:
   • Technical docs → Semantic or hierarchical
   • Code → Function/class-level chunks
   • Narrative → Sliding window with overlap
   • Structured data → Parent-child with metadata

Self-RAG and Corrective RAG: Quality Verification

Basic RAG assumes retrieved documents are relevant. This assumption fails frequently in production, causing hallucinations and poor answers. Self-correcting patterns address this with explicit quality checks.

The CRAG Pattern

Corrective RAG (CRAG) introduces a retrieval evaluator that grades document relevance before generation. Based on confidence scores, the system routes to different processing paths.

Workflow:

1. Retrieve documents
2. Grade each document’s relevance
3. Route based on aggregate confidence:
   • High confidence (>0.7): Proceed with knowledge refinement
   • Low confidence (<0.3): Trigger web search
   • Medium confidence (0.3-0.7): Combine web search + refinement

Knowledge Refinement partitions documents into “knowledge strips”, grades each strip, and filters irrelevant content before passing to the LLM.

Implementation with LangGraph

from langgraph.graph import StateGraph, END
from langchain_tavily import TavilySearch
from langchain_openai import ChatOpenAI
from typing import TypedDict, List
from langchain.schema import Document

# Define workflow state
class RAGState(TypedDict):
    query: str
    documents: List[Document]
    relevance_score: float
    web_results: List[Document]
    answer: str

# Initialize components
vectorstore = Chroma.from_documents(docs, OpenAIEmbeddings(model="text-embedding-3-small"))
llm = ChatOpenAI(model="gpt-4o-mini")
web_search = TavilySearch(max_results=3)

# Node functions
def retrieve(state):
    """Retrieve initial documents"""
    docs = vectorstore.similarity_search(state["query"], k=5)
    return {"documents": docs}

def grade_documents(state):
    """Grade document relevance"""
    prompt = f"""
    Score the relevance of this document to the query on a scale of 0-1.
    Query: {state["query"]}
    Document: {state["documents"][0].page_content[:500]}

    Return only a number between 0 and 1.
    """

    # Grade each document
    scores = []
    for doc in state["documents"]:
        response = llm.invoke(prompt)
        score = float(response.content.strip())
        scores.append(score)

    avg_score = sum(scores) / len(scores)
    return {"relevance_score": avg_score}

def route_query(state):
    """Route based on relevance score"""
    score = state["relevance_score"]

    if score > 0.7:
        return "correct"
    elif score < 0.3:
        return "incorrect"
    else:
        return "ambiguous"

def perform_web_search(state):
    """Fallback web search"""
    web_results = web_search.invoke(state["query"])
    web_docs = [
        Document(page_content=result["content"])
        for result in web_results
    ]
    return {"web_results": web_docs}

def refine_knowledge(state):
    """Partition and filter document strips"""
    refined_docs = []

    for doc in state["documents"]:
        # Simple strip partitioning (sentence-level)
        sentences = doc.page_content.split('. ')

        for sentence in sentences:
            # Grade each sentence
            grade_prompt = f"""
            Is this sentence relevant to: {state["query"]}?
            Sentence: {sentence}
            Answer only: yes or no
            """
            response = llm.invoke(grade_prompt)

            if "yes" in response.content.lower():
                refined_docs.append(sentence)

    return {"documents": [Document(page_content=". ".join(refined_docs))]}

def generate_answer(state):
    """Generate final answer"""
    context_docs = state.get("documents", [])
    web_docs = state.get("web_results", [])

    all_context = context_docs + web_docs
    context_text = "\n\n".join([doc.page_content for doc in all_context])

    prompt = f"""
    Answer the question based on this context:

    Context:
    {context_text}

    Question: {state["query"]}

    Answer:
    """

    response = llm.invoke(prompt)
    return {"answer": response.content}

# Build the graph
workflow = StateGraph(RAGState)

# Add nodes
workflow.add_node("retrieve", retrieve)
workflow.add_node("grade", grade_documents)
workflow.add_node("web_search", perform_web_search)
workflow.add_node("refine", refine_knowledge)
workflow.add_node("generate", generate_answer)

# Add edges
workflow.set_entry_point("retrieve")
workflow.add_edge("retrieve", "grade")

# Conditional routing after grading
workflow.add_conditional_edges(
    "grade",
    route_query,
    {
        "correct": "refine",
        "incorrect": "web_search",
        "ambiguous": "web_search"
    }
)

workflow.add_edge("refine", "generate")
workflow.add_edge("web_search", "generate")
workflow.add_edge("generate", END)

# Compile and run
app = workflow.compile()

# Execute
result = app.invoke({
    "query": "What are the latest AWS Lambda optimization techniques?"
})

print(result["answer"])

Where CRAG Helps

The evaluator is what changes the outcome. A document that fails grading never reaches the prompt, so the model cannot ground an answer in it. That matters most on queries where retrieval confidence is low and on questions that fall outside the knowledge base entirely, which is exactly where a plain pipeline invents an answer with full confidence.

Every document costs a grading call, and a low-confidence route adds a web-search round trip on top. A CRAG query therefore runs several times the latency and tokens of a single-pass query, which is why the grader is the first model to shrink when that bill hurts.

GraphRAG: Knowledge Graphs for Multi-Hop Reasoning

Traditional RAG retrieves based on semantic similarity to the query. This works for single-hop questions but fails when answers require connecting information across multiple documents. GraphRAG solves this through knowledge graph construction and graph-based retrieval.

When Traditional RAG Fails

Complex queries that GraphRAG handles better:

• “Which AWS services integrate with both Lambda and DynamoDB?”
• “What are the security implications of serverless database patterns?”
• “Summarize all best practices mentioned across the documentation”

These require understanding relationships and synthesizing information across documents.

GraphRAG Architecture

Phase 1: Knowledge Graph Construction

1. Extract entities (services, concepts, technologies)
2. Extract relationships (integrates_with, depends_on, alternative_to)
3. Extract claims (factual statements)
4. Build directed graph

Phase 2: Community Detection

1. Apply Leiden algorithm for hierarchical clustering
2. Create community summaries at each level
3. Build hierarchical index

Phase 3: Retrieval

• Global search: Use community summaries for broad questions
• Local search: Traverse graph for specific relationship queries
• Hybrid: Combine graph traversal with vector similarity

Implementation Pattern

from langchain_neo4j import Neo4jGraph, GraphCypherQAChain
from langchain_openai import ChatOpenAI

# Initialize Neo4j graph database
graph = Neo4jGraph(
    url="bolt://localhost:7687",
    username="neo4j",
    password="your-password"
)

# Entity and relationship extraction (simplified)
def extract_entities_relationships(text: str):
    """Use LLM to extract graph elements"""
    prompt = f"""
    Extract entities and relationships from this text.
    Format: (Entity1)-[RELATIONSHIP]->(Entity2)

    Text: {text}
    """

    llm = ChatOpenAI(model="gpt-4o")
    response = llm.invoke(prompt)
    return response.content

# Populate graph
def build_knowledge_graph(documents):
    for doc in documents:
        # Extract structured data
        graph_data = extract_entities_relationships(doc.page_content)

        # Convert to Cypher queries
        # Example: CREATE (lambda:Service {name: 'AWS Lambda'})
        # CREATE (lambda)-[:INTEGRATES_WITH]->(dynamodb)

        # Execute Cypher
        # graph.query(cypher_statement)
        pass

# Query-time retrieval
qa_chain = GraphCypherQAChain.from_llm(
    llm=ChatOpenAI(model="gpt-4o"),
    graph=graph,
    verbose=True,
    return_intermediate_steps=True
)

# Multi-hop query
response = qa_chain.invoke({
    "query": "Which serverless services integrate with DynamoDB and support event-driven architectures?"
})

print(response["result"])
print("Cypher Query:", response["intermediate_steps"][0]["query"])

Performance Trade-offs

Costs:

• Preprocessing: entity and relationship extraction runs LLM calls over every document, which dominates the ingestion bill
• Storage: a graph database to operate alongside the vector store
• Complexity: someone on the team has to own Cypher and the graph schema

Benefits:

• Multi-hop recall, because traversal follows relationships that similarity search cannot see
• Global questions (“summarize everything about X”) answered from community summaries, which a top-k window can never assemble
• Traceable answers, since the traversal path doubles as the citation trail

When to Use GraphRAG

Good fit:

• Rich relationship domains (medical, legal, enterprise knowledge)
• Multi-hop reasoning requirements
• Holistic understanding needs across large corpora
• Available preprocessing budget

Poor fit:

• Simple FAQ systems
• Primarily single-hop queries
• Limited preprocessing resources
• Small document collections (<1000 documents)

AWS Bedrock Knowledge Bases: Production Implementation

AWS Bedrock Knowledge Bases packages most of the patterns above as a managed service. Hybrid search, hierarchical chunking, and reranking arrive as configuration flags on a pipeline someone else operates.

Architecture Options

Vector Store Choices:

• Amazon OpenSearch Serverless: Most common for production RAG, supports hybrid search
• Amazon Aurora PostgreSQL: With pgvector extension, good for existing PostgreSQL users
• Amazon Neptune Analytics: For GraphRAG patterns
• Third-party: MongoDB Atlas, Pinecone, Redis Enterprise Cloud

Two API Patterns

1. RetrieveAndGenerate API (Fully Managed)

This handles the entire RAG pipeline:

import boto3

bedrock_agent_runtime = boto3.client('bedrock-agent-runtime')

response = bedrock_agent_runtime.retrieve_and_generate(
    input={
        'text': 'How do I optimize Lambda cold starts?'
    },
    retrieveAndGenerateConfiguration={
        'type': 'KNOWLEDGE_BASE',
        'knowledgeBaseConfiguration': {
            'knowledgeBaseId': 'KB123456',
            'modelArn': 'arn:aws:bedrock:us-east-1::foundation-model/anthropic.claude-sonnet-4-5-20250929-v1:0',
            'retrievalConfiguration': {
                'vectorSearchConfiguration': {
                    'numberOfResults': 10,  # Increase from default 5
                    'overrideSearchType': 'HYBRID'  # Hybrid search
                }
            }
        }
    }
)

answer = response['output']['text']
citations = response['citations']

# Citations include source attribution
for citation in citations:
    print(f"Source: {citation['retrievedReferences'][0]['location']}")

2. Retrieve API (Custom Control)

For more control over the pipeline:

# Retrieve documents
retrieve_response = bedrock_agent_runtime.retrieve(
    knowledgeBaseId='KB123456',
    retrievalQuery={
        'text': 'How do I optimize Lambda cold starts?'
    },
    retrievalConfiguration={
        'vectorSearchConfiguration': {
            'numberOfResults': 20,
            'overrideSearchType': 'HYBRID'
        }
    }
)

# Process retrieved chunks
retrieved_docs = retrieve_response['retrievalResults']
for doc in retrieved_docs:
    content = doc['content']['text']
    score = doc['score']
    source = doc['location']['s3Location']

    print(f"Score: {score:.3f} - Source: {source['uri']}")

# Now you control:
# - Custom reranking logic
# - Document filtering
# - Prompt construction
# - Model selection for generation

Advanced Chunking Configuration

# Hierarchical chunking (parent-child pattern)
chunking_config = {
    'chunkingStrategy': 'HIERARCHICAL',
    'hierarchicalChunkingConfiguration': {
        'levelConfigurations': [
            {
                'maxTokens': 1500  # Parent chunk size
            },
            {
                'maxTokens': 300  # Child chunk size
            }
        ],
        'overlapTokens': 60
    }
}

# Alternative: Semantic chunking
chunking_config = {
    'chunkingStrategy': 'SEMANTIC',
    'semanticChunkingConfiguration': {
        'maxTokens': 300,
        'bufferSize': 0,
        'breakpointPercentileThreshold': 95
    }
}

# Advanced: Custom Lambda chunking function
chunking_config = {
    'chunkingStrategy': 'NONE',  # Disable default chunking
    'customChunkingConfiguration': {
        'lambdaArn': 'arn:aws:lambda:us-east-1:123456789:function:custom-chunker'
    }
}

Reranking Integration

retrieve_config = {
    'vectorSearchConfiguration': {
        'numberOfResults': 50,  # High recall
        'overrideSearchType': 'HYBRID',
        'rerankingConfiguration': {
            'type': 'BEDROCK_RERANKING_MODEL',
            'bedrockRerankingConfiguration': {
                'numberOfResults': 10,  # Precision after reranking
                'modelConfiguration': {
                    'modelArn': 'arn:aws:bedrock:us-west-2::foundation-model/cohere.rerank-v3-5:0'
                }
            }
        }
    }
}

Production Optimization Tips

Settings that carry the most weight in a Bedrock deployment:

1. Increase numberOfResults: Default 5 often insufficient; use 10-15 for complex queries
2. Enable Hybrid Search: Improves named entity and abbreviation retrieval
3. Implement Reranking: Changes which documents fill the top-k slots on technical queries
4. Choose Appropriate Chunking: Hierarchical for technical docs, semantic for narrative
5. Monitor Token Usage: Track embedding and generation costs separately
6. Use Customer-Managed KMS: For sensitive data encryption
7. Cache Strategically: Cache embeddings and common query results

Infrastructure as Code (CDK)

from aws_cdk import (
    aws_bedrock as bedrock,
    aws_opensearchserverless as opensearch,
    aws_s3 as s3,
    aws_iam as iam,
)

# S3 bucket for documents
docs_bucket = s3.Bucket(
    self, "DocsBucket",
    versioned=True,
    encryption=s3.BucketEncryption.S3_MANAGED
)

# OpenSearch Serverless collection
vector_collection = opensearch.CfnCollection(
    self, "VectorCollection",
    name="rag-vectors",
    type="VECTORSEARCH"
)

# IAM role for Knowledge Base
kb_role = iam.Role(
    self, "KBRole",
    assumed_by=iam.ServicePrincipal("bedrock.amazonaws.com")
)

docs_bucket.grant_read(kb_role)

# Bedrock Knowledge Base
kb = bedrock.CfnKnowledgeBase(
    self, "RAGKnowledgeBase",
    name="production-rag-kb",
    role_arn=kb_role.role_arn,
    knowledge_base_configuration=bedrock.CfnKnowledgeBase.KnowledgeBaseConfigurationProperty(
        type="VECTOR",
        vector_knowledge_base_configuration=bedrock.CfnKnowledgeBase.VectorKnowledgeBaseConfigurationProperty(
            embedding_model_arn="arn:aws:bedrock:us-east-1::foundation-model/amazon.titan-embed-text-v2:0"
        )
    ),
    storage_configuration=bedrock.CfnKnowledgeBase.StorageConfigurationProperty(
        type="OPENSEARCH_SERVERLESS",
        opensearch_serverless_configuration=bedrock.CfnKnowledgeBase.OpenSearchServerlessConfigurationProperty(
            collection_arn=vector_collection.attr_arn,
            vector_index_name="bedrock-knowledge-base-index",
            field_mapping=bedrock.CfnKnowledgeBase.OpenSearchServerlessFieldMappingProperty(
                vector_field="embedding",
                text_field="text",
                metadata_field="metadata"
            )
        )
    )
)

# Data source (S3)
data_source = bedrock.CfnDataSource(
    self, "S3DataSource",
    name="s3-docs",
    knowledge_base_id=kb.attr_knowledge_base_id,
    data_source_configuration=bedrock.CfnDataSource.DataSourceConfigurationProperty(
        type="S3",
        s3_configuration=bedrock.CfnDataSource.S3DataSourceConfigurationProperty(
            bucket_arn=docs_bucket.bucket_arn
        )
    )
)

Evaluation with RAGAS: Measuring RAG Quality

With RAG systems, improvement requires measurement. The RAGAS framework provides automated, reference-free metrics for both retrieval and generation quality.

Core Metrics

Retrieval Metrics:

1. Context Precision: Are relevant chunks ranked higher than irrelevant ones?
2. Context Recall: Did we retrieve all necessary information?
3. Context Relevancy: How much of retrieved content is actually relevant?

Generation Metrics:

1. Faithfulness: Is the answer factually grounded in the retrieved context?
2. Answer Relevancy: Does the answer address the question?

Implementation

from ragas import evaluate
from ragas.metrics import (
    context_precision,
    context_recall,
    faithfulness,
    answer_relevancy,
)
from datasets import Dataset

# Prepare evaluation dataset
eval_data = {
    'question': [
        'What are AWS Lambda cold start optimization techniques?',
        'How does DynamoDB handle partition keys?'
    ],
    'answer': [
        'Lambda cold starts can be optimized using provisioned concurrency, which keeps functions initialized, and SnapStart for Java functions which reduces initialization time.',
        'DynamoDB uses partition keys to distribute data across partitions. High-cardinality partition keys ensure even distribution and optimal performance.'
    ],
    'contexts': [
        [
            'Provisioned concurrency keeps Lambda functions initialized and ready to respond.',
            'SnapStart reduces cold start times for Java functions by caching initialized state.',
            'Function optimization like reducing package size improves cold start performance.'
        ],
        [
            'Partition keys determine how DynamoDB distributes data across partitions.',
            'Choose high-cardinality partition keys to ensure even data distribution.',
            'Low-cardinality keys create hot partitions and throttling.'
        ]
    ],
    'ground_truth': [
        'Provisioned concurrency, SnapStart, and function optimization reduce Lambda cold starts',
        'Partition keys distribute data; high cardinality ensures even distribution'
    ]
}

dataset = Dataset.from_dict(eval_data)

# Evaluate
result = evaluate(
    dataset,
    metrics=[
        context_precision,
        context_recall,
        faithfulness,
        answer_relevancy,
    ],
)

print(f"Context Precision: {result['context_precision']:.3f}")
print(f"Context Recall: {result['context_recall']:.3f}")
print(f"Faithfulness: {result['faithfulness']:.3f}")
print(f"Answer Relevancy: {result['answer_relevancy']:.3f}")
print(f"\nRAGAS Score: {result.mean():.3f}")

Production Monitoring

import wandb
from ragas.integrations.wandb import log

# Initialize monitoring
wandb.init(project="rag-production-monitoring")

def monitor_rag_quality(queries, answers, contexts, ground_truths):
    """Continuous evaluation in production"""
    eval_data = Dataset.from_dict({
        'question': queries,
        'answer': answers,
        'contexts': contexts,
        'ground_truth': ground_truths
    })

    result = evaluate(
        eval_data,
        metrics=[
            context_precision,
            context_recall,
            faithfulness,
            answer_relevancy
        ]
    )

    # Log to W&B
    log(result, run=wandb.run)

    # Alert on quality degradation
    if result['faithfulness'] < 0.7:
        send_alert("Faithfulness dropped below threshold!")

    if result['context_precision'] < 0.6:
        send_alert("Context precision degraded!")

    return result

# Use in production
daily_queries = get_sample_queries()  # Sample of production queries
daily_results = monitor_rag_quality(
    queries=daily_queries['questions'],
    answers=daily_queries['answers'],
    contexts=daily_queries['contexts'],
    ground_truths=daily_queries['ground_truths']
)

Metric Interpretation

Sample scores and what each one tells you to change:

Context Precision (0.85)

• 85% of relevant documents appear in top positions
• Good ranking quality
• Lower scores indicate irrelevant results rank too highly

Context Recall (0.75)

• 75% of necessary information was retrieved
• Missing 25% of relevant content
• Increase numberOfResults or improve chunking

Faithfulness (0.92)

• 92% of answer claims are supported by context
• Low hallucination rate
• Below 0.7 indicates serious problems

Answer Relevancy (0.88)

• Answer addresses 88% of question intent
• Minimal tangential information
• Below 0.7 suggests answer drift

Production Trade-offs: The Iron Triangle

Every RAG architecture decision balances three competing factors: latency, cost, and accuracy. Improving one usually spends the other two.

Latency Breakdown

The instinct is to blame generation for a slow RAG endpoint. In a single-pass pipeline that is roughly true. In configurations that re-retrieve and re-grade, it stops being true: retrieval runs several times, and each round appends its documents to the prompt, so prefill grows with every pass. Generation ends up a minority share of the wall clock.

Measure that split before tuning anything. If retrieval and prefill dominate, swapping the generation model changes very little.

Strategy-Specific Trade-offs

Strategy	Latency	Cost	Where the gain lands
Basic vector search	Baseline	Baseline	Baseline
Hybrid search	One extra pass, run in parallel	A second index to maintain	Exact terms, names, abbreviations
Cross-encoder reranking	One inference per candidate	Reranker model calls	Ordering of the top-k
Multi-query (RAG-Fusion)	One retrieval per rewrite	Rewrite plus retrieval per query	Ambiguous or underspecified queries
GraphRAG	Mostly paid at ingestion	LLM extraction over the whole corpus	Multi-hop and global questions
Parent-child retrieval	Docstore lookup after the vector hit	Corpus stored twice	Context coherence
Self-RAG/CRAG	Grading pass plus optional web search	One grading call per document	Faithfulness when retrieval is unreliable

Optimization Strategies

1. Caching Strategy

from functools import lru_cache
import hashlib
import time

# Cache embeddings
@lru_cache(maxsize=10000)
def get_embedding(text: str):
    return embedding_model.embed(text)

# Cache retrieval results
class RAGCache:
    def __init__(self, ttl_seconds=3600):
        self.cache = {}
        self.ttl = ttl_seconds

    def get_cache_key(self, query: str, k: int):
        return hashlib.md5(f"{query}:{k}".encode()).hexdigest()

    def get(self, query: str, k: int):
        key = self.get_cache_key(query, k)
        if key in self.cache:
            result, timestamp = self.cache[key]
            if time.time() - timestamp < self.ttl:
                return result
        return None

    def set(self, query: str, k: int, result):
        key = self.get_cache_key(query, k)
        self.cache[key] = (result, time.time())

# Usage
cache = RAGCache(ttl_seconds=3600)

def cached_retrieval(query: str, k: int = 10):
    cached_result = cache.get(query, k)
    if cached_result is None:
        result = retriever.invoke(query)
        cache.set(query, k, result)
        return result
    return cached_result

2. Model Routing

class AdaptiveRAG:
    def __init__(self):
        # Use cheap models for auxiliary tasks
        self.router_model = "gpt-4o-mini"
        self.grader_model = "gpt-4o-mini"
        # Use powerful model only for generation
        self.generator_model = "claude-sonnet-4-6-20250217"

    def classify_query_complexity(self, query: str):
        """Route with small model"""
        prompt = f"Is this a simple or complex query? {query}"
        response = openai.chat.completions.create(
            model=self.router_model,
            messages=[{"role": "user", "content": prompt}],
            max_tokens=10
        )
        return response.choices[0].message.content

    def grade_documents(self, query: str, docs: list):
        """Grade with small model"""
        # Grading logic using self.grader_model
        pass

    def generate_answer(self, query: str, docs: list):
        """Generate with powerful model"""
        # Generation logic using self.generator_model
        pass

Routing and grading dominate the call count in this pipeline while generation dominates the token count. Putting the small model on the frequent calls is where the savings sit, and the answer the user reads still comes from the strong model.

Production Decision Framework

Low Latency Priority (<1s response):

• Basic vector search or lightweight hybrid
• Avoid multi-query patterns and heavy reranking
• Aggressive caching
• Quantized embedding models
• HNSW indexing with tuned parameters

High Accuracy Priority (>90% faithfulness):

• Hybrid search + cross-encoder reranking
• CRAG for quality verification
• Hierarchical/parent-child chunking
• GraphRAG for multi-hop queries
• Accept higher latency and cost

Cost-Constrained:

• Smaller embedding models
• Limited numberOfResults (5-10)
• Avoid multi-query patterns
• Use cheap LLMs for routing/grading
• Aggressive caching
• Approximate indexing (IVF+PQ)

Balanced Approach (Most Common):

• Hybrid search with RRF
• Lightweight reranking
• Parent-child chunking
• Moderate numberOfResults (10-15)
• Selective caching
• Mid-tier models (Claude Haiku, GPT-4o-mini)

The Default and Its Limits

Hybrid search with parent-child chunking is the configuration to reach for first. It repairs the two failures that hurt earliest, missed exact terms and fragmented context, and it needs no infrastructure beyond a keyword index. Everything else is an override, and each one has a condition attached. Cross-encoder reranking earns its inference budget when precision on domain-specific queries is the binding constraint. CRAG earns its extra passes when retrieval quality is unpredictable and a confident wrong answer is expensive. GraphRAG earns its ingestion bill only when questions genuinely span documents and the corpus is large enough to hold real relationships; below a thousand documents, or with mostly single-hop questions, it takes more than it returns.

Stand up RAGAS evaluation before any of these, because without a faithfulness and context-precision baseline every escalation looks like progress.

References

• Retrieval-Augmented Generation for Knowledge-Intensive NLP Tasks - Original RAG paper by Lewis et al. (2020), NeurIPS
• LangChain Retrieval Documentation - Official LangChain guide to building RAG systems
• LlamaIndex RAG - Introduction - LlamaIndex framework guide for retrieval and indexing
• Contextual Retrieval - Anthropic - Anthropic research on contextual chunking; 49-67% failure reduction
• OpenAI Vector Embeddings Guide - Embedding model documentation including Matryoshka truncation