0bc618763e
Operatiile care nu se declara la RAR (ex. ITP facturat in service) primesc o regula exclus=1 in operations_mapping, setabila din panoul de mapare al preview-ului de import si din tab-ul Mapari (optiunea 'Nu se declara la RAR'). - resolve_prestatii(excluded_ops): item nemapat cu op exclusa -> adnotat exclus, nu mai e needs_mapping; precedenta: cod explicit > exclus > mapare > reguli text - split_prestatii_excluse: itemii exclusi nu intra niciodata in payload/cheie - preview import: rand cu toate operatiile excluse -> stare 'excluded' (eticheta Nedeclarat), necomis; operatia dispare din panoul de mapat - reresolve/corectie/API: submission cu toate operatiile excluse -> needs_data cu motiv explicit; ingestia API trateaza excluderea la clasificare - migrare: coloana operations_mapping.exclus + rebuild import_rows pentru CHECK-ul resolved_status cu 'excluded' (o singura data, gardat pe sqlite_master) - fix flake: clamp similaritate embeddings la [-1,1] (float32 dadea 1.0000001) Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
356 lines
14 KiB
Python
356 lines
14 KiB
Python
"""Modul embedding in-proces pentru sugestie cod RAR.
|
|
|
|
Design:
|
|
- Model multilingv via fastembed/ONNX (~230MB pe disc, quantizat, fara torch)
|
|
- Lazy load la prima folosire, NU la import si NU pe /healthz
|
|
- Worker NU incarca modelul (API-only)
|
|
- Degradare gratioasa: daca modelul nu se incarca -> is_available()=False,
|
|
suggest_nearest() -> [] fara exceptie, ingestia NU e blocata
|
|
- Embeddings = DOAR sugestie (nu intra in lantul de enqueue/resolve_prestatii)
|
|
- NU apelat din resolve_prestatii/load_mapping
|
|
|
|
API public (nivel modul):
|
|
index_corpus(items, signature, vectors) -> None
|
|
suggest_nearest(text, top_k) -> [{cod, is_nul, similaritate}]
|
|
is_available() -> bool
|
|
|
|
Clase (pentru teste / injectare backend):
|
|
EmbeddingEngine(backend) -- motor testabil cu backend injectabil
|
|
FastEmbedBackend() -- backend real fastembed/ONNX
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import math
|
|
import threading
|
|
from typing import Protocol, runtime_checkable
|
|
|
|
try:
|
|
import numpy as _np # garantat de fastembed/onnxruntime; fallback pur-Python daca lipseste
|
|
except ImportError: # pragma: no cover
|
|
_np = None
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
# Modelul ales: paraphrase-multilingual-MiniLM-L12-v2
|
|
# ~230MB pe disc (ONNX quantizat), 384 dim, multilingv (ro/en/etc.), suportat de
|
|
# fastembed, fara torch.
|
|
FASTEMBED_MODEL = "sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2"
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Protocol backend (mockabil in teste) #
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
@runtime_checkable
|
|
class EmbeddingBackend(Protocol):
|
|
"""Interfata minimala pentru un backend de embedding."""
|
|
|
|
def embed(self, texts: list[str]) -> list[list[float]]:
|
|
"""Intoarce o lista de vectori (cate unul per text)."""
|
|
...
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Backend real: fastembed/ONNX #
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
class FastEmbedBackend:
|
|
"""Backend fastembed/ONNX. Lazy-load la constructie.
|
|
|
|
Arunca ImportError daca fastembed nu e instalat, sau orice exceptie
|
|
de la TextEmbedding (download esuat, ONNX incompatibil etc.).
|
|
Apelantul (_load_engine) prinde aceste exceptii.
|
|
"""
|
|
|
|
def __init__(self, model_name: str = FASTEMBED_MODEL):
|
|
from fastembed import TextEmbedding # import tardiv -- nu blocheaza la import modul
|
|
self._model = TextEmbedding(model_name=model_name)
|
|
|
|
def embed(self, texts: list[str]) -> list[list[float]]:
|
|
# fastembed.embed() intoarce un generator de numpy arrays
|
|
return [vec.tolist() for vec in self._model.embed(texts)]
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Motor de embedding (testabil, backend injectabil) #
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
def _cosine_similarity(a: list[float], b: list[float]) -> float:
|
|
"""Similaritate cosine intre doi vectori. Returneaza 0.0 pe vectori nuli."""
|
|
dot = sum(x * y for x, y in zip(a, b))
|
|
na = math.sqrt(sum(x * x for x in a))
|
|
nb = math.sqrt(sum(x * x for x in b))
|
|
if na == 0.0 or nb == 0.0:
|
|
return 0.0
|
|
return dot / (na * nb)
|
|
|
|
|
|
class EmbeddingEngine:
|
|
"""Motor de embedding cu corpus indexat si cautare NN cosine.
|
|
|
|
Parametri:
|
|
backend: instanta EmbeddingBackend (real sau mock).
|
|
None => degradare gratioasa (is_available=False).
|
|
"""
|
|
|
|
def __init__(self, backend: EmbeddingBackend | None = None):
|
|
self._backend = backend
|
|
self._corpus_vecs: list[list[float]] = []
|
|
self._corpus_items: list[dict] = []
|
|
self._corpus_sig: str | None = None
|
|
# Matrice numpy + norme precalculate: cautarea NN devine un matvec
|
|
# (~ms) in loc de cosine pur-Python per item (~0.5s la 17k vectori).
|
|
self._corpus_matrix = None
|
|
self._corpus_norms = None
|
|
|
|
def is_available(self) -> bool:
|
|
"""True daca backend-ul e disponibil si gata de folosire."""
|
|
return self._backend is not None
|
|
|
|
def has_corpus(self) -> bool:
|
|
"""True daca un corpus a fost indexat (suggest_nearest poate produce ceva)."""
|
|
return bool(self._corpus_items)
|
|
|
|
def corpus_signature(self) -> str | None:
|
|
"""Semnatura corpusului indexat (None daca gol). Apelantul re-indexeaza
|
|
doar cand semnatura nomenclatorului s-a schimbat (evita re-embed inutil)."""
|
|
return self._corpus_sig
|
|
|
|
def index_corpus(
|
|
self,
|
|
items: list[dict],
|
|
signature: str | None = None,
|
|
vectors: list | None = None,
|
|
) -> None:
|
|
"""Indexeaza corpus [{denumire, cod}] si il pastreaza in memorie.
|
|
|
|
`vectors`: vectori precalculati, aliniati POZITIONAL cu `items` (ex. din
|
|
embedding_cache). `None` (default) = comportamentul existent, embed complet
|
|
prin backend. Daca `vectors` e furnizat dar lungimea nu corespunde cu `items`
|
|
sau contine `None`, se ignora (log.warning) si se cade pe embed complet (A8) --
|
|
o dezaliniere silentioasa ar produce coduri sugerate GRESITE.
|
|
|
|
Ignora silentios daca backend-ul lipseste, corpus-ul e gol sau apare
|
|
orice exceptie la vectorizare (degradare gratioasa).
|
|
"""
|
|
self._corpus_vecs = []
|
|
self._corpus_items = []
|
|
self._corpus_sig = None
|
|
self._corpus_matrix = None
|
|
self._corpus_norms = None
|
|
|
|
if not items or not self.is_available():
|
|
return
|
|
|
|
if vectors is not None and (len(vectors) != len(items) or any(v is None for v in vectors)):
|
|
log.warning(
|
|
"embeddings: index_corpus vectors (%d) nealiniat cu items (%d) sau contine None -- fallback embed complet",
|
|
len(vectors), len(items),
|
|
)
|
|
vectors = None
|
|
|
|
try:
|
|
if vectors is not None:
|
|
self._corpus_vecs = list(vectors)
|
|
else:
|
|
texts = [str(item["denumire"]) for item in items]
|
|
self._corpus_vecs = self._backend.embed(texts)
|
|
self._corpus_items = list(items)
|
|
self._corpus_sig = signature
|
|
self._build_matrix()
|
|
except Exception as exc:
|
|
log.warning("embeddings: index_corpus esuat: %s", exc)
|
|
# corpus ramane gol -- suggest_nearest va returna []
|
|
|
|
def _build_matrix(self) -> None:
|
|
"""Precalculeaza matricea numpy + normele corpusului pentru cautarea NN.
|
|
|
|
Esecul (numpy lipsa, vectori neregulati) lasa matricea None --
|
|
suggest_nearest cade pe scanarea pur-Python, corpusul ramane valid.
|
|
"""
|
|
if _np is None:
|
|
return
|
|
try:
|
|
matrix = _np.asarray(self._corpus_vecs, dtype=_np.float32)
|
|
if matrix.ndim != 2:
|
|
raise ValueError(f"vectori corpus neregulati (ndim={matrix.ndim})")
|
|
norms = _np.linalg.norm(matrix, axis=1)
|
|
norms[norms == 0.0] = 1.0 # vector nul -> similaritate 0, fara div/0
|
|
self._corpus_matrix = matrix
|
|
self._corpus_norms = norms
|
|
except Exception as exc:
|
|
log.warning("embeddings: matrice numpy esuata, fallback pur-Python: %s", exc)
|
|
self._corpus_matrix = None
|
|
self._corpus_norms = None
|
|
|
|
def embed(self, texts: list[str]) -> list[list[float]]:
|
|
"""Vectorizeaza texte brute prin backend (folosit la miss-uri de cache).
|
|
|
|
Arunca daca backend-ul lipseste sau embed() esueaza -- apelantul
|
|
(sync_corpus_vectors) propaga eroarea, fara degradare gratioasa aici.
|
|
"""
|
|
if not self.is_available():
|
|
raise RuntimeError("embeddings: backend indisponibil")
|
|
return self._backend.embed(texts)
|
|
|
|
def suggest_nearest(
|
|
self,
|
|
denumire: str,
|
|
top_k: int = 3,
|
|
) -> list[dict]:
|
|
"""Returneaza top_k vecini cosine [{cod, is_nul, similaritate}].
|
|
|
|
`is_nul`: cand corpusul include exemple NUL (non-operatii),
|
|
un vecin NUL = semnal de SUPRESIE, nu cod. Default False pe corpusuri vechi
|
|
fara `is_nul` in itemi. Returneaza [] daca backend-ul lipseste, corpus-ul e gol
|
|
sau apare orice exceptie (degradare gratioasa -- nu blocheaza ingestia).
|
|
"""
|
|
if not self.is_available() or not self._corpus_items:
|
|
return []
|
|
|
|
try:
|
|
query_vecs = self._backend.embed([str(denumire)])
|
|
query_vec = query_vecs[0]
|
|
if self._corpus_matrix is not None:
|
|
q = _np.asarray(query_vec, dtype=_np.float32)
|
|
qn = float(_np.linalg.norm(q)) or 1.0
|
|
sims = (self._corpus_matrix @ q) / (self._corpus_norms * qn)
|
|
k = min(top_k, sims.shape[0])
|
|
if k <= 0:
|
|
return []
|
|
# argpartition + sort doar pe top_k: O(n) in loc de O(n log n)
|
|
idx = _np.argpartition(-sims, k - 1)[:k]
|
|
idx = idx[_np.argsort(-sims[idx])]
|
|
return [
|
|
{
|
|
"cod": self._corpus_items[i]["cod"],
|
|
"is_nul": bool(self._corpus_items[i].get("is_nul", False)),
|
|
# clamp la [-1, 1]: eroarea de rotunjire float32 poate da
|
|
# 1.0000001 pe text identic (ar afisa >100% in UI).
|
|
"similaritate": min(1.0, max(-1.0, float(sims[i]))),
|
|
}
|
|
for i in idx
|
|
]
|
|
scored = [
|
|
{
|
|
"cod": item["cod"],
|
|
"is_nul": bool(item.get("is_nul", False)),
|
|
"similaritate": _cosine_similarity(query_vec, vec),
|
|
}
|
|
for item, vec in zip(self._corpus_items, self._corpus_vecs)
|
|
]
|
|
scored.sort(key=lambda r: r["similaritate"], reverse=True)
|
|
return scored[:top_k]
|
|
except Exception as exc:
|
|
log.warning("embeddings: suggest_nearest esuat: %s", exc)
|
|
return []
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Singleton global cu lazy load (API-only, NU worker) #
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
_engine: EmbeddingEngine | None = None
|
|
_engine_lock = threading.Lock()
|
|
|
|
|
|
def _load_engine() -> EmbeddingEngine:
|
|
"""Lazy load: construieste engine-ul la prima folosire.
|
|
|
|
Captureaza ORICE exceptie la incarcare (import, download, ONNX init)
|
|
si returneaza un engine degradat (backend=None) -- ingestia continua
|
|
pe exact+fuzzy, embedding = sugestie dezactivata.
|
|
"""
|
|
try:
|
|
backend = FastEmbedBackend()
|
|
log.info("embeddings: backend fastembed incarcat (%s)", FASTEMBED_MODEL)
|
|
return EmbeddingEngine(backend=backend)
|
|
except ImportError:
|
|
log.warning(
|
|
"embeddings: fastembed nu e instalat -- sugestii NN dezactivate"
|
|
)
|
|
except Exception as exc:
|
|
log.warning(
|
|
"embeddings: incarcare backend esuata (%s) -- sugestii NN dezactivate",
|
|
exc,
|
|
)
|
|
return EmbeddingEngine(backend=None)
|
|
|
|
|
|
def _get_engine() -> EmbeddingEngine:
|
|
"""Returneaza engine-ul global (lazy-init, thread-safe).
|
|
|
|
Lock-ul previne incarcarea dubla a modelului cand warmup-ul de la startup
|
|
si un request concurent ajung aici simultan.
|
|
"""
|
|
global _engine
|
|
if _engine is None:
|
|
with _engine_lock:
|
|
if _engine is None:
|
|
_engine = _load_engine()
|
|
return _engine
|
|
|
|
|
|
def is_loaded() -> bool:
|
|
"""True daca engine-ul global a fost deja construit. NU forteaza incarcarea."""
|
|
return _engine is not None
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# API public la nivel de modul (wiring L14-S6) #
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
def is_available() -> bool:
|
|
"""True daca modelul e incarcat si gata de folosire."""
|
|
return _get_engine().is_available()
|
|
|
|
|
|
def has_corpus() -> bool:
|
|
"""True daca un corpus a fost indexat in motorul global.
|
|
|
|
NU forteaza incarcarea modelului: daca engine-ul nu a fost initializat inca
|
|
(`_engine is None`), corpus-ul e gol prin definitie -> False, fara cost.
|
|
Apelantii (ex. enrich_suggestions) folosesc asta ca poarta ieftina inainte de
|
|
a atinge calea scumpa (is_available/suggest_nearest, care lazy-load ~230MB).
|
|
"""
|
|
if _engine is None:
|
|
return False
|
|
return _engine.has_corpus()
|
|
|
|
|
|
def corpus_signature() -> str | None:
|
|
"""Semnatura corpusului global indexat (None daca engine ne-initializat/gol).
|
|
|
|
NU forteaza incarcarea modelului: `_engine is None` -> None fara cost.
|
|
"""
|
|
if _engine is None:
|
|
return None
|
|
return _engine.corpus_signature()
|
|
|
|
|
|
def embed_texts(texts: list[str]) -> list[list[float]]:
|
|
"""Vectorizeaza texte brute prin motorul global (folosit la miss-uri de cache).
|
|
|
|
Arunca daca engine-ul e indisponibil -- apelantul (embedding_cache.sync_corpus_vectors)
|
|
propaga eroarea catre ensure_embeddings_corpus (degradare gratioasa acolo).
|
|
"""
|
|
return _get_engine().embed(texts)
|
|
|
|
|
|
def index_corpus(items: list[dict], signature: str | None = None, vectors: list | None = None) -> None:
|
|
"""Vectorizeaza corpus [{denumire, cod}] in motorul global.
|
|
|
|
`vectors`: vezi EmbeddingEngine.index_corpus (precalculati, aliniati cu `items`).
|
|
Silentios pe eroare (degradare gratioasa).
|
|
"""
|
|
_get_engine().index_corpus(items, signature=signature, vectors=vectors)
|
|
|
|
|
|
def suggest_nearest(denumire: str, top_k: int = 3) -> list[dict]:
|
|
"""Returneaza top_k sugestii [{cod, is_nul, similaritate}] sau [] la eroare.
|
|
|
|
Sigur de apelat indiferent de starea backend-ului.
|
|
"""
|
|
return _get_engine().suggest_nearest(denumire, top_k=top_k)
|