File size: 3,463 Bytes
a31a731 | 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 | """Verifiable unit-test rewards for coding RL."""
from __future__ import annotations
import subprocess
import sys
from pathlib import Path
from typing import Any
def run_hidden_tests(
solution_code: str,
hidden_tests: list[str],
*,
timeout_s: float = 5.0,
python_exe: str | None = None,
) -> dict[str, Any]:
"""Execute each assert against solution_code; return pass fraction + details."""
py = python_exe or sys.executable
per_test: list[dict[str, Any]] = []
passed = 0
for i, assertion in enumerate(hidden_tests):
script = f"{solution_code}\n{assertion}\n"
try:
proc = subprocess.run(
[py, "-c", script],
capture_output=True,
text=True,
timeout=timeout_s,
)
ok = proc.returncode == 0
err = (proc.stderr or "")[:500]
except Exception as e: # noqa: BLE001
ok = False
err = f"{type(e).__name__}: {e}"
per_test.append({"i": i, "ok": ok, "stderr": err})
if ok:
passed += 1
total = len(hidden_tests)
return {
"pass_frac": (passed / total) if total else 0.0,
"passed": passed,
"total": total,
"per_test": per_test,
}
def extract_python_code(completion: str) -> str:
"""Pull first ```python fenced block, else raw text."""
if "```python" in completion:
start = completion.find("```python") + len("```python")
end = completion.find("```", start)
if end == -1:
return completion[start:].strip()
return completion[start:end].strip()
if "```" in completion:
start = completion.find("```") + 3
end = completion.find("```", start)
if end != -1:
return completion[start:end].strip()
return completion.strip()
def grade_completion(
completion: str,
hidden_tests: list[str],
*,
timeout_s: float = 5.0,
) -> dict[str, Any]:
code = extract_python_code(completion)
grade = run_hidden_tests(code, hidden_tests, timeout_s=timeout_s)
grade["extracted_code"] = code
return grade
def compute_reward(
grade: dict[str, Any],
*,
format_ok: bool = True,
ran_execute: bool = False,
) -> dict[str, float]:
components = {
"pass_frac": float(grade.get("pass_frac", 0.0)),
"format_bonus": 0.05 if format_ok and grade.get("extracted_code") else 0.0,
"tool_bonus": 0.05 if ran_execute else 0.0,
"penalties": 0.0,
}
components["reward_total"] = (
components["pass_frac"]
+ components["format_bonus"]
+ components["tool_bonus"]
- components["penalties"]
)
return components
def grade_solution_file(
solution_path: Path,
hidden_tests: list[str],
*,
timeout_s: float = 5.0,
) -> dict[str, Any]:
if not solution_path.exists():
return {
"pass_frac": 0.0,
"passed": 0,
"total": len(hidden_tests),
"per_test": [],
"error": "solution.py missing",
"extracted_code": "",
}
code = solution_path.read_text(encoding="utf-8")
grade = run_hidden_tests(code, hidden_tests, timeout_s=timeout_s)
grade["extracted_code"] = code
grade["error"] = None
return grade
|