Spaces:
Sleeping
Sleeping
| """ | |
| MIT License | |
| Autor: Fernando Lafuente | |
| Proyecto: Ejercicio opcional NLP - Chunking y Recuperación | |
| Descripción: Generación de embeddings con Google y gestión de base de datos vectorial Qdrant | |
| Este módulo implementa la generación de embeddings utilizando Google Embeddings | |
| y la gestión completa de la base de datos vectorial Qdrant. Incluye funciones | |
| para insertar chunks, realizar búsquedas por similitud y administrar colecciones. | |
| La similitud se calcula usando la distancia coseno, que es especialmente efectiva | |
| para embeddings normalizados y representa la similitud semántica entre vectores. | |
| """ | |
| import logging | |
| import time | |
| from typing import List, Tuple, Optional, Dict, Any | |
| import numpy as np | |
| from qdrant_client import QdrantClient | |
| from qdrant_client.http import models | |
| from qdrant_client.http.models import Distance, VectorParams, PointStruct | |
| from openai import OpenAI | |
| import settings | |
| from utils_logging import setup_logging, log_operation_time, format_similarity_score | |
| # Configurar logger específico para este módulo | |
| logger = setup_logging() | |
| class OpenAIEmbeddingsGenerator: | |
| """ | |
| Generador de embeddings utilizando la API de OpenAI. | |
| Esta clase encapsula la lógica para generar embeddings vectoriales usando | |
| los modelos de embeddings de OpenAI. Los embeddings son representaciones | |
| vectoriales de alta dimensionalidad que capturan el significado semántico | |
| del texto de entrada. | |
| Los embeddings de OpenAI text-embedding-3-small tienen 1536 dimensiones | |
| y ofrecen una excelente relación calidad-precio para tareas de similitud | |
| semántica y búsqueda de información. | |
| """ | |
| def __init__(self, api_key: str = settings.OPENAI_API_KEY): | |
| """ | |
| Inicializa el generador con la clave de API de OpenAI. | |
| Args: | |
| api_key: Clave de API de OpenAI | |
| Raises: | |
| ValueError: Si la clave de API no es válida | |
| """ | |
| if not api_key: | |
| raise ValueError("Clave de API de OpenAI no configurada") | |
| # Configurar cliente de OpenAI | |
| self.client = OpenAI(api_key=api_key) | |
| self.model_name = settings.OPENAI_EMBEDDING_MODEL | |
| self.logger = logging.getLogger("chunking_system.embeddings_generator") | |
| self.logger.info(f"Generador de embeddings inicializado con modelo: {self.model_name}") | |
| def generate_embedding(self, text: str) -> np.ndarray: | |
| """ | |
| Genera un vector de embedding para un texto dado. | |
| Este método utiliza la API de OpenAI para convertir texto en un vector | |
| numérico de alta dimensionalidad. El vector resultante captura las | |
| características semánticas del texto y puede ser utilizado para | |
| calcular similitudes con otros textos. | |
| Args: | |
| text: Texto para el cual generar el embedding | |
| Returns: | |
| np.ndarray: Vector de embedding de dimensión 1536 | |
| Raises: | |
| ValueError: Si el texto está vacío | |
| Exception: Si hay errores en la API de OpenAI | |
| Example: | |
| generator = OpenAIEmbeddingsGenerator() | |
| vector = generator.generate_embedding("Este es un texto de ejemplo") | |
| # vector.shape = (1536,) | |
| """ | |
| if not text or not text.strip(): | |
| raise ValueError("El texto no puede estar vacío") | |
| try: | |
| # Llamada a la API de OpenAI para generar embedding | |
| response = self.client.embeddings.create( | |
| model=self.model_name, | |
| input=text | |
| ) | |
| # Extraer el vector de embedding | |
| embedding_vector = np.array(response.data[0].embedding, dtype=np.float32) | |
| # Validar dimensionalidad | |
| if embedding_vector.shape[0] != settings.EMBEDDING_DIMENSION: | |
| raise ValueError(f"Dimensión inesperada del embedding: {embedding_vector.shape[0]}") | |
| self.logger.debug(f"Embedding generado exitosamente para texto de {len(text)} caracteres") | |
| return embedding_vector | |
| except Exception as e: | |
| self.logger.error(f"Error generando embedding: {str(e)}") | |
| raise | |
| def generate_query_embedding(self, query: str) -> np.ndarray: | |
| """ | |
| Genera embedding optimizado para consultas de búsqueda. | |
| Para OpenAI embeddings, utilizamos el mismo método que para documentos | |
| ya que el modelo está optimizado para tareas generales de similitud. | |
| Args: | |
| query: Consulta de búsqueda | |
| Returns: | |
| np.ndarray: Vector de embedding optimizado para consultas | |
| """ | |
| if not query or not query.strip(): | |
| raise ValueError("La consulta no puede estar vacía") | |
| try: | |
| # Para OpenAI, usamos el mismo método para consultas y documentos | |
| response = self.client.embeddings.create( | |
| model=self.model_name, | |
| input=query | |
| ) | |
| embedding_vector = np.array(response.data[0].embedding, dtype=np.float32) | |
| self.logger.debug(f"Embedding de consulta generado para: '{query[:50]}...'") | |
| return embedding_vector | |
| except Exception as e: | |
| self.logger.error(f"Error generando embedding de consulta: {str(e)}") | |
| raise | |
| def generate_batch_embeddings(self, texts: List[str]) -> List[np.ndarray]: | |
| """ | |
| Genera embeddings para múltiples textos de forma eficiente. | |
| Este método procesa múltiples textos secuencialmente, implementando | |
| un manejo robusto de errores y rate limiting para evitar problemas | |
| con la API de Google. | |
| Args: | |
| texts: Lista de textos para procesar | |
| Returns: | |
| List[np.ndarray]: Lista de vectores de embedding | |
| """ | |
| if not texts: | |
| return [] | |
| embeddings = [] | |
| failed_indices = [] | |
| self.logger.info(f"Generando embeddings para {len(texts)} textos") | |
| for i, text in enumerate(texts): | |
| try: | |
| # Pequeña pausa para evitar rate limiting | |
| if i > 0 and i % 10 == 0: | |
| time.sleep(1) | |
| self.logger.debug(f"Pausa preventiva después de procesar {i} textos") | |
| embedding = self.generate_embedding(text) | |
| embeddings.append(embedding) | |
| # Log progreso cada 50 textos | |
| if (i + 1) % 50 == 0: | |
| self.logger.info(f"Progreso: {i + 1}/{len(texts)} embeddings generados") | |
| except Exception as e: | |
| self.logger.warning(f"Error procesando texto {i}: {str(e)}") | |
| failed_indices.append(i) | |
| # Agregar vector cero como fallback | |
| zero_vec = np.zeros(settings.EMBEDDING_DIMENSION, dtype=np.float32) | |
| embeddings.append(zero_vec) | |
| # Log de diagnóstico: mostrar norma del primer embedding y si hay vectores cero | |
| try: | |
| nonzero_norms = [float(np.linalg.norm(v)) for v in embeddings if np.any(v)] | |
| zero_count = len([v for v in embeddings if not np.any(v)]) | |
| if nonzero_norms: | |
| self.logger.info(f"Embeddings generados: muestras - norma primera={nonzero_norms[0]:.4f}, zeros={zero_count}") | |
| else: | |
| self.logger.info(f"Embeddings generados: todos ceros, count={len(embeddings)}") | |
| except Exception: | |
| self.logger.debug("No se pudo calcular norma de embeddings para diagnóstico") | |
| if failed_indices: | |
| self.logger.warning(f"Fallos en {len(failed_indices)} textos: {failed_indices}") | |
| self.logger.info(f"Generación de embeddings completada: {len(embeddings)} vectores") | |
| return embeddings | |
| class QdrantManager: | |
| """ | |
| Gestor para la base de datos vectorial Qdrant. | |
| Esta clase encapsula todas las operaciones con Qdrant, incluyendo | |
| la creación de colecciones, inserción de vectores, búsqueda por | |
| similitud y administración general de la base de datos vectorial. | |
| Qdrant utiliza el algoritmo HNSW (Hierarchical Navigable Small World) | |
| para búsqueda aproximada de vecinos más cercanos, lo que permite | |
| búsquedas eficientes en espacios de alta dimensionalidad. | |
| """ | |
| def __init__(self, | |
| url: str = settings.QDRANT_URL, | |
| collection_name: str = settings.QDRANT_COLLECTION_NAME): | |
| """ | |
| Inicializa la conexión con Qdrant. | |
| Args: | |
| url: URL de conexión a Qdrant | |
| collection_name: Nombre de la colección a utilizar | |
| """ | |
| self.url = url | |
| self.collection_name = collection_name | |
| # Initialize Qdrant client with API key if available and extended timeouts for HF Spaces | |
| if hasattr(settings, 'QDRANT_API_KEY') and settings.QDRANT_API_KEY: | |
| self.client = QdrantClient( | |
| url=url, | |
| api_key=settings.QDRANT_API_KEY, | |
| timeout=60.0, # Extended timeout for HF Spaces | |
| prefer_grpc=False # Use REST API for better compatibility | |
| ) | |
| else: | |
| self.client = QdrantClient( | |
| url=url, | |
| timeout=60.0, # Extended timeout for HF Spaces | |
| prefer_grpc=False # Use REST API for better compatibility | |
| ) | |
| self.logger = logging.getLogger("chunking_system.qdrant_manager") | |
| self.logger.info(f"Conectando a Qdrant en {url}") | |
| # Verificar conexión | |
| try: | |
| health = self.client.get_collections() | |
| self.logger.info("Conexión con Qdrant establecida correctamente") | |
| # Inicializar todas las colecciones necesarias automáticamente | |
| self.initialize_all_collections() | |
| except Exception as e: | |
| self.logger.error(f"Error conectando con Qdrant: {str(e)}") | |
| raise | |
| def create_collection_if_not_exists(self): | |
| """ | |
| Crea la colección en Qdrant si no existe. | |
| La colección se configura para almacenar vectores de 768 dimensiones | |
| (compatibles con Google Embeddings) usando distancia coseno para | |
| calcular similitudes. El índice HNSW se optimiza para búsquedas rápidas. | |
| """ | |
| try: | |
| # Verificar si la colección ya existe | |
| collections = self.client.get_collections().collections | |
| existing_collections = [col.name for col in collections] | |
| if self.collection_name in existing_collections: | |
| self.logger.info(f"Colección '{self.collection_name}' ya existe") | |
| return | |
| # Crear nueva colección | |
| self.client.create_collection( | |
| collection_name=self.collection_name, | |
| vectors_config=VectorParams( | |
| size=settings.EMBEDDING_DIMENSION, | |
| distance=Distance.COSINE # Distancia coseno para similitud semántica | |
| ) | |
| ) | |
| self.logger.info(f"Colección '{self.collection_name}' creada exitosamente") | |
| except Exception as e: | |
| self.logger.error(f"Error creando colección: {str(e)}") | |
| raise | |
| def initialize_all_collections(self): | |
| """ | |
| Inicializa todas las colecciones necesarias para los métodos de chunking. | |
| Crea automáticamente las colecciones para: | |
| - chunks_collection_fixed (Fixed Size Chunks) | |
| - chunks_collection_sentence (Sentence-based Chunks) | |
| - chunks_collection_adaptive (Adaptive Chunks) | |
| """ | |
| required_collections = [ | |
| "chunks_collection_fixed", | |
| "chunks_collection_sentence", | |
| "chunks_collection_adaptive" | |
| ] | |
| try: | |
| # Obtener colecciones existentes | |
| collections = self.client.get_collections().collections | |
| existing_collections = [col.name for col in collections] | |
| # Crear colecciones faltantes | |
| for collection_name in required_collections: | |
| if collection_name not in existing_collections: | |
| self.logger.info(f"Creando colección faltante: {collection_name}") | |
| self.client.create_collection( | |
| collection_name=collection_name, | |
| vectors_config=VectorParams( | |
| size=settings.EMBEDDING_DIMENSION, | |
| distance=Distance.COSINE | |
| ) | |
| ) | |
| self.logger.info(f"✅ Colección '{collection_name}' creada exitosamente") | |
| else: | |
| self.logger.info(f"✓ Colección '{collection_name}' ya existe") | |
| except Exception as e: | |
| self.logger.error(f"❌ Error inicializando colecciones: {str(e)}") | |
| raise | |
| def insert_chunks(self, chunks: List[str], embeddings: List[np.ndarray], chunking_method: Optional[str] = None) -> bool: | |
| """ | |
| Inserta chunks y sus embeddings en la colección de Qdrant. | |
| Este método almacena tanto el texto original de cada chunk como su | |
| representación vectorial, permitiendo recuperar el texto original | |
| después de realizar búsquedas vectoriales. | |
| Args: | |
| chunks: Lista de textos de chunks | |
| embeddings: Lista de vectores de embedding correspondientes | |
| Returns: | |
| bool: True si la inserción fue exitosa | |
| Raises: | |
| ValueError: Si las listas no tienen la misma longitud | |
| Exception: Si hay errores en la inserción | |
| """ | |
| if len(chunks) != len(embeddings): | |
| raise ValueError("El número de chunks debe coincidir con el número de embeddings") | |
| if not chunks: | |
| self.logger.warning("No hay chunks para insertar") | |
| return True | |
| try: | |
| # Crear puntos para insertar en Qdrant | |
| points = [] | |
| # Generate a high, unique integer base id (unsigned 64-bit friendly) | |
| base_id = int(time.time() * 1e9) | |
| for i, (chunk, embedding) in enumerate(zip(chunks, embeddings)): | |
| payload = { | |
| "text": chunk, # Texto original del chunk | |
| "chunk_id": i, # ID del chunk (local al lote) | |
| "length": len(chunk), # Longitud del chunk | |
| "timestamp": time.time() # Timestamp de inserción | |
| } | |
| # Añadir metadata del método de chunking si está disponible | |
| if chunking_method: | |
| payload["chunking_method"] = chunking_method | |
| point = PointStruct( | |
| id=base_id + i, # Qdrant acepta enteros sin signo o UUID; usamos entero único | |
| vector=embedding.tolist(), # Convertir numpy array a lista | |
| payload=payload | |
| ) | |
| points.append(point) | |
| # Insertar todos los puntos en Qdrant | |
| operation_info = self.client.upsert( | |
| collection_name=self.collection_name, | |
| points=points | |
| ) | |
| self.logger.info(f"Insertados {len(chunks)} chunks en Qdrant exitosamente") | |
| self.logger.debug(f"Información de operación: {operation_info}") | |
| # Comprobar estado después de la inserción | |
| try: | |
| info = self.get_collection_info() | |
| self.logger.info(f"Estado colección '{self.collection_name}' tras upsert: puntos={info.get('points_count')}") | |
| except Exception: | |
| self.logger.debug("No fue posible obtener info de colección tras upsert") | |
| return True | |
| except Exception as e: | |
| self.logger.error(f"Error insertando chunks en Qdrant: {str(e)}") | |
| raise | |
| def search_similar_chunks(self, | |
| query_embedding: np.ndarray, | |
| top_k: int = settings.TOP_K_RESULTS, | |
| min_score: float = settings.MIN_SIMILARITY_SCORE) -> List[Tuple[str, float, Dict[str, Any]]]: | |
| """ | |
| Busca chunks similares a un embedding de consulta. | |
| Esta función implementa búsqueda por similitud vectorial usando | |
| distancia coseno. Retorna los chunks más similares junto con sus | |
| puntuaciones de similitud, filtrados por puntuación mínima. | |
| La similitud coseno se calcula como: sim(A,B) = (A·B) / (||A|| × ||B||) | |
| donde A es el vector de consulta y B es cada vector en la base de datos. | |
| Args: | |
| query_embedding: Vector de embedding de la consulta | |
| top_k: Número máximo de resultados a retornar | |
| min_score: Puntuación mínima de similitud para incluir un resultado | |
| Returns: | |
| List[Tuple[str, float]]: Lista de tuplas (texto_chunk, puntuación) | |
| ordenadas por similitud descendente | |
| Example: | |
| results = manager.search_similar_chunks(query_vector, top_k=5) | |
| for text, score in results: | |
| print(f"Similitud: {score:.4f} - {text[:100]}...") | |
| """ | |
| try: | |
| # Realizar búsqueda vectorial en Qdrant | |
| search_result = self.client.search( | |
| collection_name=self.collection_name, | |
| query_vector=query_embedding.tolist(), | |
| limit=top_k, | |
| score_threshold=min_score # Filtro por puntuación mínima | |
| ) | |
| # Extraer resultados y formatear | |
| results = [] | |
| for hit in search_result: | |
| chunk_text = hit.payload.get("text", "") | |
| similarity_score = hit.score | |
| payload = hit.payload or {} | |
| results.append((chunk_text, similarity_score, payload)) | |
| self.logger.info(f"Búsqueda completada: {len(results)} chunks encontrados " | |
| f"(umbral mínimo: {min_score})") | |
| # Log estadísticas de puntuaciones | |
| if results: | |
| scores = [score for _, score, _ in results] | |
| self.logger.debug(f"Puntuaciones - Max: {max(scores):.4f}, " | |
| f"Min: {min(scores):.4f}, " | |
| f"Promedio: {sum(scores)/len(scores):.4f}") | |
| return results | |
| except Exception as e: | |
| self.logger.error(f"Error en búsqueda vectorial: {str(e)}") | |
| raise | |
| def get_collection_info(self) -> Dict[str, Any]: | |
| """ | |
| Obtiene información estadística de la colección. | |
| Returns: | |
| Dict[str, Any]: Información de la colección (número de puntos, etc.) | |
| """ | |
| try: | |
| collection_info = self.client.get_collection(self.collection_name) | |
| info = { | |
| "name": self.collection_name, | |
| "points_count": collection_info.points_count, | |
| "vectors_count": collection_info.vectors_count, | |
| "status": collection_info.status | |
| } | |
| self.logger.debug(f"Información de colección: {info}") | |
| return info | |
| except Exception as e: | |
| self.logger.error(f"Error obteniendo información de colección: {str(e)}") | |
| return {} | |
| def clear_collection(self): | |
| """ | |
| Elimina todos los puntos de la colección. | |
| Útil para testing o para reinicializar la base de datos. | |
| """ | |
| try: | |
| self.client.delete_collection(self.collection_name) | |
| self.logger.info(f"Colección '{self.collection_name}' eliminada") | |
| # NOTA: No recreamos la colección genérica ya que solo usamos colecciones específicas por método | |
| # Las colecciones específicas se crean automáticamente cuando son necesarias | |
| except Exception as e: | |
| self.logger.error(f"Error limpiando colección: {str(e)}") | |
| raise | |
| class EmbeddingsQdrantPipeline: | |
| """ | |
| Pipeline completo para generar embeddings y gestionar Qdrant. | |
| Esta clase orquesta el proceso completo: generación de embeddings para | |
| chunks, inserción en Qdrant y búsqueda de chunks similares a consultas. | |
| Actúa como una fachada que simplifica el uso del sistema completo. | |
| """ | |
| def __init__(self): | |
| """ | |
| Inicializa el pipeline completo con generador de embeddings y Qdrant. | |
| """ | |
| self.embeddings_generator = OpenAIEmbeddingsGenerator() | |
| # Mantener un gestor genérico, pero las colecciones por método se crearán | |
| # dinámicamente cuando procesemos cada método de chunking. | |
| self.qdrant_manager = QdrantManager() | |
| self.logger = logging.getLogger("chunking_system.embeddings_pipeline") | |
| # NOTA: La colección genérica ya no se usa, solo las específicas por método | |
| # self.qdrant_manager.create_collection_if_not_exists() | |
| self.logger.info("Pipeline de embeddings y Qdrant inicializado") | |
| def process_and_store_chunks(self, chunks_by_method: Dict[str, List[str]]) -> bool: | |
| """ | |
| Procesa chunks completos: genera embeddings y los almacena en Qdrant. | |
| IMPORTANTE: Verifica si ya existen embeddings para evitar costos innecesarios de API. | |
| Este método ejecuta el pipeline completo para un conjunto de chunks: | |
| 1. Verifica si ya existen datos en Qdrant | |
| 2. Si no existen, genera embeddings para todos los chunks | |
| 3. Los almacena en Qdrant con sus metadatos | |
| 4. Verifica que la inserción fue exitosa | |
| Args: | |
| chunks: Lista de chunks de texto a procesar | |
| Returns: | |
| bool: True si el procesamiento fue exitoso | |
| """ | |
| if not chunks_by_method: | |
| self.logger.warning("No hay chunks para procesar") | |
| return True | |
| try: | |
| # Procesar cada método de chunking por separado. Cada método utiliza | |
| # su propia colección en Qdrant — esto permite evitar colisiones y | |
| # controlar regeneraciones por método para ahorrar llamadas a OpenAI. | |
| total_inserted = 0 | |
| for method, chunks in chunks_by_method.items(): | |
| if not chunks: | |
| self.logger.info(f"Método '{method}': no hay chunks, se omite") | |
| continue | |
| collection_name = f"{settings.QDRANT_COLLECTION_NAME}_{method}" | |
| manager = QdrantManager(collection_name=collection_name) | |
| manager.create_collection_if_not_exists() | |
| collection_info = manager.get_collection_info() | |
| existing_points = collection_info.get("points_count", 0) | |
| # Si está activado el reemplazo, limpiar colección del método antes de insertar | |
| if settings.REPLACE_COLLECTION_ON_PROCESS and existing_points > 0: | |
| self.logger.info(f"Reemplazo activado: limpiando colección '{collection_name}' con {existing_points} puntos") | |
| try: | |
| manager.clear_collection() | |
| existing_points = 0 | |
| except Exception as e: | |
| self.logger.warning(f"No se pudo limpiar '{collection_name}', se continuará con inserción: {e}") | |
| # Generar embeddings para los chunks de este método | |
| self.logger.info(f"Generando embeddings para método '{method}' ({len(chunks)} chunks)") | |
| embeddings = self.embeddings_generator.generate_batch_embeddings(chunks) | |
| # Insertar en la colección correspondiente incluyendo metadata | |
| self.logger.info(f"Insertando {len(chunks)} chunks en colección '{collection_name}' (existentes previos: {existing_points})") | |
| success = manager.insert_chunks(chunks, embeddings, chunking_method=method) | |
| if not success: | |
| raise RuntimeError(f"Fallo al insertar chunks para el método {method}") | |
| collection_info = manager.get_collection_info() | |
| total_inserted += collection_info.get("points_count", 0) | |
| self.logger.info(f"Procesamiento completado. Total insertados en todas las colecciones: {total_inserted}") | |
| return True | |
| except Exception as e: | |
| self.logger.error(f"Error en pipeline de procesamiento: {str(e)}") | |
| raise | |
| def search_chunks(self, query: str, top_k: int = settings.TOP_K_RESULTS) -> List[Tuple[str, float, str]]: | |
| """ | |
| Busca chunks relevantes para una consulta de texto. | |
| Este método implementa el flujo completo de búsqueda: | |
| 1. Genera embedding para la consulta | |
| 2. Busca chunks similares en Qdrant | |
| 3. Retorna resultados ordenados por relevancia | |
| Args: | |
| query: Consulta de texto del usuario | |
| top_k: Número de resultados a retornar | |
| Returns: | |
| List[Tuple[str, float]]: Chunks relevantes con puntuaciones | |
| Example: | |
| pipeline = EmbeddingsQdrantPipeline() | |
| results = pipeline.search_chunks("¿Qué es el aprendizaje automático?") | |
| for chunk, score in results: | |
| print(f"Relevancia: {score:.4f}") | |
| print(f"Contenido: {chunk}") | |
| """ | |
| if not query or not query.strip(): | |
| raise ValueError("La consulta no puede estar vacía") | |
| try: | |
| # Generar embedding para la consulta | |
| self.logger.info(f"Procesando consulta: '{query[:100]}...'") | |
| query_embedding = self.embeddings_generator.generate_query_embedding(query) | |
| # Buscar en todas las colecciones por método y agregar metadata | |
| aggregated_results: List[Tuple[str, float, str]] = [] | |
| for method in settings.CHUNKING_METHODS: | |
| collection_name = f"{settings.QDRANT_COLLECTION_NAME}_{method}" | |
| try: | |
| manager = QdrantManager(collection_name=collection_name) | |
| except Exception: | |
| # Si no existe la colección o falla la conexión, omitimos | |
| self.logger.debug(f"Colección '{collection_name}' no disponible, se omite") | |
| continue | |
| # Realizar la búsqueda en la colección del método | |
| try: | |
| hits = manager.search_similar_chunks(query_embedding=query_embedding, top_k=top_k) | |
| except Exception as e: | |
| self.logger.warning(f"Error buscando en colección '{collection_name}': {str(e)}") | |
| continue | |
| # Si no hay hits (por filtro de score), reintentar sin umbral para diagnóstico | |
| if not hits: | |
| self.logger.debug(f"No se encontraron hits en '{collection_name}' con umbral {settings.MIN_SIMILARITY_SCORE}. Reintentando sin umbral para diagnóstico") | |
| try: | |
| hits = manager.search_similar_chunks(query_embedding=query_embedding, top_k=top_k, min_score=0.0) | |
| if hits: | |
| self.logger.info(f"Hit(s) encontrados en '{collection_name}' solo al relajar el umbral: {len(hits)}") | |
| except Exception as e: | |
| self.logger.debug(f"Reintento sin umbral falló en '{collection_name}': {str(e)}") | |
| for text, score, payload in hits: | |
| # Preferir el método almacenado en el payload si existe | |
| payload_method = payload.get("chunking_method") if isinstance(payload, dict) else None | |
| origin_method = payload_method or method | |
| aggregated_results.append((text, score, origin_method)) | |
| # Ordenar todos los resultados agregados por puntuación descendente | |
| aggregated_results.sort(key=lambda x: x[1], reverse=True) | |
| # Retornar solo los top_k resultados globales | |
| final_results = aggregated_results[:top_k] | |
| self.logger.info(f"Búsqueda completada: {len(final_results)} chunks relevantes (agregados)") | |
| return final_results | |
| except Exception as e: | |
| self.logger.error(f"Error en búsqueda de chunks: {str(e)}") | |
| raise | |
| def get_embeddings_function(self): | |
| """ | |
| Retorna una función para generar embeddings individuales. | |
| Útil para chunking adaptativo que requiere generar embeddings | |
| de oraciones individuales para calcular similitudes. | |
| Returns: | |
| callable: Función que genera embeddings | |
| """ | |
| return self.embeddings_generator.generate_embedding | |
| def get_system_status(self) -> Dict[str, Any]: | |
| """ | |
| Obtiene el estado general del sistema de embeddings y Qdrant. | |
| Returns: | |
| Dict[str, Any]: Estado del sistema | |
| """ | |
| try: | |
| # Agregar estado agregando información de todas las colecciones por método | |
| total_points = 0 | |
| collections = [] | |
| for method in settings.CHUNKING_METHODS: | |
| collection_name = f"{settings.QDRANT_COLLECTION_NAME}_{method}" | |
| try: | |
| manager = QdrantManager(collection_name=collection_name) | |
| info = manager.get_collection_info() | |
| points = info.get("points_count", 0) | |
| except Exception: | |
| points = 0 | |
| total_points += points | |
| collections.append({"name": collection_name, "points": points}) | |
| status = { | |
| "embeddings_model": self.embeddings_generator.model_name, | |
| "qdrant_url": self.qdrant_manager.url, | |
| "collections": collections, | |
| "collection_name": self.qdrant_manager.collection_name, | |
| "total_chunks": total_points, | |
| "status": "operational" | |
| } | |
| return status | |
| except Exception as e: | |
| return { | |
| "status": "error", | |
| "error": str(e) | |
| } | |
| def clear_embeddings_cache(self): | |
| """ | |
| Limpia todos los embeddings existentes para forzar regeneración. | |
| CUIDADO: Esta operación eliminará todos los embeddings existentes | |
| y requerirá nuevas llamadas a la API de OpenAI (costo adicional). | |
| Returns: | |
| bool: True si se limpió exitosamente | |
| """ | |
| try: | |
| self.logger.warning("LIMPIANDO caché de embeddings. Esto requerirá regenerar todos los embeddings.") | |
| for method in settings.CHUNKING_METHODS: | |
| collection_name = f"{settings.QDRANT_COLLECTION_NAME}_{method}" | |
| try: | |
| manager = QdrantManager(collection_name=collection_name) | |
| manager.clear_collection() | |
| self.logger.info(f"Colección '{collection_name}' limpiada") | |
| except Exception as e: | |
| # Log y continuar con siguientes colecciones | |
| self.logger.warning(f"No se pudo limpiar la colección '{collection_name}': {str(e)}") | |
| self.logger.info("Caché de embeddings limpiado para todas las colecciones de métodos") | |
| return True | |
| except Exception as e: | |
| self.logger.error(f"Error limpiando caché de embeddings: {str(e)}") | |
| return False |