| |
| |
| """ |
| ============================================================================== |
| 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 = { |
| "min_instruction_len": 20, |
| "min_response_len": 100, |
| "max_response_len": 4096, |
| "blocked_phrases": [ |
| "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": "", |
| "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 |
|
|
| |
| import random |
| random.shuffle(valid_pairs) |
|
|
| |
| split_idx = int(len(valid_pairs) * 0.95) |
| train_pairs = valid_pairs[:split_idx] |
| val_pairs = valid_pairs[split_idx:] |
|
|
| |
| 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") |
|
|
| |
| 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") |
|
|
| |
| with open(output_path, "w", encoding="utf-8") as f: |
| for pair in valid_pairs: |
| f.write(json.dumps(pair, ensure_ascii=False) + "\n") |
|
|
| |
| 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) |
|
|
