Skip to content

verify

Run every query against both adapters and score retrieval parity. Raises ValueError if a query_id isn't found in the source.

Source code in vecparity/verify/parity.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
def verify_parity(
    source: VectorDBAdapter,
    target: VectorDBAdapter,
    queries: list[QueryCase],
    min_recall_at_k: float = 0.9,
) -> ParityReport:
    """Run every query against both adapters and score retrieval parity.
    Raises ValueError if a query_id isn't found in the source."""
    results: list[QueryResult] = []
    for q in queries:
        vector = q.query_vector
        if vector is None:
            if q.query_id is None:
                raise ValueError("QueryCase needs either query_vector or query_id")
            record = source.get(q.query_id)
            if record is None:
                raise ValueError(f"query_id {q.query_id!r} not found in source")
            vector = record.vector

        source_hits = source.search(vector, top_k=q.top_k)
        target_hits = target.search(vector, top_k=q.top_k)
        results.append(_compare_one(source_hits, target_hits, q.label))

    return ParityReport(results=results, min_recall_at_k=min_recall_at_k)

Bases: BaseModel

Source code in vecparity/verify/parity.py 
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
class ParityReport(BaseModel):
    results: list[QueryResult]
    min_recall_at_k: float
    """Threshold this report was evaluated against."""

    @property
    def mean_recall_at_k(self) -> float:
        if not self.results:
            return 1.0
        return sum(r.recall_at_k for r in self.results) / len(self.results)

    @property
    def worst_query(self) -> QueryResult | None:
        if not self.results:
            return None
        return min(self.results, key=lambda r: r.recall_at_k)

    @property
    def passed(self) -> bool:
        return self.mean_recall_at_k >= self.min_recall_at_k

    def summary(self) -> str:
        lines = [
            f"Parity report: {len(self.results)} queries, "
            f"mean recall@k = {self.mean_recall_at_k:.3f} "
            f"(threshold {self.min_recall_at_k:.3f}): "
            f"{'PASS' if self.passed else 'FAIL'}"
        ]
        worst = self.worst_query
        if worst is not None and not self.passed:
            lines.append(
                f"  worst query: {worst.label or '(unlabeled)'}: "
                f"recall@k={worst.recall_at_k:.3f}, "
                f"overlap={worst.jaccard_overlap:.3f}, "
                f"score_drift={worst.mean_score_drift:.4f}"
            )
        return "\n".join(lines)

min_recall_at_k instance-attribute

Threshold this report was evaluated against.

Bases: BaseModel

Source code in vecparity/verify/parity.py 
16
17
18
19
20
21
22
class QueryResult(BaseModel):
    label: str | None
    recall_at_k: float
    jaccard_overlap: float
    mean_score_drift: float
    source_top_k: list[str]
    target_top_k: list[str]