Implementing Confidence-Calibrated Retrieval: Scoring, Thresholding, and Fallback Strategies When RAG Cannot Find a Reliable Answer

Introduction

Here's a scenario that plays out in production RAG systems every day: a user asks your enterprise document assistant a question that falls just outside the coverage of your knowledge base. Maybe they're asking about a product feature that was announced last week but hasn't made it into your indexed docs. Maybe they're using different terminology than your documents do. Or maybe they're asking something that's genuinely not in your corpus at all.

What happens? In a naive RAG implementation, the retriever returns something — it always does, because cosine similarity always has a closest match — and the LLM dutifully synthesizes a confident-sounding answer from marginally relevant chunks. The user walks away with plausible-sounding misinformation. This is the RAG equivalent of a confident wrong answer, and it's often worse than saying "I don't know."

The solution is confidence-calibrated retrieval: a system that scores its own certainty about retrieval quality, makes threshold-based decisions about whether to proceed, and routes to appropriate fallback strategies when it can't find a reliable answer. By the end of this lesson, you'll have a complete, production-grade implementation of this system, including the scoring machinery, decision logic, and multiple fallback paths.

What you'll learn:

• How to construct multi-signal confidence scores that go beyond raw cosine similarity
• How to calibrate thresholds empirically using labeled evaluation sets
• How to implement tiered fallback strategies: clarification, broader retrieval, LLM-only response, and hard refusal
• How to structure retrieval responses so the LLM knows how much to trust its context
• How to monitor confidence distributions over time to detect index drift and coverage gaps

---

Prerequisites

You should be comfortable with the fundamentals of RAG architecture — how embedding-based retrieval works, what vector similarity scores mean, and how retrieved chunks get passed to an LLM. If you need to build that foundation first, start with Retrieval-Augmented Generation Explained: How RAG Works and When to Use It and then Building a Production Document Q&A System with Vector Embeddings.

You should also be comfortable writing async Python and working with OpenAI or a similar embeddings API. We'll use Pinecone for vector storage in examples, but the concepts translate directly to any vector database.

---

Why Cosine Similarity Alone Cannot Be Your Confidence Signal

Before writing any code, you need to understand what cosine similarity is actually measuring — and more importantly, what it isn't measuring.

Cosine similarity measures the angle between two vectors in embedding space. A score of 0.92 between your query vector and a document chunk vector means those representations are geometrically close. But "geometrically close in embedding space" is not the same as "this chunk actually answers the question." The embedding model that generated both vectors learned a compressed representation of semantic meaning, and while it's impressively good, it has blind spots.

Consider this pair:

• Query: "What is our refund policy for SaaS subscriptions?"
• Top chunk: "Our customer success team is committed to ensuring every subscriber gets maximum value from their investment in our platform."

That chunk might score 0.85 similarity — both are about customers and subscriptions and value. But it answers absolutely nothing about refund policy. The embedding model correctly identified topical proximity but that's not what you need.

The fundamental problem is that cosine similarity is a retrieval signal, not a relevance judgment. It tells you which chunks are most likely to be relevant, ranked, but it cannot tell you whether the best chunk you found actually contains the answer to the question.

This distinction drives the architecture we're about to build. We need to combine multiple signals into a composite confidence score that better approximates "does my index contain a reliable answer to this question?"

---

The Four Signals of Retrieval Confidence

A well-calibrated confidence score should incorporate at least these four signals, each capturing a different failure mode.

Signal 1: Top-K Score Distribution (Gap and Concentration)

When your index genuinely contains an answer, the top result tends to separate meaningfully from the pack. When it doesn't, scores are flatter — the retriever is essentially guessing among equally-plausible-but-wrong chunks.

Two metrics matter here:

Score gap: The difference between the top result's score and the second result's score. A large gap suggests the top result is distinctively relevant. A small gap suggests ambiguity.

Score concentration: How many of the top-K results have scores above a threshold, normalized. If 7 of your top-10 results score above 0.80, that suggests broad topical coverage. If only 1 does, the retriever found one somewhat-relevant chunk and padded the rest.

def compute_distribution_signals(scores: list[float]) -> dict:
    """
    scores: List of similarity scores for top-K results, descending order.
    Returns gap and concentration metrics.
    """
    if not scores:
        return {"gap": 0.0, "concentration": 0.0, "top_score": 0.0}
    
    top_score = scores[0]
    gap = (scores[0] - scores[1]) if len(scores) > 1 else scores[0]
    
    # Concentration: fraction of top-K above a meaningful threshold
    # 0.75 is a reasonable baseline for most embedding models
    threshold = 0.75
    concentration = sum(1 for s in scores if s > threshold) / len(scores)
    
    return {
        "top_score": top_score,
        "gap": gap,
        "concentration": concentration
    }

Signal 2: Cross-Chunk Semantic Coherence

If your index contains a real answer, the top chunks should discuss the same topic consistently. If retrieval is failing, the top chunks often come from disparate parts of the corpus that happened to match individual words in the query.

We can measure this by computing pairwise similarity between the retrieved chunks themselves (not between chunks and the query). Low inter-chunk similarity suggests the retriever scraped together unrelated content.

import numpy as np
from openai import OpenAI

client = OpenAI()

def get_embeddings(texts: list[str], model: str = "text-embedding-3-small") -> list[list[float]]:
    response = client.embeddings.create(input=texts, model=model)
    return [item.embedding for item in response.data]

def compute_chunk_coherence(chunk_texts: list[str]) -> float:
    """
    Measure how semantically related the retrieved chunks are to each other.
    Returns a coherence score between 0 and 1.
    """
    if len(chunk_texts) < 2:
        return 1.0  # Only one chunk; nothing to compare
    
    embeddings = get_embeddings(chunk_texts)
    embeddings_array = np.array(embeddings)
    
    # Normalize
    norms = np.linalg.norm(embeddings_array, axis=1, keepdims=True)
    normalized = embeddings_array / norms
    
    # Pairwise cosine similarities
    similarity_matrix = normalized @ normalized.T
    
    # Average of upper triangle (excluding diagonal)
    n = len(chunk_texts)
    upper_triangle_indices = np.triu_indices(n, k=1)
    pairwise_scores = similarity_matrix[upper_triangle_indices]
    
    return float(np.mean(pairwise_scores))

Signal 3: Query Coverage Estimation

This signal asks: does the top chunk actually contain the specific entities and concepts the query is asking about? We can approximate this with a lightweight keyword-overlap calculation, but a more robust version uses named entity recognition or query decomposition.

import re
from collections import Counter

def compute_query_coverage(query: str, chunk_texts: list[str], top_n: int = 3) -> float:
    """
    Estimate what fraction of meaningful query terms appear in the top chunks.
    """
    # Extract meaningful terms from query (skip stopwords)
    stopwords = {
        "what", "is", "the", "a", "an", "in", "of", "for", "to", "and",
        "or", "how", "do", "does", "can", "could", "would", "should",
        "our", "your", "their", "this", "that", "are", "was", "were"
    }
    
    query_terms = set(
        re.sub(r'[^a-z0-9\s]', '', query.lower()).split()
    ) - stopwords
    
    if not query_terms:
        return 0.5  # Can't assess; return neutral
    
    # Check top N chunks
    combined_text = " ".join(chunk_texts[:top_n]).lower()
    found_terms = sum(1 for term in query_terms if term in combined_text)
    
    return found_terms / len(query_terms)

Signal 4: LLM-Assisted Relevance Judgment (The Expensive But Accurate Signal)

For high-stakes decisions or when other signals are ambiguous, ask a small, fast model to make a binary relevance judgment. This is the most accurate signal but costs additional tokens and latency.

The key to making this practical is using a small model (GPT-4o-mini, Claude Haiku) with a tightly structured prompt that returns a JSON response.

import json
from openai import OpenAI

client = OpenAI()

def llm_relevance_check(
    query: str, 
    top_chunks: list[str],
    model: str = "gpt-4o-mini"
) -> dict:
    """
    Ask a lightweight LLM to judge whether the retrieved chunks
    can answer the question. Returns a dict with 'can_answer' (bool)
    and 'confidence' (0.0-1.0).
    """
    chunk_preview = "\n\n---\n\n".join(
        f"Chunk {i+1}:\n{chunk[:500]}" 
        for i, chunk in enumerate(top_chunks[:3])
    )
    
    prompt = f"""You are evaluating whether retrieved document chunks contain 
enough information to answer a user's question.

USER QUESTION:
{query}

RETRIEVED CHUNKS:
{chunk_preview}

Assess whether these chunks, taken together, contain sufficient information 
to answer the user's question accurately.

Respond with ONLY a JSON object in this exact format:
{{
  "can_answer": true or false,
  "confidence": 0.0 to 1.0,
  "reasoning": "one sentence explanation"
}}

Be conservative. If the chunks are tangentially related but don't actually 
contain the specific information needed, respond with can_answer: false."""

    response = client.chat.completions.create(
        model=model,
        messages=[{"role": "user", "content": prompt}],
        temperature=0.0,
        response_format={"type": "json_object"}
    )
    
    return json.loads(response.choices[0].message.content)

Using structured output for this relevance check is critical — you need reliable parseable output, not prose.

---

Building the Composite Confidence Score

Now we combine these signals into a single score. The right weighting depends on your domain, but here's a principled starting point and the machinery to tune it.

from dataclasses import dataclass
from typing import Optional
import numpy as np

@dataclass
class RetrievalConfidence:
    composite_score: float
    top_score: float
    score_gap: float
    score_concentration: float
    chunk_coherence: float
    query_coverage: float
    llm_judgment: Optional[float]  # None if not computed
    signal_breakdown: dict
    
    @property
    def passes_threshold(self) -> bool:
        return self.composite_score >= 0.65  # Default; should be calibrated

class ConfidenceScorer:
    """
    Composite confidence scorer for RAG retrieval results.
    
    Weights are calibrated against a labeled evaluation set.
    Default weights represent a balanced starting point.
    """
    
    def __init__(
        self,
        weights: Optional[dict] = None,
        use_llm_judgment: bool = False,
        llm_model: str = "gpt-4o-mini"
    ):
        self.weights = weights or {
            "top_score": 0.35,
            "score_gap": 0.20,
            "score_concentration": 0.15,
            "chunk_coherence": 0.15,
            "query_coverage": 0.15
        }
        self.use_llm_judgment = use_llm_judgment
        self.llm_model = llm_model
        
        # Validate weights sum to 1.0
        total = sum(self.weights.values())
        if abs(total - 1.0) > 0.01:
            raise ValueError(f"Weights must sum to 1.0, got {total}")
    
    def score(
        self,
        query: str,
        retrieved_chunks: list[dict]  # Each dict has 'text' and 'score'
    ) -> RetrievalConfidence:
        
        scores = [c["score"] for c in retrieved_chunks]
        texts = [c["text"] for c in retrieved_chunks]
        
        # Signal 1: Score distribution
        dist_signals = compute_distribution_signals(scores)
        
        # Signal 2: Chunk coherence
        coherence = compute_chunk_coherence(texts) if len(texts) > 1 else 1.0
        
        # Signal 3: Query coverage
        coverage = compute_query_coverage(query, texts)
        
        # Normalize top_score and gap to [0, 1] range
        # Assumes scores are already cosine similarities in [0, 1]
        normalized_signals = {
            "top_score": dist_signals["top_score"],
            "score_gap": min(dist_signals["gap"] * 5, 1.0),  # Scale gap: 0.2 → 1.0
            "score_concentration": dist_signals["concentration"],
            "chunk_coherence": coherence,
            "query_coverage": coverage
        }
        
        # Compute weighted composite
        composite = sum(
            self.weights[signal] * value 
            for signal, value in normalized_signals.items()
        )
        
        # Optional: Override with LLM judgment
        llm_confidence = None
        if self.use_llm_judgment:
            llm_result = llm_relevance_check(query, texts, self.llm_model)
            llm_confidence = llm_result["confidence"]
            if not llm_result["can_answer"]:
                # LLM says no — hard cap the composite score
                composite = min(composite, 0.45)
            else:
                # Blend LLM judgment into composite
                composite = composite * 0.6 + llm_confidence * 0.4
        
        return RetrievalConfidence(
            composite_score=composite,
            top_score=dist_signals["top_score"],
            score_gap=dist_signals["gap"],
            score_concentration=dist_signals["concentration"],
            chunk_coherence=coherence,
            query_coverage=coverage,
            llm_judgment=llm_confidence,
            signal_breakdown=normalized_signals
        )

---

Calibrating Your Thresholds Empirically

This is where most implementations fail. Teams pick a threshold like 0.7 because it "feels right" and ship it. The problem is that threshold values are entirely dependent on your specific embedding model, your document corpus, and your query distribution. A threshold that works beautifully for one combination can be catastrophically wrong for another.

Proper calibration requires a labeled evaluation set.

Building an Evaluation Dataset

You need a dataset of (query, answer_quality) pairs where you've manually labeled whether RAG produced a reliable answer. Here's a minimal structure:

# Example evaluation dataset entries
eval_dataset = [
    {
        "query": "What is the cancellation fee for annual enterprise plans?",
        "has_answer_in_corpus": True,
        "correct_answer_exists": True
    },
    {
        "query": "Does the platform support SAML 2.0 SSO?",
        "has_answer_in_corpus": True,
        "correct_answer_exists": True
    },
    {
        "query": "What are the upcoming features planned for Q3?",
        "has_answer_in_corpus": False,  # Not in docs
        "correct_answer_exists": False
    },
    {
        "query": "How do I export data to Salesforce?",
        "has_answer_in_corpus": True,
        "correct_answer_exists": True
    },
    # ... at minimum 100 labeled examples; 300+ for reliable calibration
]

A good evaluation set should have roughly 40-60% of queries that can be answered by the corpus and 40-60% that cannot. If you sample only from actual user queries, you'll likely see a skewed distribution that makes calibration harder.

Finding the Optimal Threshold

import numpy as np
from sklearn.metrics import f1_score, precision_score, recall_score

def calibrate_threshold(
    scored_examples: list[dict],  # Each has 'composite_score' and 'has_answer'
    target_metric: str = "f1",
    min_precision: float = 0.85  # We rarely want to sacrifice precision for recall
) -> dict:
    """
    Find the confidence threshold that optimizes the target metric
    while respecting the minimum precision constraint.
    
    Args:
        scored_examples: List of dicts with composite_score (float) and 
                        has_answer (bool) labels
        target_metric: 'f1', 'precision', or 'recall'
        min_precision: Minimum acceptable precision (avoid false answers)
    
    Returns:
        dict with optimal threshold and performance metrics
    """
    scores = np.array([e["composite_score"] for e in scored_examples])
    labels = np.array([int(e["has_answer"]) for e in scored_examples])
    
    # Try every percentile as a candidate threshold
    candidate_thresholds = np.percentile(scores, range(10, 95, 2))
    
    best_threshold = None
    best_metric_value = -1
    results = []
    
    for threshold in candidate_thresholds:
        predictions = (scores >= threshold).astype(int)
        
        precision = precision_score(labels, predictions, zero_division=0)
        recall = recall_score(labels, predictions, zero_division=0)
        f1 = f1_score(labels, predictions, zero_division=0)
        
        results.append({
            "threshold": threshold,
            "precision": precision,
            "recall": recall,
            "f1": f1
        })
        
        # Respect precision floor
        if precision < min_precision:
            continue
        
        metric_value = {"f1": f1, "precision": precision, "recall": recall}[target_metric]
        
        if metric_value > best_metric_value:
            best_metric_value = metric_value
            best_threshold = threshold
    
    if best_threshold is None:
        # No threshold achieved minimum precision; find least-bad option
        best_threshold = candidate_thresholds[
            np.argmax([r["precision"] for r in results])
        ]
    
    best_result = next(r for r in results if r["threshold"] == best_threshold)
    
    return {
        "optimal_threshold": float(best_threshold),
        "precision_at_threshold": best_result["precision"],
        "recall_at_threshold": best_result["recall"],
        "f1_at_threshold": best_result["f1"],
        "all_results": results
    }

The calibration run gives you a threshold value. But don't treat it as permanent. Query distributions shift over time, your corpus grows, and your embedding model might change. Build this into your CI/CD pipeline — every significant index update should trigger a re-calibration run against your evaluation set. The lesson on building a self-improving RAG pipeline has more on making this continuous.

---

Implementing Tiered Fallback Strategies

Now for the most architecturally interesting part: what happens when confidence falls below your threshold? A single "I don't know" response is both correct and useless. Real systems need tiered fallbacks that extract maximum value from partial information.

Here's the confidence band structure we'll implement:

Score ≥ HIGH_THRESHOLD (e.g., 0.80):   Proceed with standard RAG
Score in MID_BAND (e.g., 0.65–0.80):   Proceed with hedged response + citations
Score in LOW_BAND (e.g., 0.45–0.65):   Trigger query clarification or broader retrieval
Score < LOW_THRESHOLD (e.g., 0.45):    Route to LLM-only response or hard refusal

from enum import Enum
from dataclasses import dataclass
from typing import Callable, Awaitable

class RetrievalDecision(Enum):
    PROCEED_HIGH_CONFIDENCE = "proceed_high"
    PROCEED_WITH_HEDGING = "proceed_hedged"
    ATTEMPT_CLARIFICATION = "clarify"
    ATTEMPT_BROADER_RETRIEVAL = "broaden"
    FALLBACK_LLM_ONLY = "llm_only"
    HARD_REFUSAL = "refuse"

@dataclass
class ConfidenceThresholds:
    high: float = 0.80
    mid: float = 0.65
    low: float = 0.45
    
    # Domain-specific overrides
    # E.g., for medical/legal content, raise all thresholds
    @classmethod
    def for_high_stakes_domain(cls):
        return cls(high=0.88, mid=0.75, low=0.55)
    
    @classmethod
    def for_exploratory_search(cls):
        return cls(high=0.70, mid=0.50, low=0.30)

class FallbackRouter:
    """
    Routes retrieval results to appropriate handling strategies
    based on confidence score and context.
    """
    
    def __init__(self, thresholds: ConfidenceThresholds = None):
        self.thresholds = thresholds or ConfidenceThresholds()
    
    def decide(
        self, 
        confidence: RetrievalConfidence,
        query_has_clarification_potential: bool = True,
        allow_llm_only: bool = True
    ) -> RetrievalDecision:
        
        score = confidence.composite_score
        
        if score >= self.thresholds.high:
            return RetrievalDecision.PROCEED_HIGH_CONFIDENCE
        
        elif score >= self.thresholds.mid:
            return RetrievalDecision.PROCEED_WITH_HEDGING
        
        elif score >= self.thresholds.low:
            # In mid-low band: try to recover before giving up
            if confidence.query_coverage < 0.3:
                # Query terms barely appear in chunks → likely wrong domain
                # Clarification might help more than broader retrieval
                if query_has_clarification_potential:
                    return RetrievalDecision.ATTEMPT_CLARIFICATION
            
            # Try to find better chunks with an expanded search
            return RetrievalDecision.ATTEMPT_BROADER_RETRIEVAL
        
        else:
            # Very low confidence
            if allow_llm_only:
                return RetrievalDecision.FALLBACK_LLM_ONLY
            return RetrievalDecision.HARD_REFUSAL

Fallback 1: Hedged Response Generation

When confidence is in the mid-band, the LLM should know it's working from uncertain ground. The key is injecting confidence metadata into the prompt, not just the chunks.

def build_hedged_prompt(
    query: str,
    chunks: list[dict],
    confidence: RetrievalConfidence
) -> str:
    
    chunk_text = "\n\n---\n\n".join(
        f"[Source {i+1}, relevance: {chunk['score']:.2f}]\n{chunk['text']}"
        for i, chunk in enumerate(chunks)
    )
    
    return f"""You are answering a question using retrieved documents. 
The retrieval system has MODERATE confidence ({confidence.composite_score:.2f}/1.0) 
that these documents contain the answer. 

This means:
- The documents may be partially relevant but not a perfect match
- Some information may be missing or tangential
- You MUST acknowledge uncertainty in your response
- You MUST recommend the user verify critical information

RETRIEVED DOCUMENTS:
{chunk_text}

USER QUESTION: {query}

Instructions:
1. Answer based ONLY on what's explicitly stated in the documents above
2. Begin your response with a brief uncertainty acknowledgment
3. If you can only partially answer, say so clearly
4. Do NOT fill gaps with general knowledge
5. Cite which source(s) you're drawing from (e.g., "According to Source 2...")"""

Fallback 2: Query Clarification

When the query terms barely appear in retrieved content, the problem might be terminology mismatch rather than missing information. Generate a clarifying question.

def generate_clarification_question(
    query: str,
    retrieved_chunks: list[dict],
    confidence: RetrievalConfidence
) -> str:
    """
    Generate a targeted clarification question when retrieval confidence
    is low due to likely terminology mismatch or ambiguity.
    """
    
    # What topics DID we find? Use these to guide clarification.
    top_chunk_summary = retrieved_chunks[0]["text"][:300] if retrieved_chunks else ""
    
    prompt = f"""A user asked: "{query}"

The search system found potentially related content about: 
{top_chunk_summary}

However, confidence is low that this content answers the question.

Generate ONE brief clarifying question that would help determine:
1. Whether the user is asking about something in our documentation
2. What specific aspect they need help with

The clarifying question should be natural and helpful, not expose system details.
Return only the question text, no preamble."""
    
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}],
        temperature=0.3,
        max_tokens=100
    )
    
    return response.choices[0].message.content.strip()

Fallback 3: Broader Retrieval with Query Expansion

When a narrow query fails, try expanding it. Query expansion generates alternative phrasings and runs multiple retrievals.

def expand_query(query: str, n_expansions: int = 3) -> list[str]:
    """
    Generate alternative phrasings of the query to increase retrieval coverage.
    Returns original query plus n_expansions alternatives.
    """
    prompt = f"""Given this search query, generate {n_expansions} alternative 
phrasings that might retrieve relevant documents. Consider:
- Synonyms and related terminology
- More general/specific versions  
- Different ways to frame the same information need

Original query: {query}

Return ONLY a JSON array of strings (the alternative phrasings), 
no original query, no explanations."""

    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}],
        temperature=0.5,
        response_format={"type": "json_object"}
    )
    
    # Parse and return
    result = json.loads(response.choices[0].message.content)
    # Handle both {"queries": [...]} and direct array formats
    if isinstance(result, list):
        expansions = result
    else:
        expansions = next(iter(result.values()), [])
    
    return [query] + expansions[:n_expansions]

async def broader_retrieval(
    query: str,
    retrieval_fn: Callable,  # Your vector search function
    scorer: ConfidenceScorer,
    top_k: int = 5
) -> tuple[list[dict], RetrievalConfidence]:
    """
    Attempt retrieval with expanded queries, return the best result set.
    """
    expanded_queries = expand_query(query)
    
    best_confidence = None
    best_chunks = []
    
    for expanded_query in expanded_queries:
        chunks = await retrieval_fn(expanded_query, top_k=top_k)
        confidence = scorer.score(expanded_query, chunks)
        
        if best_confidence is None or confidence.composite_score > best_confidence.composite_score:
            best_confidence = confidence
            best_chunks = chunks
    
    return best_chunks, best_confidence

This integrates naturally with hybrid search approaches — when query expansion is running, you might also want to toggle between dense and sparse retrieval modes to maximize coverage.

Fallback 4: LLM-Only Response

When retrieval genuinely fails, an honest LLM-only response is often more valuable than a forced RAG answer. The key is structuring the prompt so the LLM knows it's operating without grounded context.

def build_llm_only_prompt(query: str, domain_description: str) -> str:
    return f"""You are a helpful assistant for {domain_description}.

A user has asked: "{query}"

IMPORTANT: You do not have access to specific documentation to answer this question. 
You should:
1. Acknowledge that you cannot find this in your knowledge base
2. If you can offer general guidance that might be helpful, do so with clear caveats
3. Direct the user to appropriate resources (support team, documentation URL, etc.)
4. Do NOT present general knowledge as if it were from official documentation

Your response should be helpful but honest about its limitations."""

---

Putting It All Together: The Full Pipeline

Here's how the components integrate into a complete confidence-calibrated RAG pipeline:

import asyncio
from dataclasses import dataclass
from typing import Optional

@dataclass
class RAGResponse:
    answer: str
    decision: RetrievalDecision
    confidence: RetrievalConfidence
    sources: list[dict]
    fallback_activated: bool
    fallback_type: Optional[str]

class ConfidenceCalibratedRAG:
    """
    Production-ready RAG pipeline with confidence scoring and fallback routing.
    """
    
    def __init__(
        self,
        retrieval_fn: Callable,          # Async function: (query, top_k) -> list[dict]
        generation_client: OpenAI,
        scorer: ConfidenceScorer,
        router: FallbackRouter,
        generation_model: str = "gpt-4o",
        domain_description: str = "enterprise software",
        enable_query_expansion: bool = True,
        max_expansion_attempts: int = 1
    ):
        self.retrieval_fn = retrieval_fn
        self.client = generation_client
        self.scorer = scorer
        self.router = router
        self.generation_model = generation_model
        self.domain_description = domain_description
        self.enable_query_expansion = enable_query_expansion
        self.max_expansion_attempts = max_expansion_attempts
    
    async def query(
        self,
        query: str,
        conversation_history: Optional[list] = None,
        top_k: int = 5
    ) -> RAGResponse:
        
        # Step 1: Initial retrieval
        chunks = await self.retrieval_fn(query, top_k=top_k)
        confidence = self.scorer.score(query, chunks)
        
        # Step 2: Routing decision
        decision = self.router.decide(confidence)
        
        # Step 3: Handle each decision branch
        
        if decision == RetrievalDecision.PROCEED_HIGH_CONFIDENCE:
            answer = await self._generate_standard(query, chunks, conversation_history)
            return RAGResponse(
                answer=answer,
                decision=decision,
                confidence=confidence,
                sources=chunks,
                fallback_activated=False,
                fallback_type=None
            )
        
        elif decision == RetrievalDecision.PROCEED_WITH_HEDGING:
            answer = await self._generate_hedged(query, chunks, confidence, conversation_history)
            return RAGResponse(
                answer=answer,
                decision=decision,
                confidence=confidence,
                sources=chunks,
                fallback_activated=True,
                fallback_type="hedged_generation"
            )
        
        elif decision == RetrievalDecision.ATTEMPT_CLARIFICATION:
            clarification = generate_clarification_question(query, chunks, confidence)
            return RAGResponse(
                answer=clarification,
                decision=decision,
                confidence=confidence,
                sources=[],
                fallback_activated=True,
                fallback_type="clarification_request"
            )
        
        elif decision == RetrievalDecision.ATTEMPT_BROADER_RETRIEVAL:
            if self.enable_query_expansion:
                expanded_chunks, expanded_confidence = await broader_retrieval(
                    query, self.retrieval_fn, self.scorer, top_k=top_k
                )
                
                # Re-evaluate with expanded results
                new_decision = self.router.decide(expanded_confidence)
                
                if new_decision in (
                    RetrievalDecision.PROCEED_HIGH_CONFIDENCE,
                    RetrievalDecision.PROCEED_WITH_HEDGING
                ):
                    answer = await self._generate_hedged(
                        query, expanded_chunks, expanded_confidence, conversation_history
                    )
                    return RAGResponse(
                        answer=answer,
                        decision=new_decision,
                        confidence=expanded_confidence,
                        sources=expanded_chunks,
                        fallback_activated=True,
                        fallback_type="query_expansion"
                    )
            
            # Expansion didn't help enough; fall through to LLM-only
            answer = await self._generate_llm_only(query, conversation_history)
            return RAGResponse(
                answer=answer,
                decision=RetrievalDecision.FALLBACK_LLM_ONLY,
                confidence=confidence,
                sources=[],
                fallback_activated=True,
                fallback_type="llm_only_after_expansion"
            )
        
        elif decision == RetrievalDecision.FALLBACK_LLM_ONLY:
            answer = await self._generate_llm_only(query, conversation_history)
            return RAGResponse(
                answer=answer,
                decision=decision,
                confidence=confidence,
                sources=[],
                fallback_activated=True,
                fallback_type="llm_only"
            )
        
        else:  # HARD_REFUSAL
            return RAGResponse(
                answer=self._hard_refusal_message(query),
                decision=decision,
                confidence=confidence,
                sources=[],
                fallback_activated=True,
                fallback_type="hard_refusal"
            )
    
    async def _generate_standard(self, query, chunks, history):
        messages = self._build_messages(
            query, chunks, history,
            system_note="You are a helpful assistant with access to documentation."
        )
        response = self.client.chat.completions.create(
            model=self.generation_model,
            messages=messages,
            temperature=0.2
        )
        return response.choices[0].message.content
    
    async def _generate_hedged(self, query, chunks, confidence, history):
        prompt = build_hedged_prompt(query, chunks, confidence)
        messages = (history or []) + [{"role": "user", "content": prompt}]
        response = self.client.chat.completions.create(
            model=self.generation_model,
            messages=messages,
            temperature=0.2
        )
        return response.choices[0].message.content
    
    async def _generate_llm_only(self, query, history):
        prompt = build_llm_only_prompt(query, self.domain_description)
        messages = (history or []) + [{"role": "user", "content": prompt}]
        response = self.client.chat.completions.create(
            model=self.generation_model,
            messages=messages,
            temperature=0.3
        )
        return response.choices[0].message.content
    
    def _build_messages(self, query, chunks, history, system_note):
        chunk_text = "\n\n".join(c["text"] for c in chunks)
        system = f"{system_note}\n\nContext:\n{chunk_text}"
        messages = [{"role": "system", "content": system}]
        if history:
            messages.extend(history)
        messages.append({"role": "user", "content": query})
        return messages
    
    def _hard_refusal_message(self, query):
        return (
            "I wasn't able to find reliable information to answer your question. "
            "This topic may not be covered in our documentation, or your question "
            "may need more specific details. Please contact our support team directly "
            "or try rephrasing your question."
        )

---

Monitoring Confidence Distributions in Production

A calibrated threshold that worked at launch will drift. Your corpus grows, user query patterns evolve, and you might update your embedding model. Without monitoring, you won't know when your thresholds need recalibration.

The signals to monitor are:

Confidence score distribution over time: Plot a histogram of composite scores daily. If the distribution shifts left (more low scores), your index may have coverage gaps for new query types. If it shifts right, your corpus grew and now over-retrieves on broad queries.

Fallback rate by decision type: Track what percentage of queries hit each decision branch. A sudden spike in LLM-only fallbacks is a canary for index drift or a new user query pattern.

User satisfaction signals correlated with confidence bands: If your system collects thumbs up/down feedback, segment it by confidence band. This tells you whether your hedging thresholds are well-calibrated.

import time
from collections import defaultdict

class ConfidenceMonitor:
    """
    Lightweight monitoring for confidence score distributions.
    In production, integrate with your observability stack 
    (Datadog, Grafana, etc.) instead of storing in memory.
    """
    
    def __init__(self):
        self.events = []
        self.decision_counts = defaultdict(int)
    
    def record(self, response: RAGResponse, query: str):
        self.events.append({
            "timestamp": time.time(),
            "query_length": len(query.split()),
            "composite_score": response.confidence.composite_score,
            "top_score": response.confidence.top_score,
            "decision": response.decision.value,
            "fallback_type": response.fallback_type,
            "signal_breakdown": response.confidence.signal_breakdown
        })
        self.decision_counts[response.decision.value] += 1
    
    def get_distribution_stats(self, last_n: int = 1000) -> dict:
        recent = self.events[-last_n:]
        scores = [e["composite_score"] for e in recent]
        
        if not scores:
            return {}
        
        scores_array = np.array(scores)
        total = len(recent)
        
        return {
            "sample_size": total,
            "mean_confidence": float(np.mean(scores_array)),
            "median_confidence": float(np.median(scores_array)),
            "p10": float(np.percentile(scores_array, 10)),
            "p25": float(np.percentile(scores_array, 25)),
            "p75": float(np.percentile(scores_array, 75)),
            "decision_distribution": {
                decision: count / total 
                for decision, count in self.decision_counts.items()
            },
            "fallback_rate": sum(
                1 for e in recent if e["fallback_type"] is not None
            ) / total
        }

This monitoring integrates naturally with the broader observability approach covered in implementing LLM observability. Your confidence score and fallback decision should be first-class fields in every trace, not an afterthought.

---

Hands-On Exercise

Scenario: You're building a RAG-powered support assistant for a B2B SaaS company. Your document corpus covers product documentation, pricing pages, and release notes. Users ask a mix of questions: some clearly covered by docs, some about recent features that aren't yet indexed, and some completely off-topic (competitor comparisons, general business questions).

Exercise Tasks:

Part 1: Build and score a test set

Using the provided ConfidenceScorer class, run 20 queries through your retrieval system. Categorize them manually into three groups: clearly answerable, partially answerable, and not answerable from your corpus. Record the composite confidence score for each.

For each query, identify which signal (top_score, score_gap, score_concentration, chunk_coherence, query_coverage) contributed most to the confidence assessment. Write a brief analysis of whether the signal correctly identified the query's answerability.

Part 2: Calibrate thresholds

Using your 20 labeled examples and the calibrate_threshold function, find the optimal threshold for your dataset. Try three different target metrics: F1, precision, and recall. Observe how changing the min_precision parameter from 0.80 to 0.95 changes the optimal threshold and the resulting recall.

Write a paragraph explaining which threshold you would choose and why, given the support assistant use case.

Part 3: Implement and test one fallback path

Implement the query expansion fallback. Run 5 queries from your "partially answerable" set through the full pipeline with expansion enabled. For each:

• Compare the original confidence score to the post-expansion score
• Note which query expansion improved retrieval most
• Identify cases where expansion didn't help and explain why

Part 4: Add a monitoring record

Wire up the ConfidenceMonitor to your pipeline. After running all 20 queries, call get_distribution_stats() and interpret the results. What does the fallback rate tell you about your index coverage? What would you do next to improve it?

---

Common Mistakes & Troubleshooting

Mistake: Using a single global threshold for all query types

Different query categories have very different confidence distributions. A "how-to" question asking for a specific procedure will get different signals than a conceptual question asking for an explanation. Consider training category-specific thresholds and using a query classifier to route to the appropriate threshold before scoring.

Mistake: Embedding the confidence score in the user-visible response

Telling users "I'm 73% confident" is confusing and can undermine trust in both directions. Users don't know what 73% means in this context. Use confidence scores internally to route behavior; express uncertainty to users in natural language ("I found some information that may be relevant, but I'd recommend verifying...").

Mistake: Over-relying on query expansion in low-confidence cases

Query expansion is expensive — it multiplies your retrieval calls and adds LLM latency for the expansion itself. It's most valuable when the confidence failure is likely a terminology mismatch (high top_score but low query_coverage). If the failure is more fundamental (all signals are low), expansion is unlikely to help and you should route to clarification or refusal instead.

Mistake: Ignoring chunk coherence when you have source diversity

If your RAG system retrieves from multiple disparate data sources (e.g., documentation, support tickets, marketing copy), low chunk coherence might be expected and normal, not a signal of retrieval failure. Consider computing coherence within each source rather than across all sources, or adjusting the weight of this signal for multi-source setups.

Mistake: Calibrating once and never recalibrating

This is the most common mistake in production systems. Build recalibration into your data pipeline — every time you do a significant index update or observe a meaningful shift in your confidence score distribution, run your evaluation dataset through the calibration function and update your thresholds. The evaluation set itself should also grow over time by collecting and labeling production queries.

Troubleshooting: All queries scoring below threshold

This usually means either your index is sparse relative to your query distribution, or your cosine similarity scores are systematically lower than expected (which can happen with certain embedding models or if documents are very long and chunked suboptimally). Check average top-K scores first. If they're uniformly below 0.65, review your chunking strategy — oversized chunks compress too much information and reduce per-query similarity scores.

Troubleshooting: Fallback rate spikes after index updates

Adding many new documents can paradoxically hurt confidence scores if the new content introduces a lot of topically-diverse chunks that dilute retrieval results. After a large index update, run your calibration set again before concluding there's a problem. The distribution shift may warrant a threshold adjustment rather than a rollback.

---

Summary & Next Steps

You now have a complete framework for confidence-calibrated retrieval — not as a theoretical concept but as a working system you can adapt to your production environment.

The core ideas to carry forward:

Confidence is multi-dimensional. No single metric (not even LLM-based relevance judgment) is sufficient on its own. Composite scores from distribution signals, chunk coherence, and query coverage give you a richer picture of why retrieval succeeded or failed.

Thresholds must be calibrated empirically. Pick a threshold based on your labeled evaluation set, your domain's risk tolerance, and the precision/recall tradeoff that makes sense for your use case. Recalibrate when your index or query distribution changes.

Fallbacks should be a spectrum, not a binary. The difference between a hedged response, a clarifying question, query expansion, and an LLM-only answer is not arbitrary — each is optimized for a different failure mode. Routing correctly between them is what separates a production-quality system from a demo.

Monitoring is not optional. Confidence distributions in production drift. Without monitoring your fallback rates and score distributions, you're flying blind.

For next steps, you might explore building a reranking layer to sharpen retrieval precision before the confidence gate, or look at building a citation and source attribution system to give users visibility into which sources contributed to hedged responses. Both integrate naturally with the confidence infrastructure we built here.

The agentic loop design pattern is also a natural evolution — once you have reliable confidence scoring, you can build multi-step retrieval agents that iteratively narrow uncertainty before committing to an answer, rather than making a single retrieval pass and routing to fallback.