File size: 1,781 Bytes
e70c280
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
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
"""
File: /local.py
Project: embeddings
Created Date: Sunday February 22nd 2026
Author: Christian Nonis <alch.infoemail@gmail.com>
-----
Last Modified: Sunday February 22nd 2026 5:24:22 pm
Modified By: Christian Nonis <alch.infoemail@gmail.com>
-----
"""

from __future__ import annotations

from typing import TYPE_CHECKING

from src.adapters.interfaces.embeddings import EmbeddingsClient
from src.config import config

if TYPE_CHECKING:
    from sentence_transformers import SentenceTransformer

E5_PREFIX = "passage: "

_model: SentenceTransformer | None = None


def _get_model() -> SentenceTransformer:
    global _model
    if _model is None:
        from sentence_transformers import SentenceTransformer
        _model = SentenceTransformer(config.embeddings.local_model)
    return _model


class EmbeddingError(Exception):
    pass


class LocalEmbeddingsClient(EmbeddingsClient):
    def _prefix(self, text: str) -> str:
        if text.strip().lower().startswith(E5_PREFIX.strip().lower()):
            return text
        return E5_PREFIX + text

    def embed_text(self, text: str) -> list[float]:
        try:
            prefixed = self._prefix(text)
            embedding = _get_model().encode(prefixed, convert_to_numpy=True)
            return embedding.tolist()
        except Exception as e:
            raise EmbeddingError(f"Embedding failed: {e}") from e

    def embed_texts(self, texts: list[str]) -> list[list[float]]:
        try:
            prefixed = [self._prefix(t) for t in texts]
            embeddings = _get_model().encode(prefixed, convert_to_numpy=True)
            return embeddings.tolist()
        except Exception as e:
            raise EmbeddingError(f"Embedding failed: {e}") from e


_embeddings_local_client = LocalEmbeddingsClient()