jiangzhs/OpenCompass
1
0
Fork 0
mirror of https://github.com/open-compass/opencompass.git synced 2025-05-30 16:03:24 +08:00
Code Issues Actions 2 Packages Projects Releases Wiki Activity

[Feature] Support LiveCodeBench (#1617)

Browse Source 
* Update

* Update LCB

* Update

* Update

* Update

* Update

* Update
This commit is contained in:
Songyang Zhang 2024-10-21 20:50:39 +08:00 committed by GitHub
parent 5868d5afa4
commit a4d5a6c81b
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
20 changed files with 2537 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

42
configs/datasets/livecodebench/README.md Normal file
View File

@ -0,0 +1,42 @@
# LiveCodeBench
## Dataset
LiveCodeBench provides holistic and contamination-free evaluation of coding capabilities of LLMs. Particularly, LiveCodeBench continuously collects new problems over time from contests across three competition platforms -- LeetCode, AtCoder, and CodeForces. Next, LiveCodeBench also focuses on a broader range of code-related capabilities, such as self-repair, code execution, and test output prediction, beyond just code generation. Currently, LiveCodeBench hosts four hundred high-quality coding problems that were published between May 2023 and March 2024.
- Origin Project: https://livecodebench.github.io/leaderboard.html
## Setting
| Model Type | Code Generation | Test Output Prediction | Code Execution |
|------------|--------|--------|--------|
| Base Model | ❌ | ❌ | ❌ |
| Chat Model | ✅ | ✅ | ✅ |
## Baseline Performance
| Model Type | Code Generation(pass@1) | Test Output Prediction(pass@1) | Code Execution(pass@1) |
|------------|--------|--------|--------|
| Qwen2.5-7B-Instruct(HF) | 39.25 | 48.64 | 41.96 |
| Meta-Llama-3.1-8B-Instruct(HF) | 20.25 | 24.66 | 17.12 |
## Citation
```bibtex
@article{jain2024livecodebench,
author = {Naman Jain, King Han, Alex Gu, Wen-Ding Li, Fanjia Yan, Tianjun Zhang, Sida Wang, Armando Solar-Lezama, Koushik Sen, Ion Stoica},
title = {LiveCodeBench: Holistic and Contamination Free Evaluation of Large Language Models for Code},
year = {2024},
journal = {arXiv preprint},
}
@misc{2023opencompass,
title={OpenCompass: A Universal Evaluation Platform for Foundation Models},
author={OpenCompass Contributors},
howpublished = {\url{https://github.com/open-compass/opencompass}},
year={2023}
}
```

4
configs/datasets/livecodebench/livecodebench_gen.py Normal file
View File

@ -0,0 +1,4 @@
from mmengine.config import read_base
with read_base():
from .livecodebench_gen_b2b0fd import LCB_datasets # noqa: F401, F403

163
configs/datasets/livecodebench/livecodebench_gen_b2b0fd.py Normal file
View File

@ -0,0 +1,163 @@
from opencompass.openicl.icl_prompt_template import PromptTemplate
from opencompass.openicl.icl_retriever import ZeroRetriever
from opencompass.openicl.icl_inferencer import GenInferencer
from opencompass.datasets import (
LCBCodeGenerationDataset,
LCBCodeExecutionDataset,
LCBTestOutputPredictionDataset,
LCBCodeGenerationEvaluator,
LCBCodeExecutionEvaluator,
LCBTestOutputEvaluator
)
from opencompass.datasets.livecodebench import TestOutputPromptConstants
lcb_code_generation_reader_cfg = dict(
input_columns=[
'question_content',
'format_prompt',
],
# output_column='evaluation_sample',
output_column='question_id',
)
SYSTEM_MESSAGE_GENERIC = f'You are an expert Python programmer. You will be given a question (problem specification) and will generate a correct Python program that matches the specification and passes all tests. You will NOT return anything except for the program.'
prompt_template = '### Question:\n{question_content}\n\n{format_prompt}' + \
'### Answer: (use the provided format with backticks)\n\n'
# Code Generation Tasks
lcb_code_generation_infer_cfg = dict(
prompt_template=dict(
type=PromptTemplate,
template=dict(
round=[
dict(
role='HUMAN',
prompt=prompt_template
)
]
)
),
retriever=dict(type=ZeroRetriever),
inferencer=dict(type=GenInferencer, max_out_len=1024)
)
lcb_code_generation_eval_cfg = dict(
evaluator=dict(
type=LCBCodeGenerationEvaluator,
num_process_evaluate=4,
timeout=6,
),
pred_role='BOT',
)
LCBCodeGeneration_dataset = dict(
type=LCBCodeGenerationDataset,
abbr='lcb_code_generation',
path='opencompass/code_generation_lite',
reader_cfg=lcb_code_generation_reader_cfg,
infer_cfg=lcb_code_generation_infer_cfg,
eval_cfg=lcb_code_generation_eval_cfg
)
# Code Execution Dataset
lcb_code_execution_reader_cfg = dict(
input_columns=[
'prompt',
],
output_column='evaluation_sample',
)
lcb_code_execution_infer_cfg = dict(
prompt_template=dict(
type=PromptTemplate,
template=dict(
begin=[
dict(
role='SYSTEM',
prompt='You are an expert at Python programming, code execution, test case generation, and fuzzing.'
),
],
round=[
dict(
role='HUMAN',
prompt='{prompt}'
)
]
)
),
retriever=dict(type=ZeroRetriever),
inferencer=dict(type=GenInferencer, max_out_len=1024)
)
lcb_code_execution_eval_cfg = dict(
evaluator=dict(
type=LCBCodeExecutionEvaluator,
),
pred_role='BOT',
)
LCBCodeExecution_dataset = dict(
type=LCBCodeExecutionDataset,
abbr='lcb_code_execution',
path='opencompass/execution-v2',
reader_cfg=lcb_code_execution_reader_cfg,
infer_cfg=lcb_code_execution_infer_cfg,
eval_cfg=lcb_code_execution_eval_cfg,
)
# TestOuputput Dataset
lcb_test_output_reader_cfg = dict(
input_columns=[
'prompt',
],
output_column='evaluation_sample',
)
system_prompt = 'You are an expert Python programmer. You will be given a question (problem specification) and will generate a correct Python program that matches the specification and passes all tests. You will NOT return anything except for the program.'
lcb_test_output_infer_cfg = dict(
prompt_template=dict(
type=PromptTemplate,
template=dict(
# begin=[
# dict(
# role='SYSTEM',
# prompt=system_prompt
# ),
# ],
round=[
dict(
role='HUMAN',
prompt='{prompt}'
)
]
)
),
retriever=dict(type=ZeroRetriever),
inferencer=dict(type=GenInferencer, max_out_len=1024)
)
lcb_test_output_eval_cfg = dict(
evaluator=dict(
type=LCBTestOutputEvaluator,
),
pred_role='BOT',
)
LCBTestOutput_dataset = dict(
type=LCBTestOutputPredictionDataset,
abbr='lcb_test_output',
path='opencompass/test_generation',
reader_cfg=lcb_test_output_reader_cfg,
infer_cfg=lcb_test_output_infer_cfg,
eval_cfg=lcb_test_output_eval_cfg,
)
LCB_datasets = [
LCBCodeGeneration_dataset,
LCBCodeExecution_dataset,
LCBTestOutput_dataset,
]

9
configs/models/hf_llama/hf_llama3_1_8b_instruct.py
View File

@ -1,5 +1,13 @@
from opencompass.models import HuggingFacewithChatTemplate
api_meta_template = dict(
round=[
dict(role='HUMAN', api_role='HUMAN'),
dict(role='BOT', api_role='BOT', generate=True),
],
reserved_roles=[dict(role='SYSTEM', api_role='SYSTEM')],
)
models = [
dict(
type=HuggingFacewithChatTemplate,
@ -9,5 +17,6 @@ models = [
batch_size=8,
run_cfg=dict(num_gpus=1),
stop_words=['<|end_of_text|>', '<|eot_id|>'],
meta_template=api_meta_template,
)
]

42
opencompass/configs/datasets/livecodebench/README.md Normal file
View File

@ -0,0 +1,42 @@
# LiveCodeBench
## Dataset
LiveCodeBench provides holistic and contamination-free evaluation of coding capabilities of LLMs. Particularly, LiveCodeBench continuously collects new problems over time from contests across three competition platforms -- LeetCode, AtCoder, and CodeForces. Next, LiveCodeBench also focuses on a broader range of code-related capabilities, such as self-repair, code execution, and test output prediction, beyond just code generation. Currently, LiveCodeBench hosts four hundred high-quality coding problems that were published between May 2023 and March 2024.
- Origin Project: https://livecodebench.github.io/leaderboard.html
## Setting
| Model Type | Code Generation | Test Output Prediction | Code Execution |
|------------|--------|--------|--------|
| Base Model | ❌ | ❌ | ❌ |
| Chat Model | ✅ | ✅ | ✅ |
## Baseline Performance
| Model Type | Code Generation(pass@1) | Test Output Prediction(pass@1) | Code Execution(pass@1) |
|------------|--------|--------|--------|
| Qwen2.5-7B-Instruct(HF) | 39.25 | 48.64 | 41.96 |
| Meta-Llama-3.1-8B-Instruct(HF) | 20.25 | 24.66 | 17.12 |
## Citation
```bibtex
@article{jain2024livecodebench,
author = {Naman Jain, King Han, Alex Gu, Wen-Ding Li, Fanjia Yan, Tianjun Zhang, Sida Wang, Armando Solar-Lezama, Koushik Sen, Ion Stoica},
title = {LiveCodeBench: Holistic and Contamination Free Evaluation of Large Language Models for Code},
year = {2024},
journal = {arXiv preprint},
}
@misc{2023opencompass,
title={OpenCompass: A Universal Evaluation Platform for Foundation Models},
author={OpenCompass Contributors},
howpublished = {\url{https://github.com/open-compass/opencompass}},
year={2023}
}
```

4
opencompass/configs/datasets/livecodebench/livecodebench_gen.py Normal file
View File

@ -0,0 +1,4 @@
from mmengine.config import read_base
with read_base():
from .livecodebench_gen_b2b0fd import LCB_datasets # noqa: F401, F403

163
opencompass/configs/datasets/livecodebench/livecodebench_gen_b2b0fd.py Normal file
View File

@ -0,0 +1,163 @@
from opencompass.openicl.icl_prompt_template import PromptTemplate
from opencompass.openicl.icl_retriever import ZeroRetriever
from opencompass.openicl.icl_inferencer import GenInferencer
from opencompass.datasets import (
LCBCodeGenerationDataset,
LCBCodeExecutionDataset,
LCBTestOutputPredictionDataset,
LCBCodeGenerationEvaluator,
LCBCodeExecutionEvaluator,
LCBTestOutputEvaluator
)
from opencompass.datasets.livecodebench import TestOutputPromptConstants
lcb_code_generation_reader_cfg = dict(
input_columns=[
'question_content',
'format_prompt',
],
# output_column='evaluation_sample',
output_column='question_id',
)
SYSTEM_MESSAGE_GENERIC = f'You are an expert Python programmer. You will be given a question (problem specification) and will generate a correct Python program that matches the specification and passes all tests. You will NOT return anything except for the program.'
prompt_template = '### Question:\n{question_content}\n\n{format_prompt}' + \
'### Answer: (use the provided format with backticks)\n\n'
# Code Generation Tasks
lcb_code_generation_infer_cfg = dict(
prompt_template=dict(
type=PromptTemplate,
template=dict(
round=[
dict(
role='HUMAN',
prompt=prompt_template
)
]
)
),
retriever=dict(type=ZeroRetriever),
inferencer=dict(type=GenInferencer, max_out_len=1024)
)
lcb_code_generation_eval_cfg = dict(
evaluator=dict(
type=LCBCodeGenerationEvaluator,
num_process_evaluate=4,
timeout=6,
),
pred_role='BOT',
)
LCBCodeGeneration_dataset = dict(
type=LCBCodeGenerationDataset,
abbr='lcb_code_generation',
path='opencompass/code_generation_lite',
reader_cfg=lcb_code_generation_reader_cfg,
infer_cfg=lcb_code_generation_infer_cfg,
eval_cfg=lcb_code_generation_eval_cfg
)
# Code Execution Dataset
lcb_code_execution_reader_cfg = dict(
input_columns=[
'prompt',
],
output_column='evaluation_sample',
)
lcb_code_execution_infer_cfg = dict(
prompt_template=dict(
type=PromptTemplate,
template=dict(
begin=[
dict(
role='SYSTEM',
prompt='You are an expert at Python programming, code execution, test case generation, and fuzzing.'
),
],
round=[
dict(
role='HUMAN',
prompt='{prompt}'
)
]
)
),
retriever=dict(type=ZeroRetriever),
inferencer=dict(type=GenInferencer, max_out_len=1024)
)
lcb_code_execution_eval_cfg = dict(
evaluator=dict(
type=LCBCodeExecutionEvaluator,
),
pred_role='BOT',
)
LCBCodeExecution_dataset = dict(
type=LCBCodeExecutionDataset,
abbr='lcb_code_execution',
path='opencompass/execution-v2',
reader_cfg=lcb_code_execution_reader_cfg,
infer_cfg=lcb_code_execution_infer_cfg,
eval_cfg=lcb_code_execution_eval_cfg,
)
# TestOuputput Dataset
lcb_test_output_reader_cfg = dict(
input_columns=[
'prompt',
],
output_column='evaluation_sample',
)
system_prompt = 'You are an expert Python programmer. You will be given a question (problem specification) and will generate a correct Python program that matches the specification and passes all tests. You will NOT return anything except for the program.'
lcb_test_output_infer_cfg = dict(
prompt_template=dict(
type=PromptTemplate,
template=dict(
# begin=[
# dict(
# role='SYSTEM',
# prompt=system_prompt
# ),
# ],
round=[
dict(
role='HUMAN',
prompt='{prompt}'
)
]
)
),
retriever=dict(type=ZeroRetriever),
inferencer=dict(type=GenInferencer, max_out_len=1024)
)
lcb_test_output_eval_cfg = dict(
evaluator=dict(
type=LCBTestOutputEvaluator,
),
pred_role='BOT',
)
LCBTestOutput_dataset = dict(
type=LCBTestOutputPredictionDataset,
abbr='lcb_test_output',
path='opencompass/test_generation',
reader_cfg=lcb_test_output_reader_cfg,
infer_cfg=lcb_test_output_infer_cfg,
eval_cfg=lcb_test_output_eval_cfg,
)
LCB_datasets = [
LCBCodeGeneration_dataset,
LCBCodeExecution_dataset,
LCBTestOutput_dataset,
]

9
opencompass/configs/models/hf_llama/hf_llama3_1_8b_instruct.py
View File

@ -1,5 +1,13 @@
from opencompass.models import HuggingFacewithChatTemplate
api_meta_template = dict(
round=[
dict(role='HUMAN', api_role='HUMAN'),
dict(role='BOT', api_role='BOT', generate=True),
],
reserved_roles=[dict(role='SYSTEM', api_role='SYSTEM')],
)
models = [
dict(
type=HuggingFacewithChatTemplate,
@ -9,5 +17,6 @@ models = [
batch_size=8,
run_cfg=dict(num_gpus=1),
stop_words=['<|end_of_text|>', '<|eot_id|>'],
meta_template=api_meta_template,
)
]

1
opencompass/datasets/__init__.py
View File

@ -67,6 +67,7 @@ from .lawbench import * # noqa: F401, F403
from .LCBench import * # noqa: F401, F403
from .lcsts import * # noqa: F401, F403
from .leval import * # noqa: F401, F403
from .livecodebench import * # noqa: F401, F403
from .llm_compression import LLMCompressionDataset # noqa: F401, F403
from .longbench import * # noqa: F401, F403
from .lveval import * # noqa: F401, F403

7
opencompass/datasets/livecodebench/__init__.py Normal file
View File

@ -0,0 +1,7 @@
from .evaluator import LCBCodeExecutionEvaluator # noqa: F401, F403
from .evaluator import LCBCodeGenerationEvaluator # noqa: F401, F403
from .evaluator import LCBTestOutputEvaluator # noqa: F401, F403
from .livecodebench import LCBCodeExecutionDataset # noqa: F401, F403
from .livecodebench import LCBCodeGenerationDataset # noqa: F401, F403
from .livecodebench import LCBTestOutputPredictionDataset # noqa: F401, F403
from .prompts import TestOutputPromptConstants # noqa: F401, F403

451
opencompass/datasets/livecodebench/evaluator.py Normal file
View File

@ -0,0 +1,451 @@
import ast
import json
import multiprocessing
from collections import defaultdict
from concurrent.futures import ProcessPoolExecutor, as_completed
import numpy as np
from tqdm import tqdm
from opencompass.openicl.icl_evaluator import BaseEvaluator
from opencompass.registry import ICL_EVALUATORS
from opencompass.utils import get_logger
from .execute_utils import BASE_IMPORTS, codeexecute_check_correctness
from .extract_utils import (extract_code_execution, extract_code_generation,
extract_test_output_code)
from .livecodebench import LCBCodeGenerationDataset
from .pass_k_utils import compute_metrics_from_results
def codegen_check_correctness(sample, generation, timeout, debug=True):
"""Check correctness of code generation with a global timeout.
The global timeout is to catch some extreme/rare cases not handled by the
timeouts inside `run_test`
"""
def _temp_run(sample, generation, debug, result, metadata_list, timeout):
from .testing_util import run_test
res, metadata = run_test(sample,
test=generation,
debug=debug,
timeout=timeout)
result.append(res)
metadata_list.append(metadata)
manager = multiprocessing.Manager()
result = manager.list()
metadata_list = manager.list()
p = multiprocessing.Process(
target=_temp_run,
args=(sample, generation, debug, result, metadata_list, timeout),
)
p.start()
p.join(timeout=(timeout + 1) *
len(json.loads(sample['input_output'])['inputs']) + 5)
if p.is_alive():
p.kill()
if not result:
in_outs = json.loads(sample['input_output'])
# consider that all tests failed
result = [[-1 for i in range(len(in_outs['inputs']))]]
if debug:
logger = get_logger()
logger.info('global timeout')
return result[0], metadata_list[0]
def evaluate_generations_by_problem(problem_generations: list, sample: list,
debug: bool, timeout: int):
"""Evaluate each problem.
Args:
problem_generations:
sample:
debug:
timeout
"""
# problem_generations: list[str] = args[0]
# sample = args[1]
# debug: bool = args[2]
# timeout: int = args[3]
logger = get_logger()
res = []
metadata = []
for o_idx, o in enumerate(problem_generations):
curr_res = [-2]
try:
curr_res, curr_metadata = codegen_check_correctness(
sample, o, timeout=timeout, debug=debug)
if debug:
logger.info(f'\nSuccessful compilation of task {o_idx}!')
fixed = []
for e in curr_res:
if isinstance(e, np.ndarray):
e = e.item(0)
if isinstance(e, np.bool_):
e = bool(e)
fixed.append(e)
curr_res = fixed
if not np.all(curr_res):
if debug:
logger.info(
f'Results were not True for all test cases' # noqa: F541, E501
f' {curr_res=}\n')
except Exception as e:
if debug:
logger.info(
f'Compilation failed, test framework exception' # noqa: F541, E501
f' = {repr(e)}{e}\n')
# break
curr_metadata = {}
finally:
assert isinstance(curr_res, list)
assert isinstance(curr_metadata, dict)
res.append(curr_res)
metadata.append(curr_metadata)
if debug:
for i, r in enumerate(problem_generations):
logger.info(f'Sample\n{r}\nResult\n{res[i]}')
logger.info('*' * 30 + '\n\n')
return res, metadata
def evaluate_generations(
samples_list: list,
generations_list: list[list[str]],
debug: bool = False,
num_process_evaluate: int = 16,
timeout=6,
):
"""We take the list of code generations and try to compile them and the run
their corresponding unit tests which are retrieved from the APPS dataset.
Args:
generations: list of code generations (same order as samples in APPS
dataset)
level: difficulty level used in the generation, can be "all",
"introductory", "interview" or "competition"
Returns:
results: dictionary of results, key is the problem index, value is
a list of results for each generation
[-2] = compile error, [-1] = runtime error [False] = failed test
case [True] = passed test case
"""
# generations are code generations in the same order of the dataset
inputs = [[(generations_list[index], samples_list[index], debug, timeout),
index] for index in range(len(generations_list))]
with tqdm(total=len(inputs)) as pbar:
with ProcessPoolExecutor(
max_workers=1 if debug else num_process_evaluate) as executor:
futures = {
executor.submit(evaluate_generations_by_problem,
problem_generations, sample, debug, timeout):
index
for (problem_generations, sample, debug,
timeout), index in inputs
}
results = {}
metadata = {}
for future in as_completed(futures):
index = futures[future]
results[index], metadata[index] = future.result()
pbar.update(1)
assert len(results) == len(
inputs), f'results = {len(results)} inputs = {len(inputs)} {results=}'
# results = {i: r for r, (_, i) in zip(results, inputs)}
return results, metadata
def codegen_metrics(
samples_list,
generations_list,
k_list=[1, 5, 10, 20, 40, 50, 75, 100, 125, 150, 200, 500, 1000],
num_process_evaluate=16,
timeout=6,
debug=False,
):
logger = get_logger()
samples_linear = []
generations_linear = []
remap_index = []
results = defaultdict(list)
metadatas = defaultdict(list)
for idx, (sample,
generation_list) in enumerate(zip(samples_list,
generations_list)):
assert isinstance(generation_list, list), generations_list[0]
for generation in generation_list:
assert isinstance(generation, str), generations_list[0]
samples_linear.append(sample)
generations_linear.append([generation])
remap_index.append(idx)
logger.info(f'LCBCodeGeneration: Evaluating {len(samples_linear)}...')
results_linear, metadatas_linear = evaluate_generations(
samples_linear,
generations_linear,
debug=debug,
num_process_evaluate=num_process_evaluate,
timeout=timeout,
)
for idx, sub_results in sorted(results_linear.items(), key=lambda x: x[0]):
results[remap_index[idx]].append(sub_results[0])
for idx, sub_metadatas in sorted(metadatas_linear.items(),
key=lambda x: x[0]):
metadatas[remap_index[idx]].append(sub_metadatas[0])
metrics = compute_metrics_from_results(results, k_list=k_list)
final_metadata = []
for key in sorted(list(metadatas.keys())):
final_metadata.append(metadatas[key])
for i in range(len(final_metadata)):
if type(final_metadata[i]) is not list:
final_metadata[i] = [json.dumps(final_metadata[i])]
else:
final_metadata[i] = [json.dumps(x) for x in final_metadata[i]]
assert len(final_metadata[i]) == len(
generations_list[0]), f'{len(final_metadata[i])=}'
return [metrics, results, final_metadata]
@ICL_EVALUATORS.register_module()
class LCBCodeGenerationEvaluator(BaseEvaluator):
def __init__(self, num_process_evaluate, timeout=6):
super().__init__()
self.num_process_evaluate = num_process_evaluate
self.timeout = timeout
self.dataset = LCBCodeGenerationDataset.load()['test']
def score(self, predictions, references):
predictions = [[extract_code_generation(item)] for item in predictions]
evaluation_samples = dict()
for idx in range(len(self.dataset)):
evaluation_samples[self.dataset[idx][
'question_id']] = self.dataset[idx]['evaluation_sample']
references = [evaluation_samples[item] for item in references]
references = [{'input_output': item} for item in references]
BaseEvaluator.is_num_equal(predictions, references)
results = { # noqa: F841
'pass': 0,
'timeout': 0,
'failed': 0,
'wrong_answer': 0
} # noqa: F401, F403
metrics, eval_results, final_metadata = codegen_metrics(
references,
predictions,
k_list=[1],
num_process_evaluate=self.num_process_evaluate,
timeout=self.timeout,
)
return metrics
def evaluate_score(args) -> list[bool]:
gs, (c, i, o) = args
execution_results = []
for g in gs:
if i in g:
pass
else:
code_to_execute = f'{BASE_IMPORTS}\n{c}\nassert {o} == {g}'
execution_results.append(
codeexecute_check_correctness(code_to_execute, 3))
if len(execution_results) == 0:
execution_results = [False] * len(gs)
return execution_results
def code_execution_metrics(
samples,
generations,
):
def pass_at_k(n, c, k):
if n - c < k:
return 1.0
return 1.0 - np.prod(1.0 - k / np.arange(n - c + 1, n + 1))
# execute the code
references = [(doc['code'], doc['input'], doc['output'])
for doc in samples]
with ProcessPoolExecutor() as executor:
args_list = zip(generations, references)
results = executor.map(evaluate_score, args_list)
all_results = list(results)
# serial version
# all_results = []
# for i in range(len(generations)):
# generation = generations[i]
# result = evaluate_score([generation, references[i]])
# all_results.append(result)
# compute pass@1
pass_at_1s = []
for execution_result in all_results:
c, n = execution_result.count(True), len(execution_result)
pass_at_1s.append(pass_at_k(n, c, 1))
metrics = {'pass@1': sum(pass_at_1s) / len(pass_at_1s) * 100}
results = {}
for i, r in enumerate(all_results):
r_new = []
for _r in r:
r_new.append([_r])
results[i] = r_new
return [metrics, results]
@ICL_EVALUATORS.register_module()
class LCBCodeExecutionEvaluator(BaseEvaluator):
def __init__(self):
super().__init__()
# self.num_process_evaluate = num_process_evaluate
# self.timeout = timeout
def score(self, predictions, references):
predictions = [[extract_code_execution(item)] for item in predictions]
references = [json.loads(item) for item in references]
metrics, results = code_execution_metrics(references, predictions)
return metrics
def parse_assert_statement(statement):
"""Parse a Python assert statement and extract the expected output from the
right side of the '==' operator as a string.
:param statement: A string containing the assert statement.
:return: The expected output from the assert statement as a string.
"""
try:
parsed = ast.parse(statement, mode='exec')
except SyntaxError:
return 'Invalid syntax'
if len(parsed.body) == 0:
return 'Empty statement'
if not isinstance(parsed.body[0], ast.Assert):
return 'Not an assert statement'
comparison = parsed.body[0].test
if not isinstance(comparison, ast.Compare) or not isinstance(
comparison.ops[0], ast.Eq):
return 'Not an equality assertion'
# Extract and return the right side of the '==' operator as a string
return ast.get_source_segment(statement, comparison.comparators[0])
def check_testcase_output(testcase_str, expected_output):
if len(testcase_str.splitlines()) > 1:
for line in testcase_str.splitlines():
if line.startswith('#'):
continue
if 'assert' in line:
testcase_str = line
break
testcase_str = testcase_str.strip()
if 'assert' in testcase_str:
testcase_output_str = str(parse_assert_statement(testcase_str))
else:
testcase_output_str = testcase_str
global_result = None
try:
testcase_output_eval = eval(testcase_output_str)
except Exception as e:
print(e)
global_result = False
# print("Failed to eval testcase output", testcase_output_str)
# breakpoint()
try:
expected_output_eval = json.loads(expected_output)
except Exception as e:
print(e)
global_result = False
print('Failed to eval expected testcase output', expected_output)
if global_result is None:
global_result = testcase_output_eval == expected_output_eval
return global_result
def test_output_metrics(
samples,
generations,
k_list=[1, 5],
):
num_samples = len(samples)
results = []
for idx in tqdm(list(range(num_samples))):
idx_results = []
sample = samples[idx]
extracted_generation_list = generations[idx]
for extracted_generation in extracted_generation_list:
global_result = check_testcase_output(extracted_generation,
sample['output'])
idx_results.append([global_result])
results.append(idx_results)
results = {
result_idx: results[result_idx]
for result_idx in range(len(results))
}
metrics = compute_metrics_from_results(results, k_list=k_list)
return [metrics, results]
@ICL_EVALUATORS.register_module()
class LCBTestOutputEvaluator(BaseEvaluator):
def __init__(self):
super().__init__()
def score(self, predictions, references):
predictions = [[extract_test_output_code(item)]
for item in predictions]
references = [json.loads(item) for item in references]
metrics, results = test_output_metrics(references,
predictions,
k_list=[1])
return metrics

271
opencompass/datasets/livecodebench/execute_utils.py Normal file
View File

@ -0,0 +1,271 @@
# Copyright 2020 The HuggingFace Datasets Authors and the
# current dataset script contributor.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# This code is adapted from OpenAI's release
# https://github.com/openai/human-eval/blob/master/human_eval/execution.py
import contextlib
import faulthandler
import io
import multiprocessing
import os
import platform
import signal
import tempfile
BASE_IMPORTS = """from itertools import accumulate, chain, combinations, count, permutations, product, groupby, islice, repeat
from copy import deepcopy
from string import ascii_lowercase
from math import floor, log2, log10, sqrt, comb, gcd, ceil, inf, isqrt
from collections import defaultdict, deque, Counter
from bisect import bisect, bisect_left, bisect_right, insort
from heapq import heappush, heappop, heapify, merge
from functools import reduce, cache, lru_cache
from random import randrange, shuffle
from operator import itemgetter, sub
from re import search as re_search # Assuming 're' refers to a regex search
from os.path import commonprefix
from typing import List, Tuple, Dict, Set, Optional, Union, Any, Callable, Iterable, Iterator, Generator
import copy
import string
import math
import collections
import bisect
import heapq
import functools
import random
import itertools
import operator
import re
import numpy as np
import pandas as pd
from math import log, prod # 'log' and 'prod' are functions in the math module
from collections import deque, defaultdict, Counter, OrderedDict
from itertools import accumulate, permutations, combinations, product, groupby, islice, chain, repeat, zip_longest, cycle
from functools import lru_cache, reduce, partial
# from sortedcontainers import SortedList, SortedDict, SortedSet
# import sortedcontainers
from operator import iand
import sys
""" # noqa: E501
def codeexecute_check_correctness(check_program, timeout=3):
"""Evaluates the functional correctness of a completion by running the test
suite provided in the problem.
:param completion_id: an optional completion ID so we can match
the results later even if execution finishes asynchronously.
"""
manager = multiprocessing.Manager()
result = manager.list()
p = multiprocessing.Process(target=unsafe_execute,
args=(check_program, result, timeout))
p.start()
p.join(timeout=timeout + 1)
if p.is_alive():
p.kill()
if not result:
result.append('timed out')
return result[0] == 'passed'
def unsafe_execute(check_program, result, timeout):
with create_tempdir():
# These system calls are needed when cleaning up tempdir.
import os
import shutil
rmtree = shutil.rmtree
rmdir = os.rmdir
chdir = os.chdir
# Disable functionalities that can make destructive changes
# to the test.
reliability_guard()
# Run program.
try:
exec_globals = {}
with swallow_io():
with time_limit(timeout):
exec(check_program, exec_globals)
result.append('passed')
except TimeoutException:
result.append('timed out')
except BaseException as e:
result.append(f'failed: {e}')
# Needed for cleaning up.
shutil.rmtree = rmtree
os.rmdir = rmdir
os.chdir = chdir
@contextlib.contextmanager
def time_limit(seconds):
def signal_handler(signum, frame):
raise TimeoutException('Timed out!')
signal.setitimer(signal.ITIMER_REAL, seconds)
signal.signal(signal.SIGALRM, signal_handler)
try:
yield
finally:
signal.setitimer(signal.ITIMER_REAL, 0)
@contextlib.contextmanager
def swallow_io():
stream = WriteOnlyStringIO()
with contextlib.redirect_stdout(stream):
with contextlib.redirect_stderr(stream):
with redirect_stdin(stream):
yield
@contextlib.contextmanager
def create_tempdir():
with tempfile.TemporaryDirectory() as dirname:
with chdir(dirname):
yield dirname
class TimeoutException(Exception):
pass
class WriteOnlyStringIO(io.StringIO):
"""StringIO that throws an exception when it's read from."""
def read(self, *args, **kwargs):
raise OSError
def readline(self, *args, **kwargs):
raise OSError
def readlines(self, *args, **kwargs):
raise OSError
def readable(self, *args, **kwargs):
"""Returns True if the IO object can be read."""
return False
class redirect_stdin(contextlib._RedirectStream): # type: ignore
_stream = 'stdin'
@contextlib.contextmanager
def chdir(root):
if root == '.':
yield
return
cwd = os.getcwd()
os.chdir(root)
try:
yield
except BaseException as exc:
raise exc
finally:
os.chdir(cwd)
def reliability_guard(maximum_memory_bytes=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.
"""
if maximum_memory_bytes is not None:
import resource
resource.setrlimit(resource.RLIMIT_AS,
(maximum_memory_bytes, maximum_memory_bytes))
resource.setrlimit(resource.RLIMIT_DATA,
(maximum_memory_bytes, maximum_memory_bytes))
if not platform.uname().system == 'Darwin':
resource.setrlimit(resource.RLIMIT_STACK,
(maximum_memory_bytes, maximum_memory_bytes))
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
import sys
sys.modules['ipdb'] = None
sys.modules['joblib'] = None
sys.modules['resource'] = None
sys.modules['psutil'] = None
sys.modules['tkinter'] = None

75
opencompass/datasets/livecodebench/extract_utils.py Normal file
View File

@ -0,0 +1,75 @@
# Copyright LiveCodeBench @ 2024,
import re
def extract_code_generation(model_output: str, model_type: str = 'chat'):
# modified from
outputlines = model_output.split('\n')
# TODO: handle codellama
if model_type == 'base':
return model_output.strip()
elif model_type == 'chat':
indexlines = [i for i, line in enumerate(outputlines) if '```' in line]
else:
raise ValueError(f'Invalid mode type: {model_type}')
if len(indexlines) < 2:
return ''
return '\n'.join(outputlines[indexlines[0] + 1:indexlines[1]])
def extract_code_execution(model_output: str, cot: bool = False):
pattern = r'\[PYTHON\](.*?)\[\/PYTHON\]'
matches = re.findall(pattern, model_output, re.DOTALL)
if matches:
# fetch the last one
model_output = matches[-1]
if '[PYTHON]' in model_output:
model_output
if cot:
if '[ANSWER]' in model_output:
model_output = model_output.split('[ANSWER]')[1].strip()
if '==' in model_output:
model_output = model_output.split('==')[1].strip()
if '[/ANSWER]' in model_output:
model_output = model_output.split('[/ANSWER]')[0].strip()
else:
model_output = model_output.split('\n')[0].strip()
return model_output.strip()
def extract_test_output_code(model_output: str):
outputlines = model_output.split('\n')
# find the last line startwith assert...
indexlines = [
i for i, line in enumerate(outputlines) if line.startswith('assert')
]
if indexlines:
return outputlines[indexlines[-1]]
# TODO: handle codellama format
# if lmstyle and lmstyle == LMStyle.CodeLLaMaInstruct:
# indexlines = \
# [i for i, line in enumerate(outputlines) if "PYTHON]" in line]
# else:
# first try to extract ```python if not then try ```
indexlines = [
i for i, line in enumerate(outputlines)
if '```python' in line or '```Python' in line
]
if indexlines:
start_index = indexlines[0]
else:
start_index = None
indexlines = [i for i, line in enumerate(outputlines) if '```' in line]
if start_index is not None:
indexlines = [i for i in indexlines if i > start_index]
indexlines = [start_index] + indexlines
if len(indexlines) < 2:
return ''
return '\n'.join(outputlines[indexlines[0] + 1:indexlines[1]])

217
opencompass/datasets/livecodebench/livecodebench.py Normal file
View File

@ -0,0 +1,217 @@
# Copyright (c) 2024, LiveCodeBench and its contributors.
# Copyright (c) 2023, OpenCompass and its contributors.
import base64
import json
import pickle
import zlib
from dataclasses import dataclass
from enum import Enum
from datasets import DatasetDict, load_dataset
from opencompass.utils import get_data_path # noqa: F401, F403
from ..base import BaseDataset
from .prompts import SelfRepairPromptConstants # noqa: F401, F403
from .prompts import TestOutputPromptConstants # noqa: F401, F403
from .prompts import (CodeGenerationPromptConstants,
get_generic_question_template_answer_self_repair,
get_generic_question_template_test_completion,
make_code_execution_prompt)
class Platform(Enum):
LEETCODE = 'leetcode'
CODEFORCES = 'codeforces'
ATCODER = 'atcoder'
class Difficulty(Enum):
EASY = 'easy'
MEDIUM = 'medium'
HARD = 'hard'
class TestType(Enum):
STDIN = 'stdin'
FUNCTIONAL = 'functional'
@dataclass
class Test:
input: str
output: str
testtype: TestType
def __post_init__(self):
self.testtype = TestType(self.testtype)
class LCBCodeGenerationDataset(BaseDataset):
@staticmethod
def load(path: str = 'opencompass/code_generation_lite',
local_mode: bool = False,
release_version: str = 'release_v1'):
def transform(item):
# Define the dataitem mapping logic
# starter_code
if item['starter_code']:
format_prompt = f'### Format: {CodeGenerationPromptConstants.FORMATTING_MESSAGE_WITH_STARTER_CODE}\n' # noqa: E501
format_prompt += f"```python\n{item['starter_code']}\n```\n\n"
else:
format_prompt = f'### Format: {CodeGenerationPromptConstants.FORMATTING_WITHOUT_STARTER_CODE}\n' # noqa: E501
format_prompt += '```python\n# YOUR CODE HERE\n```\n\n'
item['format_prompt'] = format_prompt
# load test cases
public_test_cases = item['public_test_cases']
public_test_cases = json.loads(item['public_test_cases'])
private_test_cases = item['private_test_cases']
try:
private_test_cases = json.loads(item['private_test_cases'])
except Exception as e: # noqa: F841
private_test_cases = json.loads(
pickle.loads(
zlib.decompress(
base64.b64decode(private_test_cases.encode(
'utf-8')) # type: ignore
))) # type: ignore
# load metadata
metadata = json.loads(item['metadata'])
evaluation_sample = json.dumps({
'inputs':
[t['input'] for t in public_test_cases + private_test_cases],
'outputs':
[t['output'] for t in public_test_cases + private_test_cases],
'fn_name':
metadata.get('func_name', None),
})
item['evaluation_sample'] = evaluation_sample
return item
path = get_data_path(path, local_mode=local_mode)
dataset = load_dataset(
path, # 'livecodebench/code_generation_lite'
split='test',
version_tag=release_version,
trust_remote_code=True)
dataset = dataset.map(transform)
return DatasetDict({'test': dataset, 'train': dataset})
class LCBCodeExecutionDataset(BaseDataset):
@staticmethod
def load(
path: str = 'opencompass/execution-v2',
local_mode: bool = False,
cot: bool = False,
# release_version: str = "release_v1"
):
# path = get_data_path(path, local_mode=local_mode)
def transform(item):
code, input = item['code'], item['input']
prompt = make_code_execution_prompt(code, input, cot=cot)
item['prompt'] = prompt
evaluation_sample = json.dumps({
'code': item['code'],
'input': item['input'],
'output': item['output']
})
item['evaluation_sample'] = evaluation_sample
return item
path = get_data_path(path, local_mode=local_mode)
dataset = load_dataset(path,
split='test') # 'livecodebench/execution-v2'
dataset = dataset.map(transform)
return DatasetDict({'test': dataset, 'train': dataset})
class LCBTestOutputPredictionDataset(BaseDataset):
@staticmethod
def load(
path: str = 'opencompass/test_generation',
local_mode: bool = False,
# release_version: str = "release_v1"
):
# path = get_data_path(path, local_mode=local_mode)
def transform(item):
question_content = item['question_content']
starter_code = item['starter_code']
test = json.loads(item['test'])
testcase_input = test[0]['input']
testcase_output = test[0]['output']
item['testcase_input'] = testcase_input
item['testcase_output'] = testcase_output
item['prompt'] = get_generic_question_template_test_completion(
question_content=question_content,
starter_code=starter_code,
testcase_input=testcase_input)
evaluation_sample = json.dumps({
'input':
item['question_content'],
'output':
json.loads(item['test'])[0]['output']
})
item['evaluation_sample'] = evaluation_sample
return item
path = get_data_path(path, local_mode=local_mode)
# 'livecodebench/test_generation',
dataset = load_dataset(path, split='test', trust_remote_code=True)
dataset = dataset.map(transform)
return DatasetDict({'test': dataset, 'train': dataset})
class LCBSelfRepairDataset(BaseDataset):
@staticmethod
def load(path: str = 'livecodebench/code_generation_lite',
local_mode: bool = False,
release_version: str = 'release_v1'):
def transform(item):
# define the data item mapping logic
question = item['question_content']
code = item['code_list'][0]
metadata = item['metadata']
prompt = get_generic_question_template_answer_self_repair(
question=question, code=code, metadata=metadata)
item['prompt'] = prompt
return
dataset = load_dataset(path,
split='test',
version_tag=release_version,
trust_remote_code=True)
dataset = dataset.map(transform)
return DatasetDict({'test': dataset, 'train': dataset})

72
opencompass/datasets/livecodebench/pass_k_utils.py Normal file
View File

@ -0,0 +1,72 @@
# Copyright LiveCodeBench @ 2024,
import numpy as np
def estimate_pass_at_k(num_samples, num_correct, k):
"""Estimates pass@k of each problem and returns them in an array."""
def estimator(n: int, c: int, k: int) -> float:
"""Calculates 1 - comb(n - c, k) / comb(n, k)."""
if n - c < k:
return 1.0 * 100
return 100 * (1.0 - np.prod(1.0 - k / np.arange(n - c + 1, n + 1)))
import itertools
if isinstance(num_samples, int):
num_samples_it = itertools.repeat(num_samples, len(num_correct))
else:
assert len(num_samples) == len(num_correct)
num_samples_it = iter(num_samples)
return np.array([
estimator(int(n), int(c), k)
for n, c in zip(num_samples_it, num_correct)
])
def compute_metrics_from_results(results, k_list=[1, 5]):
total = []
correct = []
task_ids = []
for task_id, res in results.items():
all_correct = []
for generation in res:
gen = np.array(generation)
all_correct.append(np.all(gen > 0))
task_ids.append(task_id)
total.append(len(all_correct))
correct.append(sum(all_correct))
total = np.array(total)
correct = np.array(correct)
ks = k_list
detail_pass_at_k = {
f'pass@{k}': estimate_pass_at_k(total, correct, k).tolist()
for k in ks if (total >= k).all()
}
pass_at_k = {
f'pass@{k}': estimate_pass_at_k(total, correct, k).mean()
for k in ks if (total >= k).all()
}
detail_metrics = {
k: dict(zip(task_ids, v))
for k, v in detail_pass_at_k.items()
}
pass_at_k['detail'] = detail_metrics
return pass_at_k
def extract_instance_results(results):
instance_wise_grades = {}
for task_id, res in results.items():
instance_wise_grades[task_id] = []
for generation in res:
instance_wise_grades[task_id].append(
all([g > 0 for g in generation]))
instance_wise_grades = [
v for _, v in sorted(instance_wise_grades.items(),
key=lambda item: item[0])
]
return instance_wise_grades

210
opencompass/datasets/livecodebench/prompts.py Normal file
View File

@ -0,0 +1,210 @@
# Copyright LiveCodeBench @ 2024,
import json
class CodeGenerationPromptConstants:
SYSTEM_MESSAGE_GENERIC = 'You are an expert Python programmer. You will be given a question (problem specification) and will generate a correct Python program that matches the specification and passes all tests. You will NOT return anything except for the program.' # noqa: E501
SYSTEM_MESSAGE_GEMINI = 'You are an expert Python programmer. You will be given a question (problem specification) and will generate a correct Python program that matches the specification and passes all tests. You will NOT return anything except for the program. Do NOT use system calls like `exit` in the generated program.' # noqa: E501
SYSTEM_MESSAGE_DEEPSEEK = 'You are an AI programming assistant, utilizing the DeepSeek Coder model, developed by DeepSeek Company, and you answer questions related to computer science.' # noqa: E501
SYSTEM_MESSAGE_MAGIC = 'You are an exceptionally intelligent coding assistant that consistently delivers accurate and reliable responses to user instructions.\n\n@@ Instruction\n' # noqa: E501
SYSTEM_MESSAGE_WIZARD = 'Below is an instruction that describes a task. Write a response that appropriately completes the request.' # noqa: E501
SYSTEM_MESSAGE_PHIND = """You are an expert Python programmer. You will be given a question (problem specification) and will generate a correct Python program that matches the specification and passes all tests. You will NOT return anything except for the program. Put your fixed program within code delimiters, for example:
```python
# YOUR CODE HERE
```""" # noqa: E501
SYSTEM_MESSAGE_CODEQWEN = (
'<|im_start|>system\nYou are a helpful assistant.<|im_end|>\n<|im_start|>user' # noqa: E501
)
FORMATTING_MESSAGE_WITH_STARTER_CODE = 'You will use the following starter code to write the solution to the problem and enclose your code within delimiters.' # noqa: E501
FORMATTING_WITHOUT_STARTER_CODE = '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.' # noqa: E501
PYTHON_FORMAT = '```python\n# YOUR CODE HERE\n```\n\n'
class TestOutputPromptConstants:
SYSTEM_MESSAGE_CHAT_GENERIC = 'You are a helpful programming assistant and an expert Python programmer. You are helping a user to write a test case to help to check the correctness of the function. The user has written a input for the testcase. You will calculate the output of the testcase and write the whole assertion statement in the markdown code block with the correct output.' # noqa: E501
SYSTEM_MESSAGE_COMPLETION_GENERIC = 'You are a helpful programming assistant and an expert Python programmer. You are helping a user to write a test case to help to check the correctness of the function.' # noqa: E501
SYSTEM_MESSAGE_INST_CLLAMA = 'You are a helpful programming assistant and an expert Python programmer. You are helping a user to write a test case to help to check the correctness of the function. The user has written a input for the testcase. You will calculate the output of the testcase and write out the complete assertion statement between [PYTHON] and [/PYTHON] tags.' # noqa: E501
SYSTEM_MESSAGE_WIZARD = 'Below is an instruction that describes a task. Write a response that appropriately completes the request.' # noqa: E501
SYSTEM_MESSAGE_PHIND = """You are an expert Python programmer. You will be given a question (problem specification) and will generate a correct Python program that matches the specification and passes all tests. You will NOT return anything except for the program. You must put the entire fixed program within code delimiters only for once., for example:
```python
# YOUR CODE HERE
```""" # noqa: E501
FORMATTING_MESSAGE = 'You will use the following starter code to write the solution to the problem and enclose your code within delimiters.' # noqa: E501
FORMATTING_WITHOUT_STARTER_MESSAGE = '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.' # noqa: E501
class SelfRepairPromptConstants:
SYSTEM_MESSAGE_GENERIC = 'You are a helpful programming assistant and an expert Python programmer. You are helping a user write a program to solve a problem. The user has written some code, but it has some errors and is not passing the tests. You will help the user by first giving a concise (at most 2-3 sentences) textual explanation of what is wrong with the code. After you have pointed out what is wrong with the code, you will then generate a fixed version of the program. You must put the entire fixed program within code delimiters only for once.' # noqa: E501
SYSTEM_MESSAGE_DEEPSEEK = 'You are an AI programming assistant, utilizing the DeepSeek Coder model, developed by DeepSeek Company, and you are helping a user correct a error program for code competition. The user has written some code, but it has some errors and is not passing the tests. You will help the user by first giving a concise (at most 2-3 sentences) textual explanation of what is wrong with the code. After you have pointed out what is wrong with the code, you will then generate a fixed version of the entire executable program. You must put the entire fixed executable program within code delimiters.' # noqa: E501
SYSTEM_MESSAGE_MAGIC = 'You are an exceptionally intelligent coding assistant that consistently delivers accurate and reliable responses to user instructions.\n\n@@ Instruction\n' # noqa: E501
SYSTEM_MESSAGE_WIZARD = 'Below is an instruction that describes a task. Write a response that appropriately completes the request.' # noqa: E501
SYSTEM_MESSAGE_PHIND = """You are an expert Python programmer. You will be given a question (problem specification) and will generate a correct Python program that matches the specification and passes all tests. You will NOT return anything except for the program. You must put the entire fixed program within code delimiters only for once., for example:
```python
# YOUR CODE HERE
```""" # noqa: E501
FORMATTING_REPEAT = 'First reason about the code providing a textual explanation of what is wrong with the code and then generate a fixed of the program enclosed code delimiters.' # noqa: E501
FORMATTING_MESSAGE = 'You will use the following starter code to write the solution to the problem and enclose your code within delimiters.' # noqa: E501
FORMATTING_WITHOUT_STARTER_CODE = '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.' # noqa: E501
def make_code_execution_prompt(code, input, cot):
if cot:
# make_cot_output_prompt
return f"""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. Do NOT output any extra information. Execute the program step by step before arriving at an answer, and provide the full assertion with the correct output in [ANSWER] and [/ANSWER] tags, following the examples.
[PYTHON]
def performOperation(s):
s = s + s
return "b" + s + "a"
assert performOperation(s = "hi") == ??
[/PYTHON]
[THOUGHT]
Let's execute the code step by step:
1. The function performOperation is defined, which takes a single argument s.
2. The function is called with the argument "hi", so within the function, s is initially "hi".
3. Inside the function, s is concatenated with itself, so s becomes "hihi".
4. The function then returns a new string that starts with "b", followed by the value of s (which is now "hihi"), and ends with "a".
5. The return value of the function is therefore "bhihia".
[/THOUGHT]
[ANSWER]
assert performOperation(s = "hi") == "bhihia"
[/ANSWER]
[PYTHON]
{code}
assert {input} == ??
[/PYTHON]
[THOUGHT]
""" # noqa: E501
else:
# make_direct_output_prompt
return f"""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. Do NOT output any extra information. Provide the full assertion with the correct output in [ANSWER] and [/ANSWER] tags, following the examples.
[PYTHON]
def repeatNumber(number : int) -> int:
return number
assert repeatNumber(number = 17) == ??
[/PYTHON]
[ANSWER]
assert repeatNumber(number = 17) == 17
[/ANSWER]
[PYTHON]
def addCharacterA(string : str) -> str:
return string + "a"
assert addCharacterA(string = "x9j") == ??
[/PYTHON]
[ANSWER]
assert addCharacterA(string = "x9j") == "x9ja"
[/ANSWER]
[PYTHON]
{code}
assert {input} == ??
[/PYTHON]
[ANSWER]
""" # noqa: E501
def get_generic_question_template_test_completion(question_content,
starter_code,
testcase_input: str):
def format_testcase_func_name_input(function_name, testcase):
"""use the form of "assert func_name(input) == "."""
# TODO should there be a space after the == ?
input_str = ', '.join(testcase.split('\n'))
return f'assert {function_name}({input_str}) == # TODO'
def parse_function_name_from_starter_code(starter_code):
"""
starter_code : str
"""
import ast
tree = ast.parse(starter_code)
fn = None
for node in ast.walk(tree):
if isinstance(node, ast.FunctionDef):
assert fn is None
fn = node.name
return fn
prompt = f'Problem:\n{question_content}'
prompt += f'Function:\n```\n{starter_code}\n```\n'
# parse function name from starter_code
func_name = parse_function_name_from_starter_code(starter_code)
prompt += 'Please complete the following test case:\n\n'
prompt += (
f'```\n{format_testcase_func_name_input(func_name, testcase_input)}\n```\n' # noqa: E501
)
return prompt
def get_generic_question_template_answer_self_repair(question: str, code,
metadata):
def get_check_prompt(metadata):
# def get_check_prompt(question: str, result, metadata):
# # assumes i/o examples are already truncated!
# # less pressure on storing 10 MB json because on a single large
# # input-output pair
# result_by_test_case = result
# assert len(metadata) == 1, f"metadata = {metadata}"
# metadata = metadata[0]
metadata = json.loads(metadata)
if 'error_code' not in metadata:
return ''
if metadata['error_code'] == -1:
# time limit exceeded
message = f"The above code is incorrect and got the following compilation error.\n{metadata['error']}" # noqa: E501
elif metadata['error_code'] == -2:
# wrong answer
message = f"The above code is incorrect and got a wrong answer.\nInput: {metadata['inputs']}\nGenerated Output: {metadata['output']}\nExpected: {metadata['expected']}" # noqa: E501
elif metadata['error_code'] == -3:
# time limit exceeded
message = f"The above code is incorrect and got time limit exceeded.\n{metadata['error']}\nInput: {metadata['inputs']}\nExpected: {metadata['expected']}" # noqa: E501
pass
elif metadata['error_code'] == -4:
# runtime error
message = f"The above code is incorrect and got a runtime error.\nInput: {metadata['inputs']}\nExpected: {metadata['expected']}\n{metadata['error']}" # noqa: E501
else:
raise NotImplementedError(
f"metadata['error_code'] = {metadata['error_code']} not implemented || {metadata=}" # noqa: E501
)
return message
prompt = f'### Question:\n{question}\n\n'
prompt += f'### Answer:\n```python\n{code}\n```\n\n'
# prompt += get_check_prompt(question, result, metadata) + "\n"
prompt += get_check_prompt(metadata) + '\n'
prompt += f'### Format: {SelfRepairPromptConstants.FORMATTING_WITHOUT_STARTER_CODE}\n' # noqa: E501
prompt += '```python\n# YOUR CODE HERE\n```\n\n'
prompt += '### Answer: (use the provided format with backticks)\n\n'
return prompt

762
opencompass/datasets/livecodebench/testing_util.py Normal file
View File

@ -0,0 +1,762 @@
# Copyright LiveCodeBench @ 2024,
import ast
import faulthandler
import json
import platform
# to run the solution files we're using a timing based approach
import signal
import sys
# used for debugging to time steps
from datetime import datetime
from enum import Enum
# for capturing the stdout
from io import StringIO
# used for testing the code that reads from input
from unittest.mock import mock_open, patch
import numpy as np
from pyext import RuntimeModule
def truncatefn(s, length=300):
assert isinstance(s, str)
if len(s) <= length:
return s
return s[:length // 2] + '...(truncated) ...' + s[-length // 2:]
class CODE_TYPE(Enum):
call_based = 0
standard_input = 1
# stuff for setting up signal timer
class TimeoutException(Exception):
pass
def timeout_handler(signum, frame):
print('alarm went off')
# return
raise TimeoutException
signal.signal(signal.SIGALRM, timeout_handler)
# timeout = 6 # seconds
# used to capture stdout as a list
# from https://stackoverflow.com/a/16571630/6416660
# alternative use redirect_stdout() from contextlib
class Capturing(list):
def __enter__(self):
self._stdout = sys.stdout
sys.stdout = self._stringio = StringIO()
# Make closing the StringIO a no-op
self._stringio.close = lambda x: 1
return self
def __exit__(self, *args):
self.append(self._stringio.getvalue())
del self._stringio # free up some memory
sys.stdout = self._stdout
def only_int_check(val):
return isinstance(val, int)
def string_int_check(val):
return isinstance(val, str) and val.isdigit()
def combined_int_check(val):
return only_int_check(val) or string_int_check(val)
def run_test(sample, test=None, debug=False, timeout=6):
"""if test(generated_code) is not None it'll try to run the code.
otherwise it'll just return an input and output pair.
"""
# Disable functionalities that can make destructive changes to the test.
reliability_guard()
if debug:
print(f'start = {datetime.now().time()}')
try:
in_outs = json.loads(sample['input_output'])
except ValueError:
in_outs = None
if in_outs:
if in_outs.get('fn_name') is None:
which_type = CODE_TYPE.standard_input # Standard input
method_name = None
else:
which_type = CODE_TYPE.call_based # Call-based
method_name = in_outs['fn_name']
if debug:
print(f'loaded input_output = {datetime.now().time()}')
if test is None:
assert False, 'should not happen: test code is none'
return in_outs, {'error': 'no test code provided'}
elif test is not None:
results = []
sol = 'from string import *\nfrom re import *\nfrom datetime import *\nfrom collections import *\nfrom heapq import *\nfrom bisect import *\nfrom copy import *\nfrom math import *\nfrom random import *\nfrom statistics import *\nfrom itertools import *\nfrom functools import *\nfrom operator import *\nfrom io import *\nfrom sys import *\nfrom json import *\nfrom builtins import *\nfrom typing import *\nimport string\nimport re\nimport datetime\nimport collections\nimport heapq\nimport bisect\nimport copy\nimport math\nimport random\nimport statistics\nimport itertools\nimport functools\nimport operator\nimport io\nimport sys\nimport json\nsys.setrecursionlimit(6*10**5)\n' # noqa: E501
if debug:
print(f'loading test code = {datetime.now().time()}')
if which_type == CODE_TYPE.call_based:
sol += test
if debug:
print(f'sol = {sol}')
signal.alarm(timeout)
try:
tmp_sol = RuntimeModule.from_string('tmp_sol', '', sol)
if 'class Solution' not in test:
tmp = tmp_sol
else:
tmp = tmp_sol.Solution()
signal.alarm(0)
except Exception as e:
signal.alarm(0)
if debug:
print(f'type 0 compilation error = {e}')
results.append(-2)
return results, {
'error': repr(e),
'error_code': -1,
'error_message': 'Compilation Error',
}
signal.alarm(0)
elif which_type == CODE_TYPE.standard_input:
# sol
# if code has if __name__ == "__main__": then remove it
try:
astree = ast.parse(test)
last_block = astree.body[-1]
if isinstance(last_block, ast.If):
condition = last_block.test
if ast.unparse(
condition).strip() == "__name__ == '__main__'":
test = (ast.unparse(astree.body[:-1]) + '\n' +
ast.unparse(last_block.body))
except Exception as e: # noqa:
pass
tmp_test = test.split('\n')
new_test = []
for x in tmp_test:
if (not x.startswith('from ')) and (
not x.startswith('import ')):
new_test.append('\t' + x + '\n')
else:
new_test.append(x + '\n')
tmp_test = new_test
new_test = ''
started = False
for i in tmp_test:
if i.startswith('\t') and not started:
new_test += 'stdin = sys.stdin\nstdout = sys.stdout\n'
new_test += 'def code():\n'
new_test += i
started = True
elif started and ((i.startswith('from ')) or
(i.startswith('import '))):
new_test += '\t' + i
else:
new_test += i
tmp_test = new_test
sol += tmp_test
if debug:
print(f'sol = {sol}')
method_name = 'code'
signal.alarm(timeout)
try:
tmp_sol = RuntimeModule.from_string('tmp_sol', '', sol)
tmp = tmp_sol
signal.alarm(0)
except Exception as e:
signal.alarm(0)
if debug:
print(f'type 1 compilation error = {e}')
results.append(-2)
return results, {
'error': repr(e),
'error_code': -1,
'error_message': 'Compilation Error',
}
signal.alarm(0)
if debug:
print(f'get method = {datetime.now().time()}')
try:
method = getattr(tmp,
method_name) # get_attr second arg must be str
except Exception as e:
signal.alarm(0)
e = sys.exc_info()
print(f'unable to get function error = {e}')
results.append(-2)
return results, {
'error': repr(e),
'error_code': -1,
'error_message': 'Unable to extract code',
}
for index, inputs in enumerate(in_outs['inputs']):
raw_inputs = inputs
raw_outputs = in_outs['outputs'][index]
if which_type == CODE_TYPE.call_based:
inputs = [json.loads(line) for line in inputs.split('\n')]
in_outs['outputs'][index] = json.loads(
in_outs['outputs'][index])
truncate_line_size = 300 // (raw_inputs.count('\n') + 1)
raw_inputs = '\n'.join([
truncatefn(line, truncate_line_size)
for line in raw_inputs.strip().split('\n')
])
raw_outputs = truncatefn(raw_outputs, 200)
else:
raw_inputs = truncatefn(raw_inputs)
raw_outputs = truncatefn(raw_outputs, 200)
# JSON forces dictionaries to have string keys; this undoes this
# (assuming a singleton list)
try:
if isinstance(inputs[0], dict):
inputs = [{int(k): v for k, v in inputs[0].items()}]
except Exception as e: # noqa: F841
True
try:
if isinstance(in_outs['outputs'][index], dict):
in_outs['outputs'][index] = [{
int(k): v
for k, v in in_outs['outputs'][index].items()
}]
except Exception as e: # noqa: F841
True
try:
if isinstance(in_outs['outputs'][index][0], dict):
in_outs['outputs'][index] = [{
int(k): v
for k, v in in_outs['outputs'][index][0].items()
}]
except Exception as e: # noqa: F841
True
if debug:
print(
f'time: {datetime.now().time()} testing index = {index} '
f'inputs = {inputs}, {type(inputs)}. type = {which_type}')
if which_type == CODE_TYPE.call_based: # Call-based
signal.alarm(timeout)
faulthandler.enable()
try:
output = method(*inputs)
raw_true_output = output
raw_true_output_copy = json.dumps(output)
raw_true_output_copy = truncatefn(raw_true_output_copy,
200)
# ground truth sequences are not tuples
if isinstance(output, tuple):
output = list(output)
tmp_result = output == in_outs['outputs'][index]
if (isinstance(in_outs['outputs'][index], list)
and in_outs['outputs'][index]):
tmp_result = tmp_result or (
output == in_outs['outputs'][index][0])
# ground truth sequences are not tuples
try:
if isinstance(output[0], tuple):
tmp_result = tmp_result or ([
list(x) for x in output
] == in_outs['outputs'][index][0])
except Exception as e: # noqa: F841
True
results.append(tmp_result)
if tmp_result is not True:
return results, {
'output': raw_true_output_copy,
'expected': raw_outputs,
'inputs': raw_inputs,
'error_code': -2,
'error_message': 'Wrong Answer',
}
# reset the alarm
signal.alarm(0)
except Exception as e:
signal.alarm(0)
faulthandler.disable()
if debug:
print(
f'Standard input runtime error or time limit exceeded error = {e}' # noqa: E501
)
results.append(-1)
if 'timeoutexception' in repr(e).lower():
return results, {
'error': repr(e),
'error_code': -3,
'error_message': 'Time Limit Exceeded',
'inputs': raw_inputs,
'expected': raw_outputs,
}
else:
return results, {
'error': repr(e),
'error_code': -4,
'error_message': 'Runtime Error',
'inputs': raw_inputs,
'expected': raw_outputs,
}
faulthandler.disable()
signal.alarm(0)
if debug:
print(
f"outputs = {output}, test outputs = {in_outs['outputs'][index]}, inputs = {inputs}, {type(inputs)}, {output == [in_outs['outputs'][index]]}" # noqa: E501
)
elif which_type == CODE_TYPE.standard_input: # Standard input
faulthandler.enable()
passed = False
if isinstance(inputs, list):
inputs = '\n'.join(inputs)
if isinstance(in_outs['outputs'][index], list):
in_outs['outputs'][index] = '\n'.join(
in_outs['outputs'][index])
signal.alarm(timeout)
with Capturing() as output:
try:
call_method(method, inputs)
# reset the alarm
signal.alarm(0)
passed = True
except Exception as e:
# runtime error or took too long
signal.alarm(0)
print(
f'Call-based runtime error or time limit exceeded error = {repr(e)}{e}' # noqa: E501
)
results.append(-1)
if 'timeoutexception' in repr(e).lower():
return results, {
'error': repr(e),
'error_code': -3,
'error_message': 'Time Limit Exceeded',
'inputs': raw_inputs,
'expected': raw_outputs,
}
else:
return results, {
'error': repr(e),
'error_code': -4,
'error_message': 'Runtime Error',
'inputs': raw_inputs,
'expected': raw_outputs,
}
signal.alarm(0)
raw_true_output = output[0]
raw_true_output_copy = truncatefn(raw_true_output, 200)
output = raw_true_output.splitlines()
if not passed:
if debug:
nl = '\n'
if not isinstance(inputs, list):
print(
f"not passed output = {output}, test outputs = {in_outs['outputs'][index]}, inputs = {inputs.replace(nl,' new-line ')}, {type(inputs)}, {output == [in_outs['outputs'][index]]}" # noqa: E501
)
else:
print(
f"not passed output = {output}, test outputs = {in_outs['outputs'][index]}, inputs = {inputs}, {type(inputs)}, {output == [in_outs['outputs'][index]]}" # noqa: E501
)
continue
if passed and debug:
print(
f"==> output = {output}, test outputs = {in_outs['outputs'][index]}" # noqa: E501
)
if custom_compare_(output, in_outs['outputs'][index]):
tmp_result = True
results.append(tmp_result)
continue
# ground truth sequences are expressed as lists not tuples
if isinstance(output, tuple):
output = list(output)
tmp_result = False
try:
tmp_result = output == [in_outs['outputs'][index]]
if isinstance(in_outs['outputs'][index], list):
tmp_result = tmp_result or (
output == in_outs['outputs'][index])
if isinstance(output[0], str):
tmp_result = tmp_result or ([
e.strip() for e in output
] == in_outs['outputs'][index])
except Exception as e:
if debug:
print(f'Failed check1 exception = {e}')
pass
if tmp_result is True:
results.append(tmp_result)
continue
# try one more time without \n
if isinstance(in_outs['outputs'][index], list):
for tmp_index, i in enumerate(in_outs['outputs'][index]):
in_outs['outputs'][index][tmp_index] = i.split('\n')
in_outs['outputs'][index][tmp_index] = [
x.strip()
for x in in_outs['outputs'][index][tmp_index] if x
]
else:
in_outs['outputs'][index] = in_outs['outputs'][
index].split('\n')
in_outs['outputs'][index] = list(
filter(len, in_outs['outputs'][index]))
in_outs['outputs'][index] = list(
map(lambda x: x.strip(), in_outs['outputs'][index]))
try:
tmp_result = output == [in_outs['outputs'][index]]
if isinstance(in_outs['outputs'][index], list):
tmp_result = tmp_result or (
output == in_outs['outputs'][index])
except Exception as e:
if debug:
print(f'Failed check2 exception = {e}')
pass
if tmp_result is True:
results.append(tmp_result)
continue
# try by converting the output into a split up list too
if isinstance(output, list):
output = list(filter(len, output))
if debug:
nl = '\n'
if not isinstance(inputs, list):
print(
f"@1 output = {output}, test outputs = {in_outs['outputs'][index]}, inputs = {inputs.replace(nl,' new-line ')}, {type(inputs)}, {output == [in_outs['outputs'][index]]} {tmp_result=}" # noqa: E501
)
else:
print(
f"@1 output = {output}, test outputs = {in_outs['outputs'][index]}, inputs = {inputs}, {type(inputs)}, {output == [in_outs['outputs'][index]]} {tmp_result=}" # noqa: E501
)
if tmp_result is True:
results.append(tmp_result)
continue
if debug:
print(f'{tmp_result=} @a')
try:
tmp_result = output == [in_outs['outputs'][index]]
if isinstance(in_outs['outputs'][index], list):
tmp_result = tmp_result or (
output == in_outs['outputs'][index])
except Exception as e:
if debug:
print(f'Failed check3 exception = {e}')
pass
if debug:
print(f'{tmp_result=} @b')
try:
all_ints = all(
combined_int_check(e1) and combined_int_check(e2)
for e1, e2 in zip(output, in_outs['outputs'][index]))
if not all_ints:
if debug:
print([
combined_int_check(e1)
and combined_int_check(e2) for e1, e2 in zip(
output, in_outs['outputs'][index])
])
output_float = [float(e) for e in output]
gt_float = [
float(e) for e in in_outs['outputs'][index]
]
tmp_result = tmp_result or (
(len(output_float) == len(gt_float))
and np.allclose(output_float, gt_float))
except Exception as e: # noqa: F841
pass
if debug:
print(f'{tmp_result=} @c')
try:
if isinstance(output[0], list):
all_ints = all(
combined_int_check(e1) and combined_int_check(e2)
for e1, e2 in zip(output[0], in_outs['outputs']
[index]))
if not all_ints:
output_float = [float(e) for e in output[0]]
gt_float = [
float(e) for e in in_outs['outputs'][index][0]
]
tmp_result = tmp_result or (
(len(output_float) == len(gt_float))
and np.allclose(output_float, gt_float))
except Exception as e: # noqa: F841
pass
if tmp_result is True:
results.append(tmp_result)
continue
if debug:
print(f'{tmp_result=} @d')
# try by converting the stuff into split up list
if isinstance(in_outs['outputs'][index], list):
for tmp_index, i in enumerate(in_outs['outputs'][index]):
in_outs['outputs'][index][tmp_index] = set(i.split())
else:
in_outs['outputs'][index] = set(
in_outs['outputs'][index].split())
if debug:
print(f'{tmp_result=} @e')
try:
tmp_result = output == in_outs['outputs'][index]
except Exception as e:
if debug:
print(f'Failed check4 exception = {e}')
continue
if tmp_result is True:
results.append(tmp_result)
continue
if debug:
print(f'{tmp_result=} @f')
# try by converting the output into a split up list too
if isinstance(output, list):
for tmp_index, i in enumerate(output):
output[tmp_index] = i.split()
output = list(filter(len, output))
for tmp_index, i in enumerate(output):
output[tmp_index] = set(i)
else:
output = output.split()
output = list(filter(len, output))
output = set(output)
if debug:
print(f'{tmp_result=} @g')
# try:
# tmp_result = set(frozenset(s) for s in output) == set(
# frozenset(s) for s in in_outs["outputs"][index]
# )
# except Exception as e:
# if debug:
# print(f"Failed check5 exception = {e}")
# if they are all numbers, round so that similar numbers are
# treated as identical
# try:
# all_ints = all(
# combined_int_check(e1) and combined_int_check(e2)
# for e1, e2 in zip(output, in_outs["outputs"][index])
# )
# tmp_result = tmp_result or (
# set(
# frozenset(round(float(t), 3) for t in s) for s in output)
# == set(
# frozenset(round(float(t), 3) for t in s)
# for s in in_outs["outputs"][index]
# )
# )
# except Exception as e:
# if debug:
# print(f"Failed check6 exception = {e}")
if debug:
print(f'{tmp_result=} @h')
if tmp_result is True and debug:
print('PASSED')
results.append(tmp_result)
if tmp_result is not True:
return results, {
'output': raw_true_output_copy,
'expected': raw_outputs,
'inputs': raw_inputs,
'error_code': -2,
'error_message': 'Wrong Answer',
}
if debug:
nl = '\n'
if not isinstance(inputs, list):
print(
f"@2 output = {output}, test outputs = {in_outs['outputs'][index]}, inputs = {inputs.replace(nl,' new-line ')}, {type(inputs)}, {output == [in_outs['outputs'][index]]}" # noqa: E501
)
else:
print(
f"@2 output = {output}, test outputs = {in_outs['outputs'][index]}, inputs = {inputs}, {type(inputs)}, {output == [in_outs['outputs'][index]]}" # noqa: E501
)
print(f'results = {results}')
return results, {}
def custom_compare_(output, ground_truth):
if isinstance(output, list):
output_1 = '\n'.join(output)
if stripped_string_compare(output_1, ground_truth):
return True
if isinstance(output, list):
output_2 = [o.lstrip().rstrip() for o in output]
output_2 = '\n'.join(output_2)
if stripped_string_compare(output_2, ground_truth):
return True
return False
def stripped_string_compare(s1, s2):
s1 = s1.lstrip().rstrip()
s2 = s2.lstrip().rstrip()
return s1 == s2
def call_method(method, inputs):
if isinstance(inputs, list):
inputs = '\n'.join(inputs)
inputs_line_iterator = iter(inputs.split('\n'))
# sys.setrecursionlimit(10000)
# @patch('builtins.input', side_effect=inputs.split("\n"))
@patch('builtins.open', mock_open(read_data=inputs))
@patch('sys.stdin', StringIO(inputs))
@patch('sys.stdin.readline', lambda *args: next(inputs_line_iterator))
@patch('sys.stdin.readlines', lambda *args: inputs.split('\n'))
@patch('sys.stdin.read', lambda *args: inputs)
# @patch('sys.stdout.write', print)
def _inner_call_method(_method):
try:
return _method()
except SystemExit as e: # noqa: F841
pass
finally:
pass
return _inner_call_method(method)
def reliability_guard(maximum_memory_bytes=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.
"""
if maximum_memory_bytes is not None:
import resource
resource.setrlimit(resource.RLIMIT_AS,
(maximum_memory_bytes, maximum_memory_bytes))
resource.setrlimit(resource.RLIMIT_DATA,
(maximum_memory_bytes, maximum_memory_bytes))
if not platform.uname().system == 'Darwin':
resource.setrlimit(resource.RLIMIT_STACK,
(maximum_memory_bytes, maximum_memory_bytes))
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
import sys
sys.modules['ipdb'] = None
sys.modules['joblib'] = None
sys.modules['resource'] = None
sys.modules['psutil'] = None
sys.modules['tkinter'] = None

7
opencompass/openicl/icl_evaluator/icl_base_evaluator.py
View File

@ -8,3 +8,10 @@ class BaseEvaluator:
def score(self):
raise NotImplementedError("Method hasn't been implemented yet")
@staticmethod
def is_num_equal(predictions, references):
if len(predictions) != len(references):
return {'error': 'preds and refrs have different length'}
else:
return

27
opencompass/utils/datasets_info.py
View File

@ -276,6 +276,21 @@ DATASETS_MAPPING = {
"ms_id": "",
"hf_id": "opencompass/needlebench",
"local": "./data/needlebench",
},
"opencompass/code_generation_lite": {
"ms_id": "",
"hf_id": "",
"local": "./data/code_generation_lite",
},
"opencompass/execution-v2": {
"ms_id": "",
"hf_id": "",
"local": "./data/execution-v2",
},
"opencompass/test_generation": {
"ms_id": "",
"hf_id": "",
"local": "./data/test_generation",
}
}
@ -428,4 +443,16 @@ DATASETS_URL = {
"url": "http://opencompass.oss-cn-shanghai.aliyuncs.com/datasets/data/teval.zip",
"md5": "7628ab5891a26bf96ca17becfd044867",
},
"/code_generation_lite": {
"url": "http://opencompass.oss-cn-shanghai.aliyuncs.com/datasets/data/code_generation_lite.zip",
"md5": "60103a18ca63b05ea06e98d24170f23d",
},
"/execution-v2": {
"url": "http://opencompass.oss-cn-shanghai.aliyuncs.com/datasets/data/execution-v2.zip",
"md5": "019ef1a0686ee6ca34f51c8af104fcd9",
},
"/test_generation": {
"url": "http://opencompass.oss-cn-shanghai.aliyuncs.com/datasets/data/test_generation.zip",
"md5": "918a6ea2b1eee6f2b1314db3c21cb4c7",
},
}

1
requirements/runtime.txt
View File

@ -20,6 +20,7 @@ opencv-python-headless
pandas<2.0.0
prettytable
protobuf
pyext
python-Levenshtein
rank_bm25==0.2.2
rapidfuzz