| |
| """Validate AI Puppet Theater Actor SFT JSONL files.""" |
|
|
| from __future__ import annotations |
|
|
| import argparse |
| import json |
| from collections import Counter |
| from pathlib import Path |
| from typing import Any |
|
|
|
|
| REQUIRED_TOP_LEVEL = {"id", "source_mix", "row_type", "messages"} |
| OPTIONAL_TOP_LEVEL = {"source_dataset", "transformation"} |
| ALLOWED_TOP_LEVEL = REQUIRED_TOP_LEVEL | OPTIONAL_TOP_LEVEL |
| ALLOWED_SOURCE_DATASETS = { |
| "G-reen/TheatreLM-v2.1-Characters", |
| "practical-dreamer/RPGPT_PublicDomain-alpaca", |
| } |
| ALLOWED_ROW_TYPES = { |
| "normal_reaction", |
| "prop_inspection", |
| "oracle_consult", |
| "lighting_change", |
| "memory_callback", |
| "secret_hint_or_reveal", |
| "finale", |
| "comedic_confusion", |
| } |
| OUTPUT_FIELDS = [ |
| "intent", |
| "line", |
| "emotion", |
| "gesture", |
| "stage_effect", |
| "memory_update", |
| "tool_request", |
| ] |
| ALLOWED_INTENTS = { |
| "react_to_event", |
| "clarify_problem", |
| "inspect_prop", |
| "consult_oracle", |
| "change_lighting", |
| "recall_memory", |
| "hint_secret", |
| "reveal_secret", |
| "deliver_finale", |
| "comic_confusion", |
| } |
| ALLOWED_TOOLS = {"inspect_prop", "consult_stage_oracle", "change_lighting"} |
| TOOL_ARGS = { |
| "inspect_prop": {"prop"}, |
| "consult_stage_oracle": {"question"}, |
| "change_lighting": {"mood"}, |
| } |
| V1_TOOL_ARGS = { |
| "inspect_prop": {"prop"}, |
| "consult_stage_oracle": {"question"}, |
| "change_lighting": {"mood"}, |
| } |
| V1_BANNED_ASSISTANT_FIELDS = { |
| "memory_record", |
| "memory_effect", |
| "recent_transcript", |
| "show_state", |
| "held_props", |
| "mood", |
| "name", |
| "latest_prop", |
| "latest_audience_action", |
| "tool_results", |
| } |
| BANNED_TERMS = { |
| "fuck", |
| "shit", |
| "bitch", |
| "asshole", |
| "bloodbath", |
| "gore", |
| "dismember", |
| "suicide", |
| } |
|
|
|
|
| def main() -> None: |
| parser = argparse.ArgumentParser(description=__doc__) |
| parser.add_argument("path", type=Path) |
| parser.add_argument("--max-errors", type=int, default=25) |
| args = parser.parse_args() |
|
|
| stats = validate_file(args.path, args.max_errors) |
| print(f"file: {args.path}") |
| print(f"total rows: {stats['total']}") |
| print(f"valid rows: {stats['valid']}") |
| print(f"invalid rows: {stats['invalid']}") |
| print_distribution("row_type distribution", stats["row_types"]) |
| print_distribution("tool_request distribution", stats["tools"]) |
| stem = args.path.stem |
| if stem.endswith("_train"): |
| split_stem = stem[:-6] |
| elif stem.endswith("_val"): |
| split_stem = stem[:-4] |
| else: |
| split_stem = stem |
| train_path = args.path.with_name(f"{split_stem}_train.jsonl") |
| val_path = args.path.with_name(f"{split_stem}_val.jsonl") |
| if train_path.exists() or val_path.exists(): |
| print(f"train rows: {count_lines(train_path) if train_path.exists() else 0}") |
| print(f"val rows: {count_lines(val_path) if val_path.exists() else 0}") |
| if stats["errors"]: |
| print("errors:") |
| for error in stats["errors"]: |
| print(f"- {error}") |
| if stats["invalid"]: |
| raise SystemExit(1) |
|
|
|
|
| def validate_file(path: Path, max_errors: int) -> dict[str, Any]: |
| stats: dict[str, Any] = { |
| "total": 0, |
| "valid": 0, |
| "invalid": 0, |
| "row_types": Counter(), |
| "tools": Counter(), |
| "errors": [], |
| } |
| with path.open("r", encoding="utf-8") as handle: |
| for line_number, line in enumerate(handle, start=1): |
| if not line.strip(): |
| continue |
| stats["total"] += 1 |
| try: |
| row = json.loads(line) |
| except json.JSONDecodeError as exc: |
| add_error(stats, max_errors, line_number, f"row JSON parse failed: {exc}") |
| continue |
| errors = validate_row(row, dataset_version_for(path, row)) |
| if errors: |
| stats["invalid"] += 1 |
| for error in errors: |
| add_error(stats, max_errors, line_number, error) |
| continue |
| assistant = json.loads(row["messages"][2]["content"]) |
| stats["valid"] += 1 |
| stats["row_types"][row["row_type"]] += 1 |
| tool_request = assistant["tool_request"] |
| stats["tools"][tool_request["tool"] if tool_request else "none"] += 1 |
| return stats |
|
|
|
|
| def dataset_version_for(path: Path, row: Any) -> str: |
| if "v1" in path.name: |
| return "v1" |
| if isinstance(row, dict): |
| row_id = str(row.get("id", "")) |
| source_mix = row.get("source_mix", []) |
| if row_id.startswith("actor-sft-v1-") or "synthetic_v1" in source_mix: |
| return "v1" |
| return "v0" |
|
|
|
|
| def validate_row(row: Any, version: str = "v0") -> list[str]: |
| errors: list[str] = [] |
| if not isinstance(row, dict): |
| return ["row must be an object"] |
| unknown_top_level = set(row) - ALLOWED_TOP_LEVEL |
| missing_top_level = REQUIRED_TOP_LEVEL - set(row) |
| if unknown_top_level: |
| errors.append(f"unknown top-level keys: {sorted(unknown_top_level)}") |
| if missing_top_level: |
| errors.append(f"missing top-level keys: {sorted(missing_top_level)}") |
| return errors |
| if not isinstance(row["id"], str) or not row["id"].strip(): |
| errors.append("id must be a non-empty string") |
| if row["row_type"] not in ALLOWED_ROW_TYPES: |
| errors.append(f"unknown row_type: {row['row_type']!r}") |
| if not isinstance(row["source_mix"], list) or not all(isinstance(item, str) and item for item in row["source_mix"]): |
| errors.append("source_mix must be a non-empty list of strings") |
| if "source_dataset" in row: |
| if row["source_dataset"] not in ALLOWED_SOURCE_DATASETS: |
| errors.append(f"source_dataset must be one of {sorted(ALLOWED_SOURCE_DATASETS)}") |
| if row["source_dataset"] == "G-reen/TheatreLM-v2.1-Characters" and "theatrelm_seed" not in row["source_mix"]: |
| errors.append("TheatreLM seeded rows must include theatrelm_seed in source_mix") |
| if row["source_dataset"] == "practical-dreamer/RPGPT_PublicDomain-alpaca" and "rpgpt_seed" not in row["source_mix"]: |
| errors.append("RPGPT seeded rows must include rpgpt_seed in source_mix") |
| if "transformation" in row and row["transformation"] != "seeded_synthetic_actor_json": |
| errors.append("transformation must be seeded_synthetic_actor_json when present") |
| if ("source_dataset" in row) != ("transformation" in row): |
| errors.append("source_dataset and transformation must appear together") |
| messages = row["messages"] |
| if not isinstance(messages, list) or len(messages) != 3: |
| errors.append("messages must contain exactly system, user, assistant messages") |
| return errors |
| expected_roles = ["system", "user", "assistant"] |
| for index, expected_role in enumerate(expected_roles): |
| message = messages[index] |
| if not isinstance(message, dict): |
| errors.append(f"message {index} must be an object") |
| continue |
| if set(message) != {"role", "content"}: |
| errors.append(f"message {index} must contain only role and content") |
| if message.get("role") != expected_role: |
| errors.append(f"message {index} role must be {expected_role}") |
| if not isinstance(message.get("content"), str) or not message["content"].strip(): |
| errors.append(f"message {index} content must be non-empty text") |
| user_content = messages[1]["content"] if isinstance(messages[1], dict) else "" |
| for marker in ["premise:", "show_state JSON:", "actor JSON:", "director_instruction:"]: |
| if marker not in user_content: |
| errors.append(f"user message missing {marker}") |
| errors.extend(validate_assistant_content(messages[2]["content"], row, version, user_content)) |
| return errors |
|
|
|
|
| def validate_assistant_content(content: str, row: dict[str, Any] | None = None, version: str = "v0", user_content: str = "") -> list[str]: |
| errors: list[str] = [] |
| try: |
| value = json.loads(content) |
| except json.JSONDecodeError as exc: |
| return [f"assistant content JSON parse failed: {exc}"] |
| if not isinstance(value, dict): |
| return ["assistant content must parse to an object"] |
| keys = list(value) |
| if keys != OUTPUT_FIELDS: |
| errors.append(f"assistant keys must be exactly {OUTPUT_FIELDS}; got {keys}") |
| if version == "v1": |
| copied_fields = sorted(V1_BANNED_ASSISTANT_FIELDS & set(value)) |
| if copied_fields: |
| errors.append(f"assistant must not copy state/input fields: {copied_fields}") |
| for field in OUTPUT_FIELDS: |
| if field not in value: |
| continue |
| if field == "tool_request": |
| continue |
| if field == "memory_update" and value[field] is None: |
| continue |
| if not isinstance(value[field], str): |
| errors.append(f"{field} must be a string") |
| continue |
| if not value[field].strip(): |
| errors.append(f"{field} must not be empty") |
| line = value.get("line", "") |
| if isinstance(line, str): |
| words = line.split() |
| if not 6 <= len(words) <= 18: |
| errors.append(f"line must be 6-18 words; got {len(words)}") |
| lowered = line.lower() |
| banned = sorted(term for term in BANNED_TERMS if term in lowered) |
| if banned: |
| errors.append(f"line contains banned terms: {banned}") |
| intent = value.get("intent", "") |
| if isinstance(intent, str) and intent not in ALLOWED_INTENTS: |
| errors.append(f"intent must be one of {sorted(ALLOWED_INTENTS)}") |
| gesture = value.get("gesture", "") |
| if isinstance(gesture, str) and "_" in gesture: |
| errors.append("gesture must be a short theatrical phrase, not an enum-like token") |
| memory_update = value.get("memory_update", "") |
| if isinstance(memory_update, str): |
| if not memory_update.strip(): |
| errors.append("memory_update must be null or non-empty text") |
| if len(memory_update) > 140: |
| errors.append("memory_update must be 140 characters or fewer") |
| elif memory_update is not None: |
| errors.append("memory_update must be null or a string") |
| if version == "v1" and row is not None: |
| errors.extend(validate_v1_finale_context(value, row, user_content)) |
| tool_request = value.get("tool_request") |
| errors.extend(validate_tool_request(tool_request, version)) |
| return errors |
|
|
|
|
| def validate_v1_finale_context(value: dict[str, Any], row: dict[str, Any], user_content: str) -> list[str]: |
| if value.get("intent") != "deliver_finale" and value.get("stage_effect") != "final_bow_lights": |
| return [] |
| show_state = extract_show_state(user_content) |
| finale_requested = bool(show_state.get("finale_requested")) if isinstance(show_state, dict) else False |
| story_phase = show_state.get("story_phase") if isinstance(show_state, dict) else None |
| if row.get("row_type") == "finale" or finale_requested or story_phase == "finale": |
| return [] |
| return ["deliver_finale/final_bow_lights only allowed for finale row, finale_requested, or story_phase finale"] |
|
|
|
|
| def extract_show_state(user_content: str) -> dict[str, Any] | None: |
| marker = "show_state JSON:" |
| next_marker = "\nactor JSON:" |
| if marker not in user_content: |
| return None |
| start = user_content.index(marker) + len(marker) |
| end = user_content.find(next_marker, start) |
| raw_json = user_content[start:end if end != -1 else None].strip() |
| try: |
| value = json.loads(raw_json) |
| except json.JSONDecodeError: |
| return None |
| return value if isinstance(value, dict) else None |
|
|
|
|
| def validate_tool_request(value: Any, version: str = "v0") -> list[str]: |
| if value is None: |
| return [] |
| if not isinstance(value, dict): |
| return ["tool_request must be null or an object"] |
| if set(value) != {"tool", "args", "reason"}: |
| return ["tool_request must contain exactly tool, args, and reason"] |
| tool = value["tool"] |
| if tool not in ALLOWED_TOOLS: |
| return [f"tool_request tool must be one of {sorted(ALLOWED_TOOLS)}"] |
| if not isinstance(value["reason"], str) or not value["reason"].strip(): |
| return ["tool_request reason must be non-empty text"] |
| if len(value["reason"]) > 140: |
| return ["tool_request reason must be 140 characters or fewer"] |
| args = value["args"] |
| if not isinstance(args, dict): |
| return ["tool_request args must be an object"] |
| tool_args = V1_TOOL_ARGS if version == "v1" else TOOL_ARGS |
| if set(args) != tool_args[tool]: |
| return [f"tool_request args for {tool} must be exactly {sorted(tool_args[tool])}"] |
| for arg_value in args.values(): |
| if not isinstance(arg_value, str) or not arg_value.strip(): |
| return ["tool_request arg values must be non-empty strings"] |
| if len(arg_value) > 120: |
| return ["tool_request arg values must be 120 characters or fewer"] |
| return [] |
|
|
|
|
| def add_error(stats: dict[str, Any], max_errors: int, line_number: int, message: str) -> None: |
| stats["invalid"] += 1 if message.startswith("row JSON parse failed") else 0 |
| if len(stats["errors"]) < max_errors: |
| stats["errors"].append(f"line {line_number}: {message}") |
|
|
|
|
| def count_lines(path: Path) -> int: |
| with path.open("r", encoding="utf-8") as handle: |
| return sum(1 for line in handle if line.strip()) |
|
|
|
|
| def print_distribution(title: str, values: Counter) -> None: |
| print(f"{title}:") |
| for key, count in sorted(values.items()): |
| print(f" {key}: {count}") |
|
|
|
|
| if __name__ == "__main__": |
| main() |
|
|
