Deploy document chunking system with Gradio interface

DEPLOYMENT.md

@@ -0,0 +1,91 @@
+# 🚀 Deployment Guide for Gradio Spaces
+
+## Quick Deploy Checklist
+
+### 1. Environment Variables (Required)
+In your Hugging Face Space settings, add these secrets:
+
+```
+OPENAI_API_KEY=your_openai_api_key_here
+QDRANT_URL=https://your-qdrant-cloud-instance.com
+QDRANT_API_KEY=your_qdrant_api_key (if using Qdrant Cloud)
+```
+
+### 2. Qdrant Setup Options
+
+#### Option A: Qdrant Cloud (Recommended)
+1. Sign up at [cloud.qdrant.io](https://cloud.qdrant.io)
+2. Create a cluster
+3. Get your cluster URL and API key
+4. Set QDRANT_URL and QDRANT_API_KEY in Space secrets
+
+#### Option B: Self-hosted Qdrant
+1. Deploy Qdrant on a cloud provider (DigitalOcean, AWS, etc.)
+2. Ensure it's publicly accessible
+3. Set QDRANT_URL in Space secrets
+
+### 3. Push Files to Space
+
+The following files are already copied to your Space directory:
+- `app.py` - Main Gradio application
+- `requirements.txt` - Python dependencies
+- `settings.py` - Configuration
+- `document_processor.py` - Document handling
+- `embeddings_qdrant.py` - Embeddings and Qdrant logic
+- `chunking_methods.py` - Chunking strategies
+- `pdf_loader.py` - PDF processing
+- `utils_logging.py` - Logging utilities
+
+### 4. Deploy
+
+```bash
+cd optchunkopt
+git add .
+git commit -m "Deploy chunking system to Gradio Space"
+git push origin main
+```
+
+### 5. Monitor Deployment
+
+Check the Space logs for:
+- ✅ Component initialization success
+- ⚠️ Configuration warnings
+- ❌ Connection errors
+
+## Troubleshooting
+
+### Common Issues:
+
+1. **OPENAI_API_KEY not found**
+   - Add your OpenAI API key to Space secrets
+
+2. **Qdrant connection failed**
+   - Verify QDRANT_URL is correct and accessible
+   - Check if Qdrant instance is running
+
+3. **Import errors**
+   - Check if all dependencies are in requirements.txt
+   - Verify file paths and imports
+
+### Test Locally First:
+
+```bash
+cd optchunkopt
+python app.py
+```
+
+This will help identify issues before deployment.
+
+## Features Available:
+
+- 📤 **Document Upload**: PDF, TXT, DOCX support
+- 🔄 **Multi-Method Chunking**: Fixed, sentence, adaptive strategies
+- 🔍 **Semantic Search**: OpenAI embeddings with similarity search
+- ⚙️ **System Status**: Real-time configuration monitoring
+
+## Performance Notes:
+
+- First document processing will be slower (embedding generation)
+- Search is fast once embeddings are stored
+- Consider chunking method based on document type
+- Monitor OpenAI API usage costs

app.py

@@ -0,0 +1,345 @@
+import gradio as gr
+import logging
+import os
+from typing import List, Tuple, Optional
+import pandas as pd
+
+# Import your existing modules
+from document_processor import DocumentProcessor
+from embeddings_qdrant import EmbeddingsQdrantPipeline
+import settings
+
+# Configure logging
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+# Global variables for components (will be initialized in main)
+document_processor = None
+embeddings_pipeline = None
+
+def initialize_components():
+    """Initialize the pipeline components with error handling for cloud deployment."""
+    global document_processor, embeddings_pipeline
+    
+    try:
+        document_processor = DocumentProcessor()
+        logger.info("✅ Document processor initialized")
+    except Exception as e:
+        logger.error(f"❌ Failed to initialize document processor: {str(e)}")
+        document_processor = None
+    
+    try:
+        embeddings_pipeline = EmbeddingsQdrantPipeline()
+        logger.info("✅ Embeddings pipeline initialized")
+    except Exception as e:
+        logger.error(f"❌ Failed to initialize embeddings pipeline: {str(e)}")
+        embeddings_pipeline = None
+        
+    return document_processor is not None and embeddings_pipeline is not None
+
+def process_document(file_path: str, chunking_methods: List[str]) -> Tuple[str, pd.DataFrame]:
+    """
+    Process a document with selected chunking methods.
+    
+    Args:
+        file_path: Path to the uploaded file
+        chunking_methods: List of selected chunking methods
+    
+    Returns:
+        Tuple of (status_message, chunks_dataframe)
+    """
+    try:
+        if document_processor is None:
+            return "❌ Error: Document processor not initialized. Check system configuration.", pd.DataFrame()
+            
+        if embeddings_pipeline is None:
+            return "❌ Error: Embeddings pipeline not initialized. Check Qdrant connection and OpenAI API key.", pd.DataFrame()
+            
+        if not file_path:
+            return "❌ Error: No file uploaded", pd.DataFrame()
+        
+        if not chunking_methods:
+            return "❌ Error: Please select at least one chunking method", pd.DataFrame()
+        
+        # Process the document
+        logger.info(f"Processing document: {file_path}")
+        chunks_by_method = document_processor.process_document(
+            file_path=file_path,
+            chunking_methods=chunking_methods
+        )
+        
+        if not chunks_by_method:
+            return "❌ Error: No chunks generated from document", pd.DataFrame()
+        
+        # Store embeddings in Qdrant
+        success = embeddings_pipeline.process_and_store_chunks(chunks_by_method)
+        
+        if not success:
+            return "❌ Error: Failed to store embeddings in Qdrant", pd.DataFrame()
+        
+        # Create a summary dataframe
+        summary_data = []
+        total_chunks = 0
+        for method, chunks in chunks_by_method.items():
+            summary_data.append({
+                "Chunking Method": method,
+                "Number of Chunks": len(chunks),
+                "Average Chunk Length": round(sum(len(chunk) for chunk in chunks) / len(chunks)) if chunks else 0,
+                "Sample Chunk": chunks[0][:100] + "..." if chunks else "No chunks"
+            })
+            total_chunks += len(chunks)
+        
+        summary_df = pd.DataFrame(summary_data)
+        
+        status_msg = f"✅ Success! Document processed with {len(chunking_methods)} methods. Total chunks: {total_chunks}"
+        
+        return status_msg, summary_df
+        
+    except Exception as e:
+        logger.error(f"Error processing document: {str(e)}")
+        return f"❌ Error: {str(e)}", pd.DataFrame()
+
+def search_chunks(query: str, top_k: int = None) -> Tuple[str, pd.DataFrame]:
+    """
+    Search for similar chunks across all collections.
+    
+    Args:
+        query: Search query
+        top_k: Number of results to return (uses settings default if None)
+    
+    Returns:
+        Tuple of (status_message, results_dataframe)
+    """
+    try:
+        if embeddings_pipeline is None:
+            return "❌ Error: Embeddings pipeline not initialized. Check Qdrant connection and OpenAI API key.", pd.DataFrame()
+            
+        if not query.strip():
+            return "❌ Error: Please enter a search query", pd.DataFrame()
+        
+        # Use provided top_k or default from settings
+        k = top_k if top_k is not None else settings.TOP_K_RESULTS
+        
+        # Perform search
+        logger.info(f"Searching for: '{query}' (top_k={k})")
+        results = embeddings_pipeline.search_similar_chunks(query=query, top_k=k)
+        
+        if not results:
+            return "⚠️ No results found. Try a different query or check if documents are processed.", pd.DataFrame()
+        
+        # Create results dataframe
+        results_data = []
+        for i, (text, score, method) in enumerate(results, 1):
+            results_data.append({
+                "Rank": i,
+                "Similarity Score": round(score, 4),
+                "Chunking Method": method,
+                "Text Preview": text[:200] + "..." if len(text) > 200 else text,
+                "Full Text": text
+            })
+        
+        results_df = pd.DataFrame(results_data)
+        
+        status_msg = f"✅ Found {len(results)} results for query: '{query}'"
+        
+        return status_msg, results_df
+        
+    except Exception as e:
+        logger.error(f"Error searching chunks: {str(e)}")
+        return f"❌ Error: {str(e)}", pd.DataFrame()
+
+def get_system_status() -> Tuple[str, pd.DataFrame]:
+    """
+    Get current system status including collections info.
+    
+    Returns:
+        Tuple of (status_message, status_dataframe)
+    """
+    try:
+        # Get collection statistics
+        status_data = []
+        total_chunks = 0
+        
+        for method in settings.CHUNKING_METHODS:
+            collection_name = f"{settings.QDRANT_COLLECTION_NAME}_{method}"
+            try:
+                # This would need a method to get collection info
+                # For now, we'll show configured methods
+                status_data.append({
+                    "Collection": collection_name,
+                    "Method": method,
+                    "Status": "Available",
+                    "Description": f"Collection for {method} chunking method"
+                })
+            except Exception:
+                status_data.append({
+                    "Collection": collection_name,
+                    "Method": method,
+                    "Status": "Not Available",
+                    "Description": f"Collection for {method} chunking method (not created)"
+                })
+        
+        status_df = pd.DataFrame(status_data)
+        
+        # System configuration info
+        config_info = f"""
+        📊 **System Configuration:**
+        - Embedding Model: {settings.OPENAI_EMBEDDING_MODEL}
+        - Embedding Dimensions: {settings.EMBEDDING_DIMENSIONS}
+        - Qdrant URL: {settings.QDRANT_URL}
+        - Top K Results: {settings.TOP_K_RESULTS}
+        - Min Similarity Score: {settings.MIN_SIMILARITY_SCORE}
+        - Available Chunking Methods: {', '.join(settings.CHUNKING_METHODS)}
+        """
+        
+        return config_info, status_df
+        
+    except Exception as e:
+        logger.error(f"Error getting system status: {str(e)}")
+        return f"❌ Error: {str(e)}", pd.DataFrame()
+
+# Create Gradio interface
+with gr.Blocks(title="Document Chunking & Search System", theme=gr.themes.Soft()) as demo:
+    gr.Markdown("""
+    # 📄 Document Chunking & Search System
+    
+    This system processes documents using multiple chunking strategies and enables semantic search using OpenAI embeddings stored in Qdrant.
+    
+    ## Available Chunking Methods:
+    - **Fixed**: Fixed-size chunks with overlap
+    - **Sentence**: Sentence-based chunking 
+    - **Adaptive**: Adaptive chunking based on content structure
+    """)
+    
+    with gr.Tab("📤 Document Processing"):
+        gr.Markdown("### Upload and process documents")
+        
+        with gr.Row():
+            with gr.Column():
+                file_input = gr.File(
+                    label="Upload Document", 
+                    file_types=[".pdf", ".txt", ".docx"],
+                    type="filepath"
+                )
+                
+                chunking_methods = gr.CheckboxGroup(
+                    choices=settings.CHUNKING_METHODS,
+                    value=settings.CHUNKING_METHODS,  # Select all by default
+                    label="Select Chunking Methods"
+                )
+                
+                process_btn = gr.Button("🔄 Process Document", variant="primary")
+            
+            with gr.Column():
+                process_status = gr.Textbox(
+                    label="Processing Status",
+                    placeholder="Upload a document and click 'Process Document'",
+                    interactive=False
+                )
+        
+        chunks_table = gr.Dataframe(
+            label="Chunks Summary",
+            headers=["Chunking Method", "Number of Chunks", "Average Chunk Length", "Sample Chunk"],
+            interactive=False
+        )
+        
+        process_btn.click(
+            process_document,
+            inputs=[file_input, chunking_methods],
+            outputs=[process_status, chunks_table]
+        )
+    
+    with gr.Tab("🔍 Search"):
+        gr.Markdown("### Search for similar content across all processed documents")
+        
+        with gr.Row():
+            with gr.Column():
+                search_input = gr.Textbox(
+                    label="Search Query",
+                    placeholder="Enter your search query here...",
+                    lines=2
+                )
+                
+                top_k_input = gr.Slider(
+                    minimum=1,
+                    maximum=20,
+                    value=settings.TOP_K_RESULTS,
+                    step=1,
+                    label="Number of Results"
+                )
+                
+                search_btn = gr.Button("🔍 Search", variant="primary")
+            
+            with gr.Column():
+                search_status = gr.Textbox(
+                    label="Search Status",
+                    placeholder="Enter a query and click 'Search'",
+                    interactive=False
+                )
+        
+        results_table = gr.Dataframe(
+            label="Search Results",
+            headers=["Rank", "Similarity Score", "Chunking Method", "Text Preview", "Full Text"],
+            interactive=False
+        )
+        
+        search_btn.click(
+            search_chunks,
+            inputs=[search_input, top_k_input],
+            outputs=[search_status, results_table]
+        )
+    
+    with gr.Tab("⚙️ System Status"):
+        gr.Markdown("### Current system configuration and status")
+        
+        status_btn = gr.Button("🔄 Refresh Status", variant="secondary")
+        
+        system_info = gr.Textbox(
+            label="System Information",
+            lines=10,
+            interactive=False
+        )
+        
+        collections_table = gr.Dataframe(
+            label="Collections Status",
+            headers=["Collection", "Method", "Status", "Description"],
+            interactive=False
+        )
+        
+        # Load initial status
+        demo.load(
+            get_system_status,
+            outputs=[system_info, collections_table]
+        )
+        
+        status_btn.click(
+            get_system_status,
+            outputs=[system_info, collections_table]
+        )
+
+if __name__ == "__main__":
+    # Set up environment for deployment
+    port = int(os.environ.get("PORT", 7860))
+    
+    print("🚀 Starting Document Chunking & Search System...")
+    
+    # Initialize components with error handling
+    init_success = initialize_components()
+    
+    if not init_success:
+        print("⚠️  Warning: Some components failed to initialize. Check configuration.")
+        print("� Make sure you have:")
+        print("   - OPENAI_API_KEY environment variable set")
+        print("   - Qdrant instance running (QDRANT_URL configured)")
+    else:
+        print("✅ All components initialized successfully")
+    
+    print(f"�📊 Configured for {len(settings.CHUNKING_METHODS)} chunking methods")
+    print(f"🔗 Qdrant URL: {settings.QDRANT_URL}")
+    print(f"🤖 Embedding Model: {settings.OPENAI_EMBEDDING_MODEL}")
+    
+    demo.launch(
+        server_name="0.0.0.0",
+        server_port=port,
+        share=False  # Gradio Spaces will handle public access
+    )

chunking_methods.py

@@ -0,0 +1,567 @@
+"""
+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.
+    """
+    
+    @abstractmethod
+    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
+    
+    @abstractmethod
+    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")
+    
+    @log_operation_time
+    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+|[.!?]+$')
+    
+    @log_operation_time
+    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")
+    
+    @log_operation_time
+    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}")
+    
+    @log_operation_time
+    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()

document_processor.py

@@ -0,0 +1,159 @@
+"""
+Document processor module for handling various document types and chunking methods.
+Integrates PDF loading and chunking functionality for the Gradio interface.
+"""
+
+import logging
+from typing import Dict, List, Optional
+import os
+from pathlib import Path
+
+# Import existing modules
+from pdf_loader import load_pdf_from_settings
+from chunking_methods import ChunkingManager, FixedChunking, SentenceChunking, AdaptiveChunking
+import settings
+
+logger = logging.getLogger(__name__)
+
+class DocumentProcessor:
+    """
+    Handles document loading and processing with multiple chunking methods.
+    """
+    
+    def __init__(self):
+        """Initialize the document processor with available chunking methods."""
+        self.chunking_manager = ChunkingManager()
+        
+        # Register available chunking methods
+        self.chunking_strategies = {
+            'fixed': FixedChunking(),
+            'sentence': SentenceChunking(), 
+            'adaptive': AdaptiveChunking()
+        }
+        
+        logger.info(f"DocumentProcessor initialized with {len(self.chunking_strategies)} chunking strategies")
+    
+    def load_document(self, file_path: str) -> Optional[str]:
+        """
+        Load document content based on file type.
+        
+        Args:
+            file_path: Path to the document file
+            
+        Returns:
+            Document text content or None if loading fails
+        """
+        try:
+            if not os.path.exists(file_path):
+                logger.error(f"File not found: {file_path}")
+                return None
+            
+            file_extension = Path(file_path).suffix.lower()
+            
+            if file_extension == '.pdf':
+                # Use existing PDF loader
+                # For now, we'll temporarily set the PDF path in settings
+                original_pdf_path = getattr(settings, 'PDF_CONTENT_FILE', None)
+                settings.PDF_CONTENT_FILE = file_path
+                
+                try:
+                    content = load_pdf_from_settings()
+                    return content
+                finally:
+                    # Restore original setting
+                    if original_pdf_path:
+                        settings.PDF_CONTENT_FILE = original_pdf_path
+                    
+            elif file_extension == '.txt':
+                with open(file_path, 'r', encoding='utf-8') as f:
+                    return f.read()
+                    
+            elif file_extension in ['.doc', '.docx']:
+                try:
+                    from docx import Document
+                    doc = Document(file_path)
+                    content = []
+                    for paragraph in doc.paragraphs:
+                        content.append(paragraph.text)
+                    return '\n'.join(content)
+                except ImportError:
+                    logger.error("python-docx not installed. Cannot process DOCX files.")
+                    return None
+                    
+            else:
+                logger.error(f"Unsupported file type: {file_extension}")
+                return None
+                
+        except Exception as e:
+            logger.error(f"Error loading document {file_path}: {str(e)}")
+            return None
+    
+    def process_document(self, file_path: str, chunking_methods: List[str]) -> Dict[str, List[str]]:
+        """
+        Process a document with specified chunking methods.
+        
+        Args:
+            file_path: Path to the document file
+            chunking_methods: List of chunking method names to apply
+            
+        Returns:
+            Dictionary mapping method names to lists of chunks
+        """
+        try:
+            # Load document content
+            content = self.load_document(file_path)
+            if not content:
+                logger.error(f"Failed to load content from {file_path}")
+                return {}
+            
+            logger.info(f"Loaded document with {len(content)} characters")
+            
+            # Process with each requested chunking method
+            chunks_by_method = {}
+            
+            for method_name in chunking_methods:
+                if method_name not in self.chunking_strategies:
+                    logger.warning(f"Unknown chunking method: {method_name}")
+                    continue
+                
+                try:
+                    strategy = self.chunking_strategies[method_name]
+                    chunks = strategy.chunk_text(content)
+                    chunks_by_method[method_name] = chunks
+                    
+                    logger.info(f"Generated {len(chunks)} chunks using {method_name} method")
+                    
+                except Exception as e:
+                    logger.error(f"Error processing with {method_name} method: {str(e)}")
+                    continue
+            
+            return chunks_by_method
+            
+        except Exception as e:
+            logger.error(f"Error processing document {file_path}: {str(e)}")
+            return {}
+    
+    def get_supported_extensions(self) -> List[str]:
+        """
+        Get list of supported file extensions.
+        
+        Returns:
+            List of supported file extensions
+        """
+        return ['.pdf', '.txt', '.docx']
+    
+    def validate_file(self, file_path: str) -> bool:
+        """
+        Validate if a file can be processed.
+        
+        Args:
+            file_path: Path to the file to validate
+            
+        Returns:
+            True if file can be processed, False otherwise
+        """
+        if not os.path.exists(file_path):
+            return False
+        
+        extension = Path(file_path).suffix.lower()
+        return extension in self.get_supported_extensions()

embeddings_qdrant.py

@@ -0,0 +1,710 @@
+"""
+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}")
+    
+    @log_operation_time
+    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
+    
+    @log_operation_time
+    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
+        self.client = QdrantClient(url=url)
+        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")
+        except Exception as e:
+            self.logger.error(f"Error conectando con Qdrant: {str(e)}")
+            raise
+    
+    @log_operation_time
+    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
+    
+    @log_operation_time
+    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 = []
+            
+            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=i,  # ID único para cada chunk (local a la colección)
+                    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
+    
+    @log_operation_time
+    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")
+            
+            # Recrear colección vacía
+            self.create_collection_if_not_exists()
+            
+        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")
+    
+    @log_operation_time
+    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)
+
+                if existing_points > 0:
+                    self.logger.info(f"Método '{method}': encontrados {existing_points} puntos en colección '{collection_name}'. Se omite generación para ahorrar costos.")
+                    continue
+
+                # Generar embeddings solo 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}'")
+                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
+    
+    @log_operation_time
+    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

pdf_loader.py

@@ -0,0 +1,275 @@
+"""
+MIT License
+Autor: Fernando Lafuente
+Proyecto: Ejercicio opcional NLP - Chunking y Recuperación
+Descripción: Extracción y limpieza de texto desde archivos PDF
+
+Este módulo se encarga de extraer el contenido textual de documentos PDF
+y aplicar técnicas de limpieza para preparar el texto para el chunking.
+Utiliza PyMuPDF (fitz) para la extracción y aplica filtros pedagógicos
+que mejoran la calidad del texto para procesamiento posterior.
+"""
+
+import logging
+from pathlib import Path
+from typing import Optional
+
+import fitz  # PyMuPDF
+
+import settings
+from utils_logging import setup_logging, log_operation_time, clean_text_basic, validate_text_length
+
+
+# Configurar logger específico para este módulo
+logger = setup_logging()
+
+
+class PDFLoader:
+    """
+    Clase para cargar y procesar documentos PDF.
+    
+    Esta clase encapsula la lógica de extracción de texto desde archivos PDF,
+    aplicando técnicas de limpieza y validación que aseguran que el texto
+    resultante sea apropiado para chunking y generación de embeddings.
+    
+    La extracción se realiza página por página para permitir un mejor control
+    sobre la calidad del texto y facilitar el debugging en documentos largos.
+    """
+    
+    def __init__(self):
+        """
+        Inicializa el cargador de PDF con configuración por defecto.
+        """
+        self.logger = logging.getLogger("chunking_system.pdf_loader")
+    
+    @log_operation_time
+    def load_pdf(self, pdf_path: Path) -> str:
+        """
+        Carga un archivo PDF y extrae todo su contenido textual.
+        
+        Este método abre el PDF, procesa cada página individualmente,
+        extrae el texto y lo concatena en un único string. Aplica
+        limpieza básica y validaciones para asegurar calidad del texto.
+        
+        Args:
+            pdf_path: Ruta al archivo PDF a procesar
+            
+        Returns:
+            str: Texto completo extraído del PDF
+            
+        Raises:
+            FileNotFoundError: Si el archivo PDF no existe
+            ValueError: Si el PDF está corrupto o no contiene texto
+            
+        Example:
+            loader = PDFLoader()
+            texto = loader.load_pdf(Path("documento.pdf"))
+        """
+        # Validar que el archivo existe
+        if not pdf_path.exists():
+            raise FileNotFoundError(f"Archivo PDF no encontrado: {pdf_path}")
+        
+        self.logger.info(f"Iniciando carga de PDF: {pdf_path.name}")
+        
+        try:
+            # Abrir documento PDF con PyMuPDF
+            pdf_document = fitz.open(str(pdf_path))
+            total_pages = pdf_document.page_count
+            
+            self.logger.info(f"PDF abierto correctamente. Páginas totales: {total_pages}")
+            
+            # Extraer texto de todas las páginas
+            all_text = []
+            
+            for page_num in range(total_pages):
+                page_text = self._extract_page_text(pdf_document, page_num)
+                if page_text:  # Solo agregar páginas con contenido
+                    all_text.append(page_text)
+                    self.logger.debug(f"Página {page_num + 1}: {len(page_text)} caracteres extraídos")
+            
+            # Cerrar documento
+            pdf_document.close()
+            
+            # Concatenar todo el texto
+            full_text = "\n\n".join(all_text)
+            
+            # Aplicar limpieza final
+            clean_text = self._clean_extracted_text(full_text)
+            
+            # Validar que se extrajo contenido útil
+            if not validate_text_length(clean_text):
+                raise ValueError(f"El PDF no contiene suficiente texto válido: {pdf_path}")
+            
+            self.logger.info(f"Extracción completada. Texto final: {len(clean_text)} caracteres")
+            
+            return clean_text
+            
+        except Exception as e:
+            self.logger.error(f"Error procesando PDF {pdf_path}: {str(e)}")
+            raise
+    
+    def _extract_page_text(self, pdf_document: fitz.Document, page_num: int) -> str:
+        """
+        Extrae texto de una página específica del PDF.
+        
+        Este método privado se encarga de extraer el texto de una sola página,
+        aplicando técnicas específicas de PyMuPDF para obtener la mejor calidad
+        de texto posible. Maneja casos especiales como páginas vacías o con
+        elementos gráficos únicamente.
+        
+        Args:
+            pdf_document: Documento PDF abierto con PyMuPDF
+            page_num: Número de página a procesar (base 0)
+            
+        Returns:
+            str: Texto extraído de la página, limpio y validado
+        """
+        try:
+            # Obtener la página
+            page = pdf_document[page_num]
+            
+            # Extraer texto usando el método estándar de PyMuPDF
+            # El método get_text() retorna el texto en el orden de lectura
+            page_text = page.get_text()
+            
+            # Aplicar limpieza básica específica para esta página
+            clean_page_text = clean_text_basic(page_text)
+            
+            # Validar que la página contiene texto útil
+            if not validate_text_length(clean_page_text, min_length=10):
+                self.logger.debug(f"Página {page_num + 1} descartada: contenido insuficiente")
+                return ""
+            
+            return clean_page_text
+            
+        except Exception as e:
+            self.logger.warning(f"Error extrayendo texto de página {page_num + 1}: {str(e)}")
+            return ""
+    
+    def _clean_extracted_text(self, text: str) -> str:
+        """
+        Aplica limpieza específica al texto extraído de PDF.
+        
+        Esta función implementa técnicas de limpieza especializadas para
+        texto extraído de PDFs, que suelen contener artefactos específicos
+        como headers/footers repetidos, numeración de páginas, etc.
+        
+        Args:
+            text: Texto en bruto extraído del PDF
+            
+        Returns:
+            str: Texto limpio listo para chunking
+        """
+        if not text:
+            return ""
+        
+        # Aplicar limpieza básica primero
+        clean_text = clean_text_basic(text)
+        
+        # Limpieza específica para PDFs
+        
+        # Eliminar líneas que son solo números (numeración de páginas)
+        lines = clean_text.split('\n')
+        filtered_lines = []
+        
+        for line in lines:
+            line_stripped = line.strip()
+            
+            # Descartar líneas vacías
+            if not line_stripped:
+                continue
+            
+            # Descartar líneas que son solo números (posibles números de página)
+            if line_stripped.isdigit() and len(line_stripped) <= 3:
+                continue
+            
+            # Descartar líneas muy cortas que probablemente son artefactos
+            if len(line_stripped) < 5:
+                continue
+            
+            filtered_lines.append(line_stripped)
+        
+        # Reconstruir texto
+        final_text = ' '.join(filtered_lines)
+        
+        # Normalizar espacios finales
+        final_text = ' '.join(final_text.split())
+        
+        return final_text
+    
+    def get_pdf_metadata(self, pdf_path: Path) -> dict:
+        """
+        Extrae metadatos del archivo PDF para información adicional.
+        
+        Los metadatos pueden ser útiles para logging, debugging o para
+        personalizar el procesamiento según las características del documento.
+        
+        Args:
+            pdf_path: Ruta al archivo PDF
+            
+        Returns:
+            dict: Diccionario con metadatos del PDF
+            
+        Example:
+            metadata = loader.get_pdf_metadata(pdf_path)
+            logger.info(f"Procesando: {metadata['title']}")
+        """
+        try:
+            pdf_document = fitz.open(str(pdf_path))
+            
+            metadata = {
+                'title': pdf_document.metadata.get('title', 'Sin título'),
+                'author': pdf_document.metadata.get('author', 'Desconocido'),
+                'subject': pdf_document.metadata.get('subject', ''),
+                'creator': pdf_document.metadata.get('creator', ''),
+                'pages': pdf_document.page_count,
+                'file_size': pdf_path.stat().st_size,
+                'encrypted': pdf_document.needs_pass
+            }
+            
+            pdf_document.close()
+            
+            return metadata
+            
+        except Exception as e:
+            self.logger.warning(f"Error extrayendo metadatos de {pdf_path}: {str(e)}")
+            return {
+                'title': pdf_path.name,
+                'author': 'Desconocido',
+                'pages': 0,
+                'file_size': 0,
+                'encrypted': False
+            }
+
+
+def load_pdf_from_settings() -> str:
+    """
+    Función de conveniencia que carga el PDF configurado en settings.
+    
+    Esta función utiliza la configuración centralizada para cargar
+    automáticamente el primer PDF encontrado en el directorio especificado.
+    
+    Returns:
+        str: Texto extraído del PDF
+        
+    Raises:
+        FileNotFoundError: Si no se encuentra ningún PDF
+        ValueError: Si el PDF no contiene texto válido
+        
+    Example:
+        texto_documento = load_pdf_from_settings()
+        chunks = chunker.create_chunks(texto_documento)
+    """
+    # Obtener ruta del PDF desde configuración
+    pdf_path = settings.get_pdf_file()
+    
+    # Crear instancia del loader y procesar
+    loader = PDFLoader()
+    
+    # Mostrar metadatos para información
+    metadata = loader.get_pdf_metadata(pdf_path)
+    logger.info(f"Cargando PDF: {metadata['title']} "
+               f"({metadata['pages']} páginas, {metadata['file_size']} bytes)")
+    
+    # Cargar y retornar texto
+    return loader.load_pdf(pdf_path)

requirements.txt

@@ -0,0 +1,13 @@
+gradio>=4.0.0
+langchain
+qdrant-client
+openai
+PyMuPDF
+numpy
+pandas
+scikit-learn
+typer
+python-dotenv
+rich
+python-docx
+nltk

settings.py

@@ -0,0 +1,172 @@
+"""
+MIT License
+Autor: Fernando Lafuente
+Proyecto: Ejercicio opcional NLP - Chunking y Recuperación
+Descripción: Configuración centralizada de parámetros del sistema
+
+Este módulo centraliza todos los parámetros configurables del sistema de chunking
+y recuperación semántica. Permite modificar el comportamiento sin tocar el código
+principal, siguiendo principios de arquitectura limpia.
+"""
+
+import os
+from pathlib import Path
+from dotenv import load_dotenv
+
+# Cargar variables de entorno desde archivo .env
+load_dotenv()
+
+# =============================================================================
+# CONFIGURACIÓN DE RUTAS Y ARCHIVOS
+# =============================================================================
+
+# Directorio raíz del proyecto
+PROJECT_ROOT = Path(__file__).parent.absolute()
+
+# Directorio donde se almacenan los archivos PDF
+PDF_DIRECTORY = PROJECT_ROOT / "data" / "pdf"
+
+# Crear directorio si no existe
+PDF_DIRECTORY.mkdir(parents=True, exist_ok=True)
+
+# =============================================================================
+# CONFIGURACIÓN DE OPENAI EMBEDDINGS
+# =============================================================================
+
+# Clave de API de OpenAI (debe configurarse como variable de entorno)
+# Ejecutar: export OPENAI_API_KEY="tu_clave_aqui"
+OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
+
+# Modelo de embeddings de OpenAI a utilizar
+# Opciones: "text-embedding-3-small", "text-embedding-3-large", "text-embedding-ada-002"
+OPENAI_EMBEDDING_MODEL = "text-embedding-3-large"  # Más económico y eficiente
+
+# Dimensionalidad de los embeddings de OpenAI (text-embedding-3-small)
+EMBEDDING_DIMENSION = 3072
+
+# =============================================================================
+# CONFIGURACIÓN DE QDRANT
+# =============================================================================
+
+# URL de conexión a Qdrant local
+QDRANT_URL = "http://localhost:6333"
+
+# Nombre de la colección donde se almacenan los vectores
+QDRANT_COLLECTION_NAME = "chunks_collection"
+
+# =============================================================================
+# CONFIGURACIÓN DE CHUNKING
+# =============================================================================
+
+# Método de chunking a utilizar
+# Opciones: "fixed", "sentence", "adaptive"
+CHUNKING_METHOD = "adaptive"
+
+# Parámetros para chunking fijo
+FIXED_CHUNK_SIZE = 500  # caracteres por chunk
+FIXED_CHUNK_OVERLAP = 50  # caracteres de solapamiento
+
+# Parámetros para chunking por oraciones
+SENTENCE_MIN_LENGTH = 50  # longitud mínima de oración en caracteres
+SENTENCE_MAX_CHUNK_SIZE = 800  # tamaño máximo de chunk en caracteres
+
+# Parámetros para chunking adaptativo
+ADAPTIVE_SIMILARITY_THRESHOLD = 0.7  # umbral de similitud coseno (0-1)
+ADAPTIVE_MIN_CHUNK_SIZE = 100  # tamaño mínimo de chunk en caracteres
+ADAPTIVE_MAX_CHUNK_SIZE = 1000  # tamaño máximo de chunk en caracteres
+
+# Lista de métodos de chunking soportados por el sistema
+CHUNKING_METHODS = ["fixed", "sentence", "adaptive"]
+
+# =============================================================================
+# CONFIGURACIÓN DE BÚSQUEDA Y RETRIEVAL
+# =============================================================================
+
+# Número de chunks más relevantes a retornar
+TOP_K_RESULTS = 10
+
+# Puntuación mínima de similitud para considerar un resultado relevante
+MIN_SIMILARITY_SCORE = 0.5
+
+# =============================================================================
+# CONFIGURACIÓN DE LOGGING
+# =============================================================================
+
+# Nivel de logging
+# Opciones: "DEBUG", "INFO", "WARNING", "ERROR"
+LOG_LEVEL = "DEBUG"
+
+# Formato de los mensajes de log
+LOG_FORMAT = "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
+
+# =============================================================================
+# CONFIGURACIÓN DE PROCESAMIENTO DE TEXTO
+# =============================================================================
+
+# Caracteres a eliminar durante la limpieza de texto
+CHARS_TO_REMOVE = ["\n\n", "\t", "\r"]
+
+# Caracteres a reemplazar por espacios
+CHARS_TO_REPLACE = {"\n": " ", "  ": " "}
+
+# Longitud mínima de texto válido después de limpieza
+MIN_TEXT_LENGTH = 10
+
+# =============================================================================
+# VALIDACIÓN DE CONFIGURACIÓN
+# =============================================================================
+
+def validate_settings():
+    """
+    Valida que la configuración sea correcta antes de ejecutar el sistema.
+    
+    Raises:
+        ValueError: Si algún parámetro es inválido
+        FileNotFoundError: Si no se encuentra el directorio de PDFs
+    """
+    # Validar que existe el directorio de PDFs
+    if not PDF_DIRECTORY.exists():
+        raise FileNotFoundError(f"Directorio de PDFs no encontrado: {PDF_DIRECTORY}")
+    
+    # Validar que hay al menos un PDF en el directorio
+    pdf_files = list(PDF_DIRECTORY.glob("*.pdf"))
+    if not pdf_files:
+        raise FileNotFoundError(f"No se encontraron archivos PDF en: {PDF_DIRECTORY}")
+    
+    # Validar clave de API de OpenAI
+    if not OPENAI_API_KEY:
+        raise ValueError("Variable de entorno OPENAI_API_KEY no configurada")
+    
+    # Validar método de chunking
+    valid_methods = ["fixed", "sentence", "adaptive"]
+    if CHUNKING_METHOD not in valid_methods:
+        raise ValueError(f"Método de chunking inválido: {CHUNKING_METHOD}. "
+                        f"Opciones válidas: {valid_methods}")
+    
+    # Validar parámetros numéricos
+    if TOP_K_RESULTS <= 0:
+        raise ValueError("TOP_K_RESULTS debe ser mayor que 0")
+    
+    if MIN_SIMILARITY_SCORE < 0 or MIN_SIMILARITY_SCORE > 1:
+        raise ValueError("MIN_SIMILARITY_SCORE debe estar entre 0 y 1")
+    
+    # Validar umbrales de chunking adaptativo
+    if ADAPTIVE_SIMILARITY_THRESHOLD < 0 or ADAPTIVE_SIMILARITY_THRESHOLD > 1:
+        raise ValueError("ADAPTIVE_SIMILARITY_THRESHOLD debe estar entre 0 y 1")
+
+def get_pdf_file():
+    """
+    Obtiene el primer archivo PDF encontrado en el directorio configurado.
+    
+    Returns:
+        Path: Ruta al archivo PDF
+        
+    Raises:
+        FileNotFoundError: Si no se encuentra ningún PDF
+    """
+    pdf_files = list(PDF_DIRECTORY.glob("*.pdf"))
+    if not pdf_files:
+        raise FileNotFoundError(f"No se encontraron archivos PDF en: {PDF_DIRECTORY}")
+    
+    # Retorna el primer PDF encontrado
+    return pdf_files[0]

settings_deployment.py

@@ -0,0 +1,89 @@
+# =============================================================================
+# DEPLOYMENT CONFIGURATION FOR GRADIO SPACES
+# =============================================================================
+
+import os
+from dotenv import load_dotenv
+
+# Load environment variables
+load_dotenv()
+
+# =============================================================================
+# OPENAI CONFIGURATION
+# =============================================================================
+
+# API Key (required for deployment)
+OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
+if not OPENAI_API_KEY:
+    print("⚠️  Warning: OPENAI_API_KEY not found in environment variables")
+
+# Embedding model configuration
+OPENAI_EMBEDDING_MODEL = "text-embedding-3-large"
+EMBEDDING_DIMENSIONS = 3072
+
+# =============================================================================
+# QDRANT CONFIGURATION
+# =============================================================================
+
+# Qdrant URL (cloud deployment will need external Qdrant)
+QDRANT_URL = os.getenv("QDRANT_URL", "http://localhost:6333")
+QDRANT_API_KEY = os.getenv("QDRANT_API_KEY")  # For cloud Qdrant
+
+# Collection configuration
+QDRANT_COLLECTION_NAME = "chunks_collection"
+
+# =============================================================================
+# CHUNKING CONFIGURATION
+# =============================================================================
+
+# Available chunking methods
+CHUNKING_METHODS = ["fixed", "sentence", "adaptive"]
+
+# Fixed chunking parameters
+FIXED_CHUNK_SIZE = 1000
+FIXED_CHUNK_OVERLAP = 200
+
+# Sentence chunking parameters  
+SENTENCE_MIN_CHUNK_SIZE = 100
+SENTENCE_MAX_CHUNK_SIZE = 1000
+
+# Adaptive chunking parameters
+ADAPTIVE_MIN_CHUNK_SIZE = 200
+ADAPTIVE_MAX_CHUNK_SIZE = 800
+ADAPTIVE_SIMILARITY_THRESHOLD = 0.5
+
+# =============================================================================
+# SEARCH AND RETRIEVAL CONFIGURATION
+# =============================================================================
+
+# Number of chunks to return in search results
+TOP_K_RESULTS = 10
+
+# Minimum similarity score for search results
+MIN_SIMILARITY_SCORE = 0.5
+
+# =============================================================================
+# LOGGING CONFIGURATION
+# =============================================================================
+
+# Logging level for deployment
+LOG_LEVEL = "INFO"
+
+# =============================================================================
+# DEPLOYMENT NOTES
+# =============================================================================
+
+"""
+For Gradio Spaces deployment:
+
+1. Set environment variables in Space settings:
+   - OPENAI_API_KEY: Your OpenAI API key
+   - QDRANT_URL: URL to your cloud Qdrant instance
+   - QDRANT_API_KEY: API key for cloud Qdrant (if required)
+
+2. Consider using Qdrant Cloud or self-hosted Qdrant instance
+   since Gradio Spaces don't support Docker containers
+
+3. The app will show warning messages if components fail to initialize
+   but will still launch for configuration troubleshooting
+"""

utils_logging.py

@@ -0,0 +1,258 @@
+"""
+MIT License
+Autor: Fernando Lafuente
+Proyecto: Ejercicio opcional NLP - Chunking y Recuperación
+Descripción: Configuración de sistema de logging y utilidades generales
+
+Este módulo proporciona utilidades para logging estructurado y funciones auxiliares
+que son utilizadas por múltiples componentes del sistema. Centraliza la configuración
+de logging para mantener consistencia en toda la aplicación.
+"""
+
+import logging
+import sys
+from datetime import datetime
+from typing import Any, Optional
+
+import settings
+
+
+def setup_logging(
+    level: str = settings.LOG_LEVEL,
+    format_string: str = settings.LOG_FORMAT
+) -> logging.Logger:
+    """
+    Configura el sistema de logging de la aplicación con formato estándar.
+    
+    Esta función establece un logger principal que será utilizado por todos
+    los módulos del sistema para mantener un formato consistente y facilitar
+    el debugging y monitoreo de operaciones.
+    
+    Args:
+        level: Nivel de logging (DEBUG, INFO, WARNING, ERROR)
+        format_string: Formato de los mensajes de log
+        
+    Returns:
+        logging.Logger: Logger configurado para la aplicación
+        
+    Example:
+        logger = setup_logging()
+        logger.info("Sistema de chunking inicializado correctamente")
+    """
+    # Convertir string de nivel a constante de logging
+    numeric_level = getattr(logging, level.upper(), None)
+    if not isinstance(numeric_level, int):
+        raise ValueError(f"Nivel de logging inválido: {level}")
+    
+    # Configurar formato del logger
+    formatter = logging.Formatter(format_string)
+    
+    # Configurar handler para consola
+    console_handler = logging.StreamHandler(sys.stdout)
+    console_handler.setFormatter(formatter)
+    
+    # Crear y configurar logger principal
+    logger = logging.getLogger("chunking_system")
+    logger.setLevel(numeric_level)
+    logger.addHandler(console_handler)
+    
+    # Evitar duplicación de mensajes
+    logger.propagate = False
+    
+    return logger
+
+
+def log_operation_time(func):
+    """
+    Decorador que mide y registra el tiempo de ejecución de una función.
+    
+    Este decorador es útil para monitorear el rendimiento de operaciones
+    críticas como chunking, generación de embeddings y búsquedas vectoriales.
+    
+    Args:
+        func: Función a decorar
+        
+    Returns:
+        Función decorada que registra su tiempo de ejecución
+        
+    Example:
+        @log_operation_time
+        def process_document(document):
+            # procesar documento
+            return result
+    """
+    def wrapper(*args, **kwargs):
+        logger = logging.getLogger("chunking_system")
+        start_time = datetime.now()
+        
+        logger.info(f"Iniciando operación: {func.__name__}")
+        
+        try:
+            result = func(*args, **kwargs)
+            end_time = datetime.now()
+            duration = (end_time - start_time).total_seconds()
+            
+            logger.info(f"Operación completada: {func.__name__} "
+                       f"(duración: {duration:.2f} segundos)")
+            return result
+            
+        except Exception as e:
+            end_time = datetime.now()
+            duration = (end_time - start_time).total_seconds()
+            
+            logger.error(f"Error en operación: {func.__name__} "
+                        f"(duración: {duration:.2f} segundos) - {str(e)}")
+            raise
+    
+    return wrapper
+
+
+def safe_division(numerator: float, denominator: float, default: float = 0.0) -> float:
+    """
+    Realiza división segura evitando errores por división entre cero.
+    
+    Esta función es especialmente útil al calcular métricas de similitud
+    y normalización de vectores, donde pueden aparecer casos límite.
+    
+    Args:
+        numerator: Numerador de la división
+        denominator: Denominador de la división
+        default: Valor por defecto si el denominador es cero
+        
+    Returns:
+        float: Resultado de la división o valor por defecto
+        
+    Example:
+        # Cálculo seguro de similitud coseno
+        similarity = safe_division(dot_product, norm_product, 0.0)
+    """
+    if abs(denominator) < 1e-10:  # Evitar divisiones por números muy pequeños
+        return default
+    return numerator / denominator
+
+
+def clean_text_basic(text: str) -> str:
+    """
+    Aplica limpieza básica a un texto eliminando caracteres problemáticos.
+    
+    Esta función preprocesa el texto para mejorar la calidad del chunking
+    y la generación de embeddings, eliminando caracteres que pueden introducir
+    ruido en el procesamiento posterior.
+    
+    Args:
+        text: Texto a limpiar
+        
+    Returns:
+        str: Texto limpio y normalizado
+        
+    Example:
+        texto_limpio = clean_text_basic("Texto con\n\nsaltos\t\tde línea")
+        # Resultado: "Texto con saltos de línea"
+    """
+    if not text or not isinstance(text, str):
+        return ""
+    
+    # Eliminar caracteres especificados en configuración
+    for char in settings.CHARS_TO_REMOVE:
+        text = text.replace(char, "")
+    
+    # Reemplazar caracteres especificados
+    for old_char, new_char in settings.CHARS_TO_REPLACE.items():
+        text = text.replace(old_char, new_char)
+    
+    # Normalizar espacios múltiples
+    while "  " in text:
+        text = text.replace("  ", " ")
+    
+    # Eliminar espacios al inicio y final
+    text = text.strip()
+    
+    return text
+
+
+def validate_text_length(text: str, min_length: int = settings.MIN_TEXT_LENGTH) -> bool:
+    """
+    Valida que un texto tenga longitud suficiente para ser procesado.
+    
+    Esta función evita procesar chunks demasiado cortos que no aportan
+    información semántica útil y pueden introducir ruido en la búsqueda.
+    
+    Args:
+        text: Texto a validar
+        min_length: Longitud mínima requerida
+        
+    Returns:
+        bool: True si el texto es válido, False en caso contrario
+        
+    Example:
+        if validate_text_length(chunk):
+            # Procesar chunk válido
+            embeddings = generate_embeddings(chunk)
+    """
+    if not text or not isinstance(text, str):
+        return False
+    
+    # Limpiar texto antes de validar longitud
+    clean_text = clean_text_basic(text)
+    
+    return len(clean_text) >= min_length
+
+
+def format_similarity_score(score: float, decimals: int = 4) -> str:
+    """
+    Formatea una puntuación de similitud para mostrar al usuario.
+    
+    Args:
+        score: Puntuación de similitud (0-1)
+        decimals: Número de decimales a mostrar
+        
+    Returns:
+        str: Puntuación formateada como string
+        
+    Example:
+        formatted = format_similarity_score(0.85739)
+        # Resultado: "0.8574"
+    """
+    if not isinstance(score, (int, float)):
+        return "N/A"
+    
+    return f"{score:.{decimals}f}"
+
+
+def truncate_text(text: str, max_length: int = 100, suffix: str = "...") -> str:
+    """
+    Trunca un texto a una longitud máxima agregando sufijo si es necesario.
+    
+    Útil para mostrar previews de chunks en logs o resultados sin saturar
+    la consola con textos muy largos.
+    
+    Args:
+        text: Texto a truncar
+        max_length: Longitud máxima permitida
+        suffix: Sufijo a agregar si se trunca
+        
+    Returns:
+        str: Texto truncado o original si no excede la longitud
+        
+    Example:
+        preview = truncate_text(long_chunk, 50)
+        logger.info(f"Chunk procesado: {preview}")
+    """
+    if not text or len(text) <= max_length:
+        return text
+    
+    return text[:max_length - len(suffix)] + suffix
+
+
+def get_current_timestamp() -> str:
+    """
+    Obtiene timestamp actual en formato legible para logs.
+    
+    Returns:
+        str: Timestamp formateado
+        
+    Example:
+        timestamp = get_current_timestamp()
+        # Resultado: "2024-01-15 14:30:45"
+    """
+    return datetime.now().strftime("%Y-%m-%d %H:%M:%S")