1#!/usr/bin/env python3
2"""Validate DPO (Direct Preference Optimization) JSONL files for Azure AI Foundry.
3 
4Adapted from foundry-ft agent with additional checks:
5- Identical preferred/non_preferred detection
6- DPO overtraining risk (small dataset warning)
7"""
8import json
9import sys
10 
11 
12 
13try:
14    sys.stdout.reconfigure(encoding="utf-8")
15    sys.stderr.reconfigure(encoding="utf-8")
16except (AttributeError, OSError):
17    pass  # Stream not reconfigurable (older Python or non-tty); default encoding is fine
18def validate_dpo(filepath: str) -> None:
19    errors = []
20    warnings = []
21    total = 0
22 
23    with open(filepath, "r", encoding="utf-8") as f:
24        for line_num, line in enumerate(f, 1):
25            line = line.strip()
26            if not line:
27                continue
28            total += 1
29 
30            try:
31                record = json.loads(line)
32            except json.JSONDecodeError as e:
33                errors.append(f"Line {line_num}: Invalid JSON — {e}")
34                continue
35 
36            for field in ["input", "preferred_output", "non_preferred_output"]:
37                if field not in record:
38                    errors.append(f"Line {line_num}: Missing '{field}' field")
39 
40            if "input" not in record:
41                continue
42 
43            inp = record["input"]
44            if "messages" not in inp:
45                errors.append(f"Line {line_num}: 'input' missing 'messages' field")
46            else:
47                msgs = inp["messages"]
48                if not any(m.get("role") == "user" for m in msgs):
49                    errors.append(f"Line {line_num}: 'input.messages' has no 'user' message")
50 
51            for output_field in ["preferred_output", "non_preferred_output"]:
52                if output_field in record:
53                    out = record[output_field]
54                    if not isinstance(out, list) or len(out) == 0:
55                        errors.append(f"Line {line_num}: '{output_field}' must be a non-empty array")
56                    elif not any(m.get("role") == "assistant" for m in out):
57                        errors.append(f"Line {line_num}: '{output_field}' has no 'assistant' message")
58 
59            if "preferred_output" in record and "non_preferred_output" in record:
60                pref = json.dumps(record["preferred_output"], sort_keys=True)
61                non_pref = json.dumps(record["non_preferred_output"], sort_keys=True)
62                if pref == non_pref:
63                    warnings.append(f"Line {line_num}: preferred and non_preferred outputs are identical")
64 
65    print(f"\n{'='*60}")
66    print(f"DPO Validation Report: {filepath}")
67    print(f"{'='*60}")
68    print(f"Total records: {total}")
69    print(f"Errors: {len(errors)}")
70    print(f"Warnings: {len(warnings)}")
71 
72    # DPO-specific guidance from our experiments
73    if total < 500 and total > 0:
74        print(f"\n⚠️  DPO tip: With {total} pairs, use n_epochs=1-2 max (Azure defaults to 3, which causes overtraining on small datasets).")
75    if total > 0:
76        print(f"\n💡 DPO tip: If your base model already scores >9/10 on this task, DPO may hurt more than help.")
77 
78    if errors:
79        print(f"\n❌ ERRORS (must fix):")
80        for e in errors[:20]:
81            print(f"  • {e}")
82        if len(errors) > 20:
83            print(f"  ... and {len(errors) - 20} more errors")
84 
85    if warnings:
86        print(f"\n⚠️  WARNINGS:")
87        for w in warnings[:10]:
88            print(f"  • {w}")
89 
90    if not errors:
91        print(f"\n✅ Data is valid for DPO fine-tuning!")
92    else:
93        print(f"\n❌ Fix {len(errors)} error(s) before submitting.")
94        sys.exit(1)
95 
96 
97if __name__ == "__main__":
98    if len(sys.argv) != 2:
99        print("Usage: python validate_dpo.py <path-to-jsonl>")
100        sys.exit(1)
101    validate_dpo(sys.argv[1])
102