"""A content-bound embedding interface audit, not an embedding quality certificate.""" from dataclasses import asdict, dataclass, fields import hashlib import json import math import sys def text(value, *, allow_empty=False): if type(value) is not str or (not allow_empty and not value.strip()) or len(value) > 4096: raise ValueError("bounded built-in string required") return value def sha256_hex(value): if type(value) is not str or len(value) != 64 or any(c not in "0123456789abcdef" for c in value): raise ValueError("lowercase SHA-256 required") return value def integer(value, lower, upper): if type(value) is not int or not lower <= value <= upper: raise ValueError("bounded integer, not bool, required") return value def number(value, lower, upper): if type(value) not in (int, float) or not math.isfinite(value): raise ValueError("finite real, not bool, required") if value != 0 and abs(value) < sys.float_info.min: raise ValueError("subnormal rejected") if not lower <= value <= upper: raise ValueError("number out of range") return float(value) def digest(value): return hashlib.sha256( json.dumps(value, sort_keys=True, separators=(",", ":"), allow_nan=False).encode() ).hexdigest() @dataclass(frozen=True) class EmbeddingContract: contract_id: str task: str model: str model_revision: str preprocessing_revision: str tokenizer_revision: str pooling: str metric: str normalization: str dimension: int max_input_tokens: int query_prefix: str document_prefix: str owner: str rollback: str def __post_init__(self): for name in ( "contract_id", "task", "model", "model_revision", "preprocessing_revision", "tokenizer_revision", "pooling", "metric", "normalization", "query_prefix", "document_prefix", "owner", "rollback", ): text(getattr(self, name)) if self.pooling not in ("mean", "cls", "last-token"): raise ValueError("unsupported pooling") if self.metric not in ("cosine", "inner-product", "squared-l2"): raise ValueError("unsupported metric") if self.normalization not in ("l2", "none"): raise ValueError("unsupported normalization") if self.metric == "cosine" and self.normalization != "l2": raise ValueError("teaching cosine contract requires l2 normalization") integer(self.dimension, 1, 65_536) integer(self.max_input_tokens, 1, 1_000_000) def contract_digest(contract): if type(contract) is not EmbeddingContract: raise ValueError("concrete EmbeddingContract required") checked = EmbeddingContract(**{field.name: getattr(contract, field.name) for field in fields(EmbeddingContract)}) return digest(asdict(checked)) @dataclass(frozen=True) class EmbeddingRecord: embedding_id: str role: str source_id: str source_revision: str input_digest: str contract_digest: str vector: tuple[float, ...] def __post_init__(self): for name in ("embedding_id", "role", "source_id", "source_revision"): text(getattr(self, name)) if self.role not in ("query", "document"): raise ValueError("role must be query or document") sha256_hex(self.input_digest) sha256_hex(self.contract_digest) if type(self.vector) not in (tuple, list) or not 1 <= len(self.vector) <= 65_536: raise ValueError("bounded vector required") object.__setattr__(self, "vector", tuple(number(value, -1_000_000, 1_000_000) for value in self.vector)) @dataclass(frozen=True) class RankedItem: embedding_id: str score: float @dataclass(frozen=True) class EmbeddingAudit: contract_digest: str dimension: int metric: str ranked: tuple[RankedItem, ...] claim: str def _rehydrate_record(record): if type(record) is not EmbeddingRecord: raise ValueError("concrete EmbeddingRecord required") return EmbeddingRecord(**{field.name: getattr(record, field.name) for field in fields(EmbeddingRecord)}) def _score(metric, query, document): if metric in ("cosine", "inner-product"): return sum(a * b for a, b in zip(query, document)) return -sum((a - b) ** 2 for a, b in zip(query, document)) def audit(contract, query, documents): if type(contract) is not EmbeddingContract: raise ValueError("concrete EmbeddingContract required") contract = EmbeddingContract(**{field.name: getattr(contract, field.name) for field in fields(EmbeddingContract)}) identity = digest(asdict(contract)) query = _rehydrate_record(query) if type(documents) not in (tuple, list) or not 1 <= len(documents) <= 10_000: raise ValueError("one through ten thousand documents required") documents = tuple(_rehydrate_record(item) for item in documents) if query.role != "query" or any(item.role != "document" for item in documents): raise ValueError("one query and document candidates required") all_records = (query,) + documents if len({item.embedding_id for item in all_records}) != len(all_records): raise ValueError("duplicate embedding identity") if any(item.contract_digest != identity for item in all_records): raise ValueError("embedding contract binding mismatch") if any(len(item.vector) != contract.dimension for item in all_records): raise ValueError("embedding dimension mismatch") if contract.normalization == "l2": for item in all_records: norm = math.sqrt(sum(value * value for value in item.vector)) if not math.isclose(norm, 1.0, rel_tol=0.0, abs_tol=1e-6): raise ValueError("l2-normalized vector required") ranked = tuple( RankedItem(item.embedding_id, _score(contract.metric, query.vector, item.vector)) for item in documents ) ranked = tuple(sorted(ranked, key=lambda item: (-item.score, item.embedding_id))) return EmbeddingAudit(identity, contract.dimension, contract.metric, ranked, "STRUCTURAL_COMPATIBILITY_ONLY") def example_contract(): return EmbeddingContract( contract_id="support-search-v1", task="illustrative support retrieval", model="example-dual-encoder", model_revision="sha-example-001", preprocessing_revision="unicode-nfc-v1", tokenizer_revision="tokenizer-example-001", pooling="mean", metric="cosine", normalization="l2", dimension=3, max_input_tokens=128, query_prefix="query: ", document_prefix="document: ", owner="example-search-team", rollback="restore support-search-v0", ) def example_records(contract=None): contract = example_contract() if contract is None else contract identity = contract_digest(contract) query = EmbeddingRecord( "query-refund", "query", "query-1", "1", digest({"text": "query: refund policy"}), identity, (1.0, 0.0, 0.0) ) documents = ( EmbeddingRecord("doc-policy", "document", "policy", "7", digest({"text": "document: returns and refunds"}), identity, (0.8, 0.6, 0.0)), EmbeddingRecord("doc-hours", "document", "hours", "3", digest({"text": "document: store opening hours"}), identity, (0.0, 1.0, 0.0)), ) return query, documents def main(): contract = example_contract() query, documents = example_records(contract) result = audit(contract, query, documents) print("example=illustrative_only") print("contract=VALID") print(f"dimension={result.dimension}") print("metric=" + result.metric) print("top_item=" + result.ranked[0].embedding_id) print("claim=" + result.claim) if __name__ == "__main__": main()