Spaces:
Paused
Paused
File size: 2,183 Bytes
6b3874e | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 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 62 63 64 65 | """Embedding layer with an honest, documented tradeoff.
Preferred path: chromadb's bundled embedder (all-MiniLM-L6-v2 via ONNX) when
chromadb is installed — real semantic embeddings.
Fallback path (always available): a deterministic hashed TF vector embedder.
Tradeoff: the hashing embedder is purely LEXICAL — it captures word overlap
(with sublinear term-frequency weighting and stable feature hashing) but has
zero semantic generalization ("pedestrian" and "person on foot" do not
match). For this package's seed corpus, which uses consistent engineering
vocabulary, lexical retrieval is adequate and — critically — fully
deterministic and dependency-free, which the test suite relies on.
"""
from __future__ import annotations
import hashlib
import math
import re
from typing import List
import numpy as np
_TOKEN_RE = re.compile(r"[a-z0-9]+")
# Common words that would otherwise dominate hashed-TF cosine similarity.
_STOPWORDS = frozenset(
"a an and are as at be by for from has have in is it its of on or that the "
"this to was were will with within must shall should not no".split())
class HashedTfEmbedder:
"""Deterministic hashed term-frequency embedder (numpy only)."""
name = "hashed-tf-fallback"
def __init__(self, dim: int = 1024):
self.dim = dim
def _bucket(self, token: str) -> int:
return int.from_bytes(hashlib.md5(token.encode()).digest()[:4], "little") % self.dim
def embed(self, texts: List[str]) -> np.ndarray:
out = np.zeros((len(texts), self.dim), dtype=np.float64)
for i, text in enumerate(texts):
for tok in _TOKEN_RE.findall(text.lower()):
if tok in _STOPWORDS:
continue
out[i, self._bucket(tok)] += 1.0
# sublinear tf then L2 normalize
nz = out[i] > 0
out[i, nz] = 1.0 + np.log(out[i, nz])
norm = math.sqrt(float(np.dot(out[i], out[i])))
if norm > 0:
out[i] /= norm
return out
def chromadb_available() -> bool:
try:
import chromadb # noqa: F401
return True
except ImportError:
return False
|