|
|
|
|
| """
|
| 10_code_hard.py - harder coding benchmarks behind `07_domain_eval.py run --suite codehard --hard-n N` (N tasks per benchmark, nested prefixes):
|
| lcb LiveCodeBench code_generation_lite, release 6: 175 AtCoder and LeetCode contest problems from 2025-01 to 2025-04 (newer than most training
|
| data), about 40 hidden tests each; the model writes a whole program (stdin/stdout) or a LeetCode class, every test must pass.
|
| The task id carries the difficulty: lcb_easy_..., lcb_medium_..., lcb_hard_...
|
| crux CRUXEval (800 short Python functions), half output prediction (CRUXEval-O) and half input prediction (CRUXEval-I): code understanding,
|
| checked by running `assert f(...) == ...` on the model's answer.
|
| Public data, no token. Needs `pip install datasets` and network (the LiveCodeBench file is about 1 GB). THE MODEL-WRITTEN CODE IS EXECUTED (see 07).
|
| Use `--max-tokens 4096 --ctx 8192` for the lcb tasks (long answers). Checks:
|
| python 10_code_hard.py selftest lookup-table solutions pass every test of a few lcb problems, wrong ones fail; crux references pass
|
| python 10_code_hard.py validate runs all 800 crux references and prints the ids to put in BAD_CRUX
|
| """
|
| import base64
|
| import io
|
| import json
|
| import os
|
| import pickle
|
| import random
|
| import re
|
| import subprocess
|
| import sys
|
| import tempfile
|
| import time
|
| import zlib
|
| from pathlib import Path
|
|
|
| LCB_FILES = ("test6",)
|
| BAD_CRUX = set()
|
| HEADER = ("from string import *\nfrom re import *\nfrom datetime import *\nfrom collections import *\nfrom heapq import *\nfrom bisect import *\n"
|
| "from copy import deepcopy\nfrom math import *\nfrom typing import *\nfrom functools import *\nfrom itertools import *\nfrom operator import *\n"
|
| "from random import *\nfrom statistics import *\nimport sys, heapq, bisect, math, collections, itertools, functools, random, string, re, json, copy\n")
|
| FUNC_RUNNER = '''import json, sys
|
| ns = {}
|
| exec(compile(open("sol.py", encoding="utf-8").read(), "sol.py", "exec"), ns)
|
| f = getattr(ns["Solution"](), sys.argv[1])
|
| for inp, out in json.load(open("tests.json", encoding="utf-8")):
|
| got = f(*[json.loads(l) for l in inp.split("\\n") if l.strip()])
|
| if json.loads(json.dumps(got)) != json.loads(out):
|
| print("FAIL")
|
| sys.exit(1)
|
| print("OK")
|
| '''
|
|
|
|
|
|
|
| class _Plain(pickle.Unpickler):
|
| """The private tests are a pickled JSON string; a string needs no globals, so refuse every one (a dataset must not run code while loading)."""
|
| def find_class(self, module, name):
|
| raise pickle.UnpicklingError(f"global {module}.{name} refused")
|
|
|
|
|
| def decode_private(s):
|
| return json.loads(_Plain(io.BytesIO(zlib.decompress(base64.b64decode(s)))).load())
|
|
|
|
|
| def lcb_tests(row):
|
| return json.loads(row["public_test_cases"]) + decode_private(row["private_test_cases"])
|
|
|
|
|
| def _same(got, want):
|
| g, w = [l.rstrip() for l in got.strip().splitlines()], [l.rstrip() for l in want.strip().splitlines()]
|
| if len(g) != len(w):
|
| return False
|
| for a, b in zip(g, w):
|
| if a == b:
|
| continue
|
| ta, tb = a.split(), b.split()
|
| if len(ta) != len(tb):
|
| return False
|
| for x, y in zip(ta, tb):
|
| if x == y:
|
| continue
|
| if re.fullmatch(r"-?\d+", x) and re.fullmatch(r"-?\d+", y):
|
| return False
|
| try:
|
| if abs(float(x) - float(y)) > 1e-6 * max(1.0, abs(float(y))):
|
| return False
|
| except ValueError:
|
| return False
|
| return True
|
|
|
|
|
| def _limits():
|
| if os.name != "posix":
|
| return None
|
| import resource
|
| return lambda: resource.setrlimit(resource.RLIMIT_AS, (2 << 30, 2 << 30))
|
|
|
|
|
| def run_lcb(code, tests, func, total=90, per=8):
|
| """True when `code` passes every test: func None = a stdin/stdout program (one process per test), else a LeetCode Solution method."""
|
| env = {**os.environ, "OPENBLAS_NUM_THREADS": "1", "OMP_NUM_THREADS": "1"}
|
| with tempfile.TemporaryDirectory() as d:
|
| d = Path(d)
|
| (d / "sol.py").write_text(HEADER + code, encoding="utf-8")
|
| t0 = time.time()
|
| if func is None:
|
| for t in tests:
|
| if time.time() - t0 > total:
|
| return False
|
| try:
|
| r = subprocess.run([sys.executable, "-I", "sol.py"], input=t["input"], capture_output=True, text=True, timeout=per, cwd=d, env=env,
|
| preexec_fn=_limits())
|
| except (subprocess.TimeoutExpired, UnicodeDecodeError):
|
| return False
|
| if r.returncode != 0 or not _same(r.stdout, t["output"]):
|
| return False
|
| return True
|
| (d / "tests.json").write_text(json.dumps([[t["input"], t["output"]] for t in tests]), encoding="utf-8")
|
| (d / "run.py").write_text(FUNC_RUNNER, encoding="utf-8")
|
| try:
|
| r = subprocess.run([sys.executable, "-I", "run.py", func], capture_output=True, text=True, timeout=total, cwd=d, env=env, preexec_fn=_limits())
|
| except (subprocess.TimeoutExpired, UnicodeDecodeError):
|
| return False
|
| return r.returncode == 0 and r.stdout.strip().endswith("OK")
|
|
|
|
|
| FUNC_DEBUG = '''import json, sys, traceback
|
| ns = {}
|
| exec(compile(open("sol.py", encoding="utf-8").read(), "sol.py", "exec"), ns)
|
| f = getattr(ns["Solution"](), sys.argv[1])
|
| for inp, out in json.load(open("tests.json", encoding="utf-8")):
|
| try:
|
| got = f(*[json.loads(l) for l in inp.split("\\n") if l.strip()])
|
| except Exception:
|
| print("Input:\\n" + inp.strip()[:300] + "\\nRuntime error:\\n" + traceback.format_exc()[-600:])
|
| sys.exit(1)
|
| if json.loads(json.dumps(got)) != json.loads(out):
|
| print("Input:\\n" + inp.strip()[:300] + "\\nExpected output:\\n" + out.strip()[:300] + "\\nYour output:\\n" + json.dumps(got)[:300])
|
| sys.exit(1)
|
| print("OK")
|
| '''
|
|
|
|
|
| def first_failure(code, tests, func, per=8):
|
| """None when `code` passes every test in `tests` (the VISIBLE sample tests), else a short text on the first failure, for a repair prompt."""
|
| env = {**os.environ, "OPENBLAS_NUM_THREADS": "1", "OMP_NUM_THREADS": "1"}
|
| with tempfile.TemporaryDirectory() as d:
|
| d = Path(d)
|
| (d / "sol.py").write_text(HEADER + code, encoding="utf-8")
|
| if func is None:
|
| for t in tests:
|
| shown = "Input:\n" + t["input"].strip()[:300]
|
| try:
|
| r = subprocess.run([sys.executable, "-I", "sol.py"], input=t["input"], capture_output=True, text=True, timeout=per, cwd=d, env=env,
|
| preexec_fn=_limits())
|
| except subprocess.TimeoutExpired:
|
| return f"{shown}\nThe program did not finish within {per} seconds."
|
| except UnicodeDecodeError:
|
| return f"{shown}\nThe program printed bytes that are not valid text."
|
| if r.returncode != 0:
|
| return f"{shown}\nRuntime error:\n{r.stderr[-600:]}"
|
| if not _same(r.stdout, t["output"]):
|
| return f"{shown}\nExpected output:\n{t['output'].strip()[:300]}\nYour output:\n{r.stdout.strip()[:300]}"
|
| return None
|
| (d / "tests.json").write_text(json.dumps([[t["input"], t["output"]] for t in tests]), encoding="utf-8")
|
| (d / "run.py").write_text(FUNC_DEBUG, encoding="utf-8")
|
| try:
|
| r = subprocess.run([sys.executable, "-I", "run.py", func], capture_output=True, text=True, timeout=per * 4, cwd=d, env=env, preexec_fn=_limits())
|
| except subprocess.TimeoutExpired:
|
| return "The code did not finish in time on the sample tests."
|
| except UnicodeDecodeError:
|
| return "The code printed bytes that are not valid text."
|
| if r.returncode == 0 and r.stdout.strip().endswith("OK"):
|
| return None
|
| return (r.stdout or r.stderr).strip()[-800:] or "The code crashed."
|
|
|
|
|
| def lcb_feedback(resp, row, func):
|
| """What to tell the model after a failed attempt: only the visible sample tests are used (never the hidden ones); None = no new information."""
|
| code = last_block(resp)
|
| if not code.strip():
|
| return "I could not find a ```python code block in your answer. Give the complete solution in one ```python block."
|
| bad = first_failure(code, json.loads(row["public_test_cases"]), func)
|
| if bad is None:
|
| return None
|
| return "Your code fails this sample test.\n\n" + bad + "\n\nFix the code. Reply with the complete corrected solution in one ```python block."
|
|
|
|
|
| def last_block(resp):
|
| blocks = re.findall(r"```(?:python|py|Python)?[ \t]*\n(.*?)```", resp, flags=re.S)
|
| return blocks[-1] if blocks else ""
|
|
|
|
|
| def lcb_prompt(row):
|
| if row["starter_code"]:
|
| fmt = "You will use the following starter code to write the solution to the problem and enclose your code within delimiters."
|
| block = row["starter_code"]
|
| else:
|
| fmt = ("Read the inputs from stdin solve the problem and write the answer to stdout (do not directly test on the sample inputs). Enclose your "
|
| "code within delimiters as follows. Ensure that when the python program runs, it reads the inputs, runs the algorithm and writes output "
|
| "to STDOUT.")
|
| block = "# YOUR CODE HERE"
|
| return f"### Question:\n{row['question_content']}\n\n### Format: {fmt}\n```python\n{block}\n```\n\n### Answer: (use the provided format with backticks)\n"
|
|
|
|
|
| def lcb_rows():
|
| from datasets import load_dataset
|
| rows = []
|
| for f in LCB_FILES:
|
| rows += list(load_dataset("json", data_files=f"hf://datasets/livecodebench/code_generation_lite/{f}.jsonl", split="train"))
|
| return rows
|
|
|
|
|
| def pool_lcb(rnd):
|
| rows = lcb_rows()
|
| rnd.shuffle(rows)
|
| T = []
|
| for r in rows:
|
| func = json.loads(r["metadata"] or "{}").get("func_name") if r["starter_code"] else None
|
|
|
| def check(resp, r=r, func=func, cache={}):
|
| code = last_block(resp)
|
| if not code.strip():
|
| return False
|
| if r["question_id"] not in cache:
|
| cache.clear()
|
| cache[r["question_id"]] = lcb_tests(r)
|
| return run_lcb(code, cache[r["question_id"]], func)
|
| check.feedback = lambda resp, r=r, func=func: lcb_feedback(resp, r, func)
|
| T.append((f"lcb_{r['difficulty']}_{r['question_id']}", "lcb", lcb_prompt(r), check, None))
|
| return T
|
|
|
|
|
|
|
| def pool_crux(rnd, run_code):
|
| from datasets import load_dataset
|
| rows = [r for r in load_dataset("cruxeval-org/cruxeval", split="test")]
|
| ro, ri = rows[:], rows[:]
|
| rnd.shuffle(ro)
|
| rnd.shuffle(ri)
|
| O, I = [], []
|
| for r in ro:
|
| if f"crux_o_{r['id']}" in BAD_CRUX:
|
| continue
|
| pro = ("You are given a Python function and an assertion containing an input to the function. Complete the assertion with a literal (no "
|
| "unsimplified expressions, no function calls) containing the output when executing the provided code on the given input, even if the "
|
| "function is incorrect or incomplete. Think step by step, then end with exactly one line in this form: [ANSWER] assert f(...) == <literal> "
|
| f"[/ANSWER]\n\n```python\n{r['code']}\nassert f({r['input']}) == ??\n```")
|
|
|
| def check_o(resp, r=r):
|
| m = re.findall(r"\[ANSWER\]\s*assert\s+f\(.*?\)\s*==\s*(.*?)\s*\[/ANSWER\]", resp, flags=re.S)
|
| return bool(m) and run_code(r["code"], f"assert f({r['input']}) == ({m[-1]})", 10)
|
| O.append((f"crux_o_{r['id']}", "crux", pro, check_o, f"[ANSWER] assert f({r['input']}) == {r['output']} [/ANSWER]"))
|
| for r in ri:
|
| if f"crux_i_{r['id']}" in BAD_CRUX:
|
| continue
|
| pri = ("You are given a Python function and an assertion containing an output of the function. Find any input such that executing the function "
|
| "on the input leads to the given output. There may be several answers, give only one. Think step by step, then end with exactly one line "
|
| f"in this form: [ANSWER] assert f(<your input>) == {r['output']} [/ANSWER]\n\n```python\n{r['code']}\nassert f(??) == {r['output']}\n```")
|
|
|
| def check_i(resp, r=r):
|
| m = re.findall(r"\[ANSWER\]\s*assert\s+f\((.*)\)\s*==", resp, flags=re.S)
|
| return bool(m) and run_code(r["code"], f"assert f({m[-1]}) == {r['output']}", 10)
|
| I.append((f"crux_i_{r['id']}", "crux", pri, check_i, f"[ANSWER] assert f({r['input']}) == {r['output']} [/ANSWER]"))
|
| mixed, i = [], 0
|
| while i < min(len(O), len(I)):
|
| mixed += [O[i], I[i]]
|
| i += 1
|
| return mixed
|
|
|
|
|
|
|
| def build_codehard(seed, n, run_code, code_of, num):
|
| """[(id, category, prompt, check(response) -> bool, reference response or None)]: the first n lcb problems and the first n crux tasks."""
|
| T = pool_lcb(random.Random(f"{seed}lcb"))[:n] + pool_crux(random.Random(f"{seed}crux"), run_code)[:n]
|
| assert len({t[0] for t in T}) == len(T), "duplicate task ids"
|
| return T
|
|
|
|
|
| def _lookup_solution(row, tests):
|
| """A 'solution' that only looks the answers up: it passes iff the runner, the decoding and the comparison are right."""
|
| if not row["starter_code"]:
|
| return "import sys\nD = " + repr({t["input"]: t["output"] for t in tests}) + "\nprint(D[sys.stdin.read()], end='')\n", None
|
| func = json.loads(row["metadata"])["func_name"]
|
| D = {json.dumps([json.loads(l) for l in t["input"].split("\n") if l.strip()]): t["output"] for t in tests}
|
| return f"import json\nD = {D!r}\nclass Solution:\n def {func}(self, *a):\n return json.loads(D[json.dumps(list(a))])\n", func
|
|
|
|
|
| if __name__ == "__main__" and sys.argv[1:] in (["selftest"], ["validate"]):
|
| import importlib
|
| sys.path.insert(0, Path(__file__).resolve().parent.as_posix())
|
| m = importlib.import_module("07_domain_eval")
|
| if sys.argv[1] == "selftest":
|
| rows = lcb_rows()
|
| stdin, func = [r for r in rows if not r["starter_code"]][:4], [r for r in rows if r["starter_code"]][:4]
|
| for r in stdin + func:
|
| tests = lcb_tests(r)
|
| code, fn = _lookup_solution(r, tests)
|
| assert run_lcb(code, tests, fn), ("lookup solution failed", r["question_id"])
|
| wrong = "print(0)\n" if fn is None else f"class Solution:\n def {fn}(self, *a):\n return None\n"
|
| assert not run_lcb(wrong, tests, fn), ("wrong solution passed", r["question_id"])
|
| assert last_block("text\n```python\nA\n```\nmore\n```python\nB\n```") == "B\n"
|
| for r in stdin + func:
|
| tests = json.loads(r["public_test_cases"])
|
| good, fn = _lookup_solution(r, tests)
|
| assert lcb_feedback("```python\n" + good + "```", r, fn) is None, ("feedback for a correct solution", r["question_id"])
|
| bad_code = "print(0)\n" if fn is None else f"class Solution:\n def {fn}(self, *a):\n return None\n"
|
| crash = "raise ValueError('boom')\n" if fn is None else f"class Solution:\n def {fn}(self, *a):\n raise ValueError('boom')\n"
|
| assert "sample test" in (lcb_feedback("```python\n" + bad_code + "```", r, fn) or ""), ("no feedback for a wrong answer", r["question_id"])
|
| assert "boom" in (lcb_feedback("```python\n" + crash + "```", r, fn) or ""), ("no error text for a crash", r["question_id"])
|
| assert "could not find" in lcb_feedback("no code here", r, fn)
|
| T = build_codehard(1, 10, m.run_code, m.code_of, m.num)
|
| crux = [t for t in T if t[1] == "crux"]
|
| bad = [t[0] for t in crux if not t[3](t[4])]
|
| junk = [t[0] for t in T if t[1] == "crux" and t[3]("zzz 987654321")] + [t[0] for t in T if t[1] == "lcb" and t[3]("zzz 987654321")]
|
| per = {c: sum(t[1] == c for t in T) for c in ("lcb", "crux")}
|
| assert not bad and not junk and per == {"lcb": 10, "crux": 10}, (bad, junk, per)
|
| print(f"selftest OK: {len(T)} tasks {per}; {len(stdin)} stdin + {len(func)} functional lcb problems pass their lookup solution and fail a wrong one; "
|
| f"crux references pass, garbage fails")
|
| else:
|
| from datasets import load_dataset
|
| bad = []
|
| for r in load_dataset("cruxeval-org/cruxeval", split="test"):
|
| if not m.run_code(r["code"], f"assert f({r['input']}) == ({r['output']})", 10):
|
| bad += [f"crux_o_{r['id']}", f"crux_i_{r['id']}"]
|
| print(f"{len(bad) // 2} crux references fail: BAD_CRUX = {set(bad)!r}")
|
| n_ok = sum(1 for r in lcb_rows() if lcb_tests(r))
|
| print(f"lcb: {n_ok} problems decode their tests")
|
|
|
