Module 8: Final Capstone Project - Complete RAG System
Document Ingestion Pipeline
Description
In this capsule you'll implement the first component of the production-ready RAG system: the Document Ingestion Pipeline. This system loads documents from multiple formats (TXT, MD, JSON, PDF), extracts metadata automatically, and prepares the data for the Module 8 chunking pipeline.
This component is critical because the quality of the RAG depends directly on the quality of the ingestion. A robust loader handles edge cases (empty files, wrong encoding, corrupt formats) and extracts useful metadata for later filtering and ranking.
By the end of this capsule you'll have a production-ready DocumentLoader that you can reuse in any RAG project.
Objectives
By completing this capsule, you'll be able to:
- ✅ Implement loaders for TXT, MD, JSON, and PDF
- ✅ Extract metadata automatically (title, author, date, word count)
- ✅ Handle encoding errors and corrupt formats
- ✅ Load entire directories recursively
- ✅ Validate loaded documents
- ✅ Implement structured logging
Component architecture
Document Ingestion Pipeline
│
├── DocumentLoader (main class)
│ ├── load(file_path) → Document
│ ├── load_directory(dir_path) → List[Document]
│ └── validate(document) → bool
│
├── Format Loaders (private methods)
│ ├── _load_txt()
│ ├── _load_markdown()
│ ├── _load_json()
│ └── _load_pdf()
│
└── MetadataExtractor (helper class)
├── extract_basic(text) → Dict
├── extract_from_filename(path) → Dict
└── estimate_tokens(text) → int
Step 1: Environment setup
1.1: Dependencies
pip install pypdf2 python-dotenvWhy:
pypdf2: To load PDF filespython-dotenv: For configuration (although this component doesn't use API keys)
1.2: File structure
rag-system/
├── src/
│ ├── ingestion/
│ │ ├── __init__.py
│ │ ├── document_loader.py # Main loader
│ │ └── metadata_extractor.py # Metadata extractor
│ └── utils/
│ └── logger.py # Structured logging
├── data/
│ └── documents/ # Corpus for testing
│ ├── file1.txt
│ ├── file2.md
│ ├── file3.json
│ └── file4.pdf
└── tests/
└── test_document_loader.pyStep 2: Implement DocumentLoader
2.1: Base class with format detection
Create src/ingestion/document_loader.py:
"""
Document Loader - RAG System
Loads documents from multiple formats with metadata extraction
"""
from pathlib import Path
from typing import List, Dict, Optional
import json
import logging
# PDF support (optional, install with: pip install pypdf2)
try:
import PyPDF2
PDF_SUPPORT = True
except ImportError:
PDF_SUPPORT = False
logging.warning("PyPDF2 not installed. PDF support disabled.")
class Document:
"""
Representation of a loaded document
Attributes:
text: Document content
source: Path to the original file
format: File format (txt, md, json, pdf)
metadata: Additional metadata (title, date, 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-generate basic metadata if it doesn't exist
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:
"""Convert to dictionary for serialization"""
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:
"""
Loads documents from multiple formats with robust error handling
Supported formats:
- TXT: Plain text
- MD: Markdown (extracts the title automatically)
- JSON: {"text": "...", "metadata": {...}}
- PDF: Extracts text from PDFs (requires PyPDF2)
Example:
loader = DocumentLoader()
# Load a single file
doc = loader.load("document.txt")
# Load an entire directory
docs = loader.load_directory("./data/documents")
"""
# Supported formats
SUPPORTED_FORMATS = {'.txt', '.md', '.json'}
def __init__(self, encoding: str = 'utf-8', errors: str = 'replace'):
"""
Initialize DocumentLoader
Args:
encoding: Encoding for text files (default: utf-8)
errors: How to handle encoding errors (replace/ignore/strict)
"""
self.encoding = encoding
self.errors = errors
# Add PDF if it's available
if PDF_SUPPORT:
self.SUPPORTED_FORMATS.add('.pdf')
# Setup logging
self.logger = logging.getLogger(__name__)
def load(self, file_path: str) -> Document:
"""
Load a document from a file
Args:
file_path: Path to the file
Returns:
Document object
Raises:
FileNotFoundError: If the file doesn't exist
ValueError: If the format is unsupported
"""
path = Path(file_path)
# Validate that the file exists
if not path.exists():
raise FileNotFoundError(f"File not found: {file_path}")
# Validate that it's a file (not a directory)
if not path.is_file():
raise ValueError(f"Path is not a file: {file_path}")
# Detect format
suffix = path.suffix.lower()
if suffix not in self.SUPPORTED_FORMATS:
raise ValueError(
f"Unsupported format: {suffix}. "
f"Supported: {', '.join(self.SUPPORTED_FORMATS)}"
)
# Delegate to the specific loader
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:
"""
Load a plain text file
Args:
path: Path to the .txt file
Returns:
Document object
"""
try:
with open(path, 'r', encoding=self.encoding, errors=self.errors) as f:
text = f.read()
# Validate that it's not empty
if not text.strip():
self.logger.warning(f"Empty file: {path}")
# Basic metadata
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:
"""
Load a Markdown file and extract the title
Args:
path: Path to the .md file
Returns:
Document object with the extracted title
"""
try:
with open(path, 'r', encoding=self.encoding, errors=self.errors) as f:
text = f.read()
# Extract the title (first # header found)
title = self._extract_markdown_title(text)
# Metadata with the title
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:
"""
Extract the Markdown title (first # header)
Args:
content: Content of the MD file
Returns:
Extracted title or 'Untitled'
"""
lines = content.split('\n')
for line in lines:
# Look for a line that starts with #
if line.strip().startswith('#'):
# Remove # and spaces
title = line.strip().lstrip('#').strip()
if title:
return title
return 'Untitled'
def _load_json(self, path: Path) -> Document:
"""
Load a structured JSON file
Expected format:
{
"text": "document content",
"metadata": {
"title": "...",
"author": "...",
...
}
}
Args:
path: Path to the .json file
Returns:
Document object
"""
try:
with open(path, 'r', encoding=self.encoding) as f:
data = json.load(f)
# Validate the structure
if 'text' not in data:
raise ValueError(f"JSON missing 'text' field in {path}")
text = data['text']
# Metadata: combine the JSON metadata + basic metadata
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:
"""
Load a PDF file and extract the text
Args:
path: Path to the .pdf file
Returns:
Document object with the extracted text
Raises:
ImportError: If PyPDF2 is not installed
"""
if not PDF_SUPPORT:
raise ImportError(
"PDF support requires PyPDF2. Install with: pip install pypdf2"
)
try:
# Open the PDF
with open(path, 'rb') as f:
pdf_reader = PyPDF2.PdfReader(f)
# Validate that it has pages
num_pages = len(pdf_reader.pages)
if num_pages == 0:
raise ValueError(f"PDF has no pages: {path}")
# Extract text from all pages
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)
# PDF metadata
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]:
"""
Load all documents from a directory
Args:
dir_path: Path to the directory
recursive: If True, searches subdirectories
Returns:
List of 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 based on recursive
pattern = '**/*' if recursive else '*'
for file_path in path.glob(pattern):
# Only files with a supported format
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:
"""
Validate that a document is well-formed
Args:
document: Document to validate
Returns:
True if valid, False otherwise
"""
# Validate that it has text
if not document.text or not document.text.strip():
self.logger.warning(f"Document has no text: {document.source}")
return False
# Validate the minimum length (e.g. 10 characters)
if len(document.text) < 10:
self.logger.warning(f"Document too short: {document.source}")
return False
# Validate the metadata
if not document.metadata:
self.logger.warning(f"Document missing metadata: {document.source}")
return False
return True
# Usage demo
if __name__ == "__main__":
# Setup logging
logging.basicConfig(level=logging.INFO)
# Create the loader
loader = DocumentLoader()
# Load the example directory
try:
docs = loader.load_directory("./data/documents")
print(f"\n✅ Loaded {len(docs)} documents")
# Show info for each document
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}")Step 3: Advanced Metadata Extractor
Create src/ingestion/metadata_extractor.py:
"""
Metadata Extractor - RAG System
Extracts useful metadata from documents for filtering and ranking
"""
from typing import Dict
import re
from datetime import datetime
class MetadataExtractor:
"""
Extracts additional metadata from documents
Extracted metadata:
- Estimated token count (for cost estimation)
- Language detection (basic)
- Content type (code, prose, mixed)
- Creation/modification date
"""
# Tokens per word (approximate for OpenAI)
TOKENS_PER_WORD = 1.3
def extract(self, document) -> Dict:
"""
Extract all available metadata
Args:
document: Document object
Returns:
Dict with the complete metadata
"""
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:
"""
Estimate the number of tokens (OpenAI tokenization)
Args:
text: Text to analyze
Returns:
Estimated number of tokens
"""
word_count = len(text.split())
return int(word_count * self.TOKENS_PER_WORD)
def detect_code(self, text: str) -> bool:
"""
Detect whether the document contains code
Args:
text: Text to analyze
Returns:
True if it contains code blocks
"""
# Detect markdown code blocks
if '```' in text:
return True
# Detect common code patterns
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:
"""
Classify the content type
Args:
text: Text to analyze
Returns:
'code', 'documentation', 'prose', 'mixed'
"""
has_code = self.detect_code(text)
# Detect whether it's documentation (many headers)
header_count = len(re.findall(r'^#+\s', text, re.MULTILINE))
is_documentation = header_count > 3
if has_code and is_documentation:
return 'mixed' # Documentation with code examples
elif has_code:
return 'code'
elif is_documentation:
return 'documentation'
else:
return 'prose'
# Demo
if __name__ == "__main__":
from document_loader import Document
# Create an example document
doc = Document(
text="""
# Python Tutorial
```python
def hello():
print("Hello World")
```
""",
source="example.md",
format="markdown"
)
# Extract metadata
extractor = MetadataExtractor()
metadata = extractor.extract(doc)
print("Extracted metadata:")
for key, value in metadata.items():
print(f" {key}: {value}")Troubleshooting
Problem 1: UnicodeDecodeError
Cause: The file has an encoding other than UTF-8
Solution:
# Option 1: Use errors='replace' (replaces invalid characters)
loader = DocumentLoader(encoding='utf-8', errors='replace')
# Option 2: Detect the encoding automatically
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()Problem 2: Empty PDF extraction
Cause: The PDF is a scanned image (has no extractable text)
Solution:
# Option 1: Use 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])
# Option 2: Validate that the PDF has text
text = pdf_reader.pages[0].extract_text()
if not text.strip():
raise ValueError("PDF has no extractable text (might be scanned)")Problem 3: FileNotFoundError with relative paths
Cause: The working directory is different from what's expected
Solution:
# Always use absolute paths
from pathlib import Path
# Option 1: Resolve a relative path to an absolute one
abs_path = Path(file_path).resolve()
# Option 2: Path relative to the script location
script_dir = Path(__file__).parent
data_path = script_dir / 'data' / 'documents'Unit tests
Create tests/test_document_loader.py:
import pytest
from src.ingestion.document_loader import DocumentLoader, Document
def test_load_txt():
"""Test loading a TXT file"""
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 loading a directory"""
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 document validation"""
loader = DocumentLoader()
# Valid document
valid_doc = Document("Valid content", "test.txt", "txt")
assert loader.validate(valid_doc) == True
# Empty document
empty_doc = Document("", "empty.txt", "txt")
assert loader.validate(empty_doc) == FalseSummary
In this capsule you implemented:
- ✅
DocumentLoaderwith support for TXT, MD, JSON, PDF - ✅ Automatic metadata extraction (title, word count, tokens)
- ✅ Robust error handling (encoding, corrupt formats)
- ✅ Recursive directory loading
- ✅ Document validation
- ✅ Structured logging
Next capsule: Chunking Pipeline - Implement SmartChunker with recursive strategies and metadata enrichment.
Additional Resources
- PyPDF2 Documentation - PDF extraction library
- Python pathlib - File path manipulation
- Chardet - Character encoding detection
- Python logging - Structured logging
- pytest - Testing framework
- LangChain Document Loaders - Alternative with more formats
- LlamaIndex SimpleDirectoryReader - Loader with 40+ formats
Module 8 - Capsule 02