Spaces:
Sleeping
Sleeping
| """ | |
| HuggingFace Embedding Service using Inference API | |
| """ | |
| import logging | |
| import os | |
| from typing import List | |
| import requests | |
| logger = logging.getLogger(__name__) | |
| class HFEmbeddingService: | |
| """HuggingFace Embedding Service using Inference API""" | |
| def __init__(self, model_name: str = "intfloat/multilingual-e5-large"): | |
| self.model_name = model_name | |
| self.api_url = f"https://router.huggingface.co/hf-inference/models/{model_name}" | |
| self.hf_token = os.getenv("HF_TOKEN") | |
| if not self.hf_token: | |
| logger.warning("⚠️ HF_TOKEN not found - service will return empty embeddings") | |
| self.headers = {} | |
| else: | |
| self.headers = { | |
| "Authorization": f"Bearer {self.hf_token}", | |
| "Content-Type": "application/json", | |
| } | |
| logger.info(f"✅ HF Embedding Service initialized: {model_name}") | |
| def get_embeddings(self, texts: List[str]) -> List[List[float]]: | |
| """Get embeddings for texts""" | |
| if not texts: | |
| return [] | |
| # Return empty embeddings if no token (for testing) | |
| if not self.hf_token: | |
| logger.warning("⚠️ No HF_TOKEN - returning empty embeddings") | |
| return [[0.0] * 1024 for _ in texts] | |
| try: | |
| payload = {"inputs": texts} | |
| response = requests.post(self.api_url, headers=self.headers, json=payload, timeout=30) | |
| if response.status_code == 200: | |
| try: | |
| embeddings = response.json() | |
| # Validate that embeddings is a list of lists | |
| if isinstance(embeddings, list) and all(isinstance(emb, list) for emb in embeddings): | |
| logger.debug(f"✅ Generated {len(embeddings)} embeddings") | |
| return embeddings | |
| else: | |
| logger.error(f"❌ Invalid embedding format: {type(embeddings)}") | |
| return [[0.0] * 1024 for _ in texts] | |
| except (ValueError, TypeError) as e: | |
| logger.error(f"❌ JSON decode error: {e}") | |
| return [[0.0] * 1024 for _ in texts] | |
| else: | |
| logger.error(f"❌ HF API error: {response.status_code}") | |
| # Return empty embeddings on error | |
| return [[0.0] * 1024 for _ in texts] | |
| except Exception as e: | |
| logger.error(f"❌ Embedding error: {e}") | |
| # Return empty embeddings on error | |
| return [[0.0] * 1024 for _ in texts] | |
| def get_embedding(self, text: str) -> List[float]: | |
| """Get embedding for single text""" | |
| embeddings = self.get_embeddings([text]) | |
| return embeddings[0] if embeddings else [0.0] * 1024 | |
| def embed_text(self, text: str) -> List[float]: | |
| """Get embedding for single text (SearchService compatibility)""" | |
| return self.get_embedding(text) | |
| def get_embedding_dimension(self) -> int: | |
| """Get embedding dimension""" | |
| return 1024 # Known for multilingual-e5-large | |
| def health_check(self) -> bool: | |
| """Health check""" | |
| try: | |
| if not self.hf_token: | |
| return False | |
| test_embedding = self.get_embedding("test") | |
| return len(test_embedding) == 1024 | |
| except Exception: | |
| return False | |