Spaces:
Running
Running
| """ | |
| embeddings.py — Vectorización de textos via HuggingFace Inference API. | |
| Convierte textos en vectores numéricos (embeddings) usando el modelo | |
| paraphrase-multilingual-MiniLM-L12-v2 corriendo en la nube de HuggingFace. | |
| No requiere GPU ni modelos locales — todo corre remotamente. | |
| Manejo de errores: | |
| - 503 (modelo cargando): espera HF_RETRY_DELAY segundos, reintenta hasta 3 veces | |
| - 429 (rate limit): espera HF_RATELIMIT_DELAY segundos, reintenta | |
| - Otros errores: loguea y devuelve lista vacía para ese batch | |
| """ | |
| import logging | |
| import time | |
| from typing import Any | |
| import requests | |
| import sys | |
| from pathlib import Path | |
| sys.path.insert(0, str(Path(__file__).parent.parent)) | |
| from config import ( | |
| HUGGINGFACE_API_KEY, | |
| HF_API_URL_EMBEDDINGS, | |
| HF_BATCH_SIZE, | |
| HF_RETRY_DELAY, | |
| HF_RATELIMIT_DELAY, | |
| SCRAPING_MAX_RETRIES, | |
| LOG_FORMAT, | |
| LOG_DATE_FORMAT, | |
| LOG_LEVEL, | |
| ) | |
| # ── Logging ─────────────────────────────────────────────────────────────────── | |
| logging.basicConfig(format=LOG_FORMAT, datefmt=LOG_DATE_FORMAT) | |
| logger = logging.getLogger("embeddings") | |
| logger.setLevel(getattr(logging, LOG_LEVEL, logging.INFO)) | |
| def _headers() -> dict: | |
| """Construye los headers HTTP para la API de HuggingFace.""" | |
| headers = {"Content-Type": "application/json"} | |
| if HUGGINGFACE_API_KEY: | |
| headers["Authorization"] = f"Bearer {HUGGINGFACE_API_KEY}" | |
| return headers | |
| def _fetch_batch(textos: list[str]) -> list[list[float]]: | |
| """ | |
| Envía un batch de textos a HuggingFace y devuelve sus embeddings. | |
| Maneja reintentos automáticos para errores 503 y 429. | |
| Devuelve lista vacía si todos los intentos fallan. | |
| """ | |
| payload = {"inputs": textos, "options": {"wait_for_model": True}} | |
| for intento in range(1, SCRAPING_MAX_RETRIES + 1): | |
| try: | |
| t0 = time.time() | |
| resp = requests.post( | |
| HF_API_URL_EMBEDDINGS, | |
| headers=_headers(), | |
| json=payload, | |
| timeout=60, | |
| ) | |
| elapsed = time.time() - t0 | |
| if resp.status_code == 200: | |
| resultado = resp.json() | |
| logger.info( | |
| f"[embeddings] batch {len(textos)} textos — " | |
| f"{elapsed:.1f}s — dim={len(resultado[0]) if resultado else 0}" | |
| ) | |
| return resultado | |
| elif resp.status_code == 503: | |
| logger.warning( | |
| f"[embeddings] 503 modelo cargando (intento {intento}/" | |
| f"{SCRAPING_MAX_RETRIES}) — esperando {HF_RETRY_DELAY}s" | |
| ) | |
| time.sleep(HF_RETRY_DELAY) | |
| elif resp.status_code == 429: | |
| logger.warning( | |
| f"[embeddings] 429 rate limit — esperando {HF_RATELIMIT_DELAY}s" | |
| ) | |
| time.sleep(HF_RATELIMIT_DELAY) | |
| else: | |
| logger.error( | |
| f"[embeddings] error {resp.status_code}: {resp.text[:200]}" | |
| ) | |
| return [] | |
| except requests.exceptions.Timeout: | |
| logger.warning(f"[embeddings] timeout en intento {intento}/{SCRAPING_MAX_RETRIES}") | |
| if intento < SCRAPING_MAX_RETRIES: | |
| time.sleep(HF_RETRY_DELAY) | |
| except Exception as e: | |
| logger.error(f"[embeddings] error inesperado: {e}") | |
| return [] | |
| logger.error(f"[embeddings] todos los reintentos fallaron para batch de {len(textos)} textos") | |
| return [] | |
| def get_embeddings(texts: list[str]) -> list[list[float]]: | |
| """ | |
| Vectoriza una lista de textos usando HuggingFace Inference API. | |
| Procesa en batches de HF_BATCH_SIZE para respetar los límites del free tier. | |
| Los textos vacíos se saltan y se devuelve lista vacía en su posición. | |
| Parámetros: | |
| texts: lista de strings a vectorizar | |
| Devuelve: | |
| Lista de embeddings (lista de floats). Mismo orden que la entrada. | |
| Las posiciones con error devuelven lista vacía []. | |
| """ | |
| if not texts: | |
| return [] | |
| todos_los_embeddings: list[list[float]] = [] | |
| for i in range(0, len(texts), HF_BATCH_SIZE): | |
| batch = texts[i: i + HF_BATCH_SIZE] | |
| # Filtrar textos vacíos, preservar posiciones | |
| indices_validos = [j for j, t in enumerate(batch) if t and t.strip()] | |
| textos_validos = [batch[j] for j in indices_validos] | |
| if not textos_validos: | |
| todos_los_embeddings.extend([[] for _ in batch]) | |
| continue | |
| embeddings_batch = _fetch_batch(textos_validos) | |
| # Validar que la API devolvió el mismo número de embeddings que textos | |
| if embeddings_batch and len(embeddings_batch) != len(textos_validos): | |
| logger.warning( | |
| f"[embeddings] mismatch: enviados {len(textos_validos)} textos, " | |
| f"recibidos {len(embeddings_batch)} embeddings — se completan con []" | |
| ) | |
| # Completar o recortar para que el zip funcione correctamente | |
| if len(embeddings_batch) < len(textos_validos): | |
| embeddings_batch.extend([] for _ in range(len(textos_validos) - len(embeddings_batch))) | |
| else: | |
| embeddings_batch = embeddings_batch[:len(textos_validos)] | |
| # Reconstruir con posiciones originales | |
| resultado_batch: list[list[float]] = [[] for _ in batch] | |
| for pos, emb in zip(indices_validos, embeddings_batch): | |
| resultado_batch[pos] = emb | |
| todos_los_embeddings.extend(resultado_batch) | |
| logger.info( | |
| f"[embeddings] completado — {len(texts)} textos, " | |
| f"{sum(1 for e in todos_los_embeddings if e)} con embedding válido" | |
| ) | |
| return todos_los_embeddings | |