Módulo 8: Proyecto Final Integrador - RAG System Completo

Document Ingestion Pipeline

Descripción

En esta cápsula implementarás el primer componente del sistema RAG production-ready: el Document Ingestion Pipeline. Este sistema carga documentos de múltiples formatos (TXT, MD, JSON, PDF), extrae metadata automáticamente, y prepara los datos para el chunking pipeline del Módulo 8.

Este componente es crítico porque la calidad del RAG depende directamente de la calidad de la ingestion. Un loader robusto maneja edge cases (archivos vacíos, encoding incorrecto, formatos corruptos) y extrae metadata útil para filtrado y ranking posterior.

Al final de esta cápsula tendrás un DocumentLoader production-ready que podrás reutilizar en cualquier proyecto RAG.

Objetivos

Al completar esta cápsula, serás capaz de:

✅ Implementar loaders para TXT, MD, JSON, y PDF
✅ Extraer metadata automáticamente (título, autor, fecha, word count)
✅ Manejar errores de encoding y formatos corruptos
✅ Cargar directorios completos recursivamente
✅ Validar documentos cargados
✅ Implementar logging estructurado

Arquitectura del componente

Document Ingestion Pipeline
│
├── DocumentLoader (clase principal)
│   ├── load(file_path) → Document
│   ├── load_directory(dir_path) → List[Document]
│   └── validate(document) → bool
│
├── Format Loaders (métodos privados)
│   ├── _load_txt()
│   ├── _load_markdown()
│   ├── _load_json()
│   └── _load_pdf()
│
└── MetadataExtractor (clase auxiliar)
    ├── extract_basic(text) → Dict
    ├── extract_from_filename(path) → Dict
    └── estimate_tokens(text) → int

Paso 1: Setup del entorno

1.1: Dependencias

pip install pypdf2 python-dotenv

Por qué:

pypdf2: Para cargar archivos PDF
python-dotenv: Para configuración (aunque este componente no usa API keys)

1.2: Estructura de archivos

rag-system/
├── src/
│   ├── ingestion/
│   │   ├── __init__.py
│   │   ├── document_loader.py      # Loader principal
│   │   └── metadata_extractor.py   # Extractor de metadata
│   └── utils/
│       └── logger.py                # Logging estructurado
├── data/
│   └── documents/                   # Corpus para testing
│       ├── file1.txt
│       ├── file2.md
│       ├── file3.json
│       └── file4.pdf
└── tests/
    └── test_document_loader.py

Paso 2: Implementar DocumentLoader

2.1: Clase base con detección de formato

Crear src/ingestion/document_loader.py:

"""
Document Loader - RAG System
Carga documentos de múltiples formatos con metadata extraction
"""

from pathlib import Path
from typing import List, Dict, Optional
import json
import logging

# PDF support (opcional, instalar con: pip install pypdf2)
try:
    import PyPDF2
    PDF_SUPPORT = True
except ImportError:
    PDF_SUPPORT = False
    logging.warning("PyPDF2 not installed. PDF support disabled.")


class Document:
    """
    Representación de un documento cargado
    
    Attributes:
        text: Contenido del documento
        source: Path al archivo original
        format: Formato del archivo (txt, md, json, pdf)
        metadata: Metadata adicional (título, fecha, etc.)
    """
    
    def __init__(self, text: str, source: str, format: str, metadata: Optional[Dict] = None):
        self.text = text
        self.source = source
        self.format = format
        self.metadata = metadata or {}
        
        # Auto-generar metadata básico si no existe
        if 'word_count' not in self.metadata:
            self.metadata['word_count'] = len(text.split())
        if 'char_count' not in self.metadata:
            self.metadata['char_count'] = len(text)
    
    def to_dict(self) -> Dict:
        """Convertir a diccionario para serialización"""
        return {
            'text': self.text,
            'source': self.source,
            'format': self.format,
            'metadata': self.metadata
        }
    
    def __repr__(self) -> str:
        return f"Document(source='{self.source}', format='{self.format}', words={self.metadata.get('word_count', 0)})"


class DocumentLoader:
    """
    Carga documentos de múltiples formatos con error handling robusto
    
    Formatos soportados:
    - TXT: Plain text
    - MD: Markdown (extrae título automáticamente)
    - JSON: {"text": "...", "metadata": {...}}
    - PDF: Extrae texto de PDFs (requiere PyPDF2)
    
    Example:
        loader = DocumentLoader()
        
        # Cargar archivo individual
        doc = loader.load("document.txt")
        
        # Cargar directorio completo
        docs = loader.load_directory("./data/documents")
    """
    
    # Formatos soportados
    SUPPORTED_FORMATS = {'.txt', '.md', '.json'}
    
    def __init__(self, encoding: str = 'utf-8', errors: str = 'replace'):
        """
        Inicializar DocumentLoader
        
        Args:
            encoding: Encoding para archivos de texto (default: utf-8)
            errors: Cómo manejar errores de encoding (replace/ignore/strict)
        """
        self.encoding = encoding
        self.errors = errors
        
        # Agregar PDF si está disponible
        if PDF_SUPPORT:
            self.SUPPORTED_FORMATS.add('.pdf')
        
        # Setup logging
        self.logger = logging.getLogger(__name__)
    
    def load(self, file_path: str) -> Document:
        """
        Cargar un documento desde archivo
        
        Args:
            file_path: Path al archivo
        
        Returns:
            Document object
        
        Raises:
            FileNotFoundError: Si archivo no existe
            ValueError: Si formato no soportado
        """
        path = Path(file_path)
        
        # Validar que archivo existe
        if not path.exists():
            raise FileNotFoundError(f"File not found: {file_path}")
        
        # Validar que es un archivo (no directorio)
        if not path.is_file():
            raise ValueError(f"Path is not a file: {file_path}")
        
        # Detectar formato
        suffix = path.suffix.lower()
        
        if suffix not in self.SUPPORTED_FORMATS:
            raise ValueError(
                f"Unsupported format: {suffix}. "
                f"Supported: {', '.join(self.SUPPORTED_FORMATS)}"
            )
        
        # Delegar a loader específico
        if suffix == '.txt':
            return self._load_txt(path)
        elif suffix == '.md':
            return self._load_markdown(path)
        elif suffix == '.json':
            return self._load_json(path)
        elif suffix == '.pdf':
            return self._load_pdf(path)
        else:
            raise ValueError(f"No loader for format: {suffix}")
    
    def _load_txt(self, path: Path) -> Document:
        """
        Cargar archivo de texto plano
        
        Args:
            path: Path al archivo .txt
        
        Returns:
            Document object
        """
        try:
            with open(path, 'r', encoding=self.encoding, errors=self.errors) as f:
                text = f.read()
            
            # Validar que no está vacío
            if not text.strip():
                self.logger.warning(f"Empty file: {path}")
            
            # Metadata básico
            metadata = {
                'filename': path.name,
                'extension': path.suffix
            }
            
            return Document(
                text=text,
                source=str(path.absolute()),
                format='txt',
                metadata=metadata
            )
        
        except UnicodeDecodeError as e:
            self.logger.error(f"Encoding error in {path}: {e}")
            raise
        except Exception as e:
            self.logger.error(f"Error loading {path}: {e}")
            raise
    
    def _load_markdown(self, path: Path) -> Document:
        """
        Cargar archivo Markdown y extraer título
        
        Args:
            path: Path al archivo .md
        
        Returns:
            Document object con título extraído
        """
        try:
            with open(path, 'r', encoding=self.encoding, errors=self.errors) as f:
                text = f.read()
            
            # Extraer título (primer header # encontrado)
            title = self._extract_markdown_title(text)
            
            # Metadata con título
            metadata = {
                'filename': path.name,
                'extension': path.suffix,
                'title': title
            }
            
            return Document(
                text=text,
                source=str(path.absolute()),
                format='markdown',
                metadata=metadata
            )
        
        except Exception as e:
            self.logger.error(f"Error loading markdown {path}: {e}")
            raise
    
    @staticmethod
    def _extract_markdown_title(content: str) -> str:
        """
        Extraer título de Markdown (primer # header)
        
        Args:
            content: Contenido del archivo MD
        
        Returns:
            Título extraído o 'Untitled'
        """
        lines = content.split('\n')
        
        for line in lines:
            # Buscar línea que empiece con #
            if line.strip().startswith('#'):
                # Remover # y espacios
                title = line.strip().lstrip('#').strip()
                if title:
                    return title
        
        return 'Untitled'
    
    def _load_json(self, path: Path) -> Document:
        """
        Cargar archivo JSON estructurado
        
        Expected format:
        {
            "text": "contenido del documento",
            "metadata": {
                "title": "...",
                "author": "...",
                ...
            }
        }
        
        Args:
            path: Path al archivo .json
        
        Returns:
            Document object
        """
        try:
            with open(path, 'r', encoding=self.encoding) as f:
                data = json.load(f)
            
            # Validar estructura
            if 'text' not in data:
                raise ValueError(f"JSON missing 'text' field in {path}")
            
            text = data['text']
            
            # Metadata: combinar metadata del JSON + metadata básico
            metadata = data.get('metadata', {})
            metadata['filename'] = path.name
            metadata['extension'] = path.suffix
            
            return Document(
                text=text,
                source=str(path.absolute()),
                format='json',
                metadata=metadata
            )
        
        except json.JSONDecodeError as e:
            self.logger.error(f"Invalid JSON in {path}: {e}")
            raise ValueError(f"Invalid JSON format: {e}")
        except Exception as e:
            self.logger.error(f"Error loading JSON {path}: {e}")
            raise
    
    def _load_pdf(self, path: Path) -> Document:
        """
        Cargar archivo PDF y extraer texto
        
        Args:
            path: Path al archivo .pdf
        
        Returns:
            Document object con texto extraído
        
        Raises:
            ImportError: Si PyPDF2 no está instalado
        """
        if not PDF_SUPPORT:
            raise ImportError(
                "PDF support requires PyPDF2. Install with: pip install pypdf2"
            )
        
        try:
            # Abrir PDF
            with open(path, 'rb') as f:
                pdf_reader = PyPDF2.PdfReader(f)
                
                # Validar que tiene páginas
                num_pages = len(pdf_reader.pages)
                if num_pages == 0:
                    raise ValueError(f"PDF has no pages: {path}")
                
                # Extraer texto de todas las páginas
                text_parts = []
                for page_num in range(num_pages):
                    page = pdf_reader.pages[page_num]
                    text_parts.append(page.extract_text())
                
                text = '\n\n'.join(text_parts)
                
                # Metadata del PDF
                pdf_info = pdf_reader.metadata or {}
                metadata = {
                    'filename': path.name,
                    'extension': path.suffix,
                    'num_pages': num_pages,
                    'title': pdf_info.get('/Title', 'Untitled'),
                    'author': pdf_info.get('/Author', 'Unknown'),
                    'creator': pdf_info.get('/Creator', 'Unknown')
                }
                
                return Document(
                    text=text,
                    source=str(path.absolute()),
                    format='pdf',
                    metadata=metadata
                )
        
        except Exception as e:
            self.logger.error(f"Error loading PDF {path}: {e}")
            raise
    
    def load_directory(self, dir_path: str, recursive: bool = True) -> List[Document]:
        """
        Cargar todos los documentos de un directorio
        
        Args:
            dir_path: Path al directorio
            recursive: Si True, busca en subdirectorios
        
        Returns:
            Lista de Documents
        """
        path = Path(dir_path)
        
        if not path.exists():
            raise FileNotFoundError(f"Directory not found: {dir_path}")
        
        if not path.is_dir():
            raise ValueError(f"Path is not a directory: {dir_path}")
        
        documents = []
        
        # Glob pattern según recursive
        pattern = '**/*' if recursive else '*'
        
        for file_path in path.glob(pattern):
            # Solo archivos con formato soportado
            if file_path.is_file() and file_path.suffix.lower() in self.SUPPORTED_FORMATS:
                try:
                    doc = self.load(str(file_path))
                    documents.append(doc)
                    self.logger.info(f"Loaded: {file_path.name}")
                
                except Exception as e:
                    self.logger.warning(f"Skipping {file_path}: {e}")
        
        self.logger.info(f"Loaded {len(documents)} documents from {dir_path}")
        return documents
    
    def validate(self, document: Document) -> bool:
        """
        Validar que documento está bien formado
        
        Args:
            document: Document a validar
        
        Returns:
            True si válido, False si no
        """
        # Validar que tiene texto
        if not document.text or not document.text.strip():
            self.logger.warning(f"Document has no text: {document.source}")
            return False
        
        # Validar longitud mínima (ej: 10 caracteres)
        if len(document.text) < 10:
            self.logger.warning(f"Document too short: {document.source}")
            return False
        
        # Validar metadata
        if not document.metadata:
            self.logger.warning(f"Document missing metadata: {document.source}")
            return False
        
        return True


# Demo de uso
if __name__ == "__main__":
    # Setup logging
    logging.basicConfig(level=logging.INFO)
    
    # Crear loader
    loader = DocumentLoader()
    
    # Cargar directorio de ejemplo
    try:
        docs = loader.load_directory("./data/documents")
        print(f"\n✅ Loaded {len(docs)} documents")
        
        # Mostrar info de cada documento
        for doc in docs:
            print(f"\n{doc}")
            print(f"  Words: {doc.metadata['word_count']}")
            print(f"  Preview: {doc.text[:100]}...")
    
    except Exception as e:
        print(f"❌ Error: {e}")

Paso 3: Metadata Extractor avanzado

Crear src/ingestion/metadata_extractor.py:

"""
Metadata Extractor - RAG System
Extrae metadata útil de documentos para filtrado y ranking
"""

from typing import Dict
import re
from datetime import datetime


class MetadataExtractor:
    """
    Extrae metadata adicional de documentos
    
    Metadata extraído:
    - Token count estimado (para cost estimation)
    - Language detection (básico)
    - Tipo de contenido (código, prosa, mixto)
    - Fecha de creación/modificación
    """
    
    # Tokens por word (aproximado para OpenAI)
    TOKENS_PER_WORD = 1.3
    
    def extract(self, document) -> Dict:
        """
        Extraer todo el metadata disponible
        
        Args:
            document: Document object
        
        Returns:
            Dict con metadata completo
        """
        text = document.text
        
        return {
            **document.metadata,
            'estimated_tokens': self.estimate_tokens(text),
            'has_code': self.detect_code(text),
            'content_type': self.classify_content(text),
            'extracted_at': datetime.now().isoformat()
        }
    
    def estimate_tokens(self, text: str) -> int:
        """
        Estimar cantidad de tokens (OpenAI tokenization)
        
        Args:
            text: Texto a analizar
        
        Returns:
            Cantidad estimada de tokens
        """
        word_count = len(text.split())
        return int(word_count * self.TOKENS_PER_WORD)
    
    def detect_code(self, text: str) -> bool:
        """
        Detectar si el documento contiene código
        
        Args:
            text: Texto a analizar
        
        Returns:
            True si contiene bloques de código
        """
        # Detectar bloques de código markdown
        if '```' in text:
            return True
        
        # Detectar patterns de código comunes
        code_patterns = [
            r'def\s+\w+\(',      # Python functions
            r'function\s+\w+\(', # JS functions
            r'class\s+\w+',      # Class definitions
            r'import\s+\w+',     # Imports
            r'from\s+\w+\s+import' # Python imports
        ]
        
        for pattern in code_patterns:
            if re.search(pattern, text):
                return True
        
        return False
    
    def classify_content(self, text: str) -> str:
        """
        Clasificar tipo de contenido
        
        Args:
            text: Texto a analizar
        
        Returns:
            'code', 'documentation', 'prose', 'mixed'
        """
        has_code = self.detect_code(text)
        
        # Detectar si es documentation (muchos headers)
        header_count = len(re.findall(r'^#+\s', text, re.MULTILINE))
        is_documentation = header_count > 3
        
        if has_code and is_documentation:
            return 'mixed'  # Documentación con ejemplos de código
        elif has_code:
            return 'code'
        elif is_documentation:
            return 'documentation'
        else:
            return 'prose'


# Demo
if __name__ == "__main__":
    from document_loader import Document
    
    # Crear documento de ejemplo
    doc = Document(
        text="""
        # Python Tutorial
        
        ```python
        def hello():
            print("Hello World")
        ```
        """,
        source="example.md",
        format="markdown"
    )
    
    # Extraer metadata
    extractor = MetadataExtractor()
    metadata = extractor.extract(doc)
    
    print("Metadata extraído:")
    for key, value in metadata.items():
        print(f"  {key}: {value}")

Troubleshooting

Problema 1: UnicodeDecodeError

Causa: Archivo tiene encoding diferente a UTF-8

Solución:

# Opción 1: Usar errors='replace' (reemplaza caracteres inválidos)
loader = DocumentLoader(encoding='utf-8', errors='replace')

# Opción 2: Detectar encoding automáticamente
import chardet

with open(file_path, 'rb') as f:
    raw_data = f.read()
    detected = chardet.detect(raw_data)
    encoding = detected['encoding']

with open(file_path, 'r', encoding=encoding) as f:
    text = f.read()

Problema 2: PDF extraction vacío

Causa: PDF es imagen escaneada (no tiene texto extraíble)

Solución:

# Opción 1: Usar OCR (pytesseract + pdf2image)
from pdf2image import convert_from_path
import pytesseract

images = convert_from_path(pdf_path)
text = '\n'.join([pytesseract.image_to_string(img) for img in images])

# Opción 2: Validar que PDF tiene texto
text = pdf_reader.pages[0].extract_text()
if not text.strip():
    raise ValueError("PDF has no extractable text (might be scanned)")

Problema 3: FileNotFoundError con paths relativos

Causa: Working directory diferente al esperado

Solución:

# Siempre usar paths absolutos
from pathlib import Path

# Opción 1: Resolver path relativo a absoluto
abs_path = Path(file_path).resolve()

# Opción 2: Path relativo a script location
script_dir = Path(__file__).parent
data_path = script_dir / 'data' / 'documents'

Tests unitarios

Crear tests/test_document_loader.py:

import pytest
from src.ingestion.document_loader import DocumentLoader, Document

def test_load_txt():
    """Test cargar archivo TXT"""
    loader = DocumentLoader()
    doc = loader.load("data/test.txt")
    
    assert isinstance(doc, Document)
    assert doc.format == 'txt'
    assert len(doc.text) > 0

def test_load_directory():
    """Test cargar directorio"""
    loader = DocumentLoader()
    docs = loader.load_directory("data/documents")
    
    assert len(docs) > 0
    assert all(isinstance(d, Document) for d in docs)

def test_validate():
    """Test validación de documentos"""
    loader = DocumentLoader()
    
    # Documento válido
    valid_doc = Document("Contenido válido", "test.txt", "txt")
    assert loader.validate(valid_doc) == True
    
    # Documento vacío
    empty_doc = Document("", "empty.txt", "txt")
    assert loader.validate(empty_doc) == False

Resumen

En esta cápsula implementaste:

✅ DocumentLoader con soporte para TXT, MD, JSON, PDF
✅ Metadata extraction automático (título, word count, tokens)
✅ Error handling robusto (encoding, formatos corruptos)
✅ Carga de directorios recursiva
✅ Validación de documentos
✅ Logging estructurado

Próxima cápsula: Chunking Pipeline - Implementar SmartChunker con estrategias recursivas y metadata enrichment.

Recursos Adicionales

PyPDF2 Documentation - PDF extraction library
Python pathlib - File path manipulation
Chardet - Character encoding detection
Python logging - Structured logging
pytest - Testing framework
LangChain Document Loaders - Alternativa con más formatos
LlamaIndex SimpleDirectoryReader - Loader con 40+ formatos

Módulo 8 - Cápsula 02

← AnteriorMódulo 8: Proyecto Final Integrador - RAG System Completo Siguiente →Chunking & Embedding Pipeline Integration