Add extended task for LiveCodeBench codegeneration

src/lighteval/tasks/extended/__init__.py

@@ -25,12 +25,13 @@
 
 if can_load_extended_tasks():
     import lighteval.tasks.extended.ifeval.main as ifeval
+    import lighteval.tasks.extended.lcb.main as lcb
     import lighteval.tasks.extended.mix_eval.main as mix_eval
     import lighteval.tasks.extended.mt_bench.main as mt_bench
     import lighteval.tasks.extended.olympiade_bench.main as olympiad_bench
     import lighteval.tasks.extended.tiny_benchmarks.main as tiny_benchmarks
 
-    AVAILABLE_EXTENDED_TASKS_MODULES = [ifeval, tiny_benchmarks, mt_bench, mix_eval, olympiad_bench]
+    AVAILABLE_EXTENDED_TASKS_MODULES = [ifeval, tiny_benchmarks, mt_bench, mix_eval, olympiad_bench, lcb]
 
 else:
     AVAILABLE_EXTENDED_TASKS_MODULES = []

src/lighteval/tasks/extended/lcb/lcb_utils.py

@@ -0,0 +1,292 @@
+# MIT License
+
+# Copyright (c) 2025 The HuggingFace Team
+
+# Permission is hereby granted, free of charge, to any person obtaining a copy
+# of this software and associated documentation files (the "Software"), to deal
+# in the Software without restriction, including without limitation the rights
+# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+# copies of the Software, and to permit persons to whom the Software is
+# furnished to do so, subject to the following conditions:
+
+# The above copyright notice and this permission notice shall be included in all
+# copies or substantial portions of the Software.
+
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+# SOFTWARE.
+
+import base64
+import copy
+import faulthandler
+import io
+import json
+import multiprocessing
+import pickle
+import sys
+import time
+import zlib
+from typing import Dict, Optional
+
+
+def translate_private_test_cases(encoded_data: str) -> Dict[str, str]:
+    decoded_data = base64.b64decode(encoded_data)
+    decompressed_data = zlib.decompress(decoded_data)
+    original_data = pickle.loads(decompressed_data)
+    return json.loads(original_data)
+
+
+def reliability_guard(maximum_memory_bytes: Optional[int] = None):
+    """
+    This disables various destructive functions and prevents the generated code
+    from interfering with the test (e.g. fork bomb, killing other processes,
+    removing filesystem files, etc.)
+
+    WARNING
+    This function is NOT a security sandbox. Untrusted code, including, model-
+    generated code, should not be blindly executed outside of one. See the
+    Codex paper for more information about OpenAI's code sandbox, and proceed
+    with caution.
+    """
+
+    faulthandler.disable()
+
+    import builtins
+
+    builtins.exit = None
+    builtins.quit = None
+
+    import os
+
+    os.environ["OMP_NUM_THREADS"] = "1"
+
+    os.kill = None
+    os.system = None
+    os.putenv = None
+    os.remove = None
+    os.removedirs = None
+    os.rmdir = None
+    os.fchdir = None
+    os.setuid = None
+    os.fork = None
+    os.forkpty = None
+    os.killpg = None
+    os.rename = None
+    os.renames = None
+    os.truncate = None
+    os.replace = None
+    os.unlink = None
+    os.fchmod = None
+    os.fchown = None
+    os.chmod = None
+    os.chown = None
+    os.chroot = None
+    os.fchdir = None
+    os.lchflags = None
+    os.lchmod = None
+    os.lchown = None
+    os.getcwd = None
+    os.chdir = None
+
+    import shutil
+
+    shutil.rmtree = None
+    shutil.move = None
+    shutil.chown = None
+
+    import subprocess
+
+    subprocess.Popen = None  # type: ignore
+
+    # __builtins__["help"] = None   # this line is commented out as it results into error
+
+    import sys
+
+    sys.modules["ipdb"] = None
+    sys.modules["joblib"] = None
+    sys.modules["resource"] = None
+    sys.modules["psutil"] = None
+    sys.modules["tkinter"] = None
+
+
+def post_process_code(code: str) -> str:
+    code = code.split("</code>")[0]
+    code = code.replace("```python", "")
+    code = code.split("```")[0]
+    code = code.replace("<code>", "")
+    return code
+
+
+def prepare_test_input_output_std(test_case):
+    test_input = test_case["input"]
+    test_output = test_case["output"].strip()
+    if test_output.endswith("-"):
+        test_output = test_output[: test_output.rfind("-")].rstrip()  # Remove '-' if present and trailing
+    return test_input, test_output
+
+
+def run_test_func(completion, is_extracted: bool, test_input, test_output):
+    namespace = {}
+    exec(completion, namespace)
+    func_name = completion.split("(")[0].split()[-1]
+
+    output = io.StringIO()
+    sys.stdout = output
+
+    try:
+        if not is_extracted:
+            if isinstance(test_input, dict):
+                result_output = namespace[func_name](**test_input)
+            else:
+                result_output = namespace[func_name](test_input)
+        else:
+            result_output = namespace[func_name](*test_input)
+
+        if result_output != test_output:
+            return False, result_output
+
+        return True, result_output
+
+    except Exception as e:
+        error_msg = f"Error: {str(e)}" if not is_extracted else str(e)
+        return False, error_msg
+
+    finally:
+        sys.stdout = sys.__stdout__
+
+
+def run_test_std(completion, test_input, test_output):
+    with io.StringIO() as output:
+        sys.stdout = output
+        sys.stdin = io.StringIO(test_input)
+        try:
+            exec(f'__name__ = "__main__"\n{completion}' if '__name__ == "__main__"' in completion else completion, {})
+            return output.getvalue().strip() == test_output, output.getvalue().strip()
+        finally:
+            sys.stdout = sys.__stdout__
+
+
+def prepare_test_input_output_functional(test_case, is_extracted: bool):  # noqa: C901
+    if not is_extracted:
+        # Extract input and expected output from JSON directly
+        test_input = test_case["input"]
+        test_output = test_case["output"]
+        return test_input, test_output
+    else:
+        # Robustly process complex inputs
+        input_str = test_case["input"]
+        expected_output = test_case["output"].strip()
+        inputs = []
+
+        if "=" in input_str:
+            parts = input_str.split(",") if "," in input_str else [input_str]
+            for part in parts:
+                _, value = map(str.strip, part.split("="))
+                try:
+                    value = int(value)
+                except ValueError:
+                    try:
+                        value = float(value)
+                    except ValueError:
+                        value = value.strip('"')
+                inputs.append(value)
+        else:
+            for line in input_str.split("\n"):
+                line = line.strip()
+                if not line:
+                    continue
+                if line.startswith('"') and line.endswith('"'):
+                    inputs.append(line.strip('"'))
+                    continue
+                if line.startswith("[") and line.endswith("]"):
+                    inputs.append(json.loads(line))
+                    continue
+                try:
+                    inputs.append(int(line))
+                except ValueError:
+                    try:
+                        inputs.append(float(line))
+                    except ValueError:
+                        inputs.append(line)
+
+        try:
+            expected_output = json.loads(expected_output)
+        except json.JSONDecodeError:
+            expected_output = expected_output.strip()
+        return inputs, expected_output
+
+
+def run_tests_for_one_example(test_cases, completion, result_list, is_extracted: bool):
+    time_elapsed = float("inf")
+    test_type = test_cases[0]["testtype"]
+    reliability_guard()
+    for i, test_case in enumerate(test_cases):
+        output_error = ""
+        output_value = ""
+        try:
+            time_start = time.time()
+            if test_type == "functional":
+                test_input, test_output = prepare_test_input_output_functional(test_case, is_extracted)
+                passed, output_value = run_test_func(
+                    completion, is_extracted, copy.deepcopy(test_input), copy.deepcopy(test_output)
+                )
+            else:
+                test_input, test_output = prepare_test_input_output_std(test_case)
+                passed, output_value = run_test_std(completion, copy.deepcopy(test_input), copy.deepcopy(test_output))
+            time_elapsed = time.time() - time_start
+            if not passed:
+                output_error = (
+                    f"For test input: {test_input}. Expected output is: {test_output}, but got: {output_value}."
+                )
+
+        except Exception as e:
+            passed = False
+            output_error = f"For test input: {test_input}. Expected output is: {test_output}, but got error: {e}."
+            output_value = f"Error: {e}."
+        if output_error == "":
+            output_error = f"For test input: {test_input}. Expected output is: {test_output}, your solution correctly passes this test with output {output_value}."
+        result_list.append((passed, output_error, output_value, time_elapsed))
+        if not passed:
+            return
+
+
+def lcb_run(problem: Dict[str, str], completion, timeout: float, is_extracted: bool):
+    test_cases = problem["test"]
+    manager = multiprocessing.Manager()
+    result = manager.list()
+    p = multiprocessing.Process(target=run_tests_for_one_example, args=(test_cases, completion, result, is_extracted))
+    p.start()
+    p.join(timeout=(timeout + 1) * len(test_cases) + 5)
+    if p.is_alive():
+        p.kill()
+
+    for _ in range(len(test_cases) - len(result)):
+        result.append((False, "Time out!.", "Error: Time out!", float("inf")))
+    return result
+
+
+def check_correctness(problem, completion, timeout: float, is_extracted: bool = False) -> bool:
+    """Evaluates the functional correctness of a completion by running the test
+    suite provided in the problem.
+    """
+    result_list = lcb_run(problem, completion, timeout, is_extracted)
+    details = [r[0] for r in result_list]
+    all_passed = all(details)
+
+    result = ""
+    if result_list and all_passed:
+        result = "passed"
+
+    return result == "passed"
+
+
+def extract_code(model_output: str) -> str:
+    outputlines = model_output.split("\n")
+    indexlines = [i for i, line in enumerate(outputlines) if "```" in line]
+    if len(indexlines) < 2:
+        return ""
+    return "\n".join(outputlines[indexlines[-2] + 1 : indexlines[-1]])