Build completeness checks for embedding coverage and knowledge graph gaps

Introduction

Engineers often discover retrieval gaps only after users report missing results — by then, documents that were never embedded have been silently excluded from every RAG query for hours or days. Completeness checks catch those gaps proactively by comparing your source catalog against your vector store and knowledge graph. By the end of this lesson, you'll be able to build async completeness checkers that measure embedding coverage and surface knowledge graph gaps through structured reports and Prometheus metrics, so your pipeline can alert before users ever notice a problem.

Key Terminology

• Embedding coverage — the fraction of active source documents that have a corresponding entry in the vector store, computed as the size of the intersection of source_ids and embedded_ids divided by the total number of active source document IDs; the primary health signal produced by EmbeddingCoverageChecker.check_coverage.
• CheckStatus — a three-valued enum (PASS, FAIL, WARN) that classifies a completeness check outcome with enough granularity to distinguish hard failures from degraded-but-acceptable states; stored on every CompletenessCheck instance and compared against a caller-supplied threshold.
• CompletenessCheck — the Pydantic model that records a single store's validation result, including coverage_ratio (bounded 0.0–1.0), expected_count, actual_count, and up to 100 missing_items identifiers so on-call engineers can identify which document IDs need re-ingestion without querying the database directly.
• CompletenessReport — the aggregate Pydantic model that collects all per-store CompletenessCheck results into a single overall_completeness score and an all_passed boolean, giving downstream alerting logic a single field to query rather than iterating every check.
• Set-difference gap detection — the algorithmic pattern used in check_coverage to find unembedded documents: missing = source_ids - embedded_ids, where source_ids is the set of active document IDs from the source catalog and embedded_ids is the set of IDs present in the embeddings table; the cardinality of this difference drives both the coverage ratio and the missing_items list.
• DATA_COMPLETENESS gauge — a Prometheus Gauge metric labeled by store and check that EmbeddingCoverageChecker updates on every check_coverage call, enabling the monitoring stack to fire alerts when coverage falls below the configured threshold without requiring a human to inspect database tables.

Concepts

The Silent Exclusion Problem

A RAG pipeline has no built-in mechanism to tell users when a document was never embedded. From the user's perspective, a missing document looks identical to a document that simply does not match their query — both produce a gap in results, and both are invisible at query time. This means a pipeline that stops ingesting new documents, or that drops a batch of documents during an embedding run, can silently degrade retrieval quality for hours or days before anyone notices.

Completeness checks invert the detection direction: instead of waiting for user-reported gaps, you continuously compare the authoritative source catalog to the contents of the vector store and surface the discrepancy as a structured signal. The source catalog is the ground truth for what should be embedded; the vector store is the record of what is embedded. Any document present in the source but absent from the store is a silent exclusion that this lesson teaches you to expose proactively.

Set-Difference Coverage Logic

The core algorithm in check_coverage treats both the source catalog and the embedding store as sets of document IDs, then computes their difference. source_ids is fetched from source_documents WHERE active = true — only documents currently considered live are checked, so soft-deleted records do not inflate the gap count. embedded_ids is fetched from the embeddings table with a DISTINCT to collapse multiple embedding rows per document. The set difference source_ids - embedded_ids yields exactly the document IDs that are live in the catalog but absent from the store.

The coverage ratio is then len(embedded_ids & source_ids) / len(source_ids) — the intersection (documents that are both live and embedded) divided by the total live count. Using the intersection rather than the raw embedded_ids count ensures that orphaned embeddings (embeddings for documents that have since been deactivated) do not inflate the numerator and mask real gaps. The missing_items list is capped at 100 via Pydantic's max_length=100 field constraint, keeping the report payload bounded even when an entire ingestion batch fails (see Code Walkthrough).

Threshold-Based Status and Prometheus Integration

A raw coverage ratio becomes actionable only when compared to an expected bar. check_coverage accepts a threshold parameter (defaulting to 0.95) and maps the ratio to a CheckStatus: at or above the threshold the check is PASS; below it the check is FAIL. The WARN value in the enum is available for intermediate logic — for example, a second threshold below the primary one — without requiring callers to restructure the model.

Publishing the ratio to the DATA_COMPLETENESS Prometheus gauge closes the alerting loop. Because the gauge carries store and check labels, a single metrics endpoint can expose coverage for pgvector, a knowledge graph store, and any other store you add later, and a single Prometheus alerting rule can fire on any of them by matching data_completeness_ratio < 0.95. This means the monitoring stack reacts to coverage drops in near-real time rather than waiting for a scheduled report to be manually reviewed. The CompletenessReport.all_passed boolean serves the complementary role inside application code — a health-check endpoint or an integration test can query it directly without parsing metric labels.

Code Walkthrough

Now that you understand how CompletenessCheck and CompletenessReport model validation results, you can implement the checkers that populate them.

The first block defines the Pydantic data models used throughout both checkers:

Code snippetpython
1from datetime import datetime
2from pydantic import BaseModel, Field
3from enum import Enum
4
5class CheckStatus(str, Enum):
6    PASS = "pass"
7    FAIL = "fail"
8    WARN = "warn"
9
10class CompletenessCheck(BaseModel):
11    check_name: str
12    store: str
13    status: CheckStatus
14    coverage_ratio: float = Field(ge=0.0, le=1.0)
15    expected_count: int
16    actual_count: int
17    missing_items: list[str] = Field(
18        default_factory=list, max_length=100
19    )
20    checked_at: datetime = Field(default_factory=datetime.utcnow)
21
22class CompletenessReport(BaseModel):
23    checks: list[CompletenessCheck]
24    overall_completeness: float
25    all_passed: bool
26    generated_at: datetime = Field(default_factory=datetime.utcnow)

CheckStatus classifies outcomes as PASS, FAIL, or WARN with enough granularity to distinguish hard failures from degraded-but-acceptable states. CompletenessCheck captures the coverage ratio (0.0–1.0), the gap between expected and actual counts, and up to 100 missing item identifiers for debugging. The max_length=100 guard keeps the report payload bounded even for severely incomplete stores. CompletenessReport aggregates all per-store checks into a single overall score plus an all_passed boolean that downstream alerting logic can query directly.

The second block implements EmbeddingCoverageChecker, which queries the source document catalog and pgvector to compute actual coverage and expose it to Prometheus:

Code snippetpython
1import asyncpg
2from prometheus_client import Gauge
3
4DATA_COMPLETENESS = Gauge(
5    "data_completeness_ratio",
6    "Data completeness ratio per check",
7    ["store", "check"],
8)
9
10class EmbeddingCoverageChecker:
11    def __init__(
12        self,
13        source_pool: asyncpg.Pool,
14        vector_pool: asyncpg.Pool,
15    ):
16        self._source = source_pool
17        self._vector = vector_pool
18
19    async def check_coverage(
20        self,
21        threshold: float = 0.95,
22    ) -> CompletenessCheck:
23        async with self._source.acquire() as src:
24            source_docs = await src.fetch(
25                "SELECT document_id FROM source_documents WHERE active = true"
26            )
27        source_ids = {str(r["document_id"]) for r in source_docs}
28
29        async with self._vector.acquire() as vec:
30            embedded_docs = await vec.fetch(
31                "SELECT DISTINCT document_id FROM embeddings"
32            )
33        embedded_ids = {str(r["document_id"]) for r in embedded_docs}
34
35        missing = source_ids - embedded_ids
36        coverage = (
37            len(embedded_ids & source_ids) / len(source_ids)
38            if source_ids else 1.0
39        )
40        status = CheckStatus.PASS if coverage >= threshold else CheckStatus.FAIL
41
42        DATA_COMPLETENESS.labels(
43            store="pgvector",
44            check="embedding_coverage",
45        ).set(coverage)
46
47        return CompletenessCheck(
48            check_name="embedding_coverage",
49            store="pgvector",
50            status=status,
51            coverage_ratio=coverage,
52            expected_count=len(source_ids),
53            actual_count=len(embedded_ids & source_ids),
54            missing_items=list(missing)[:100],
55        )

EmbeddingCoverageChecker holds two asyncpg.Pool connections — one for the source document catalog and one for the pgvector embedding store. check_coverage computes the set difference between active source document IDs and embedded document IDs, then publishes the ratio to the DATA_COMPLETENESS Prometheus gauge so your monitoring stack can fire an alert when coverage drops below threshold. The returned CompletenessCheck carries missing_items so on-call engineers know exactly which document IDs need re-ingestion without querying the database manually.

You'll know it works when check_coverage returns a CompletenessCheck with status=CheckStatus.PASS against a fully embedded store and status=CheckStatus.FAIL with a non-empty missing_items list when you deliberately omit one document ID from the embeddings table.

Do's and Don'ts

Having walked through the material above, the following Do's and Don'ts distill it into practice.

Do's

1. ✓Do cap missing_items at 100 entries in CompletenessCheck — the max_length=100 Pydantic guard keeps report payloads bounded even when your vector store is severely incomplete; without it, a large gap (thousands of missing document IDs) can bloat the response and overwhelm downstream alerting consumers.
2. ✓Do pass separate asyncpg.Pool instances for the source catalog and pgvector store to EmbeddingCoverageChecker — keeping the pools independent lets each connection target the correct database and isolates failures, so a pgvector connection error doesn't also bring down source document queries.
3. ✓Do publish the computed coverage ratio to the DATA_COMPLETENESS Prometheus gauge with both store and check labels — the label pair ("pgvector", "embedding_coverage") is what lets your alerting rules target embedding coverage specifically, rather than firing on every completeness check across all stores.

Don'ts

1. ✗Don't compute coverage by counting rows without taking the intersection (embedded_ids & source_ids) — documents in the embeddings table that have no matching active source record inflate the apparent coverage ratio, masking real gaps; actual_count must reflect only source-matched embeddings, not the raw embedded row count.
2. ✗Don't hard-code the threshold parameter inside check_coverage rather than accepting it as an argument — embedding coverage requirements differ across stores and pipelines; a fixed threshold prevents callers from tuning the PASS/FAIL boundary without modifying the checker class itself.
3. ✗Don't rely on all_passed in CompletenessReport as your only alert signal without also reading per-check coverage_ratio values — all_passed flips to False on a single FAIL, but the Prometheus DATA_COMPLETENESS gauge is what gives your monitoring stack the continuous signal needed to page on gradual coverage decay before it crosses the failure threshold.