wuyang
702 lines
20 KiB
Python
702 lines
20 KiB
Python
#!/usr/bin/env python3
|
|
"""Evaluate completion state, GSM8K answers, and HumanEval code safely.
|
|
|
|
The input is a raw JSON emitted by
|
|
v2_lite_chat_special_token_behavior_probe.py. Generated code is executed only
|
|
inside one fresh, networkless, read-only Docker container per candidate. The
|
|
result preserves fixed-budget and strict-completion metrics separately.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import ast
|
|
import gzip
|
|
import hashlib
|
|
import json
|
|
import re
|
|
import subprocess
|
|
import time
|
|
from collections import Counter
|
|
from decimal import Decimal, InvalidOperation
|
|
from pathlib import Path
|
|
from statistics import mean
|
|
from typing import Any
|
|
|
|
|
|
NUMBER = r"[-+]?(?:\d[\d,]*\.?\d*|\.\d+)"
|
|
NUMBER_PATTERN = re.compile(NUMBER)
|
|
BOXED_PATTERN = re.compile(
|
|
rf"\\boxed\s*\{{\s*({NUMBER})\s*\}}",
|
|
flags=re.IGNORECASE,
|
|
)
|
|
HASH_PATTERN = re.compile(
|
|
rf"####\s*({NUMBER})",
|
|
flags=re.IGNORECASE,
|
|
)
|
|
ANSWER_PATTERN = re.compile(
|
|
rf"(?:final\s+answer|answer|result|total|profit)"
|
|
rf"(?:\s+(?:is|equals|will\s+be))?\s*[:=]?\s*"
|
|
rf"(?:\$|USD\s*)?({NUMBER})",
|
|
flags=re.IGNORECASE,
|
|
)
|
|
FENCE_PATTERN = re.compile(
|
|
r"```(?P<lang>[A-Za-z0-9_+-]*)[ \t]*\n?"
|
|
r"(?P<body>.*?)(?:```|$)",
|
|
flags=re.DOTALL,
|
|
)
|
|
|
|
|
|
def parse_args() -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("--behavior-json", type=Path, required=True)
|
|
parser.add_argument("--baseline-json", type=Path)
|
|
parser.add_argument("--human-eval", type=Path, required=True)
|
|
parser.add_argument("--gsm8k", type=Path, required=True)
|
|
parser.add_argument("--output", type=Path, required=True)
|
|
parser.add_argument(
|
|
"--sandbox-image",
|
|
required=True,
|
|
help="Pinned image reference including @sha256 digest.",
|
|
)
|
|
parser.add_argument("--timeout-seconds", type=float, default=5.0)
|
|
parser.add_argument("--skip-code-execution", action="store_true")
|
|
return parser.parse_args()
|
|
|
|
|
|
def sha256_bytes(payload: bytes) -> str:
|
|
return hashlib.sha256(payload).hexdigest()
|
|
|
|
|
|
def sha256_file(path: Path) -> str:
|
|
digest = hashlib.sha256()
|
|
with path.open("rb") as handle:
|
|
while chunk := handle.read(16 * 1024 * 1024):
|
|
digest.update(chunk)
|
|
return digest.hexdigest()
|
|
|
|
|
|
def canonical_hash(value: Any) -> str:
|
|
return sha256_bytes(
|
|
json.dumps(
|
|
value,
|
|
ensure_ascii=False,
|
|
sort_keys=True,
|
|
separators=(",", ":"),
|
|
).encode()
|
|
)
|
|
|
|
|
|
def load_jsonl(path: Path) -> list[dict[str, Any]]:
|
|
with path.open(encoding="utf-8") as handle:
|
|
return [json.loads(line) for line in handle if line.strip()]
|
|
|
|
|
|
def load_tasks(
|
|
human_eval_path: Path,
|
|
gsm8k_path: Path,
|
|
) -> tuple[dict[str, dict[str, Any]], dict[str, dict[str, Any]]]:
|
|
with gzip.open(human_eval_path, "rt", encoding="utf-8") as handle:
|
|
human_eval = {
|
|
row["task_id"]: row
|
|
for line in handle
|
|
if line.strip()
|
|
for row in [json.loads(line)]
|
|
}
|
|
gsm8k = {
|
|
f"gsm8k/test/{index:04d}": row
|
|
for index, row in enumerate(load_jsonl(gsm8k_path))
|
|
}
|
|
return human_eval, gsm8k
|
|
|
|
|
|
def normalize_number(value: str | None) -> str | None:
|
|
if value is None:
|
|
return None
|
|
cleaned = value.replace(",", "").strip().rstrip(".")
|
|
try:
|
|
number = Decimal(cleaned)
|
|
except InvalidOperation:
|
|
return None
|
|
normalized = format(number.normalize(), "f")
|
|
if "." in normalized:
|
|
normalized = normalized.rstrip("0").rstrip(".")
|
|
return "0" if normalized in {"-0", "+0"} else normalized
|
|
|
|
|
|
def last_match(
|
|
pattern: re.Pattern[str],
|
|
text: str,
|
|
) -> str | None:
|
|
matches = list(pattern.finditer(text))
|
|
return matches[-1].group(1) if matches else None
|
|
|
|
|
|
def evaluate_math(
|
|
text: str,
|
|
hit_eos: bool,
|
|
gold_answer: str,
|
|
) -> dict[str, Any]:
|
|
candidates = [
|
|
("boxed", last_match(BOXED_PATTERN, text)),
|
|
("hash_marker", last_match(HASH_PATTERN, text)),
|
|
("answer_phrase", last_match(ANSWER_PATTERN, text)),
|
|
]
|
|
method = "none"
|
|
extracted = None
|
|
for name, value in candidates:
|
|
if value is not None:
|
|
method = name
|
|
extracted = value
|
|
break
|
|
explicit_terminal = extracted is not None
|
|
if extracted is None:
|
|
values = NUMBER_PATTERN.findall(text)
|
|
if values:
|
|
method = "last_number_fallback"
|
|
extracted = values[-1]
|
|
|
|
gold_values = NUMBER_PATTERN.findall(
|
|
gold_answer.rsplit("####", 1)[-1]
|
|
)
|
|
gold = normalize_number(gold_values[-1] if gold_values else None)
|
|
predicted = normalize_number(extracted)
|
|
fixed_budget_exact = (
|
|
gold is not None
|
|
and predicted is not None
|
|
and gold == predicted
|
|
)
|
|
semantic_terminal = bool(hit_eos or explicit_terminal)
|
|
return {
|
|
"gold_final": gold,
|
|
"predicted_final": predicted,
|
|
"extraction_method": method,
|
|
"explicit_final_marker": explicit_terminal,
|
|
"semantic_terminal": semantic_terminal,
|
|
"evaluator_covered": predicted is not None,
|
|
"fixed_budget_numeric_exact": fixed_budget_exact,
|
|
"strict_complete_numeric_exact": (
|
|
fixed_budget_exact and semantic_terminal
|
|
),
|
|
}
|
|
|
|
|
|
def code_fences(text: str) -> list[dict[str, str | bool]]:
|
|
rows = []
|
|
for match in FENCE_PATTERN.finditer(text):
|
|
language = match.group("lang").lower()
|
|
body = match.group("body").strip()
|
|
rows.append(
|
|
{
|
|
"language": language,
|
|
"body": body,
|
|
"closed": text[match.start():match.end()].rstrip().endswith(
|
|
"```"
|
|
),
|
|
}
|
|
)
|
|
return rows
|
|
|
|
|
|
def extract_code(
|
|
text: str,
|
|
prompt: str,
|
|
entry_point: str,
|
|
) -> dict[str, Any]:
|
|
target = re.compile(
|
|
rf"(?m)^\s*(?:async\s+)?def\s+{re.escape(entry_point)}\s*\("
|
|
)
|
|
fences = code_fences(text)
|
|
chosen = next(
|
|
(
|
|
row for row in fences
|
|
if target.search(str(row["body"]))
|
|
and row["language"] in {"", "py", "python", "python3"}
|
|
),
|
|
None,
|
|
)
|
|
if chosen is None:
|
|
chosen = next(
|
|
(
|
|
row for row in fences
|
|
if row["language"] in {"", "py", "python", "python3"}
|
|
),
|
|
None,
|
|
)
|
|
|
|
if chosen is not None:
|
|
raw = str(chosen["body"])
|
|
mode = "fenced"
|
|
fence_closed = bool(chosen["closed"])
|
|
else:
|
|
match = target.search(text)
|
|
if match:
|
|
raw = text[match.start():].strip()
|
|
mode = "direct_definition"
|
|
else:
|
|
raw = text.strip()
|
|
mode = "prompt_completion"
|
|
fence_closed = False
|
|
|
|
contains_entry_point = target.search(raw) is not None
|
|
candidate = raw if contains_entry_point else prompt + raw
|
|
try:
|
|
ast.parse(candidate)
|
|
ast_ok = True
|
|
ast_error = None
|
|
except SyntaxError as error:
|
|
ast_ok = False
|
|
ast_error = {
|
|
"line": error.lineno,
|
|
"offset": error.offset,
|
|
"type": type(error).__name__,
|
|
}
|
|
return {
|
|
"candidate": candidate,
|
|
"candidate_sha256": sha256_bytes(candidate.encode()),
|
|
"extraction_mode": mode,
|
|
"contains_entry_point_definition": contains_entry_point,
|
|
"closed_code_fence": fence_closed,
|
|
"python_ast_parse": ast_ok,
|
|
"parse_error": ast_error,
|
|
}
|
|
|
|
|
|
def sandbox_harness(
|
|
candidate: str,
|
|
test: str,
|
|
entry_point: str,
|
|
) -> str:
|
|
return (
|
|
"import sys\n"
|
|
"try:\n"
|
|
+ "\n".join(
|
|
f" {line}" if line else ""
|
|
for line in candidate.splitlines()
|
|
)
|
|
+ "\n"
|
|
+ "\n".join(
|
|
f" {line}" if line else ""
|
|
for line in test.splitlines()
|
|
)
|
|
+ f"\n check({entry_point})\n"
|
|
+ "except AssertionError:\n"
|
|
+ " raise SystemExit(10)\n"
|
|
+ "except BaseException:\n"
|
|
+ " raise SystemExit(11)\n"
|
|
+ "raise SystemExit(0)\n"
|
|
)
|
|
|
|
|
|
def execute_code(
|
|
candidate: str,
|
|
task: dict[str, Any],
|
|
image: str,
|
|
timeout_seconds: float,
|
|
) -> dict[str, Any]:
|
|
harness = sandbox_harness(
|
|
candidate,
|
|
task["test"],
|
|
task["entry_point"],
|
|
)
|
|
command = [
|
|
"docker",
|
|
"run",
|
|
"--rm",
|
|
"-i",
|
|
"--network",
|
|
"none",
|
|
"--read-only",
|
|
"--tmpfs",
|
|
"/tmp:rw,noexec,nosuid,size=16m",
|
|
"--memory",
|
|
"256m",
|
|
"--memory-swap",
|
|
"256m",
|
|
"--pids-limit",
|
|
"64",
|
|
"--cpus",
|
|
"0.5",
|
|
"--cap-drop",
|
|
"ALL",
|
|
"--security-opt",
|
|
"no-new-privileges",
|
|
"--user",
|
|
"65534:65534",
|
|
image,
|
|
"python",
|
|
"-I",
|
|
"-",
|
|
]
|
|
started = time.perf_counter()
|
|
try:
|
|
completed = subprocess.run(
|
|
command,
|
|
input=harness.encode(),
|
|
stdout=subprocess.DEVNULL,
|
|
stderr=subprocess.DEVNULL,
|
|
timeout=timeout_seconds,
|
|
check=False,
|
|
)
|
|
except subprocess.TimeoutExpired:
|
|
return {
|
|
"status": "timeout",
|
|
"return_code": None,
|
|
"runtime_ms": (time.perf_counter() - started) * 1000,
|
|
"harness_sha256": sha256_bytes(harness.encode()),
|
|
}
|
|
runtime_ms = (time.perf_counter() - started) * 1000
|
|
statuses = {
|
|
0: "passed",
|
|
10: "assertion_failed",
|
|
11: "runtime_error",
|
|
}
|
|
return {
|
|
"status": statuses.get(
|
|
completed.returncode,
|
|
"sandbox_error",
|
|
),
|
|
"return_code": completed.returncode,
|
|
"runtime_ms": runtime_ms,
|
|
"harness_sha256": sha256_bytes(harness.encode()),
|
|
}
|
|
|
|
|
|
def evaluate_code(
|
|
text: str,
|
|
hit_eos: bool,
|
|
task: dict[str, Any],
|
|
image: str,
|
|
timeout_seconds: float,
|
|
skip_execution: bool,
|
|
) -> dict[str, Any]:
|
|
extracted = extract_code(
|
|
text,
|
|
task["prompt"],
|
|
task["entry_point"],
|
|
)
|
|
execution = {
|
|
"status": "not_run",
|
|
"return_code": None,
|
|
"runtime_ms": None,
|
|
"harness_sha256": None,
|
|
}
|
|
if extracted["python_ast_parse"] and not skip_execution:
|
|
execution = execute_code(
|
|
extracted["candidate"],
|
|
task,
|
|
image,
|
|
timeout_seconds,
|
|
)
|
|
passed = execution["status"] == "passed"
|
|
semantic_terminal = bool(
|
|
hit_eos
|
|
or extracted["closed_code_fence"]
|
|
or passed
|
|
)
|
|
return {
|
|
key: value
|
|
for key, value in extracted.items()
|
|
if key != "candidate"
|
|
} | {
|
|
"semantic_terminal": semantic_terminal,
|
|
"evaluator_covered": (
|
|
extracted["python_ast_parse"]
|
|
and execution["status"] != "not_run"
|
|
),
|
|
"execution": execution,
|
|
"fixed_budget_tests_pass": passed,
|
|
"strict_complete_tests_pass": passed and semantic_terminal,
|
|
}
|
|
|
|
|
|
def completion_class(
|
|
output: dict[str, Any],
|
|
task_evaluation: dict[str, Any] | None,
|
|
) -> str:
|
|
if output["hit_eos"]:
|
|
return "NATURAL_EOS"
|
|
if (
|
|
task_evaluation is not None
|
|
and task_evaluation.get("semantic_terminal")
|
|
):
|
|
return "TASK_TERMINAL_BEFORE_EOS"
|
|
if output["stopped_at_max_new_tokens"]:
|
|
if (
|
|
task_evaluation is not None
|
|
and task_evaluation.get("extraction_method")
|
|
== "last_number_fallback"
|
|
):
|
|
return "BUDGET_TRUNCATED_WITH_FALLBACK_ONLY"
|
|
return "BUDGET_TRUNCATED_UNRESOLVED"
|
|
return "OTHER_STOP"
|
|
|
|
|
|
def baseline_index(
|
|
baseline: dict[str, Any] | None,
|
|
) -> dict[tuple[str, str], dict[str, Any]]:
|
|
if baseline is None:
|
|
return {}
|
|
return {
|
|
(source["id"], output["condition"]): output
|
|
for source in baseline["sources"]
|
|
for output in source["outputs"]
|
|
}
|
|
|
|
|
|
def prefix_audit(
|
|
source_id: str,
|
|
output: dict[str, Any],
|
|
baseline_rows: dict[tuple[str, str], dict[str, Any]],
|
|
) -> dict[str, Any] | None:
|
|
prior = baseline_rows.get((source_id, output["condition"]))
|
|
if prior is None:
|
|
return None
|
|
reference = prior["generated_token_ids"]
|
|
observed = output["generated_token_ids"][: len(reference)]
|
|
return {
|
|
"baseline_generated_tokens": len(reference),
|
|
"compared_tokens": min(
|
|
len(reference),
|
|
len(output["generated_token_ids"]),
|
|
),
|
|
"prompt_hash_exact": (
|
|
prior["prompt_token_ids_sha256"]
|
|
== output["prompt_token_ids_sha256"]
|
|
),
|
|
"generated_prefix_exact": observed == reference,
|
|
"baseline_hit_eos": prior["hit_eos"],
|
|
}
|
|
|
|
|
|
def summarize(rows: list[dict[str, Any]]) -> dict[str, Any]:
|
|
by_condition: dict[str, dict[str, Any]] = {}
|
|
conditions = sorted({row["condition"] for row in rows})
|
|
for condition in conditions:
|
|
subset = [
|
|
row for row in rows
|
|
if row["condition"] == condition
|
|
]
|
|
math_rows = [
|
|
row for row in subset
|
|
if row["domain"] == "math"
|
|
]
|
|
code_rows = [
|
|
row for row in subset
|
|
if row["domain"] == "code"
|
|
]
|
|
by_condition[condition] = {
|
|
"outputs": len(subset),
|
|
"natural_eos": sum(row["hit_eos"] for row in subset),
|
|
"budget_truncated": sum(
|
|
row["stopped_at_max_new_tokens"]
|
|
for row in subset
|
|
),
|
|
"mean_generated_tokens": mean(
|
|
row["generated_tokens"] for row in subset
|
|
),
|
|
"completion_classes": dict(
|
|
Counter(row["completion_class"] for row in subset)
|
|
),
|
|
"math_fixed_budget_exact": sum(
|
|
row["task_evaluation"][
|
|
"fixed_budget_numeric_exact"
|
|
]
|
|
for row in math_rows
|
|
),
|
|
"math_strict_complete_exact": sum(
|
|
row["task_evaluation"][
|
|
"strict_complete_numeric_exact"
|
|
]
|
|
for row in math_rows
|
|
),
|
|
"math_sources": len(math_rows),
|
|
"code_ast_parse": sum(
|
|
row["task_evaluation"]["python_ast_parse"]
|
|
for row in code_rows
|
|
),
|
|
"code_executed": sum(
|
|
row["task_evaluation"]["execution"]["status"]
|
|
!= "not_run"
|
|
for row in code_rows
|
|
),
|
|
"code_tests_pass": sum(
|
|
row["task_evaluation"]["fixed_budget_tests_pass"]
|
|
for row in code_rows
|
|
),
|
|
"code_sources": len(code_rows),
|
|
}
|
|
prefix_rows = [
|
|
row["baseline_prefix_audit"]
|
|
for row in rows
|
|
if row["baseline_prefix_audit"] is not None
|
|
]
|
|
return {
|
|
"outputs": len(rows),
|
|
"natural_eos": sum(row["hit_eos"] for row in rows),
|
|
"budget_truncated": sum(
|
|
row["stopped_at_max_new_tokens"] for row in rows
|
|
),
|
|
"completion_classes": dict(
|
|
Counter(row["completion_class"] for row in rows)
|
|
),
|
|
"baseline_prefix": {
|
|
"cells": len(prefix_rows),
|
|
"prompt_hash_exact": sum(
|
|
row["prompt_hash_exact"] for row in prefix_rows
|
|
),
|
|
"generated_prefix_exact": sum(
|
|
row["generated_prefix_exact"] for row in prefix_rows
|
|
),
|
|
},
|
|
"by_condition": by_condition,
|
|
}
|
|
|
|
|
|
def main() -> None:
|
|
args = parse_args()
|
|
for path in (
|
|
args.behavior_json,
|
|
args.human_eval,
|
|
args.gsm8k,
|
|
):
|
|
if not path.is_file():
|
|
raise FileNotFoundError(path)
|
|
if args.baseline_json is not None and not args.baseline_json.is_file():
|
|
raise FileNotFoundError(args.baseline_json)
|
|
if "@sha256:" not in args.sandbox_image:
|
|
raise ValueError("--sandbox-image must include an immutable digest")
|
|
if args.timeout_seconds <= 0:
|
|
raise ValueError("--timeout-seconds must be positive")
|
|
|
|
behavior = json.loads(
|
|
args.behavior_json.read_text(encoding="utf-8")
|
|
)
|
|
baseline = (
|
|
json.loads(args.baseline_json.read_text(encoding="utf-8"))
|
|
if args.baseline_json is not None
|
|
else None
|
|
)
|
|
baseline_rows = baseline_index(baseline)
|
|
human_eval, gsm8k = load_tasks(args.human_eval, args.gsm8k)
|
|
|
|
rows = []
|
|
for source in behavior["sources"]:
|
|
for output in source["outputs"]:
|
|
evaluation = None
|
|
if source["domain"] == "math":
|
|
evaluation = evaluate_math(
|
|
output["text"],
|
|
output["hit_eos"],
|
|
gsm8k[source["id"]]["answer"],
|
|
)
|
|
elif source["domain"] == "code":
|
|
evaluation = evaluate_code(
|
|
output["text"],
|
|
output["hit_eos"],
|
|
human_eval[source["id"]],
|
|
args.sandbox_image,
|
|
args.timeout_seconds,
|
|
args.skip_code_execution,
|
|
)
|
|
rows.append(
|
|
{
|
|
"source_id": source["id"],
|
|
"domain": source["domain"],
|
|
"condition": output["condition"],
|
|
"generated_tokens": output["generated_tokens"],
|
|
"hit_eos": output["hit_eos"],
|
|
"stopped_at_max_new_tokens": (
|
|
output["stopped_at_max_new_tokens"]
|
|
),
|
|
"prompt_token_ids_sha256": (
|
|
output["prompt_token_ids_sha256"]
|
|
),
|
|
"generated_token_ids_sha256": (
|
|
output["generated_token_ids_sha256"]
|
|
),
|
|
"text_sha256": output["text_sha256"],
|
|
"task_evaluation": evaluation,
|
|
"completion_class": completion_class(
|
|
output,
|
|
evaluation,
|
|
),
|
|
"baseline_prefix_audit": prefix_audit(
|
|
source["id"],
|
|
output,
|
|
baseline_rows,
|
|
),
|
|
}
|
|
)
|
|
|
|
result = {
|
|
"schema_version": 1,
|
|
"input": {
|
|
"behavior_path": str(args.behavior_json),
|
|
"behavior_sha256": sha256_file(args.behavior_json),
|
|
"baseline_path": (
|
|
str(args.baseline_json)
|
|
if args.baseline_json is not None
|
|
else None
|
|
),
|
|
"baseline_sha256": (
|
|
sha256_file(args.baseline_json)
|
|
if args.baseline_json is not None
|
|
else None
|
|
),
|
|
"human_eval_sha256": sha256_file(args.human_eval),
|
|
"gsm8k_sha256": sha256_file(args.gsm8k),
|
|
"model_revision": behavior["model"]["revision"],
|
|
"max_new_tokens": behavior[
|
|
"generation_contract"
|
|
]["max_new_tokens"],
|
|
},
|
|
"sandbox": {
|
|
"image": args.sandbox_image,
|
|
"timeout_seconds": args.timeout_seconds,
|
|
"code_execution_skipped": args.skip_code_execution,
|
|
"network": "none",
|
|
"filesystem": "read-only",
|
|
"user": "65534:65534",
|
|
"capabilities": "ALL dropped",
|
|
"memory": "256m",
|
|
"memory_swap": "256m",
|
|
"pids_limit": 64,
|
|
"cpus": 0.5,
|
|
"tmpfs": "/tmp:rw,noexec,nosuid,size=16m",
|
|
"host_mounts": 0,
|
|
},
|
|
"rows": rows,
|
|
"summary": summarize(rows),
|
|
"claim_boundary": [
|
|
"Four math and four code sources are not benchmark estimates.",
|
|
"Completion-conditioned metrics are selection-biased diagnostics.",
|
|
"A passing HumanEval test is functional evidence, not code-safety evidence.",
|
|
"Fallback last-number extraction is not strict completion.",
|
|
"Counterfactual token sequences are not official-valid chats.",
|
|
],
|
|
"content_hash": canonical_hash(rows),
|
|
}
|
|
args.output.parent.mkdir(parents=True, exist_ok=True)
|
|
args.output.write_text(
|
|
json.dumps(result, ensure_ascii=False, indent=2) + "\n",
|
|
encoding="utf-8",
|
|
)
|
|
payload = args.output.read_bytes()
|
|
print(
|
|
json.dumps(
|
|
{
|
|
"output": str(args.output),
|
|
"sha256": sha256_bytes(payload),
|
|
"bytes": len(payload),
|
|
"summary": result["summary"],
|
|
},
|
|
ensure_ascii=False,
|
|
indent=2,
|
|
)
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|