| """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) |
|
|
