| |
| from __future__ import annotations |
|
|
| import argparse |
| import json |
| from collections import Counter |
| from pathlib import Path |
| from typing import Any |
|
|
|
|
| ROOT = Path(__file__).resolve().parents[1] |
| INVENTORY = ROOT / "data" / "swival_tool_inventory.json" |
|
|
|
|
| def load_inventory() -> dict[str, dict[str, Any]]: |
| blob = json.loads(INVENTORY.read_text()) |
| return {tool["name"]: tool for tool in blob["tools"]} |
|
|
|
|
| def iter_jsonl(path: Path): |
| with path.open() as f: |
| for line_no, line in enumerate(f, 1): |
| if line.strip(): |
| yield line_no, json.loads(line) |
|
|
|
|
| def validate_training(path: Path, inventory: dict[str, dict[str, Any]]) -> list[str]: |
| errors: list[str] = [] |
| coverage: Counter[str] = Counter() |
| for line_no, row in iter_jsonl(path): |
| messages = row.get("messages") |
| if not isinstance(messages, list) or not messages: |
| errors.append(f"{path}:{line_no}: missing messages") |
| continue |
| for msg in messages: |
| if msg.get("role") == "assistant": |
| for tc in msg.get("tool_calls") or []: |
| fn = tc.get("function") or {} |
| name = fn.get("name") |
| if name not in inventory: |
| errors.append(f"{path}:{line_no}: unknown tool {name!r}") |
| continue |
| coverage[name] += 1 |
| raw_args = fn.get("arguments") |
| try: |
| args = json.loads(raw_args) |
| except Exception as exc: |
| errors.append(f"{path}:{line_no}: bad JSON arguments for {name}: {exc}") |
| continue |
| if not isinstance(args, dict): |
| errors.append(f"{path}:{line_no}: arguments for {name} must be an object") |
| continue |
| for required in inventory[name].get("required", []): |
| if required not in args: |
| errors.append(f"{path}:{line_no}: {name} missing required argument {required}") |
| if not any((msg.get("tool_calls") for msg in messages if msg.get("role") == "assistant")): |
| errors.append(f"{path}:{line_no}: example has no assistant tool call") |
|
|
| missing = sorted(set(inventory) - set(coverage)) |
| if missing: |
| errors.append(f"{path}: missing tool coverage: {', '.join(missing)}") |
| return errors |
|
|
|
|
| def validate_eval(path: Path, inventory: dict[str, dict[str, Any]]) -> list[str]: |
| errors: list[str] = [] |
| for line_no, row in iter_jsonl(path): |
| if not row.get("id") or not row.get("prompt"): |
| errors.append(f"{path}:{line_no}: eval row needs id and prompt") |
| if not isinstance(row.get("files"), dict): |
| errors.append(f"{path}:{line_no}: eval row needs files object") |
| expected = row.get("expected_tools") or [] |
| for name in expected: |
| if name not in inventory: |
| errors.append(f"{path}:{line_no}: unknown expected tool {name}") |
| if not any(key in row for key in ["verify", "verify_contains", "verify_missing", "verify_command"]): |
| errors.append(f"{path}:{line_no}: eval row has no verifier") |
| return errors |
|
|
|
|
| def main() -> int: |
| parser = argparse.ArgumentParser() |
| parser.add_argument("path", type=Path) |
| args = parser.parse_args() |
|
|
| inventory = load_inventory() |
| if args.path.name.endswith("_eval.jsonl"): |
| errors = validate_eval(args.path, inventory) |
| else: |
| errors = validate_training(args.path, inventory) |
|
|
| if errors: |
| for err in errors: |
| print(err) |
| return 1 |
| print(f"ok: {args.path}") |
| return 0 |
|
|
|
|
| if __name__ == "__main__": |
| raise SystemExit(main()) |
|
|
