Spaces:
Sleeping
Sleeping
| """ | |
| MIT License | |
| Autor: Fernando Lafuente | |
| Proyecto: Ejercicio opcional NLP - Chunking y Recuperación | |
| Descripción: Implementación de múltiples métodos de chunking para segmentación textual | |
| Este módulo implementa tres estrategias diferentes de chunking (segmentación textual): | |
| 1. Chunking fijo: División por tamaño fijo con solapamiento | |
| 2. Chunking por oraciones: División en oraciones naturales agrupadas | |
| 3. Chunking adaptativo: División basada en similitud semántica entre oraciones | |
| Cada método tiene ventajas específicas según el tipo de documento y la aplicación. | |
| El chunking adaptativo utiliza embeddings para agrupar oraciones semánticamente | |
| relacionadas, lo que puede mejorar la coherencia de los chunks resultantes. | |
| """ | |
| import logging | |
| import re | |
| from typing import List, Tuple | |
| from abc import ABC, abstractmethod | |
| import numpy as np | |
| from sklearn.metrics.pairwise import cosine_similarity | |
| import settings | |
| from utils_logging import setup_logging, log_operation_time, validate_text_length | |
| # Configurar logger específico para este módulo | |
| logger = setup_logging() | |
| class ChunkingStrategy(ABC): | |
| """ | |
| Clase abstracta base para todas las estrategias de chunking. | |
| Define la interfaz común que deben implementar todos los métodos de chunking. | |
| Esto permite intercambiar estrategias fácilmente y mantener código limpio | |
| siguiendo el patrón Strategy de diseño. | |
| """ | |
| def create_chunks(self, text: str) -> List[str]: | |
| """ | |
| Método abstracto para crear chunks desde un texto. | |
| Args: | |
| text: Texto a segmentar | |
| Returns: | |
| List[str]: Lista de chunks generados | |
| """ | |
| pass | |
| def get_strategy_name(self) -> str: | |
| """ | |
| Método abstracto para obtener el nombre de la estrategia. | |
| Returns: | |
| str: Nombre identificativo de la estrategia | |
| """ | |
| pass | |
| class FixedChunking(ChunkingStrategy): | |
| """ | |
| Implementa chunking por tamaño fijo con solapamiento. | |
| Este método divide el texto en chunks de tamaño fijo (en caracteres), | |
| con un solapamiento configurable entre chunks consecutivos. Es el método | |
| más simple y predecible, útil cuando se requiere uniformidad en el tamaño. | |
| Ventajas: | |
| - Predecible y rápido | |
| - Chunks de tamaño uniforme | |
| - Fácil de implementar y debuggear | |
| Desventajas: | |
| - Puede cortar oraciones o conceptos a la mitad | |
| - No considera la estructura semántica del texto | |
| """ | |
| def __init__(self, | |
| chunk_size: int = settings.FIXED_CHUNK_SIZE, | |
| overlap: int = settings.FIXED_CHUNK_OVERLAP): | |
| """ | |
| Inicializa el chunking fijo con parámetros configurables. | |
| Args: | |
| chunk_size: Tamaño en caracteres de cada chunk | |
| overlap: Número de caracteres de solapamiento entre chunks | |
| """ | |
| self.chunk_size = chunk_size | |
| self.overlap = overlap | |
| self.logger = logging.getLogger("chunking_system.fixed_chunking") | |
| # Validar parámetros | |
| if self.overlap >= self.chunk_size: | |
| raise ValueError("El solapamiento debe ser menor que el tamaño del chunk") | |
| def create_chunks(self, text: str) -> List[str]: | |
| """ | |
| Crea chunks de tamaño fijo con solapamiento del texto de entrada. | |
| El algoritmo avanza por el texto con pasos de (chunk_size - overlap), | |
| tomando ventanas de tamaño chunk_size. Esto asegura que la información | |
| en los límites de chunks se preserve en múltiples chunks. | |
| Args: | |
| text: Texto a segmentar | |
| Returns: | |
| List[str]: Lista de chunks de tamaño aproximadamente uniforme | |
| """ | |
| if not text or not validate_text_length(text): | |
| self.logger.warning("Texto vacío o demasiado corto para chunking") | |
| return [] | |
| chunks = [] | |
| start = 0 | |
| step = self.chunk_size - self.overlap | |
| while start < len(text): | |
| end = min(start + self.chunk_size, len(text)) | |
| chunk = text[start:end].strip() | |
| # Solo agregar chunks que tengan contenido válido | |
| if validate_text_length(chunk): | |
| chunks.append(chunk) | |
| self.logger.debug(f"Chunk creado: posición {start}-{end}, " | |
| f"longitud {len(chunk)} caracteres") | |
| # Salir si hemos llegado al final | |
| if end == len(text): | |
| break | |
| start += step | |
| self.logger.info(f"Chunking fijo completado: {len(chunks)} chunks generados") | |
| return chunks | |
| def get_strategy_name(self) -> str: | |
| return f"fixed_chunking_{self.chunk_size}_{self.overlap}" | |
| class SentenceChunking(ChunkingStrategy): | |
| """ | |
| Implementa chunking basado en oraciones y párrafos naturales. | |
| Este método respeta la estructura natural del texto, dividiendo en oraciones | |
| y agrupándolas hasta alcanzar un tamaño objetivo. Preserva la coherencia | |
| semántica mejor que el chunking fijo al no cortar oraciones. | |
| Ventajas: | |
| - Respeta límites naturales de oraciones | |
| - Mantiene coherencia semántica local | |
| - Más legible para humanos | |
| Desventajas: | |
| - Chunks de tamaño variable | |
| - Puede generar chunks muy largos o muy cortos | |
| - Depende de la calidad de detección de oraciones | |
| """ | |
| def __init__(self, | |
| min_length: int = settings.SENTENCE_MIN_LENGTH, | |
| max_chunk_size: int = settings.SENTENCE_MAX_CHUNK_SIZE): | |
| """ | |
| Inicializa el chunking por oraciones con parámetros configurables. | |
| Args: | |
| min_length: Longitud mínima para considerar una oración válida | |
| max_chunk_size: Tamaño máximo permitido para un chunk | |
| """ | |
| self.min_length = min_length | |
| self.max_chunk_size = max_chunk_size | |
| self.logger = logging.getLogger("chunking_system.sentence_chunking") | |
| # Patrón regex para detectar finales de oración | |
| # Busca punto, signo de interrogación o exclamación seguido de espacio o final de línea | |
| self.sentence_pattern = re.compile(r'[.!?]+\s+|[.!?]+$') | |
| def create_chunks(self, text: str) -> List[str]: | |
| """ | |
| Crea chunks agrupando oraciones hasta alcanzar el tamaño objetivo. | |
| El algoritmo identifica oraciones usando expresiones regulares, | |
| las valida y las agrupa secuencialmente hasta alcanzar el tamaño | |
| máximo permitido. Cada chunk contiene oraciones completas. | |
| Args: | |
| text: Texto a segmentar | |
| Returns: | |
| List[str]: Lista de chunks basados en oraciones | |
| """ | |
| if not text or not validate_text_length(text): | |
| self.logger.warning("Texto vacío o demasiado corto para chunking") | |
| return [] | |
| # Dividir texto en oraciones | |
| sentences = self._split_sentences(text) | |
| if not sentences: | |
| self.logger.warning("No se pudieron extraer oraciones válidas") | |
| return [] | |
| # Agrupar oraciones en chunks | |
| chunks = self._group_sentences_into_chunks(sentences) | |
| self.logger.info(f"Chunking por oraciones completado: {len(chunks)} chunks " | |
| f"generados desde {len(sentences)} oraciones") | |
| return chunks | |
| def _split_sentences(self, text: str) -> List[str]: | |
| """ | |
| Divide el texto en oraciones individuales usando regex. | |
| Este método utiliza patrones regex para identificar finales de oración | |
| y divide el texto en consecuencia. Aplica validaciones para asegurar | |
| que las oraciones resultantes sean significativas. | |
| Args: | |
| text: Texto a dividir en oraciones | |
| Returns: | |
| List[str]: Lista de oraciones válidas | |
| """ | |
| # Dividir por patrones de final de oración | |
| potential_sentences = self.sentence_pattern.split(text) | |
| sentences = [] | |
| for sentence in potential_sentences: | |
| # Limpiar y validar cada oración | |
| clean_sentence = sentence.strip() | |
| if validate_text_length(clean_sentence, self.min_length): | |
| sentences.append(clean_sentence) | |
| self.logger.debug(f"Oración válida: {len(clean_sentence)} caracteres") | |
| return sentences | |
| def _group_sentences_into_chunks(self, sentences: List[str]) -> List[str]: | |
| """ | |
| Agrupa oraciones secuencialmente en chunks respetando tamaño máximo. | |
| El algoritmo va agregando oraciones a un chunk actual hasta que | |
| agregar la siguiente oración excedería el tamaño máximo. En ese | |
| punto, finaliza el chunk actual e inicia uno nuevo. | |
| Args: | |
| sentences: Lista de oraciones a agrupar | |
| Returns: | |
| List[str]: Lista de chunks formados por oraciones agrupadas | |
| """ | |
| chunks = [] | |
| current_chunk = [] | |
| current_length = 0 | |
| for sentence in sentences: | |
| sentence_length = len(sentence) | |
| # Verificar si agregar esta oración excedería el tamaño máximo | |
| if (current_length + sentence_length > self.max_chunk_size and | |
| current_chunk): | |
| # Finalizar chunk actual | |
| chunk_text = ' '.join(current_chunk) | |
| chunks.append(chunk_text) | |
| self.logger.debug(f"Chunk completado: {len(current_chunk)} oraciones, " | |
| f"{current_length} caracteres") | |
| # Iniciar nuevo chunk | |
| current_chunk = [sentence] | |
| current_length = sentence_length | |
| else: | |
| # Agregar oración al chunk actual | |
| current_chunk.append(sentence) | |
| current_length += sentence_length + 1 # +1 por el espacio separador | |
| # Agregar último chunk si tiene contenido | |
| if current_chunk: | |
| chunk_text = ' '.join(current_chunk) | |
| chunks.append(chunk_text) | |
| self.logger.debug(f"Chunk final: {len(current_chunk)} oraciones, " | |
| f"{current_length} caracteres") | |
| return chunks | |
| def get_strategy_name(self) -> str: | |
| return f"sentence_chunking_{self.min_length}_{self.max_chunk_size}" | |
| class AdaptiveChunking(ChunkingStrategy): | |
| """ | |
| Implementa chunking adaptativo basado en similitud semántica. | |
| Este método utiliza embeddings para calcular la similitud entre oraciones | |
| consecutivas y agrupa aquellas que son semánticamente similares. Requiere | |
| un modelo de embeddings externo pero puede producir chunks más coherentes | |
| temáticamente. | |
| Ventajas: | |
| - Chunks semánticamente coherentes | |
| - Adaptativo al contenido del texto | |
| - Preserva temas y conceptos relacionados | |
| Desventajas: | |
| - Computacionalmente más costoso | |
| - Requiere modelo de embeddings | |
| - Tamaños de chunk muy variables | |
| """ | |
| def __init__(self, | |
| similarity_threshold: float = settings.ADAPTIVE_SIMILARITY_THRESHOLD, | |
| min_chunk_size: int = settings.ADAPTIVE_MIN_CHUNK_SIZE, | |
| max_chunk_size: int = settings.ADAPTIVE_MAX_CHUNK_SIZE): | |
| """ | |
| Inicializa el chunking adaptativo con parámetros configurables. | |
| Args: | |
| similarity_threshold: Umbral de similitud coseno para agrupar oraciones | |
| min_chunk_size: Tamaño mínimo de chunk en caracteres | |
| max_chunk_size: Tamaño máximo de chunk en caracteres | |
| """ | |
| self.similarity_threshold = similarity_threshold | |
| self.min_chunk_size = min_chunk_size | |
| self.max_chunk_size = max_chunk_size | |
| self.logger = logging.getLogger("chunking_system.adaptive_chunking") | |
| # Validar umbral de similitud | |
| if not 0 <= similarity_threshold <= 1: | |
| raise ValueError("similarity_threshold debe estar entre 0 y 1") | |
| def create_chunks(self, text: str, embeddings_generator=None) -> List[str]: | |
| """ | |
| Crea chunks basados en similitud semántica entre oraciones. | |
| Este algoritmo: | |
| 1. Divide el texto en oraciones | |
| 2. Genera embeddings para cada oración | |
| 3. Calcula similitud coseno entre oraciones consecutivas | |
| 4. Agrupa oraciones con similitud superior al umbral | |
| 5. Respeta límites de tamaño mínimo y máximo | |
| Args: | |
| text: Texto a segmentar | |
| embeddings_generator: Función para generar embeddings de oraciones | |
| Returns: | |
| List[str]: Lista de chunks semánticamente coherentes | |
| """ | |
| if not text or not validate_text_length(text): | |
| self.logger.warning("Texto vacío o demasiado corto para chunking") | |
| return [] | |
| if embeddings_generator is None: | |
| self.logger.warning("No se proporcionó generador de embeddings. " | |
| "Fallback a chunking por oraciones") | |
| fallback = SentenceChunking() | |
| return fallback.create_chunks(text) | |
| # Dividir en oraciones usando la misma lógica que SentenceChunking | |
| sentence_chunker = SentenceChunking() | |
| sentences = sentence_chunker._split_sentences(text) | |
| if len(sentences) < 2: | |
| self.logger.info("Pocas oraciones para chunking adaptativo") | |
| return sentences | |
| # Generar embeddings para cada oración | |
| sentence_embeddings = self._generate_sentence_embeddings( | |
| sentences, embeddings_generator | |
| ) | |
| # Calcular similitudes y agrupar | |
| chunks = self._group_by_similarity(sentences, sentence_embeddings) | |
| self.logger.info(f"Chunking adaptativo completado: {len(chunks)} chunks " | |
| f"generados desde {len(sentences)} oraciones") | |
| return chunks | |
| def _generate_sentence_embeddings(self, | |
| sentences: List[str], | |
| embeddings_generator) -> np.ndarray: | |
| """ | |
| Genera embeddings para una lista de oraciones. | |
| Args: | |
| sentences: Lista de oraciones | |
| embeddings_generator: Función que genera embeddings | |
| Returns: | |
| np.ndarray: Array de embeddings de forma (n_sentences, embedding_dim) | |
| """ | |
| embeddings = [] | |
| for i, sentence in enumerate(sentences): | |
| try: | |
| embedding = embeddings_generator(sentence) | |
| embeddings.append(embedding) | |
| self.logger.debug(f"Embedding generado para oración {i+1}") | |
| except Exception as e: | |
| self.logger.error(f"Error generando embedding para oración {i+1}: {e}") | |
| # Usar vector cero como fallback | |
| embeddings.append(np.zeros(settings.EMBEDDING_DIMENSION)) | |
| return np.array(embeddings) | |
| def _group_by_similarity(self, | |
| sentences: List[str], | |
| embeddings: np.ndarray) -> List[str]: | |
| """ | |
| Agrupa oraciones basándose en similitud de sus embeddings. | |
| El algoritmo utiliza similitud coseno para determinar si dos oraciones | |
| consecutivas deben ser agrupadas en el mismo chunk. La fórmula de | |
| similitud coseno es: sim(A,B) = (A·B) / (||A|| × ||B||) | |
| Args: | |
| sentences: Lista de oraciones | |
| embeddings: Embeddings correspondientes a cada oración | |
| Returns: | |
| List[str]: Lista de chunks agrupados por similitud | |
| """ | |
| if len(sentences) != len(embeddings): | |
| raise ValueError("Número de oraciones y embeddings no coincide") | |
| chunks = [] | |
| current_chunk = [sentences[0]] | |
| current_length = len(sentences[0]) | |
| for i in range(1, len(sentences)): | |
| # Calcular similitud coseno entre oración actual y anterior | |
| similarity = cosine_similarity( | |
| embeddings[i-1].reshape(1, -1), | |
| embeddings[i].reshape(1, -1) | |
| )[0, 0] | |
| sentence_length = len(sentences[i]) | |
| new_length = current_length + sentence_length + 1 | |
| # Decidir si agregar a chunk actual o crear nuevo chunk | |
| should_group = ( | |
| similarity >= self.similarity_threshold and | |
| new_length <= self.max_chunk_size | |
| ) | |
| if should_group: | |
| # Agregar a chunk actual | |
| current_chunk.append(sentences[i]) | |
| current_length = new_length | |
| self.logger.debug(f"Oraciones {i} y {i+1} agrupadas " | |
| f"(similitud: {similarity:.4f})") | |
| else: | |
| # Finalizar chunk actual. Se añade siempre para asegurar | |
| # que el documento quede completamente cubierto, incluso | |
| # si el chunk es menor que el tamaño mínimo configurado. | |
| chunk_text = ' '.join(current_chunk) | |
| chunks.append(chunk_text) | |
| self.logger.debug(f"Chunk completado: {len(current_chunk)} oraciones, " | |
| f"{current_length} caracteres") | |
| # Iniciar nuevo chunk | |
| current_chunk = [sentences[i]] | |
| current_length = sentence_length | |
| if similarity < self.similarity_threshold: | |
| self.logger.debug(f"Nueva sección detectada en oración {i+1} " | |
| f"(similitud: {similarity:.4f})") | |
| # Agregar último chunk (siempre que tenga contenido) | |
| if current_chunk: | |
| chunk_text = ' '.join(current_chunk) | |
| chunks.append(chunk_text) | |
| self.logger.debug(f"Chunk final: {len(current_chunk)} oraciones, " | |
| f"{current_length} caracteres") | |
| return chunks | |
| def get_strategy_name(self) -> str: | |
| return f"adaptive_chunking_{self.similarity_threshold}_{self.min_chunk_size}_{self.max_chunk_size}" | |
| class ChunkingManager: | |
| """ | |
| Gestor principal para seleccionar y ejecutar estrategias de chunking. | |
| Esta clase actúa como fachada para todos los métodos de chunking, | |
| permitiendo seleccionar la estrategia desde configuración y proporcionando | |
| una interfaz unificada para el sistema principal. | |
| """ | |
| def __init__(self): | |
| """ | |
| Inicializa el gestor con la estrategia configurada en settings. | |
| """ | |
| self.logger = logging.getLogger("chunking_system.manager") | |
| self.strategy = self._create_strategy() | |
| def _create_strategy(self) -> ChunkingStrategy: | |
| """ | |
| Crea la estrategia de chunking según la configuración. | |
| Returns: | |
| ChunkingStrategy: Instancia de la estrategia configurada | |
| Raises: | |
| ValueError: Si el método configurado no es válido | |
| """ | |
| method = settings.CHUNKING_METHOD.lower() | |
| if method == "fixed": | |
| return FixedChunking() | |
| elif method == "sentence": | |
| return SentenceChunking() | |
| elif method == "adaptive": | |
| return AdaptiveChunking() | |
| else: | |
| raise ValueError(f"Método de chunking no válido: {method}") | |
| def create_chunks(self, text: str, embeddings_generator=None) -> List[str]: | |
| """ | |
| Crea chunks usando la estrategia configurada. | |
| Args: | |
| text: Texto a segmentar | |
| embeddings_generator: Generador de embeddings (requerido para adaptativo) | |
| Returns: | |
| List[str]: Lista de chunks generados | |
| """ | |
| self.logger.info(f"Iniciando chunking con estrategia: {self.strategy.get_strategy_name()}") | |
| # Para chunking adaptativo, pasar el generador de embeddings | |
| if isinstance(self.strategy, AdaptiveChunking): | |
| chunks = self.strategy.create_chunks(text, embeddings_generator) | |
| else: | |
| chunks = self.strategy.create_chunks(text) | |
| self.logger.info(f"Chunking completado: {len(chunks)} chunks generados") | |
| # Log estadísticas de chunks | |
| if chunks: | |
| lengths = [len(chunk) for chunk in chunks] | |
| self.logger.info(f"Estadísticas de chunks - " | |
| f"Min: {min(lengths)}, Max: {max(lengths)}, " | |
| f"Promedio: {sum(lengths)/len(lengths):.1f} caracteres") | |
| return chunks | |
| def get_current_strategy_name(self) -> str: | |
| """ | |
| Obtiene el nombre de la estrategia actualmente configurada. | |
| Returns: | |
| str: Nombre de la estrategia actual | |
| """ | |
| return self.strategy.get_strategy_name() |