Spaces:
Sleeping
Sleeping
File size: 2,675 Bytes
f16b727 | 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 | """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)
|