| """ |
| Dataset Validator - checks JSONL training dataset quality. |
| |
| Validates format, structure, duplicates, length, diversity, |
| and can auto-filter to produce a clean dataset. |
| """ |
|
|
| from __future__ import annotations |
|
|
| import argparse |
| import hashlib |
| import json |
| import os |
| import re |
| import sys |
| from collections import Counter, defaultdict |
| from pathlib import Path |
| from typing import Any, Dict, List, Optional, Set, Tuple |
|
|
| _THIS_DIR = Path(__file__).resolve().parent |
| _PROJECT_ROOT = _THIS_DIR.parent |
| if str(_PROJECT_ROOT) not in sys.path: |
| sys.path.insert(0, str(_PROJECT_ROOT)) |
|
|
|
|
| |
| |
| |
|
|
| def _text_hash(text: str) -> str: |
| """SHA-256 of normalised text for exact duplicate detection.""" |
| normalised = re.sub(r"\s+", " ", text.strip().lower()) |
| return hashlib.sha256(normalised.encode("utf-8")).hexdigest() |
|
|
|
|
| def _word_set(text: str) -> Set[str]: |
| """Set of lowercase words for Jaccard similarity.""" |
| return set(re.findall(r"[a-z]{2,}", text.lower())) |
|
|
|
|
| def _jaccard_similarity(a: Set[str], b: Set[str]) -> float: |
| if not a and not b: |
| return 1.0 |
| union = a | b |
| if not union: |
| return 0.0 |
| return len(a & b) / len(union) |
|
|
|
|
| def _extract_topic_words(text: str, top_n: int = 5) -> List[str]: |
| """Extract dominant topic words from text.""" |
| stop = { |
| "the", "a", "an", "is", "are", "was", "were", "be", "been", |
| "have", "has", "had", "do", "does", "did", "will", "would", |
| "to", "of", "in", "for", "on", "with", "at", "by", "from", |
| "as", "and", "but", "or", "if", "that", "this", "what", |
| "which", "it", "its", "they", "them", "their", "not", "you", |
| "your", "can", "could", "should", "may", "might", "must", |
| "how", "why", "when", "where", "who", "whom", "about", |
| } |
| words = re.findall(r"[a-z]{3,}", text.lower()) |
| filtered = [w for w in words if w not in stop] |
| counts = Counter(filtered) |
| return [w for w, _ in counts.most_common(top_n)] |
|
|
|
|
| |
| |
| |
|
|
| class ValidationIssue: |
| """Represents a single validation problem.""" |
|
|
| def __init__(self, line_num: int, severity: str, code: str, message: str): |
| self.line_num = line_num |
| self.severity = severity |
| self.code = code |
| self.message = message |
|
|
| def __repr__(self) -> str: |
| return f"[{self.severity.upper()}] Line {self.line_num}: {self.code} - {self.message}" |
|
|
|
|
| |
| |
| |
|
|
| class DatasetValidator: |
| """Validate and clean JSONL training datasets.""" |
|
|
| REQUIRED_ROLES = {"system", "user", "assistant"} |
|
|
| def __init__( |
| self, |
| min_response_length: int = 50, |
| max_response_length: int = 10000, |
| near_duplicate_threshold: float = 0.85, |
| ): |
| self.min_response_length = min_response_length |
| self.max_response_length = max_response_length |
| self.near_duplicate_threshold = near_duplicate_threshold |
|
|
| def validate(self, filepath: str) -> Dict[str, Any]: |
| """Validate a JSONL dataset file. |
| |
| Returns a comprehensive report dict with: |
| - statistics (total, valid, invalid, duplicate, etc.) |
| - issues list |
| - per-line validity |
| """ |
| filepath = Path(filepath) |
| if not filepath.exists(): |
| raise FileNotFoundError(f"Dataset file not found: {filepath}") |
|
|
| issues: List[ValidationIssue] = [] |
| entries: List[Dict[str, Any]] = [] |
| valid_entries: List[Dict[str, Any]] = [] |
| line_validity: List[bool] = [] |
|
|
| |
| exact_hashes: Dict[str, int] = {} |
| near_dup_sets: List[Tuple[int, Set[str]]] = [] |
|
|
| |
| stats = { |
| "total_lines": 0, |
| "valid": 0, |
| "invalid": 0, |
| "parse_errors": 0, |
| "missing_roles": 0, |
| "exact_duplicates": 0, |
| "near_duplicates": 0, |
| "too_short": 0, |
| "too_long": 0, |
| "empty_content": 0, |
| "response_lengths": [], |
| "topic_words": [], |
| } |
|
|
| with open(filepath, "r", encoding="utf-8") as f: |
| for line_num, raw_line in enumerate(f, start=1): |
| stats["total_lines"] += 1 |
| raw_line = raw_line.strip() |
|
|
| if not raw_line: |
| issues.append(ValidationIssue( |
| line_num, "warning", "EMPTY_LINE", "Empty line" |
| )) |
| line_validity.append(False) |
| stats["invalid"] += 1 |
| continue |
|
|
| |
| try: |
| entry = json.loads(raw_line) |
| except json.JSONDecodeError as e: |
| issues.append(ValidationIssue( |
| line_num, "error", "PARSE_ERROR", |
| f"Invalid JSON: {e}" |
| )) |
| line_validity.append(False) |
| stats["parse_errors"] += 1 |
| stats["invalid"] += 1 |
| continue |
|
|
| entries.append(entry) |
| entry_valid = True |
|
|
| |
| messages = entry.get("messages") |
| if not isinstance(messages, list): |
| issues.append(ValidationIssue( |
| line_num, "error", "NO_MESSAGES", |
| "Missing or invalid 'messages' field" |
| )) |
| entry_valid = False |
| stats["invalid"] += 1 |
| line_validity.append(False) |
| continue |
|
|
| |
| roles_present = set() |
| assistant_content = "" |
| user_content = "" |
| has_empty = False |
|
|
| for msg in messages: |
| role = msg.get("role", "") |
| content = msg.get("content", "") |
| roles_present.add(role) |
|
|
| if role == "assistant": |
| assistant_content = content or "" |
| elif role == "user": |
| user_content = content or "" |
|
|
| if not content or not content.strip(): |
| has_empty = True |
|
|
| missing_roles = self.REQUIRED_ROLES - roles_present |
| if missing_roles: |
| issues.append(ValidationIssue( |
| line_num, "error", "MISSING_ROLES", |
| f"Missing roles: {missing_roles}" |
| )) |
| entry_valid = False |
| stats["missing_roles"] += 1 |
|
|
| if has_empty: |
| issues.append(ValidationIssue( |
| line_num, "warning", "EMPTY_CONTENT", |
| "One or more messages have empty content" |
| )) |
| stats["empty_content"] += 1 |
|
|
| |
| resp_len = len(assistant_content.split()) |
| stats["response_lengths"].append(resp_len) |
|
|
| if resp_len < self.min_response_length: |
| issues.append(ValidationIssue( |
| line_num, "warning", "TOO_SHORT", |
| f"Assistant response too short: {resp_len} words " |
| f"(min: {self.min_response_length})" |
| )) |
| stats["too_short"] += 1 |
|
|
| if resp_len > self.max_response_length: |
| issues.append(ValidationIssue( |
| line_num, "warning", "TOO_LONG", |
| f"Assistant response too long: {resp_len} words " |
| f"(max: {self.max_response_length})" |
| )) |
| stats["too_long"] += 1 |
|
|
| |
| combined_text = user_content + " " + assistant_content |
| h = _text_hash(combined_text) |
| if h in exact_hashes: |
| issues.append(ValidationIssue( |
| line_num, "warning", "EXACT_DUPLICATE", |
| f"Exact duplicate of line {exact_hashes[h]}" |
| )) |
| stats["exact_duplicates"] += 1 |
| entry_valid = False |
| else: |
| exact_hashes[h] = line_num |
|
|
| |
| if user_content: |
| user_words = _word_set(user_content) |
| for prev_line, prev_words in near_dup_sets: |
| sim = _jaccard_similarity(user_words, prev_words) |
| if sim >= self.near_duplicate_threshold: |
| issues.append(ValidationIssue( |
| line_num, "info", "NEAR_DUPLICATE", |
| f"Near-duplicate of line {prev_line} " |
| f"(Jaccard: {sim:.3f})" |
| )) |
| stats["near_duplicates"] += 1 |
| break |
| near_dup_sets.append((line_num, user_words)) |
|
|
| |
| topic_words = _extract_topic_words(user_content + " " + assistant_content) |
| stats["topic_words"].extend(topic_words) |
|
|
| if entry_valid: |
| stats["valid"] += 1 |
| valid_entries.append(entry) |
| line_validity.append(True) |
| else: |
| stats["invalid"] += 1 |
| line_validity.append(False) |
|
|
| |
| topic_counts = Counter(stats["topic_words"]) |
| total_topics = len(set(stats["topic_words"])) |
| top_topics = topic_counts.most_common(20) |
|
|
| |
| if topic_counts: |
| top3_count = sum(c for _, c in topic_counts.most_common(3)) |
| total_count = sum(topic_counts.values()) |
| concentration = top3_count / total_count if total_count else 0 |
| else: |
| concentration = 0 |
|
|
| if concentration > 0.5: |
| top_kw = ", ".join(w for w, _ in topic_counts.most_common(3)) |
| issues.append(ValidationIssue( |
| 0, "warning", "LOW_DIVERSITY", |
| f"Dataset is concentrated on few topics ({concentration:.0%} " |
| f"in top-3: {top_kw}). Consider adding more diverse examples." |
| )) |
|
|
| |
| lengths = stats["response_lengths"] |
| length_stats = {} |
| if lengths: |
| lengths_sorted = sorted(lengths) |
| length_stats = { |
| "min": lengths_sorted[0], |
| "max": lengths_sorted[-1], |
| "mean": round(sum(lengths) / len(lengths), 1), |
| "median": lengths_sorted[len(lengths) // 2], |
| "p10": lengths_sorted[int(len(lengths) * 0.1)], |
| "p90": lengths_sorted[int(len(lengths) * 0.9)], |
| } |
|
|
| report = { |
| "filepath": str(filepath), |
| "total_lines": stats["total_lines"], |
| "valid": stats["valid"], |
| "invalid": stats["invalid"], |
| "parse_errors": stats["parse_errors"], |
| "missing_roles": stats["missing_roles"], |
| "exact_duplicates": stats["exact_duplicates"], |
| "near_duplicates": stats["near_duplicates"], |
| "too_short": stats["too_short"], |
| "too_long": stats["too_long"], |
| "empty_content": stats["empty_content"], |
| "unique_topics": total_topics, |
| "topic_concentration": round(concentration, 4), |
| "top_topics": top_topics, |
| "response_length_stats": length_stats, |
| "issues": issues, |
| "line_validity": line_validity, |
| "valid_entries": valid_entries, |
| } |
|
|
| return report |
|
|
| |
|
|
| def filter_dataset( |
| self, |
| filepath: str, |
| output_path: str, |
| remove_duplicates: bool = True, |
| remove_short: bool = True, |
| remove_long: bool = True, |
| remove_invalid: bool = True, |
| ) -> Dict[str, int]: |
| """Validate and write a cleaned dataset. |
| |
| Returns stats about the filtering. |
| """ |
| report = self.validate(filepath) |
| issues_by_line: Dict[int, List[ValidationIssue]] = defaultdict(list) |
| for issue in report["issues"]: |
| issues_by_line[issue.line_num].append(issue) |
|
|
| kept = 0 |
| removed = 0 |
| reasons: Dict[str, int] = defaultdict(int) |
|
|
| os.makedirs(os.path.dirname(output_path) or ".", exist_ok=True) |
|
|
| with open(filepath, "r", encoding="utf-8") as fin, \ |
| open(output_path, "w", encoding="utf-8") as fout: |
|
|
| seen_hashes: Set[str] = set() |
|
|
| for line_num, raw_line in enumerate(fin, start=1): |
| raw_line = raw_line.strip() |
| if not raw_line: |
| removed += 1 |
| reasons["empty_line"] += 1 |
| continue |
|
|
| try: |
| entry = json.loads(raw_line) |
| except json.JSONDecodeError: |
| if remove_invalid: |
| removed += 1 |
| reasons["parse_error"] += 1 |
| continue |
|
|
| messages = entry.get("messages", []) |
| if not isinstance(messages, list): |
| if remove_invalid: |
| removed += 1 |
| reasons["no_messages"] += 1 |
| continue |
|
|
| roles = {m.get("role") for m in messages} |
| if self.REQUIRED_ROLES - roles: |
| if remove_invalid: |
| removed += 1 |
| reasons["missing_roles"] += 1 |
| continue |
|
|
| |
| assistant_text = "" |
| user_text = "" |
| for m in messages: |
| if m.get("role") == "assistant": |
| assistant_text = m.get("content", "") |
| elif m.get("role") == "user": |
| user_text = m.get("content", "") |
|
|
| |
| word_count = len(assistant_text.split()) |
| if remove_short and word_count < self.min_response_length: |
| removed += 1 |
| reasons["too_short"] += 1 |
| continue |
| if remove_long and word_count > self.max_response_length: |
| removed += 1 |
| reasons["too_long"] += 1 |
| continue |
|
|
| |
| if remove_duplicates: |
| h = _text_hash(user_text + " " + assistant_text) |
| if h in seen_hashes: |
| removed += 1 |
| reasons["duplicate"] += 1 |
| continue |
| seen_hashes.add(h) |
|
|
| fout.write(json.dumps(entry, ensure_ascii=False) + "\n") |
| kept += 1 |
|
|
| return { |
| "input_lines": report["total_lines"], |
| "kept": kept, |
| "removed": removed, |
| "removal_reasons": dict(reasons), |
| } |
|
|
| |
|
|
| def format_report(self, report: Dict[str, Any]) -> str: |
| """Format validation report as readable text.""" |
| lines: List[str] = [] |
| lines.append("=" * 70) |
| lines.append(" DATASET VALIDATION REPORT") |
| lines.append("=" * 70) |
| lines.append(f" File: {report['filepath']}") |
| lines.append("") |
|
|
| |
| lines.append("-" * 70) |
| lines.append(" SUMMARY") |
| lines.append("-" * 70) |
| lines.append(f" Total lines: {report['total_lines']}") |
| lines.append(f" Valid: {report['valid']}") |
| lines.append(f" Invalid: {report['invalid']}") |
| lines.append(f" Parse errors: {report['parse_errors']}") |
| lines.append(f" Missing roles: {report['missing_roles']}") |
| lines.append(f" Exact duplicates: {report['exact_duplicates']}") |
| lines.append(f" Near duplicates: {report['near_duplicates']}") |
| lines.append(f" Too short: {report['too_short']}") |
| lines.append(f" Too long: {report['too_long']}") |
| lines.append(f" Empty content: {report['empty_content']}") |
|
|
| |
| ls = report.get("response_length_stats", {}) |
| if ls: |
| lines.append("") |
| lines.append("-" * 70) |
| lines.append(" RESPONSE LENGTH (words)") |
| lines.append("-" * 70) |
| lines.append(f" Min: {ls.get('min', 'N/A')}") |
| lines.append(f" Max: {ls.get('max', 'N/A')}") |
| lines.append(f" Mean: {ls.get('mean', 'N/A')}") |
| lines.append(f" Median: {ls.get('median', 'N/A')}") |
| lines.append(f" P10: {ls.get('p10', 'N/A')}") |
| lines.append(f" P90: {ls.get('p90', 'N/A')}") |
|
|
| |
| lines.append("") |
| lines.append("-" * 70) |
| lines.append(" TOPIC DIVERSITY") |
| lines.append("-" * 70) |
| lines.append(f" Unique topic words: {report.get('unique_topics', 0)}") |
| lines.append(f" Top-3 concentration: {report.get('topic_concentration', 0):.1%}") |
| top_topics = report.get("top_topics", []) |
| if top_topics: |
| lines.append(" Top topics:") |
| for word, count in top_topics[:10]: |
| lines.append(f" {word:<20s} {count}") |
|
|
| |
| issues = report.get("issues", []) |
| error_issues = [i for i in issues if i.severity == "error"] |
| warning_issues = [i for i in issues if i.severity == "warning"] |
|
|
| if error_issues: |
| lines.append("") |
| lines.append("-" * 70) |
| lines.append(f" ERRORS ({len(error_issues)})") |
| lines.append("-" * 70) |
| for issue in error_issues[:20]: |
| lines.append(f" {issue}") |
| if len(error_issues) > 20: |
| lines.append(f" ... and {len(error_issues) - 20} more errors") |
|
|
| if warning_issues: |
| lines.append("") |
| lines.append("-" * 70) |
| lines.append(f" WARNINGS ({len(warning_issues)})") |
| lines.append("-" * 70) |
| for issue in warning_issues[:20]: |
| lines.append(f" {issue}") |
| if len(warning_issues) > 20: |
| lines.append(f" ... and {len(warning_issues) - 20} more warnings") |
|
|
| |
| lines.append("") |
| lines.append("-" * 70) |
| if (report["invalid"] == 0 |
| and report["exact_duplicates"] == 0 |
| and report.get("near_duplicates", 0) == 0 |
| and report.get("too_short", 0) == 0 |
| and report.get("empty_content", 0) == 0): |
| lines.append(" VERDICT: PASS - Dataset is clean") |
| elif report["invalid"] > report["total_lines"] * 0.1: |
| lines.append(" VERDICT: FAIL - Too many invalid entries (>10%)") |
| else: |
| lines.append(" VERDICT: WARN - Some issues found, consider filtering") |
| lines.append("-" * 70) |
|
|
| lines.append("=" * 70) |
| return "\n".join(lines) |
|
|
|
|
| |
| |
| |
|
|
| def main() -> None: |
| parser = argparse.ArgumentParser( |
| description="Codette Dataset Validator - check and clean JSONL training data" |
| ) |
| parser.add_argument( |
| "dataset", |
| help="Path to JSONL dataset file", |
| ) |
| parser.add_argument( |
| "--filter", "-f", |
| metavar="OUTPUT", |
| default=None, |
| help="Auto-filter and write clean dataset to OUTPUT path", |
| ) |
| parser.add_argument( |
| "--min-length", |
| type=int, |
| default=50, |
| help="Minimum assistant response length in words (default: 50)", |
| ) |
| parser.add_argument( |
| "--max-length", |
| type=int, |
| default=10000, |
| help="Maximum assistant response length in words (default: 10000)", |
| ) |
| parser.add_argument( |
| "--duplicate-threshold", |
| type=float, |
| default=0.85, |
| help="Jaccard similarity threshold for near-duplicates (default: 0.85)", |
| ) |
| parser.add_argument( |
| "--json-report", |
| metavar="PATH", |
| default=None, |
| help="Save report as JSON to this path", |
| ) |
|
|
| args = parser.parse_args() |
|
|
| validator = DatasetValidator( |
| min_response_length=args.min_length, |
| max_response_length=args.max_length, |
| near_duplicate_threshold=args.duplicate_threshold, |
| ) |
|
|
| print(f"Validating: {args.dataset}\n") |
| report = validator.validate(args.dataset) |
| print(validator.format_report(report)) |
|
|
| if args.json_report: |
| |
| save_report = {k: v for k, v in report.items() |
| if k not in ("issues", "line_validity", "valid_entries")} |
| save_report["issue_count"] = len(report["issues"]) |
| save_report["issues_summary"] = [repr(i) for i in report["issues"][:50]] |
| os.makedirs(os.path.dirname(args.json_report) or ".", exist_ok=True) |
| with open(args.json_report, "w", encoding="utf-8") as f: |
| json.dump(save_report, f, indent=2, default=str) |
| print(f"\nJSON report saved to: {args.json_report}") |
|
|
| if args.filter: |
| print(f"\nFiltering dataset -> {args.filter}") |
| filter_stats = validator.filter_dataset(args.dataset, args.filter) |
| print(f" Input lines: {filter_stats['input_lines']}") |
| print(f" Kept: {filter_stats['kept']}") |
| print(f" Removed: {filter_stats['removed']}") |
| for reason, count in filter_stats["removal_reasons"].items(): |
| print(f" - {reason}: {count}") |
|
|
|
|
| if __name__ == "__main__": |
| main() |
|
|
