Ethio-BBPE / scripts /bbpe_trainer.py
Nexuss0781's picture
Upload scripts/bbpe_trainer.py with huggingface_hub
6611338 verified
Raw
History Blame
27.7 kB
#!/usr/bin/env python3
"""
EthioBBPE: Production-Ready Byte-Level BPE Tokenizer Trainer
Advanced Features:
- Checkpointing with metadata and resume capability
- Multi-format compression (gzip, lzma, bz2)
- Model quantization for deployment
- Training metrics tracking
- Automatic backups
- Model validation utilities
- Multiple export formats
"""
import os
import json
import gzip
import bz2
import lzma
import shutil
import hashlib
import logging
from pathlib import Path
from typing import List, Optional, Union, Dict, Any, Tuple
from dataclasses import dataclass, field, asdict
from datetime import datetime
import tempfile
import struct
from tokenizers import ByteLevelBPETokenizer, trainers, Tokenizer
from tokenizers.implementations import BaseTokenizer
# Configure logging
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger("EthioBBPE")
@dataclass
class BBPEConfig:
"""Configuration for EthioBBPE training."""
vocab_size: int = 30000
min_frequency: int = 2
show_progress: bool = True
special_tokens: List[str] = field(default_factory=lambda: ["<pad>", "<unk>", "<s>", "</s>"])
lowercase: bool = False
dropout: Optional[float] = None
# File paths
data_dir: str = "./data"
model_save_dir: str = "./models"
model_name: str = "EthioBBPE"
# Advanced features
use_checkpoint: bool = True
checkpoint_dir: str = "./models/checkpoints"
save_compressed: bool = True
compression_format: str = "gzip" # Options: gzip, bz2, lzma
compression_level: int = 9 # 1-9, higher = better compression but slower
checkpoint_steps: Optional[int] = None
num_threads: int = -1 # -1 for auto
# Backup and versioning
enable_backup: bool = True
max_checkpoints: int = 5 # Keep only last N checkpoints
# Quantization (for deployment optimization)
enable_quantization: bool = False
quantization_bits: int = 8 # 8-bit or 4-bit quantization
def save(self, path: str):
"""Save configuration to JSON."""
with open(path, 'w', encoding='utf-8') as f:
json.dump(asdict(self), f, indent=2)
logger.info(f"Configuration saved to {path}")
@classmethod
def load(cls, path: str) -> "BBPEConfig":
"""Load configuration from JSON."""
with open(path, 'r', encoding='utf-8') as f:
data = json.load(f)
return cls(**data)
@dataclass
class CheckpointMetadata:
"""Metadata for checkpoint management."""
checkpoint_id: str
timestamp: str
vocab_size: int
training_step: int
is_final: bool
checksum: str
config: Dict[str, Any]
metrics: Optional[Dict[str, float]] = None
def to_dict(self) -> Dict[str, Any]:
return asdict(self)
@classmethod
def from_dict(cls, data: Dict[str, Any]) -> "CheckpointMetadata":
return cls(**data)
class EthioBBPETrainer:
"""
Production-ready trainer for Byte-Level BPE with advanced features:
- Checkpointing with metadata tracking
- Multi-format compression (gzip, bz2, lzma)
- Model quantization for deployment
- Automatic backup management
- Training metrics collection
"""
def __init__(self, config: BBPEConfig = None):
self.config = config or BBPEConfig()
self.output_dir = Path(self.config.model_save_dir)
self.checkpoint_dir = Path(self.config.checkpoint_dir)
self.tokenizer = None
self.is_trained = False
self.training_metrics = {}
self.start_time = None
# Create directories
self.output_dir.mkdir(parents=True, exist_ok=True)
self.checkpoint_dir.mkdir(parents=True, exist_ok=True)
logger.info(f"Initialized EthioBBPETrainer with output dir: {self.output_dir}")
logger.info(f"Compression format: {self.config.compression_format}, Level: {self.config.compression_level}")
if self.config.enable_quantization:
logger.info(f"Quantization enabled: {self.config.quantization_bits}-bit")
def _initialize_tokenizer(self):
"""Initialize the ByteLevelBPETokenizer."""
self.tokenizer = ByteLevelBPETokenizer(
add_prefix_space=False,
trim_offsets=True,
lowercase=self.config.lowercase
)
logger.info("Tokenizer initialized")
def train(self, files: Union[str, List[str]] = None, use_checkpoint: bool = None):
"""
Train the tokenizer on a list of files or a directory.
Args:
files: Path to a file, list of files, or directory containing text files.
If None, uses files from config.data_dir
use_checkpoint: If True, attempts to resume from the latest checkpoint.
Defaults to config.use_checkpoint
Returns:
Trained tokenizer object
"""
self.start_time = datetime.now()
if self.tokenizer is None:
self._initialize_tokenizer()
# Use config default if not specified
use_checkpoint = use_checkpoint if use_checkpoint is not None else self.config.use_checkpoint
# Resolve file paths
if files is None:
# Use data_dir from config
data_path = Path(self.config.data_dir)
if data_path.is_dir():
file_paths = [str(f) for f in data_path.glob("**/*.txt")]
file_paths.extend([str(f) for f in data_path.glob("**/*.jsonl")])
file_paths.extend([str(f) for f in data_path.glob("**/*.json")])
else:
raise FileNotFoundError(f"Data directory not found: {self.config.data_dir}")
elif isinstance(files, str):
path = Path(files)
if path.is_dir():
file_paths = [str(f) for f in path.glob("**/*.txt")]
file_paths.extend([str(f) for f in path.glob("**/*.jsonl")])
file_paths.extend([str(f) for f in path.glob("**/*.json")])
else:
file_paths = [str(path)]
else:
file_paths = files
if not file_paths:
raise ValueError("No valid training files found.")
logger.info(f"Found {len(file_paths)} files for training.")
logger.info(f"Training started at: {self.start_time}")
# Checkpoint logic
start_from_scratch = True
resumed_from = None
if use_checkpoint:
latest_ckpt = self._get_latest_checkpoint()
if latest_ckpt:
logger.info(f"Resuming from checkpoint: {latest_ckpt}")
self.tokenizer = Tokenizer.from_file(str(latest_ckpt))
start_from_scratch = False
resumed_from = str(latest_ckpt)
else:
logger.info("No checkpoint found. Starting from scratch.")
# Calculate initial metrics
total_files = len(file_paths)
total_size = sum(Path(f).stat().st_size for f in file_paths)
self.training_metrics['initial'] = {
'num_files': total_files,
'total_bytes': total_size,
'resumed_from': resumed_from
}
# Train
logger.info("Starting training...")
train_start = datetime.now()
# ByteLevelBPETokenizer.train() accepts parameters directly, not a trainer object
self.tokenizer.train(
files=file_paths,
vocab_size=self.config.vocab_size,
min_frequency=self.config.min_frequency,
special_tokens=self.config.special_tokens,
show_progress=self.config.show_progress
)
train_end = datetime.now()
train_duration = (train_end - train_start).total_seconds()
self.is_trained = True
logger.info("Training completed successfully.")
# Record final metrics
vocab = self.tokenizer.get_vocab()
self.training_metrics['final'] = {
'vocab_size': len(vocab),
'training_duration_sec': train_duration,
'end_time': train_end.isoformat()
}
logger.info(f"Final vocabulary size: {len(vocab)}")
logger.info(f"Training duration: {train_duration:.2f} seconds")
# Auto-save checkpoint after training
self._save_checkpoint("final_pre_compress")
return self.tokenizer
def _get_latest_checkpoint(self) -> Optional[Path]:
"""Find the latest checkpoint file."""
ckpts = list(self.checkpoint_dir.glob("checkpoint_*.json"))
if not ckpts:
return None
# Sort by modification time
ckpts.sort(key=lambda p: p.stat().st_mtime, reverse=True)
return ckpts[0]
def _save_checkpoint(self, name: str = "latest", save_metadata: bool = True):
"""
Save current tokenizer state to checkpoint with metadata.
Args:
name: Checkpoint name identifier
save_metadata: Whether to save metadata JSON alongside checkpoint
"""
if self.tokenizer is None:
return
ckpt_path = self.checkpoint_dir / f"checkpoint_{name}.json"
self.tokenizer.save(str(ckpt_path))
# Calculate checksum for integrity verification
checksum = self._calculate_file_checksum(ckpt_path)
# Save metadata if requested
if save_metadata:
vocab = self.tokenizer.get_vocab()
metadata = CheckpointMetadata(
checkpoint_id=f"{name}_{datetime.now().strftime('%Y%m%d_%H%M%S')}",
timestamp=datetime.now().isoformat(),
vocab_size=len(vocab),
training_step=-1, # Not used in HF tokenizers but kept for compatibility
is_final=(name == "final_pre_compress"),
checksum=checksum,
config=asdict(self.config),
metrics=self.training_metrics.copy()
)
metadata_path = self.checkpoint_dir / f"checkpoint_{name}_metadata.json"
with open(metadata_path, 'w', encoding='utf-8') as f:
json.dump(metadata.to_dict(), f, indent=2)
logger.info(f"Checkpoint metadata saved to {metadata_path}")
# Manage checkpoint rotation (keep only last N checkpoints)
if self.config.enable_backup and self.config.max_checkpoints > 0:
self._rotate_checkpoints()
logger.info(f"Checkpoint saved to {ckpt_path} (checksum: {checksum[:8]}...)")
def _calculate_file_checksum(self, filepath: Path) -> str:
"""Calculate SHA256 checksum of a file."""
sha256_hash = hashlib.sha256()
with open(filepath, "rb") as f:
for byte_block in iter(lambda: f.read(4096), b""):
sha256_hash.update(byte_block)
return sha256_hash.hexdigest()
def _rotate_checkpoints(self):
"""Keep only the most recent checkpoints based on max_checkpoints config."""
ckpts = list(self.checkpoint_dir.glob("checkpoint_*.json"))
# Exclude metadata files
ckpts = [c for c in ckpts if "_metadata" not in c.name]
if len(ckpts) > self.config.max_checkpoints:
# Sort by modification time, oldest first
ckpts.sort(key=lambda p: p.stat().st_mtime)
# Remove oldest checkpoints
num_to_remove = len(ckpts) - self.config.max_checkpoints
for i in range(num_to_remove):
old_ckpt = ckpts[i]
old_metadata = old_ckpt.parent / f"{old_ckpt.stem}_metadata.json"
try:
old_ckpt.unlink()
logger.info(f"Removed old checkpoint: {old_ckpt.name}")
if old_metadata.exists():
old_metadata.unlink()
logger.info(f"Removed old metadata: {old_metadata.name}")
except Exception as e:
logger.warning(f"Failed to remove old checkpoint {old_ckpt}: {e}")
def save(self, model_name: str = None, compress: bool = None,
compression_format: str = None, export_formats: List[str] = None):
"""
Save the trained tokenizer with advanced options.
Args:
model_name: Name of the model folder. Defaults to config.model_name
compress: If True, saves vocab in compressed format.
Defaults to config.save_compressed.
compression_format: Format for compression ('gzip', 'bz2', 'lzma').
Defaults to config.compression_format.
export_formats: List of export formats to generate.
Options: ['tokenizer.json', 'vocab_compressed', 'quantized', 'hf_export']
Defaults to all available formats.
Returns:
Path to saved model directory
"""
if not self.is_trained and self.tokenizer is None:
raise RuntimeError("Tokenizer not trained yet.")
model_name = model_name or self.config.model_name
compress = compress if compress is not None else self.config.save_compressed
compression_format = compression_format or self.config.compression_format
model_path = self.output_dir / model_name
model_path.mkdir(parents=True, exist_ok=True)
logger.info(f"Saving model to {model_path} (compressed={compress}, format={compression_format})...")
# Determine which formats to export
if export_formats is None:
export_formats = ['tokenizer.json', 'vocab_compressed', 'hf_export']
# Always save standard tokenizer.json (required for HF loading)
if 'tokenizer.json' in export_formats:
tokenizer_file = model_path / "tokenizer.json"
self.tokenizer.save(str(tokenizer_file))
logger.info(f"Standard tokenizer.json saved: {tokenizer_file}")
# Save compressed vocab if requested
if compress and 'vocab_compressed' in export_formats:
self._save_compressed_vocab(model_path, compression_format)
# Save quantized version if enabled
if self.config.enable_quantization and 'quantized' in export_formats:
self._save_quantized_model(model_path)
# Generate Hugging Face export files
if 'hf_export' in export_formats:
self._save_hf_export_files(model_path)
# Save config
self.config.save(str(model_path / "config.json"))
# Save training metrics
self._save_training_metrics(model_path)
# Save metadata card for Hugging Face
self._save_model_card(model_path)
# Log size comparison
self._log_size_comparison(model_path)
logger.info(f"Model successfully saved to {model_path}")
return model_path
def _get_compression_handler(self, format: str):
"""Get the appropriate compression handler based on format."""
handlers = {
'gzip': (gzip.open, '.gz', 'wt'),
'bz2': (bz2.open, '.bz2', 'wt'),
'lzma': (lzma.open, '.xz', 'wt')
}
if format not in handlers:
logger.warning(f"Unknown compression format '{format}', falling back to gzip")
format = 'gzip'
return handlers[format]
def _save_compressed_vocab(self, model_path: Path, format: str = 'gzip'):
"""Save vocabulary in compressed format."""
vocab = self.tokenizer.get_vocab()
open_func, ext, mode = self._get_compression_handler(format)
vocab_path = model_path / f"vocab.json{ext}"
# Use compression level from config
if format == 'gzip':
with open_func(vocab_path, mode, encoding='utf-8', compresslevel=self.config.compression_level) as f:
json.dump(vocab, f)
elif format == 'bz2':
with open_func(vocab_path, mode, encoding='utf-8') as f:
json.dump(vocab, f)
elif format == 'lzma':
with open_func(vocab_path, mode, encoding='utf-8', preset=self.config.compression_level) as f:
json.dump(vocab, f)
logger.info(f"Compressed vocab saved ({format}): {vocab_path}")
# Calculate and log compression ratio
original_size = sum(len(k) + len(str(v)) for k, v in vocab.items())
compressed_size = vocab_path.stat().st_size
ratio = (1 - compressed_size / (original_size * 2)) * 100 if original_size > 0 else 0
logger.info(f"Compression ratio: {ratio:.1f}% (original ~{original_size*2} bytes -> {compressed_size} bytes)")
def _save_quantized_model(self, model_path: Path):
"""Save a quantized version of the tokenizer for deployment."""
vocab = self.tokenizer.get_vocab()
# Create quantized vocabulary mapping
if self.config.quantization_bits == 8:
# 8-bit quantization: map tokens to uint8 range
dtype = 'uint8'
max_val = 255
elif self.config.quantization_bits == 4:
# 4-bit quantization: pack two tokens per byte
dtype = 'uint4'
max_val = 15
else:
logger.warning(f"Unsupported quantization bits: {self.config.quantization_bits}, using 8-bit")
dtype = 'uint8'
max_val = 255
# Sort vocab by ID for consistent ordering
sorted_vocab = sorted(vocab.items(), key=lambda x: x[1])
# Create quantized lookup table
quantized_data = {
'vocab_size': len(sorted_vocab),
'quantization_bits': self.config.quantization_bits,
'dtype': dtype,
'tokens': [token for token, _ in sorted_vocab],
'ids': [idx for _, idx in sorted_vocab]
}
# Save quantized model
quantized_path = model_path / f"tokenizer_quantized_{self.config.quantization_bits}bit.json.gz"
with gzip.open(quantized_path, 'wt', encoding='utf-8', compresslevel=9) as f:
json.dump(quantized_data, f)
logger.info(f"Quantized model saved ({self.config.quantization_bits}-bit): {quantized_path}")
def _save_hf_export_files(self, model_path: Path):
"""Save files needed for Hugging Face Hub export."""
# Save merges.txt if available (for BPE models)
try:
merges = self.tokenizer.model.get_merges()
if merges:
merges_path = model_path / "merges.txt"
with open(merges_path, 'w', encoding='utf-8') as f:
f.write("#version: 0.2\n")
for merge in merges:
f.write(f"{merge[0]} {merge[1]}\n")
logger.info(f"Merges file saved: {merges_path}")
except Exception as e:
logger.debug(f"No merges available or error saving: {e}")
# Save special tokens mapping
special_tokens_path = model_path / "special_tokens_map.json"
special_tokens = {
"unk_token": "<unk>",
"pad_token": "<pad>",
"bos_token": "<s>",
"eos_token": "</s>"
}
with open(special_tokens_path, 'w', encoding='utf-8') as f:
json.dump(special_tokens, f, indent=2)
logger.info(f"Special tokens map saved: {special_tokens_path}")
def _save_training_metrics(self, model_path: Path):
"""Save training metrics to JSON."""
metrics_path = model_path / "training_metrics.json"
# Add additional metrics
full_metrics = {
**self.training_metrics,
'config': asdict(self.config),
'saved_at': datetime.now().isoformat()
}
with open(metrics_path, 'w', encoding='utf-8') as f:
json.dump(full_metrics, f, indent=2)
logger.info(f"Training metrics saved: {metrics_path}")
def _log_size_comparison(self, model_path: Path):
"""Log size comparison between different saved formats."""
files_info = []
for f in model_path.iterdir():
if f.is_file():
size_kb = f.stat().st_size / 1024
files_info.append((f.name, size_kb))
files_info.sort(key=lambda x: x[1])
logger.info("Model artifacts size summary:")
for name, size in files_info:
logger.info(f" {name}: {size:.2f} KB")
def _save_model_card(self, path: Path):
"""Generate and save a README.md for Hugging Face Hub."""
card_content = f"""---
language:
- multilingual
tags:
- ethiobbpe
- bpe
- tokenizer
- byte-level
license: apache-2.0
datasets:
- user-provided
---
# EthioBBPE Tokenizer
This is a production-ready Byte-Level BPE tokenizer with advanced features for deployment.
## Features
- **Byte-Level**: Handles any Unicode character without <UNK>.
- **Multi-format Compression**: Supports gzip, bz2, and lzma compression.
- **Checkpointing**: Built-in safety checkpoints with metadata tracking.
- **Quantization**: Optional 8-bit/4-bit quantization for efficient deployment.
- **Training Metrics**: Comprehensive metrics tracking and logging.
- **Automatic Backup**: Checkpoint rotation to manage disk space.
## Usage
### Transformers
```python
from transformers import AutoTokenizer
tokenizer = AutoTokenizer.from_pretrained("{path.name}")
```
### Tokenizers Library
```python
from tokenizers import Tokenizer
tokenizer = Tokenizer.from_file("tokenizer.json")
```
### Loading Compressed Vocab
```python
import gzip
import json
# Load compressed vocabulary
with gzip.open("vocab.json.gz", 'rt', encoding='utf-8') as f:
vocab = json.load(f)
```
## Training Configuration
```json
{json.dumps(asdict(self.config), indent=2)}
```
## Model Files
- `tokenizer.json`: Standard tokenizer file (required)
- `vocab.json.gz`: Compressed vocabulary (optional, smaller size)
- `config.json`: Training configuration
- `training_metrics.json`: Training statistics
- `special_tokens_map.json`: Special tokens mapping
- `README.md`: This file
## Checkpoints
Checkpoints are saved in the `{self.checkpoint_dir}` directory with metadata including:
- Checkpoint ID and timestamp
- Vocabulary size
- SHA256 checksum for integrity verification
- Training metrics at checkpoint time
"""
with open(path / "README.md", 'w', encoding='utf-8') as f:
f.write(card_content)
def tokenize(self, text: str) -> List[str]:
if self.tokenizer is None:
raise RuntimeError("Tokenizer not initialized")
return self.tokenizer.encode(text).tokens
def encode(self, text: str) -> List[int]:
if self.tokenizer is None:
raise RuntimeError("Tokenizer not initialized")
return self.tokenizer.encode(text).ids
def decode(self, ids: List[int]) -> str:
if self.tokenizer is None:
raise RuntimeError("Tokenizer not initialized")
return self.tokenizer.decode(ids)
def get_vocab_size(self) -> int:
"""Get the current vocabulary size."""
if self.tokenizer is None:
return 0
return len(self.tokenizer.get_vocab())
def get_training_metrics(self) -> Dict[str, Any]:
"""Get training metrics collected during training."""
return self.training_metrics.copy()
def validate_checkpoint(self, checkpoint_path: Union[str, Path]) -> bool:
"""
Validate checkpoint integrity using checksum.
Args:
checkpoint_path: Path to checkpoint JSON file
Returns:
True if valid, False otherwise
"""
checkpoint_path = Path(checkpoint_path)
metadata_path = checkpoint_path.parent / f"{checkpoint_path.stem}_metadata.json"
if not metadata_path.exists():
logger.warning(f"No metadata found for checkpoint: {checkpoint_path}")
return True # Can't validate without metadata
try:
with open(metadata_path, 'r', encoding='utf-8') as f:
metadata = json.load(f)
expected_checksum = metadata.get('checksum')
if not expected_checksum:
logger.warning("No checksum in metadata")
return True
actual_checksum = self._calculate_file_checksum(checkpoint_path)
is_valid = expected_checksum == actual_checksum
if is_valid:
logger.info(f"✓ Checkpoint validated: {checkpoint_path.name}")
else:
logger.error(f"✗ Checkpoint validation FAILED: {checkpoint_path.name}")
logger.error(f" Expected: {expected_checksum}")
logger.error(f" Actual: {actual_checksum}")
return is_valid
except Exception as e:
logger.error(f"Error validating checkpoint: {e}")
return False
def list_checkpoints(self) -> List[Dict[str, Any]]:
"""
List all available checkpoints with their metadata.
Returns:
List of checkpoint info dictionaries
"""
checkpoints = []
ckpt_files = sorted(self.checkpoint_dir.glob("checkpoint_*.json"))
for ckpt_file in ckpt_files:
if "_metadata" in ckpt_file.name:
continue
info = {
'path': str(ckpt_file),
'name': ckpt_file.name,
'size_kb': ckpt_file.stat().st_size / 1024,
'modified': datetime.fromtimestamp(ckpt_file.stat().st_mtime).isoformat()
}
# Try to load metadata
metadata_file = ckpt_file.parent / f"{ckpt_file.stem}_metadata.json"
if metadata_file.exists():
try:
with open(metadata_file, 'r', encoding='utf-8') as f:
metadata = json.load(f)
info['metadata'] = metadata
info['vocab_size'] = metadata.get('vocab_size', 'N/A')
info['is_final'] = metadata.get('is_final', False)
except Exception as e:
info['error'] = str(e)
checkpoints.append(info)
return checkpoints
@staticmethod
def load_compressed_vocab(vocab_path: Union[str, Path]) -> Dict[str, int]:
"""
Load vocabulary from a compressed file.
Args:
vocab_path: Path to compressed vocab file (.gz, .bz2, .xz)
Returns:
Vocabulary dictionary
"""
vocab_path = Path(vocab_path)
if vocab_path.suffix == '.gz':
with gzip.open(vocab_path, 'rt', encoding='utf-8') as f:
return json.load(f)
elif vocab_path.suffix == '.bz2':
with bz2.open(vocab_path, 'rt', encoding='utf-8') as f:
return json.load(f)
elif vocab_path.suffix in ['.xz', '.lzma']:
with lzma.open(vocab_path, 'rt', encoding='utf-8') as f:
return json.load(f)
else:
# Assume uncompressed JSON
with open(vocab_path, 'r', encoding='utf-8') as f:
return json.load(f)