Phase 3: RAG & Knowledge Systems

Handling no results, conflicting sources & stale data

Advanced ~17 min read
Think of it this way A friendly analogy. Read this if the technical version feels dense. Show Hide

Imagine you’re building a super helpful digital librarian. You want it to answer any question someone asks by finding the best books in its giant digital library. That sounds cool, right? But what if your librarian runs into a few common problems? If you ask "What's the best way to train a dragon?", and your librarian can't find any books about dragons, or only finds a really fuzzy, unhelpful pamphlet, it's not going to give a great answer. This is like your librarian coming up empty-handed, or only finding super weak clues.

Or what if you ask "How do I grow a giant pumpkin?", and one book says "Plant seeds in April" but another book says "Plant seeds in June"? Now your librarian has found information that completely disagrees! Which one should you trust? It's like having two expert chefs give you totally different instructions for the same recipe. And finally, imagine your librarian finds a book that says "Pluto is a planet." That was true a long time ago, but now we know scientists reclassified it! This is like having really old information that used to be correct but isn't anymore.

When you're building clever computer programs that talk to people, like our digital librarian, you don't want them to just make things up or give out wrong answers because of these problems. We need to teach our digital librarian to be super smart about how it finds and uses information. For no good books, it could say, "I didn't find anything solid, maybe try asking about gardening instead?" For conflicting books, it could show both ideas and say, "Book A says this, Book B says that – what do you think?" And for old books, it would always check the "publication date" to make sure it's giving you the freshest, most up-to-date facts.

So, when you're designing your own amazing AI software, you'll learn how to build in these clever checks. You'll teach your AI librarian to be picky about how good its sources are, to spot when different sources are arguing, and to always prefer the newest, most current information. This means your computer programs won't just be smart, they'll be trustworthy and truly helpful, giving people answers they can rely on every single time, not just on the easy questions.

The mental model: RAG failure modes as a pipeline control flow problem

Think of retrieval as a gated pipeline with three checkpoints before generation. Checkpoint one: did we retrieve anything with a score above the minimum acceptable relevance? Checkpoint two: are the retrieved chunks mutually consistent, or do they contradict each other on key facts? Checkpoint three: is the retrieved content still temporally valid? Each checkpoint has a pass, a warn, and a fail path. Most teams wire up only the happy path (pass) and send everything else straight to the LLM. The LLM then hallucinates a confident-sounding answer from bad input, and the user blames the AI when the real failure was in the pipeline logic upstream.

Similarity scores from vector search (cosine, dot product, or L2 depending on your database) are noisy signals. A chunk scoring 0.61 in a Qdrant search might be genuinely relevant for a narrow technical query, or it might be a tangentially related paragraph that shares vocabulary but not meaning. You cannot pick a universal threshold. Instead, calibrate per collection or per query domain by sampling 50-100 representative queries, manually labeling the top-3 results, and plotting score vs. relevance. You'll usually find a natural inflection point. For OpenAI text-embedding-3-small, scores above roughly 0.75 are usually safe; below 0.55 is noise. But those are illustrative starting points, not production config.

Real-world scenario: a B2B SaaS internal knowledge base

Imagine a company uses RAG over its internal Confluence and Notion exports to answer employee questions. Three problems surface in the first month. First, questions about new product features return zero results because the docs haven't been ingested yet, and the LLM invents answers. Second, the pricing page was updated, but the old version is still in the index alongside the new one. Employees get a mix of old and new pricing in the same response. Third, a regulatory compliance doc was updated 14 months ago but is indexed without a timestamp, so the pipeline treats it as authoritative.

A senior engineer approaches this methodically. For no-results: they set a threshold of 0.70, and when all retrieved chunks fall below it, they trigger query rewriting via a small, cheap LLM call (GPT-4o-mini or Claude Haiku) that expands acronyms, adds synonyms, and strips filler words. If the rewritten query also fails, the pipeline returns a structured fallback message that says what it searched for and suggests the user contact the relevant team. For conflicts: they store source_url, doc_version, and ingested_at in the metadata at ingestion time. Before sending chunks to generation, a comparison step checks if two or more chunks cover the same entity (same URL prefix or same doc ID) but disagree on numeric values or boolean facts. If they do, they pick the chunk with the most recent ingested_at and annotate the response with a warning that a conflict was detected. For staleness: they set a TTL of 30 days per document type (7 days for pricing pages, 90 days for architecture docs), and a background job checks ingested_at against TTL nightly and flags expired documents for re-ingestion.

Tradeoffs vs alternative approaches

Query rewriting adds a round-trip LLM call (50-150ms and a few hundred tokens). The alternative is retrieval with HyDE (Hypothetical Document Embeddings), where you generate a fake answer and embed it as the query. HyDE often retrieves better on vague queries but is slower and burns more tokens per request. For no-results fallback, some teams fall back to a pure LLM response without context. That works if your LLM has strong world knowledge about the domain, but it breaks the RAG contract: users expect answers grounded in your documents. A middle ground is to fall back to a broader BM25 keyword search before giving up on retrieval entirely, since BM25 handles exact-match queries that dense retrieval misses.

For conflict detection, you can either do it in the pipeline (pre-generation) or in the LLM prompt (post-generation). Pre-generation conflict detection is more reliable because you can act on it deterministically. Post-generation detection (instructing the LLM to flag contradictions) is probabilistic and depends on the model noticing the conflict in context, which it doesn't always do.

What changes at scale

At 10 users, you can re-index everything nightly and do conflict detection synchronously in the request path. At 10k users, nightly full re-index is too slow for dynamic content. You need an event-driven ingestion pipeline: a webhook or change-data-capture (CDC) stream from your source system triggers incremental upserts to the vector store. Conflict detection starts to be too expensive per request if you're running an LLM call for it. Move to a metadata-only heuristic: sort chunks by ingested_at descending, take only the most recent per source document. At 10M users, your vector database query latency becomes the bottleneck. You'll want read replicas, sharded collections by domain, and a cache layer (Redis with TTL-matched to your staleness budget) for queries that repeat within a short window. The no-results fallback path must be async and cheap because it's now handling thousands of requests per minute.

Key Takeaways

  • Gate retrieval on a minimum similarity score threshold before sending context to the LLM.
  • Store source metadata (timestamp, authority rank, document version) at ingestion time, not as an afterthought.
  • Detect contradictions by comparing retrieved chunks explicitly before generation, not after.
  • Schedule incremental re-indexing and use TTL-based cache invalidation to bound maximum staleness.

Pro tips

  • Calibrate your similarity threshold on a per-collection basis using a labeled eval set, not a single global constant. A dense technical collection and a broad FAQ collection have very different score distributions, and the same cutoff will either flood one with noise or starve the other of results.
  • Store ingested_at as an ISO 8601 UTC string in every chunk's metadata at write time, not query time. If you try to add it later via a bulk update, you'll find vector databases make metadata-only updates painful and you'll regret the shortcut.
  • When you detect conflicting sources, prefer the more recently ingested chunk for generation but still surface the conflict flag to the caller. Silently picking one without disclosure is how you build a system that gives confidently wrong answers when old data sneaks back in.
  • Query rewriting via a small LLM costs roughly 200-400 tokens per call. Put it in a lazy fallback branch that only fires after the primary retrieval fails the threshold check, not as a preprocessing step on every query. At high traffic the token cost of universal rewriting adds up fast.

Common pitfalls

  • Mistake: Passing empty or near-zero-score chunks to the LLM and hoping it ignores them. Fix: Gate on a score threshold before building the context window; return a structured no-results response instead.
  • Mistake: Detecting conflicts only in the LLM prompt instruction. Fix: Do metadata-based conflict detection in pipeline code before generation so you can act on it deterministically, not probabilistically.
  • Mistake: Omitting timestamps from chunk metadata at ingestion time and trying to infer freshness from document content. Fix: Always write ingested_at and source_last_modified to metadata during the ingestion pipeline, before embeddings are generated.
  • Mistake: Using a single global TTL for all document types. Fix: Assign TTL per document category at ingestion (e.g., 7 days for pricing, 90 days for architecture docs) and store it in metadata so the staleness check is self-describing.

When to use which no-results strategy

Option Use when Avoid when
Score threshold + hard stop The domain is narrow and a wrong answer is worse than no answer (compliance, legal, medical). Users expect a best-effort response even when docs are sparse; too many hard stops destroy UX.
Query rewriting fallback Users phrase queries informally or use internal jargon that doesn't match indexed terminology. Latency budget is tight (under 300ms end-to-end); the extra LLM round-trip will blow it.
BM25 keyword search fallback Queries include exact product names, error codes, or SKUs that dense retrieval misses. Your index has no BM25 capability and adding a separate keyword index is out of scope.
LLM-only response (no context) The domain is general knowledge and your LLM's parametric memory is accurate for it. Your RAG system exists precisely because the LLM lacks reliable knowledge of this domain.
Conflict-aware synthesis with both sources The question is about evolving facts (pricing, policy) and showing both versions helps the user decide. The conflict is between a clearly outdated and a clearly current version; just use the current one.

Code Example

python
# openai>=1.0.0, qdrant-client>=1.7.0
from qdrant_client import QdrantClient
from openai import OpenAI

MIN_SCORE_THRESHOLD = 0.72

client = QdrantClient(":memory:")
oai = OpenAI()  # reads OPENAI_API_KEY from env

def retrieve_with_fallback(query: str, collection: str, top_k: int = 5) -> dict:
    embedding = oai.embeddings.create(
        model="text-embedding-3-small", input=query
    ).data[0].embedding

    results = client.search(
        collection_name=collection,
        query_vector=embedding,
        limit=top_k,
        with_payload=True,
        score_threshold=MIN_SCORE_THRESHOLD,
    )

    if not results:
        return {"status": "no_results", "chunks": [], "query_used": query}

    return {
        "status": "ok",
        "chunks": [{"text": r.payload["text"], "score": r.score, "source": r.payload.get("source_url")} for r in results],
        "query_used": query,
    }

How this code works

This code defines a core function, retrieve_with_fallback, designed to intelligently fetch information for a RAG (Retrieval-Augmented Generation) pipeline. Its main job is to search a knowledge base, represented by QdrantClient, for content relevant to a user's query. Crucially, it's built to prevent the RAG system from receiving low-quality or completely absent information, ensuring that subsequent steps (like generating a response with an LLM) have reliable input.

The function first converts the text query into a numerical embedding using oai.embeddings.create from the OpenAI API, capturing the query's meaning. It then uses this embedding to client.search the specified data collection. A subtle but important detail here is score_threshold=MIN_SCORE_THRESHOLD. This line silently filters out any search results that don't meet a minimum relevance score (0.72), ensuring that only strong matches are considered. If, after this filtering, no relevant results are found, the if not results: check handles this gracefully by returning a "status": "no_results". This prevents the RAG pipeline from processing an empty or irrelevant result set, signaling instead that no suitable information was available. If good matches are found, it packages them with their text, score, and source information.

Production-grade example

Adds retries, rewrite fallback, staleness detection, conflict detection, structured logging, and latency observability.

python
# openai>=1.0.0, qdrant-client>=1.7.0, tenacity>=8.2.0
import logging
import os
import time
from datetime import datetime, timezone
from typing import Any

from openai import OpenAI, RateLimitError, APITimeoutError
from qdrant_client import QdrantClient
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type

log = logging.getLogger(__name__)
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")

MIN_SCORE = float(os.environ.get("RAG_MIN_SCORE", "0.72"))
STALE_DAYS = int(os.environ.get("RAG_STALE_DAYS", "30"))
EMBED_MODEL = "text-embedding-3-small"
REWRITE_MODEL = "gpt-4o-mini"

oai = OpenAI(api_key=os.environ["OPENAI_API_KEY"], timeout=10.0)
qdrant = QdrantClient(url=os.environ["QDRANT_URL"], api_key=os.environ.get("QDRANT_API_KEY"))

@retry(stop=stop_after_attempt(3), wait=wait_exponential(min=1, max=8),
       retry=retry_if_exception_type((RateLimitError, APITimeoutError)))
def _embed(text: str) -> list[float]:
    resp = oai.embeddings.create(model=EMBED_MODEL, input=text)
    log.info("embed tokens=%d", resp.usage.total_tokens)
    return resp.data[0].embedding

def _rewrite_query(query: str) -> str:
    msg = oai.chat.completions.create(
        model=REWRITE_MODEL,
        messages=[
            {"role": "system", "content": "Rewrite the user query to improve vector search recall. Expand acronyms, add synonyms, remove filler. Return only the rewritten query."},
            {"role": "user", "content": query},
        ],
        max_tokens=80,
        timeout=8.0,
    )
    return msg.choices[0].message.content.strip()

def _is_stale(payload: dict) -> bool:
    ts = payload.get("ingested_at")
    if not ts:
        return True
    age_days = (datetime.now(timezone.utc) - datetime.fromisoformat(ts)).days
    return age_days > STALE_DAYS

def _detect_conflicts(chunks: list[dict]) -> bool:
    seen: dict[str, Any] = {}
    for c in chunks:
        doc_id = c.get("doc_id")
        version = c.get("doc_version")
        if doc_id and doc_id in seen and seen[doc_id] != version:
            return True
        if doc_id:
            seen[doc_id] = version
    return False

def retrieve(
    query: str, collection: str, top_k: int = 5
) -> dict:
    start = time.monotonic()
    try:
        vec = _embed(query)
        hits = qdrant.search(collection_name=collection, query_vector=vec,
                             limit=top_k, with_payload=True,
                             score_threshold=MIN_SCORE)
        if not hits:
            log.warning("no_results query=%r; attempting rewrite", query)
            rewritten = _rewrite_query(query)
            vec2 = _embed(rewritten)
            hits = qdrant.search(collection_name=collection, query_vector=vec2,
                                 limit=top_k, with_payload=True,
                                 score_threshold=MIN_SCORE)
            if not hits:
                log.warning("no_results_after_rewrite original=%r rewritten=%r", query, rewritten)
                return {"status": "no_results", "chunks": [], "query_used": rewritten, "latency_ms": int((time.monotonic()-start)*1000)}

        chunks = [
            {"text": h.payload["text"], "score": h.score,
             "source_url": h.payload.get("source_url"),
             "doc_id": h.payload.get("doc_id"),
             "doc_version": h.payload.get("doc_version"),
             "ingested_at": h.payload.get("ingested_at"),
             "stale": _is_stale(h.payload)}
            for h in hits
        ]
        stale_count = sum(1 for c in chunks if c["stale"])
        conflict = _detect_conflicts(chunks)

        if stale_count:
            log.warning("stale_chunks count=%d collection=%s", stale_count, collection)
        if conflict:
            log.warning("conflicting_sources detected collection=%s query=%r", collection, query)

        chunks.sort(key=lambda c: (not c["stale"], c["score"]), reverse=True)

        return {
            "status": "ok",
            "chunks": chunks,
            "has_stale": stale_count > 0,
            "has_conflict": conflict,
            "query_used": query,
            "latency_ms": int((time.monotonic()-start)*1000),
        }
    except Exception as exc:
        log.error("retrieval_error query=%r error=%s", query, exc, exc_info=True)
        return {"status": "error", "chunks": [], "error": str(exc), "latency_ms": int((time.monotonic()-start)*1000)}

How this code works

This code defines a robust retrieve function, a core component of a Retrieval-Augmented Generation (RAG) pipeline. Its primary role is to fetch relevant document chunks from a QdrantClient vector database based on a user's query. It intelligently uses OpenAI models to process queries and critically, it's built to address common issues: gracefully handling scenarios where initial searches yield no results, identifying when retrieved information is outdated, and flagging potential conflicts between document versions. This ensures the RAG system provides reliable, fresh, and consistent context to the AI.

To achieve this, the retrieve function first embeds the original query using _embed and performs a qdrant.search. A crucial subtlety is MIN_SCORE: if hits don't meet this threshold or are entirely absent, the code considers it "no results." In such cases, _rewrite_query leverages gpt-4o-mini to rephrase the question for a second search attempt. For data freshness, _is_stale checks the ingested_at timestamp in each chunk's payload against STALE_DAYS, marking older content. The _detect_conflicts function identifies if chunks with the same doc_id have differing doc_versions, indicating potential inconsistencies. Finally, chunks.sort prioritizes fresher, higher-scoring data.

Practice & master

Try the exercise, check your understanding, then mark this lesson mastered to track your path to pro.

Exercise

Build a retrieve_and_classify function that queries a Qdrant collection, detects whether the results are absent, stale (older than 14 days), or conflicting (same doc_id with different doc_version values), and returns a status dict with a warning field populated accordingly. Use the in-memory Qdrant client and seed it with at least two conflicting test documents.

python
# openai>=1.0.0, qdrant-client>=1.7.0
from datetime import datetime, timezone, timedelta
from qdrant_client import QdrantClient
from qdrant_client.models import Distance, VectorParams, PointStruct

client = QdrantClient(":memory:")
MIN_SCORE = 0.70
STALE_DAYS = 14

# Seed collection with test data including a conflict
client.recreate_collection("docs", vectors_config=VectorParams(size=4, distance=Distance.COSINE))
client.upsert("docs", points=[
    PointStruct(id=1, vector=[0.1, 0.9, 0.2, 0.4], payload={
        "text": "Price is $99/month", "doc_id": "pricing",
        "doc_version": "v1", "ingested_at": (datetime.now(timezone.utc) - timedelta(days=20)).isoformat()
    }),
    PointStruct(id=2, vector=[0.1, 0.85, 0.25, 0.4], payload={
        "text": "Price is $129/month", "doc_id": "pricing",
        "doc_version": "v2", "ingested_at": datetime.now(timezone.utc).isoformat()
    }),
])

def retrieve_and_classify(query_vector: list[float], collection: str) -> dict:
    hits = client.search(collection_name=collection, query_vector=query_vector,
                         limit=5, with_payload=True, score_threshold=MIN_SCORE)
    warnings = []
    # TODO: return {"status": "no_results", ...} if hits is empty
    # TODO: detect stale chunks using ingested_at and STALE_DAYS
    # TODO: detect conflicts using doc_id and doc_version
    # TODO: return status dict with "warnings" list populated
    return {"status": "ok", "chunks": [], "warnings": warnings}

result = retrieve_and_classify([0.1, 0.88, 0.22, 0.4], "docs")
print(result)

Quick check

  1. Why is it dangerous to pass low-score retrieval results directly to an LLM without filtering?

  2. A chunk has doc_id 'policy-2024' with version 'v3' and a chunk with the same doc_id with version 'v4'. What should the pipeline do?

  3. What is the primary advantage of doing conflict detection in pipeline code before generation rather than in the LLM prompt?

Self-check: Describe the three pipeline checkpoints you would add to a RAG system to handle no results, conflicting sources, and stale data. For each, explain what metadata you need to store at ingestion time to make the check possible, and what specific action the pipeline should take when the check fails.