"""Text embedding interface for SFX/music matching. The production target is a small sentence-embedding model (e.g. bge-small, 33M). To keep the fixture pipeline dependency-free and offline, the default `HashingEmbedder` produces stable, comparable vectors with no model download. Both satisfy the `Embedder` protocol, so the real model drops in later. """ from __future__ import annotations import hashlib import re from typing import Protocol import numpy as np _TOKEN_RE = re.compile(r"[a-z0-9]+") class Embedder(Protocol): """Anything that turns text into a fixed-length, comparable vector.""" def embed(self, text: str) -> np.ndarray: ... class HashingEmbedder: """Deterministic bag-of-words hashing embedder, L2-normalized. Shared tokens raise cosine similarity, so semantically similar prompts land near each other without any learned model. Hashing uses md5 (not Python's salted `hash`) so vectors are stable across processes and runs. """ def __init__(self, dim: int = 128) -> None: self.dim = dim def embed(self, text: str) -> np.ndarray: vector = np.zeros(self.dim, dtype=np.float32) for token in _TOKEN_RE.findall(text.lower()): digest = hashlib.md5(token.encode("utf-8")).digest() index = int.from_bytes(digest[:4], "big") % self.dim sign = 1.0 if digest[4] & 1 else -1.0 vector[index] += sign return _l2_normalize(vector) def _l2_normalize(vector: np.ndarray) -> np.ndarray: norm = float(np.linalg.norm(vector)) if norm == 0.0: return vector return (vector / norm).astype(np.float32) def cosine_similarity(a: np.ndarray, b: np.ndarray) -> float: """Cosine similarity of two vectors (0.0 when either is empty/zero).""" if a.size == 0 or b.size == 0: return 0.0 denom = float(np.linalg.norm(a) * np.linalg.norm(b)) if denom == 0.0: return 0.0 return float(np.dot(a, b) / denom)