#!/usr/bin/env python3 """Validate a chat-format JSONL dataset before you upload it for fine-tuning or evals. Expected line shape: {"messages": [{"role": "system", "content": "..."}, {"role": "user", "content": "..."}, {"role": "assistant", "content": "..."}]} Usage: python3 scripts/validate_chat_jsonl.py DATA.jsonl [--min-examples N] [--max-tokens N] [--chars-per-token R] [--allow-roles system,user,assistant] [--json-report report.json] Errors (make the file fail): - line is not valid JSON, or not an object with a "messages" list - message is not an object, role not in the allowed set, content not a non-empty string - "system" appears anywhere other than the first position - no user message, or no assistant message - the last message is not from the assistant - two consecutive messages share a role (user,user or assistant,assistant) - fewer than --min-examples valid examples (default 10) Warnings (reported, do not fail): - exact duplicate examples - example longer than --max-tokens (default 16000) by a chars/token estimate. This estimate is approximate; check your provider's real limit. - assistant content that contains "As an AI" style boilerplate you may not want to train in Content given as a list of parts (multimodal format) is accepted if every part is an object with a "type" key; text parts are length-checked. Requirements: Python 3.8+, no packages, no environment variables. Exit codes: 0 valid (warnings may be printed) 1 one or more errors 2 usage error (file not found) """ import argparse import hashlib import json import math import re import sys BOILERPLATE = re.compile(r"(?i)\bas an ai(?: language model)?\b|\bi am an ai\b|\bi'm just an ai\b") def content_text(content): if isinstance(content, str): return content if isinstance(content, list) and all(isinstance(p, dict) and "type" in p for p in content): return "".join(p.get("text", "") for p in content if isinstance(p.get("text"), str)) return None def check_example(obj, roles): errors = [] if not isinstance(obj, dict) or not isinstance(obj.get("messages"), list) or not obj["messages"]: return ["expected an object with a non-empty 'messages' list"], "" messages = obj["messages"] texts = [] previous = None for index, message in enumerate(messages): where = "messages[%d]" % index if not isinstance(message, dict): errors.append("%s is not an object" % where) continue role = message.get("role") if role not in roles: errors.append("%s has role %r (allowed: %s)" % (where, role, ", ".join(sorted(roles)))) if role == "system" and index != 0: errors.append("%s: system message must be first" % where) text = content_text(message.get("content")) if text is None: errors.append("%s content must be a string or a list of typed parts" % where) elif not text.strip() and isinstance(message.get("content"), str): errors.append("%s content is empty" % where) else: texts.append(text) if role == previous and role in ("user", "assistant"): errors.append("%s repeats role %r from the previous message" % (where, role)) previous = role role_list = [m.get("role") for m in messages if isinstance(m, dict)] if "user" not in role_list: errors.append("no user message") if "assistant" not in role_list: errors.append("no assistant message") if role_list and role_list[-1] != "assistant": errors.append("last message must be from the assistant") return errors, "\n".join(texts) def main(): parser = argparse.ArgumentParser(description="Validate chat-format JSONL datasets.") parser.add_argument("path") parser.add_argument("--min-examples", type=int, default=10) parser.add_argument("--max-tokens", type=int, default=16000) parser.add_argument("--chars-per-token", type=float, default=4.0) parser.add_argument("--allow-roles", default="system,user,assistant") parser.add_argument("--json-report") args = parser.parse_args() roles = {r.strip() for r in args.allow_roles.split(",") if r.strip()} try: handle = open(args.path, encoding="utf-8") except OSError as exc: print("error: cannot open %s: %s" % (args.path, exc.strerror), file=sys.stderr) return 2 errors, warnings = [], [] seen = {} valid = total = 0 lengths = [] with handle: for number, line in enumerate(handle, start=1): if not line.strip(): continue total += 1 try: obj = json.loads(line) except json.JSONDecodeError as exc: errors.append((number, "invalid JSON: %s" % exc.msg)) continue problems, text = check_example(obj, roles) for problem in problems: errors.append((number, problem)) if problems: continue valid += 1 digest = hashlib.sha256(json.dumps(obj, sort_keys=True).encode("utf-8")).hexdigest() if digest in seen: warnings.append((number, "duplicate of line %d" % seen[digest])) else: seen[digest] = number estimate = math.ceil(len(text) / args.chars_per_token) lengths.append(estimate) if estimate > args.max_tokens: warnings.append((number, "~%d tokens (estimate) exceeds --max-tokens %d" % (estimate, args.max_tokens))) for message in obj["messages"]: if message.get("role") == "assistant" and BOILERPLATE.search(content_text(message.get("content")) or ""): warnings.append((number, "assistant reply contains 'as an AI' style boilerplate")) break if valid < args.min_examples: errors.append((0, "only %d valid example(s); --min-examples is %d" % (valid, args.min_examples))) for number, message in errors[:200]: print("ERROR line %d: %s" % (number, message) if number else "ERROR: %s" % message) if len(errors) > 200: print("... %d more errors" % (len(errors) - 200)) for number, message in warnings[:100]: print("WARN line %d: %s" % (number, message)) if len(warnings) > 100: print("... %d more warnings" % (len(warnings) - 100)) summary = { "lines": total, "valid_examples": valid, "errors": len(errors), "warnings": len(warnings), "approx_tokens_total": sum(lengths), "approx_tokens_max": max(lengths) if lengths else 0, "approx_tokens_mean": round(sum(lengths) / len(lengths), 1) if lengths else 0, } print("\n" + ", ".join("%s=%s" % item for item in summary.items())) if args.json_report: with open(args.json_report, "w", encoding="utf-8") as out: json.dump({"summary": summary, "errors": [{"line": n, "message": m} for n, m in errors], "warnings": [{"line": n, "message": m} for n, m in warnings]}, out, indent=2) return 1 if errors else 0 if __name__ == "__main__": sys.exit(main())