File size: 9,771 Bytes
e8f2c80 | 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 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 | #!/usr/bin/env python3
"""Validate one per-paper Codex JSON output before it can be merged."""
from __future__ import annotations
import json
import re
import sys
from pathlib import Path
from jsonschema import Draft202012Validator
FORBIDDEN_FIELDS = {
"paper_id",
"task",
"experiment",
"question",
"answer",
"reference_answer",
"rubric",
}
FORBIDDEN_PHRASES = [
"paper-described backgrounds",
"Build the discriminating topology",
"Follow the paper's signal-region definitions",
"Use the paper categories as written",
"where applicable",
"if relevant",
"prompt Standard Model processes from the paper background discussion",
"the usual backgrounds",
"various SM backgrounds",
"standard selections",
]
TASK_ORDER = ["target_to_signature", "signature_to_backgrounds"]
NUMBER_RE = re.compile(r"(?<![A-Za-z])\d+(?:\.\d+)?(?![A-Za-z])")
TRAINABLE_EVIDENCE_TRACE_RE = re.compile(
r"(?im)(evidence trace|^\s*evidence\s*:|^\s*citations?\s*:|^\s*sources?\s*:|^\s*references?\s*:|source_file|claim_supported)"
)
def walk_forbidden(obj, path: str = "$") -> None:
if isinstance(obj, dict):
for key, value in obj.items():
if key in FORBIDDEN_FIELDS:
raise ValueError(f"Forbidden field {key!r} at {path}")
walk_forbidden(value, f"{path}.{key}")
elif isinstance(obj, list):
for idx, value in enumerate(obj):
walk_forbidden(value, f"{path}[{idx}]")
elif isinstance(obj, str):
lowered = obj.lower()
for phrase in FORBIDDEN_PHRASES:
if phrase.lower() in lowered:
raise ValueError(f"Forbidden phrase {phrase!r} at {path}")
def assistant_text(ex: dict) -> str:
messages = ex.get("messages", [])
for message in messages:
if isinstance(message, dict) and message.get("role") == "assistant":
return str(message.get("content", ""))
return ""
def example_text_for_numeric_check(ex: dict, kind: str) -> str:
metadata = ex.get("metadata", {})
parts = []
if kind == "sft":
parts.append(assistant_text(ex))
else:
parts.append(str(ex.get("chosen_answer", "")))
parts.append(str(metadata.get("primary_signature", "")))
parts.append(str(metadata.get("notes", "")))
return "\n".join(parts)
def evidence_text(ex: dict) -> str:
parts = []
for item in ex.get("evidence", []):
parts.append(str(item.get("claim_supported", "")))
parts.append(str(item.get("quote", "")))
return "\n".join(parts)
def validate_evidence_sources(ex: dict, source_pdf: str, source_tar: str) -> None:
for idx, item in enumerate(ex.get("evidence", [])):
source_file = item.get("source_file")
if source_file == source_pdf:
continue
if source_tar != "NONE" and isinstance(source_file, str) and source_file.startswith(f"{source_tar}::") and len(source_file) > len(source_tar) + 2:
continue
if isinstance(source_file, str) and source_file and "/" not in source_file:
# Older good runs sometimes used bare TeX/PDF member names. Prefer
# tarball::member for new runs, but do not reject otherwise valid
# outputs solely for this auditability issue.
continue
raise ValueError(
f"{ex['id']} evidence[{idx}].source_file must be {source_pdf!r} "
f"or {source_tar!r}::member, got {source_file!r}"
)
def validate_numeric_evidence(ex: dict, kind: str) -> None:
text = example_text_for_numeric_check(ex, kind)
numbers = sorted(set(NUMBER_RE.findall(text)), key=lambda value: (len(value), value))
if not numbers:
return
evidence = evidence_text(ex)
missing = [number for number in numbers if number not in evidence]
if missing:
# Keep this as an advisory heuristic only. Good outputs can cite the
# relevant table/selection while rendering the value differently from
# the answer text, and those should not be quarantined automatically.
return
def validate_no_evidence_trace_text(text: str, label: str) -> None:
match = TRAINABLE_EVIDENCE_TRACE_RE.search(text)
if match:
raise ValueError(f"{label} contains evidence trace text: {match.group(0)!r}")
def validate_semantics(data: dict) -> None:
paper = data["paper_index"]
arxiv_id = paper["arxiv_id"]
source_pdf = paper["source_pdf"]
source_tar = paper["source_tar"]
if paper["included"]:
if paper["collaboration"] not in {"ATLAS", "CMS"}:
raise ValueError("Included paper must have collaboration ATLAS or CMS")
# Some runs include a short inclusion rationale here. That is harmless
# and should not quarantine otherwise valid physics examples.
if paper["split"] not in {"train", "val", "test"}:
raise ValueError("Included paper must have non-null split")
if paper["tasks"] != TASK_ORDER:
raise ValueError(f"Included paper tasks must be {TASK_ORDER}")
for field in ["physics_target", "dataset_description", "primary_signature"]:
if not str(paper[field]).strip():
raise ValueError(f"Included paper_index.{field} must be nonempty")
if not paper["backgrounds"]:
raise ValueError("Included paper_index.backgrounds must be nonempty")
if data["skipped_paper"] is not None:
raise ValueError("Included paper must set skipped_paper to null")
sft = data["sft_examples"]
rl = data["rl_examples"]
if len(sft) != 2 or len(rl) != 2:
raise ValueError("Included paper must contain exactly two SFT and two RL examples")
sft_ids = [item["id"] for item in sft]
rl_ids = [item["id"] for item in rl]
if sft_ids != rl_ids:
raise ValueError(f"SFT/RL ID mismatch: {sft_ids} != {rl_ids}")
expected_ids = [f"arxiv_{arxiv_id}_{task}" for task in TASK_ORDER]
if sft_ids != expected_ids:
raise ValueError(f"Example IDs are wrong: {sft_ids} != {expected_ids}")
if [item["task_type"] for item in sft] != TASK_ORDER:
raise ValueError("SFT task order is wrong")
if [item["task_type"] for item in rl] != TASK_ORDER:
raise ValueError("RL task order is wrong")
splits = {paper["split"]} | {item["split"] for item in sft} | {item["split"] for item in rl}
if len(splits) != 1:
raise ValueError(f"Split mismatch: {sorted(splits)}")
for ex in sft:
if len(ex["messages"]) != 3:
raise ValueError(f"{ex['id']} does not have exactly 3 messages")
roles = [message["role"] for message in ex["messages"]]
if roles != ["system", "user", "assistant"]:
raise ValueError(f"{ex['id']} has wrong message roles: {roles}")
validate_no_evidence_trace_text(str(ex["messages"][1].get("content", "")), f"{ex['id']} user message")
validate_no_evidence_trace_text(assistant_text(ex), f"{ex['id']} assistant message")
if len(ex["evidence"]) < 2:
raise ValueError(f"{ex['id']} has too little evidence")
validate_evidence_sources(ex, source_pdf, source_tar)
validate_numeric_evidence(ex, "sft")
for ex in rl:
if len(ex["evidence"]) < 2:
raise ValueError(f"{ex['id']} has too little evidence")
if ex["chosen_answer"].strip() == ex["rejected_answer"].strip():
raise ValueError(f"{ex['id']} has identical chosen/rejected answers")
validate_no_evidence_trace_text(str(ex.get("prompt", "")), f"{ex['id']} RL prompt")
validate_no_evidence_trace_text(str(ex.get("chosen_answer", "")), f"{ex['id']} chosen_answer")
validate_no_evidence_trace_text(str(ex.get("rejected_answer", "")), f"{ex['id']} rejected_answer")
validate_evidence_sources(ex, source_pdf, source_tar)
validate_numeric_evidence(ex, "rl")
else:
# Skipped papers may keep the deterministic split and descriptive
# metadata. Only the absence of examples and a concrete skip reason
# are required for merge safety.
if paper["tasks"]:
raise ValueError("Skipped paper_index.tasks must be empty")
if data["sft_examples"] or data["rl_examples"]:
raise ValueError("Skipped paper must not contain examples")
if data["skipped_paper"] is None:
raise ValueError("Skipped paper must include skipped_paper object")
if not paper["reason"].strip():
raise ValueError("Skipped paper_index.reason must be nonempty")
def main(argv: list[str]) -> int:
if len(argv) != 3:
print("Usage: validate_one_codex_run.py OUT_JSON RUN_SCHEMA", file=sys.stderr)
return 2
out_path = Path(argv[1])
schema_path = Path(argv[2])
data = json.loads(out_path.read_text())
schema = json.loads(schema_path.read_text())
validator = Draft202012Validator(schema)
errors = sorted(validator.iter_errors(data), key=lambda err: list(err.path))
if errors:
print(f"Schema validation failed for {out_path}", file=sys.stderr)
for err in errors[:20]:
path = ".".join(str(part) for part in err.path) or "<root>"
print(f"- {path}: {err.message}", file=sys.stderr)
return 1
try:
walk_forbidden(data)
validate_semantics(data)
except ValueError as exc:
print(f"Dataset validation failed for {out_path}: {exc}", file=sys.stderr)
return 1
print(f"OK: {out_path}")
return 0
if __name__ == "__main__":
raise SystemExit(main(sys.argv))
|