Spaces:
Sleeping
Sleeping
| """Stateless inference. This service does not read or write the demo corpus.""" | |
| import math | |
| from api.schemas.archive import Embedding, EmbeddingsResponse, TranslationResponse | |
| class ProviderUnavailable(Exception): | |
| pass | |
| class ArchiveInferenceService: | |
| def __init__(self, embedder, translator): | |
| self.embedder = embedder | |
| self.translator = translator | |
| def embed(self, texts, requested_model=None): | |
| model_id = getattr(self.embedder, "compatibility_id", self.embedder.model_name) | |
| if requested_model not in (None, self.embedder.model_name, model_id): | |
| raise ValueError("Requested model is not configured") | |
| output = self.embedder.encode_documents(texts, batch_size=1, max_length=512) | |
| dense_rows = output["dense_vecs"] | |
| sparse_rows = output["lexical_weights"] | |
| if len(dense_rows) != len(texts) or len(sparse_rows) != len(texts): | |
| raise RuntimeError("Invalid embedding result count") | |
| results = [] | |
| for dense, sparse in zip(dense_rows, sparse_rows): | |
| vector = [float(value) for value in dense] | |
| if not vector or any(not math.isfinite(value) for value in vector): | |
| raise RuntimeError("Invalid embedding result") | |
| lexical = {} | |
| # BGEM3FlagModel returns vocabulary IDs, not token strings. | |
| for token_id, value in sparse.items(): | |
| index = int(token_id) | |
| weight = float(value) | |
| if index < 0 or not math.isfinite(weight) or weight < 0: | |
| raise RuntimeError("Invalid sparse embedding") | |
| lexical[str(index)] = max(lexical.get(str(index), 0), weight) | |
| results.append(Embedding(dense=vector, sparse=lexical)) | |
| dimension = len(results[0].dense) | |
| if any(len(item.dense) != dimension for item in results): | |
| raise RuntimeError("Inconsistent embedding dimensions") | |
| return EmbeddingsResponse(model=model_id, | |
| dimension=dimension, embeddings=results) | |
| def translate(self, texts, source_locale, target_locale): | |
| if source_locale == target_locale: | |
| return TranslationResponse(model="identity", translations=list(texts)) | |
| model, translations = self.translator.translate(texts, source_locale, target_locale) | |
| if len(translations) != len(texts) or any(not isinstance(text, str) or not text.strip() | |
| or len(text) > 24000 for text in translations): | |
| raise RuntimeError("Invalid translation response") | |
| return TranslationResponse(model=model, translations=translations) | |