Spaces:
Running
Running
File size: 2,598 Bytes
a3aa0f2 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 | #!/usr/bin/env python3
"""
Validate Thinking Dial think_rank distribution in training data.
Checks:
1. All think_rank values are in [0, 3]
2. Distribution is not degenerate (all same value)
3. Each rank has >= 5% representation
4. No duplicate prompts across different ranks
Usage:
python scripts/validate_think_rank.py data/mini_data.json
Author: Zhu Zizhan
Project: Fusion-LLM
License: Apache 2.0
"""
import json
import sys
from collections import Counter
from pathlib import Path
def validate_think_rank(data_path: str) -> bool:
"""Validate think_rank distribution in a dataset."""
with open(data_path, 'r', encoding='utf-8') as f:
data = json.load(f)
total = len(data)
ranks = [item.get('think_rank', -1) for item in data]
counter = Counter(ranks)
print(f"Dataset: {data_path}")
print(f"Total samples: {total}")
print(f"Think rank distribution: {dict(sorted(counter.items()))}")
issues = []
# Check 1: All ranks in valid range
invalid = [r for r in ranks if r not in (0, 1, 2, 3)]
if invalid:
issues.append(f"Invalid think_rank values: {Counter(invalid)}")
# Check 2: Not degenerate
if len(counter) <= 1:
issues.append(f"Degenerate distribution - only rank {list(counter.keys())}")
# Check 3: Each rank >= 5%
for rank in range(4):
pct = counter.get(rank, 0) / total * 100
if pct > 0 and pct < 5:
issues.append(f"Rank {rank} underrepresented: {pct:.1f}% (need >=5%)")
# Check 4: No same prompt with different ranks
prompt_ranks = {}
for item in data:
p = item.get('prompt', '')
r = item.get('think_rank', -1)
if p in prompt_ranks and prompt_ranks[p] != r:
issues.append(f"Prompt '{p[:30]}...' has conflicting ranks: {prompt_ranks[p]} vs {r}")
prompt_ranks[p] = r
# Summary
if issues:
print(f"\nISSUES FOUND ({len(issues)}):")
for issue in issues:
print(f" - {issue}")
return False
else:
print(f"\nAll checks passed!")
# Print distribution visualization
print("\nDistribution:")
for rank in range(4):
count = counter.get(rank, 0)
pct = count / total * 100
bar = '#' * int(pct / 2)
print(f" Rank {rank}: {count:3d} ({pct:5.1f}%) {bar}")
return True
if __name__ == '__main__':
path = sys.argv[1] if len(sys.argv) > 1 else 'data/mini_data.json'
success = validate_think_rank(path)
sys.exit(0 if success else 1)
|