(
input_file: str = None,
tmp_dir: str = "./",
n_workers: int = 32,
timeout: float = 10.0,
problem_file: str = "../data/humaneval_python.jsonl.gz",
result_path: str = None,
k: List[int] = [1, 10, 100],
test_groundtruth: bool = False,
example_test: bool = False,
is_mbpp: bool = False,
language: str = "python",
)
| 184 | |
| 185 | |
| 186 | def evaluate_functional_correctness( |
| 187 | input_file: str = None, |
| 188 | tmp_dir: str = "./", |
| 189 | n_workers: int = 32, |
| 190 | timeout: float = 10.0, |
| 191 | problem_file: str = "../data/humaneval_python.jsonl.gz", |
| 192 | result_path: str = None, |
| 193 | k: List[int] = [1, 10, 100], |
| 194 | test_groundtruth: bool = False, |
| 195 | example_test: bool = False, |
| 196 | is_mbpp: bool = False, |
| 197 | language: str = "python", |
| 198 | ): |
| 199 | if example_test: |
| 200 | print("Example test...") |
| 201 | |
| 202 | problems = read_dataset(problem_file, dataset_type="humaneval") |
| 203 | sample_jsonl = stream_jsonl_all(input_file) |
| 204 | with ThreadPoolExecutor(max_workers=n_workers) as executor: |
| 205 | futures = [] |
| 206 | completion_id = Counter() |
| 207 | n_samples = 0 |
| 208 | results = defaultdict(list) |
| 209 | |
| 210 | if test_groundtruth: |
| 211 | print("Testing ground truth...") |
| 212 | for sample in tqdm(problems.values()): |
| 213 | task_id = sample["task_id"] |
| 214 | lang = task_id.split("/")[0].lower() |
| 215 | if lang == "javascript": |
| 216 | lang = "js" |
| 217 | tmp_dir_ = os.path.join(tmp_dir, lang, "evaluation") |
| 218 | sample["generation"] = sample["canonical_solution"] |
| 219 | sample["test_code"] = process_humaneval_test(sample, problems, example_test, language) |
| 220 | if sample["test_code"] is None: |
| 221 | continue |
| 222 | args = (task_id, sample, lang, timeout, tmp_dir_, completion_id[task_id]) |
| 223 | future = executor.submit(check_correctness, *args) |
| 224 | futures.append(future) |
| 225 | completion_id[task_id] += 1 |
| 226 | n_samples += 1 |
| 227 | else: |
| 228 | print("Reading Samples...") |
| 229 | id2samples = {} |
| 230 | for sample in tqdm(sample_jsonl): |
| 231 | task_id = sample["task_id"] |
| 232 | |
| 233 | if not is_mbpp: |
| 234 | lang = language |
| 235 | if not is_mbpp and lang == "javascript": |
| 236 | lang = "js" |
| 237 | if is_mbpp: |
| 238 | lang = "python" |
| 239 | tmp_dir_ = os.path.join(tmp_dir, lang, "evaluation") |
| 240 | sample["task_id"] = task_id |
| 241 | sample["test_code"] = process_humaneval_test(sample, problems, example_test, is_mbpp, language) |
| 242 | if sample["test_code"] is None: |
| 243 | continue |
no test coverage detected