Embedding质量评估与版本管理：给向量空间做“图书质检”

import numpy as npfrom collections import Counter  class EmbeddingQualityEvaluator:    def __init__(self, dim=128):        self.dim = dim        self.version = "v0.1"        self.vector_store = {}     def add_vectors(self, ids, vectors):        for vid, vec in zip(ids, vectors):            self.vector_store[vid] = np.asarray(vec, dtype=np.float32)     def compute_recall_at_k(self, query_vec, relevant_ids, k=10):        dists = []        for vid, vec in self.vector_store.items():            d = np.linalg.norm(query_vec - vec)            dists.append((d, vid))        dists.sort(key=lambda x: x[0])        retrieved = set(vid for _, vid in dists[:k])        hits = retrieved & set(relevant_ids)        return len(hits) / len(relevant_ids)     def compute_precision_at_k(self, query_vec, relevant_ids, k=10):        dists = []        for vid, vec in self.vector_store.items():            d = np.linalg.norm(query_vec - vec)            dists.append((d, vid))        dists.sort(key=lambda x: x[0])        retrieved = set(vid for _, vid in dists[:k])        hits = retrieved & set(relevant_ids)        return len(hits) / k     def compute_distribution_uniformity(self):        if len(self.vector_store) < 2:            return 0.0        arr = np.array(list(self.vector_store.values()))        dists = []        for i in range(min(200, len(arr))):            diff = arr - arr[i]            norms = np.linalg.norm(diff, axis=1)            nn = norms[norms > 0]            if len(nn) > 0:                dists.append(nn.min())        dists = np.array(dists)        if dists.mean() == 0:            return 1.0        return 1.0 - (dists.std() / dists.mean())     def check_semantic_consistency(self, query_vec, anchor_vec, threshold=0.7):        cos_sim = np.dot(query_vec, anchor_vec) / (            np.linalg.norm(query_vec) * np.linalg.norm(anchor_vec)        )        return float(cos_sim) >= threshold, float(cos_sim)     def report(self, query_vec, relevant_ids, anchor_vec=None):        r_at_k = self.compute_recall_at_k(query_vec, relevant_ids)        p_at_k = self.compute_precision_at_k(query_vec, relevant_ids)        u = self.compute_distribution_uniformity()        print("=== Embedding Quality Report ===")        print("Version:", self.version)        print("Recall@10: %.3f" % r_at_k)        print("Precision@10: %.3f" % p_at_k)        print("Uniformity: %.3f" % u)        if anchor_vec is not None:            ok, sim = self.check_semantic_consistency(query_vec, anchor_vec)            print("Semantic consistency: %.3f (%s)" % (sim, "PASS" if ok else "FAIL"))  def main():    np.random.seed(42)    evaluator = EmbeddingQualityEvaluator(dim=16)    ids = ["doc_%d" % i for i in range(50)]    vectors = [np.random.randn(16) for _ in range(50)]    evaluator.add_vectors(ids, vectors)    query = np.random.randn(16)    relevant = ["doc_1", "doc_2", "doc_3"]    evaluator.report(query, relevant, anchor_vec=vectors[1])  if __name__ == "__main__":    main()