Embedder backends and embedding cache.
SentenceTransformerEmbedder is intentionally not imported here so that importing
nlql.embed never triggers the heavy optional dependency; import it explicitly from
nlql.embed.sentence_transformers when needed.
module-attribute
__all__ = ['Embedder', 'BaseEmbedder', 'normalize_rows', 'FakeEmbedder', 'OpenAIEmbedder', 'EmbeddingCache', 'CachedEmbedder', 'cache_key', 'MultimodalEmbedder', 'FakeMultimodalEmbedder', 'DoubaoVisionEmbedder', 'supports_images']
Bases: ABC
Base class handling the empty-input case and row normalization.
abstractmethod
property
abstractmethod
property
embed(texts: Sequence[str]) -> ndarray
源代码位于： src/nlql/embed/base.py
| def embed(self, texts: Sequence[str]) -> np.ndarray:
items = list(texts)
if not items:
return np.empty((0, self.dim), dtype=np.float32)
return normalize_rows(self._embed_raw(items))
|
Bases: Protocol
Structural type for anything that turns texts into vectors.
property
Stable identifier of the model/config; part of the cache key.
property
Output vector dimensionality.
embed(texts: Sequence[str]) -> ndarray
Return an (n, dim) float32 matrix of unit-normalized row vectors.
源代码位于： src/nlql/embed/base.py
| def embed(self, texts: Sequence[str]) -> np.ndarray:
"""Return an ``(n, dim)`` float32 matrix of unit-normalized row vectors."""
...
|
Wraps an embedder so repeated texts are embedded at most once.
源代码位于： src/nlql/embed/cache.py
| def __init__(
self,
inner: Embedder,
cache: EmbeddingCache | None = None,
*,
modality: str = "text",
) -> None:
self._inner = inner
self._cache = cache if cache is not None else EmbeddingCache()
self._modality = modality
|
property
The wrapped embedder (used to detect multimodal support).
embed(texts: Sequence[str]) -> ndarray
源代码位于： src/nlql/embed/cache.py
| def embed(self, texts: Sequence[str]) -> np.ndarray:
return self._cached(list(texts), self._key, self._inner.embed) # type: ignore[arg-type]
|
embed_images(images: Sequence[bytes | str]) -> ndarray
源代码位于： src/nlql/embed/cache.py
| def embed_images(self, images: Sequence[bytes | str]) -> np.ndarray:
compute = getattr(self._inner, "embed_images", None)
if not callable(compute):
raise NLQLEmbeddingError(f"{self._inner.model_id} cannot embed images")
return self._cached(list(images), self._image_key, compute) # type: ignore[arg-type]
|
An in-memory embedding cache with optional on-disk persistence.
源代码位于： src/nlql/embed/cache.py
| def __init__(self) -> None:
self._store: dict[str, np.ndarray] = {}
|
get(key: str) -> ndarray | None
源代码位于： src/nlql/embed/cache.py
| def get(self, key: str) -> np.ndarray | None:
return self._store.get(key)
|
set(key: str, vector: ndarray) -> None
源代码位于： src/nlql/embed/cache.py
| def set(self, key: str, vector: np.ndarray) -> None:
self._store[key] = np.asarray(vector, dtype=np.float32)
|
源代码位于： src/nlql/embed/cache.py
| def __len__(self) -> int:
return len(self._store)
|
__contains__(key: str) -> bool
源代码位于： src/nlql/embed/cache.py
| def __contains__(self, key: str) -> bool:
return key in self._store
|
save(path: str | Path) -> None
Persist to a .npz file (keys + stacked vectors).
源代码位于： src/nlql/embed/cache.py
| def save(self, path: str | Path) -> None:
"""Persist to a ``.npz`` file (keys + stacked vectors)."""
keys = list(self._store)
matrix = (
np.stack([self._store[k] for k in keys])
if keys
else np.empty((0, 0), dtype=np.float32)
)
np.savez(path, keys=np.array(keys, dtype=object), vectors=matrix)
|
load(path: str | Path) -> None
Load entries from a .npz file, merging into the current cache.
源代码位于： src/nlql/embed/cache.py
| def load(self, path: str | Path) -> None:
"""Load entries from a ``.npz`` file, merging into the current cache."""
data = np.load(path, allow_pickle=True)
keys = data["keys"]
vectors = data["vectors"]
for i, key in enumerate(keys):
self._store[str(key)] = np.asarray(vectors[i], dtype=np.float32)
|
DoubaoVisionEmbedder(*, api_key: str | None = None, base_url: str = 'https://ark.cn-beijing.volces.com/api/v3', model: str = 'doubao-embedding-vision', dim: int = 2048, timeout: float = 60.0, client: Client | None = None)
Bases: BaseEmbedder
Multimodal embedder backed by Volcengine Ark's doubao-embedding-vision.
源代码位于： src/nlql/embed/doubao.py
| def __init__(
self,
*,
api_key: str | None = None,
base_url: str = "https://ark.cn-beijing.volces.com/api/v3",
model: str = "doubao-embedding-vision",
dim: int = 2048,
timeout: float = 60.0,
client: httpx.Client | None = None,
) -> None:
self._model = model
self._dim = dim
if client is not None:
self._client = client
else:
key = api_key or os.environ.get("NLQL_ARK_API_KEY") or os.environ.get("ARK_API_KEY")
if not key:
raise NLQLEmbeddingError(
"DoubaoVisionEmbedder needs an api_key (or NLQL_ARK_API_KEY / ARK_API_KEY env var)"
)
self._client = httpx.Client(
base_url=base_url.rstrip("/"),
headers={"Authorization": f"Bearer {key}"},
timeout=timeout,
)
|
embed_images(images: Sequence[bytes | str]) -> ndarray
Embed images (bytes, file/URL string, or data URI) into the shared text space.
源代码位于： src/nlql/embed/doubao.py
| def embed_images(self, images: Sequence[bytes | str]) -> np.ndarray:
"""Embed images (bytes, file/URL string, or data URI) into the shared text space."""
items = list(images)
if not items:
return np.empty((0, self._dim), dtype=np.float32)
vectors = np.stack(
[
self._embed_one([{"type": "image_url", "image_url": {"url": _data_uri(img)}}])
for img in items
]
)
return normalize_rows(vectors)
|
源代码位于： src/nlql/embed/doubao.py
| def close(self) -> None:
self._client.close()
|
FakeEmbedder(dim: int = 64, model_id: str = 'fake')
Bases: BaseEmbedder
A hash-based embedder producing stable vectors without any model.
源代码位于： src/nlql/embed/fake.py
| def __init__(self, dim: int = 64, model_id: str = "fake") -> None:
self._dim = dim
self._model_id = f"{model_id}:{dim}"
|
FakeMultimodalEmbedder(dim: int = 64)
Bases: FakeEmbedder
Deterministic multimodal fake: an image's bytes are read as a caption and embedded in
the same bag-of-tokens space as text.
源代码位于： src/nlql/embed/multimodal.py
| def __init__(self, dim: int = 64) -> None:
super().__init__(dim=dim, model_id="fake-mm")
|
embed_images(images: Sequence[bytes | str]) -> ndarray
源代码位于： src/nlql/embed/multimodal.py
| def embed_images(self, images: Sequence[bytes | str]) -> np.ndarray:
captions = [
img.decode("utf-8", "ignore") if isinstance(img, (bytes, bytearray)) else str(img)
for img in images
]
return self.embed(captions)
|
Bases: Embedder, Protocol
An embedder that can also embed images into the same space as text.
embed_images(images: Sequence[bytes | str]) -> ndarray
Return an (n, dim) matrix of unit-normalized image vectors.
源代码位于： src/nlql/embed/multimodal.py
| def embed_images(self, images: Sequence[bytes | str]) -> np.ndarray:
"""Return an ``(n, dim)`` matrix of unit-normalized image vectors."""
...
|
OpenAIEmbedder(model: str = 'text-embedding-3-small', *, api_key: str | None = None, base_url: str = 'https://api.openai.com/v1', dimensions: int | None = None, timeout: float = 30.0, client: Client | None = None)
Bases: BaseEmbedder
Embedder backed by an OpenAI-compatible embeddings endpoint.
源代码位于： src/nlql/embed/openai.py
| def __init__(
self,
model: str = "text-embedding-3-small",
*,
api_key: str | None = None,
base_url: str = "https://api.openai.com/v1",
dimensions: int | None = None,
timeout: float = 30.0,
client: httpx.Client | None = None,
) -> None:
self._model = model
self._dimensions = dimensions
self._dim = dimensions or _DEFAULT_DIMS.get(model, 1536)
if client is not None:
self._client = client
else:
key = api_key or os.environ.get("NLQL_OPENAI_API_KEY") or os.environ.get("OPENAI_API_KEY")
if not key:
raise NLQLEmbeddingError(
"OpenAIEmbedder needs an api_key (or NLQL_OPENAI_API_KEY / OPENAI_API_KEY env var)"
)
self._client = httpx.Client(
base_url=base_url.rstrip("/"),
headers={"Authorization": f"Bearer {key}"},
timeout=timeout,
)
|
源代码位于： src/nlql/embed/openai.py
| def close(self) -> None:
self._client.close()
|
normalize_rows(matrix: ndarray) -> ndarray
Unit-normalize each row; zero rows are left as zeros.
源代码位于： src/nlql/embed/base.py
| def normalize_rows(matrix: np.ndarray) -> np.ndarray:
"""Unit-normalize each row; zero rows are left as zeros."""
m = np.asarray(matrix, dtype=np.float32)
if m.ndim == 1:
m = m.reshape(1, -1)
norms = np.linalg.norm(m, axis=1, keepdims=True)
norms[norms == 0.0] = 1.0
return np.asarray(m / norms, dtype=np.float32)
|
cache_key(model_id: str, dim: int, modality: str, text: str) -> str
Stable content-addressed key for one embedding.
源代码位于： src/nlql/embed/cache.py
| def cache_key(model_id: str, dim: int, modality: str, text: str) -> str:
"""Stable content-addressed key for one embedding."""
payload = f"{model_id}\x00{dim}\x00{modality}\x00{text}".encode()
return hashlib.sha256(payload).hexdigest()
|
supports_images(embedder: Embedder) -> bool
Whether an embedder (unwrapping a cache wrapper) can embed images.
源代码位于： src/nlql/embed/multimodal.py
| def supports_images(embedder: Embedder) -> bool:
"""Whether an embedder (unwrapping a cache wrapper) can embed images."""
inner = getattr(embedder, "inner", embedder)
return callable(getattr(inner, "embed_images", None))
|