#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ ============================================================================== Manara-3B | Step 1b: Dataset Preprocessing & Validation ============================================================================== This script validates, cleans, and formats the raw synthetic dataset into the Alpaca instruction format required for LoRA fine-tuning. Output Format (Alpaca): { "instruction": "...", "input": "", "output": "...", "domain": "...", "language": "ar" | "en" } Usage: python preprocess_dataset.py \ --input_file ../../data/processed/manara_train.jsonl \ --output_file ../../data/processed/manara_alpaca.jsonl \ --min_instruction_len 20 \ --min_response_len 100 ============================================================================== """ import json import argparse import logging from pathlib import Path from collections import Counter logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s") logger = logging.getLogger(__name__) # --------------------------------------------------------------------------- # Quality Filters # --------------------------------------------------------------------------- QUALITY_FILTERS = { "min_instruction_len": 20, # Minimum characters in instruction "min_response_len": 100, # Minimum characters in response "max_response_len": 4096, # Maximum characters in response "blocked_phrases": [ # Phrases indicating low-quality or off-topic responses "I cannot", "I'm unable", "As an AI", "I don't know", "لا أستطيع", "لا أعرف", "كذكاء اصطناعي", ], } def is_valid_pair(pair: dict, filters: dict) -> tuple[bool, str]: """Validates a single instruction-response pair against quality filters.""" instruction = pair.get("instruction", "") response = pair.get("response", "") if len(instruction) < filters["min_instruction_len"]: return False, f"Instruction too short ({len(instruction)} chars)" if len(response) < filters["min_response_len"]: return False, f"Response too short ({len(response)} chars)" if len(response) > filters["max_response_len"]: return False, f"Response too long ({len(response)} chars)" for phrase in filters["blocked_phrases"]: if phrase.lower() in response.lower(): return False, f"Blocked phrase found: '{phrase}'" return True, "OK" def convert_to_alpaca_format(pair: dict) -> dict: """Converts a raw pair to the Alpaca instruction format.""" return { "instruction": pair["instruction"], "input": "", # No additional context input for this dataset "output": pair["response"], "domain": pair.get("domain", "unknown"), "language": pair.get("language", "ar"), "id": pair.get("id", ""), "metadata": pair.get("metadata", {}), } def preprocess( input_file: str, output_file: str, filters: dict, ) -> dict: """ Main preprocessing function. Returns a dictionary of statistics about the preprocessing run. """ input_path = Path(input_file) output_path = Path(output_file) output_path.parent.mkdir(parents=True, exist_ok=True) stats = { "total_input": 0, "total_valid": 0, "total_filtered": 0, "filter_reasons": Counter(), "by_domain": Counter(), "by_language": Counter(), } valid_pairs = [] logger.info(f"Reading input file: {input_path}") with open(input_path, "r", encoding="utf-8") as f: for line in f: line = line.strip() if not line: continue stats["total_input"] += 1 try: pair = json.loads(line) except json.JSONDecodeError as e: logger.warning(f"Skipping malformed JSON line: {e}") stats["total_filtered"] += 1 stats["filter_reasons"]["malformed_json"] += 1 continue is_valid, reason = is_valid_pair(pair, filters) if is_valid: alpaca_pair = convert_to_alpaca_format(pair) valid_pairs.append(alpaca_pair) stats["total_valid"] += 1 stats["by_domain"][pair.get("domain", "unknown")] += 1 stats["by_language"][pair.get("language", "unknown")] += 1 else: stats["total_filtered"] += 1 stats["filter_reasons"][reason] += 1 # Shuffle the dataset for training import random random.shuffle(valid_pairs) # Split into train/validation sets (95/5 split) split_idx = int(len(valid_pairs) * 0.95) train_pairs = valid_pairs[:split_idx] val_pairs = valid_pairs[split_idx:] # Save train set train_path = output_path.parent / "manara_train_alpaca.jsonl" with open(train_path, "w", encoding="utf-8") as f: for pair in train_pairs: f.write(json.dumps(pair, ensure_ascii=False) + "\n") # Save validation set val_path = output_path.parent / "manara_val_alpaca.jsonl" with open(val_path, "w", encoding="utf-8") as f: for pair in val_pairs: f.write(json.dumps(pair, ensure_ascii=False) + "\n") # Save full processed set with open(output_path, "w", encoding="utf-8") as f: for pair in valid_pairs: f.write(json.dumps(pair, ensure_ascii=False) + "\n") # Save stats stats_path = output_path.parent / "preprocessing_stats.json" with open(stats_path, "w", encoding="utf-8") as f: json.dump( { **stats, "filter_reasons": dict(stats["filter_reasons"]), "by_domain": dict(stats["by_domain"]), "by_language": dict(stats["by_language"]), "train_size": len(train_pairs), "val_size": len(val_pairs), }, f, indent=2, ensure_ascii=False, ) logger.info("=" * 60) logger.info("Preprocessing complete.") logger.info(f" Total input : {stats['total_input']}") logger.info(f" Total valid : {stats['total_valid']}") logger.info(f" Total filtered: {stats['total_filtered']}") logger.info(f" Train size : {len(train_pairs)}") logger.info(f" Val size : {len(val_pairs)}") logger.info(f" By domain : {dict(stats['by_domain'])}") logger.info(f" By language : {dict(stats['by_language'])}") logger.info("=" * 60) return stats def parse_args(): parser = argparse.ArgumentParser(description="Manara-3B: Dataset Preprocessing") parser.add_argument("--input_file", required=True, help="Path to raw JSONL dataset") parser.add_argument("--output_file", required=True, help="Path for processed JSONL output") parser.add_argument("--min_instruction_len", type=int, default=20) parser.add_argument("--min_response_len", type=int, default=100) parser.add_argument("--max_response_len", type=int, default=4096) return parser.parse_args() if __name__ == "__main__": args = parse_args() filters = { "min_instruction_len": args.min_instruction_len, "min_response_len": args.min_response_len, "max_response_len": args.max_response_len, "blocked_phrases": QUALITY_FILTERS["blocked_phrases"], } preprocess(args.input_file, args.output_file, filters)