wehub-resource-sync
236 lines
8.3 KiB
Python
236 lines
8.3 KiB
Python
import copy
|
|
import json
|
|
import re
|
|
import sys
|
|
from argparse import ArgumentParser
|
|
from datasets import load_dataset
|
|
from collections import defaultdict
|
|
from concurrent.futures import ThreadPoolExecutor, as_completed
|
|
from functools import partial
|
|
from glob import glob
|
|
import os
|
|
from pympler import asizeof
|
|
|
|
from tqdm import tqdm
|
|
|
|
sys.set_int_max_str_digits(0)
|
|
|
|
sys.path.append(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))))
|
|
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
print(sys.path)
|
|
|
|
from apps.utils_execute import run_inference_process
|
|
|
|
|
|
def _worker(item, test_case_field: str):
|
|
results = []
|
|
full_results = []
|
|
all_outputs = []
|
|
all_errors = []
|
|
if not item["pred"]:
|
|
if "res" in item:
|
|
item.pop("res")
|
|
if "full_res" in item:
|
|
item.pop("full_res")
|
|
if "outputs" in item:
|
|
item.pop("outputs")
|
|
if "errors" in item:
|
|
item.pop("errors")
|
|
return item
|
|
|
|
for pred in item["pred"]:
|
|
gen_solution = pred
|
|
if gen_solution is None:
|
|
results.append(False)
|
|
# full_results.append([False] * 3)
|
|
full_results.append([-2] * 21)
|
|
all_outputs.append([None] * 21)
|
|
continue
|
|
|
|
if "Hello, World!" in gen_solution:
|
|
results.append(False)
|
|
full_results.append([-2] * 21)
|
|
all_outputs.append([None] * 21)
|
|
continue
|
|
|
|
if not item[test_case_field]:
|
|
continue
|
|
|
|
all_results = run_inference_process(item[test_case_field], gen_solution, timeout=1, debug=False, return_output=True)
|
|
res, outputs, errors = all_results
|
|
for tmp in res:
|
|
if (not isinstance(tmp, bool)) and (not isinstance(tmp, int)):
|
|
print(tmp, tmp.__class__.__name__)
|
|
res = [bool(tmp) if (not isinstance(tmp, bool)) and (not isinstance(tmp, int)) else tmp for tmp in res]
|
|
if all(item is True for item in res) is True:
|
|
results.append(True)
|
|
else:
|
|
results.append(False)
|
|
full_results.append(res)
|
|
try:
|
|
json.dumps(outputs)
|
|
all_outputs.append(outputs)
|
|
all_errors.append(errors)
|
|
except:
|
|
print(f"Cannot dump outputs for {outputs}")
|
|
all_outputs.append([])
|
|
all_errors.append([])
|
|
|
|
if results:
|
|
item["res"] = results
|
|
item["full_res"] = full_results
|
|
item["outputs"] = all_outputs
|
|
item["errors"] = all_errors
|
|
|
|
return item
|
|
|
|
|
|
# Function to check if a string contains surrogate characters
|
|
def has_surrogate_characters(text):
|
|
return bool(re.search(r'[\ud800-\udfff]', text))
|
|
|
|
|
|
def load_files(file_path):
|
|
data = []
|
|
if os.path.exists(file_path):
|
|
print(f"Loading pseudo test cases from {file_path}")
|
|
if file_path.endswith(".json"):
|
|
data.extend(json.load(open(file_path)))
|
|
else:
|
|
data.extend([json.loads(line) for line in open(file_path).readlines()])
|
|
else:
|
|
for file in glob(file_path):
|
|
print(file)
|
|
if file.endswith(".json"):
|
|
data.extend(json.load(open(file)))
|
|
else:
|
|
data.extend([json.loads(line) for line in open(file).readlines()])
|
|
|
|
return data
|
|
|
|
|
|
def merge_key(item, value):
|
|
assert isinstance(item, list)
|
|
if isinstance(value, list):
|
|
item = item + value
|
|
else:
|
|
item.append(value)
|
|
return item
|
|
|
|
|
|
def merge_seed_sampled_data(data, id_field: str = "id"):
|
|
id2data = {}
|
|
for item in data:
|
|
if isinstance(item["response"], str):
|
|
print(f"Warning: {item[id_field]} has only one response. ---- {item['response']} \n\n {item['pred']}")
|
|
item["response"] = [item["response"]]
|
|
assert isinstance(item["pred"], str) or item["pred"] is None
|
|
item["pred"] = [item["pred"]]
|
|
|
|
if item[id_field] not in id2data:
|
|
id2data[item[id_field]] = item
|
|
continue
|
|
|
|
tmp = id2data[item[id_field]]
|
|
|
|
tmp["response"] = merge_key(tmp["response"], item["response"])
|
|
tmp["pred"] = merge_key(tmp["pred"], item["pred"])
|
|
assert isinstance(tmp["pred"], list), tmp["pred"]
|
|
id2data[item[id_field]] = tmp
|
|
|
|
return list(id2data.values())
|
|
|
|
|
|
def main():
|
|
parser = ArgumentParser()
|
|
parser.add_argument("--completion_file", type=str)
|
|
parser.add_argument("--output_file", type=str)
|
|
parser.add_argument("--num_workers", type=int, default=4)
|
|
parser.add_argument("--id_field", type=str, default="id")
|
|
parser.add_argument("--test_case_field", type=str, default="test_cases")
|
|
args = parser.parse_args()
|
|
|
|
data = load_files(args.completion_file)
|
|
data = merge_seed_sampled_data(data, args.id_field)
|
|
# for item in data:
|
|
# item["response"] = list(set(item["response"])) # No need to remove duplicates, since the `pred` field is aligned.
|
|
|
|
print(f"Total number of items: {len(data)}")
|
|
|
|
missing = 0
|
|
corr = 0
|
|
corr_at_k = 0
|
|
pbar = tqdm(data)
|
|
|
|
outputs = []
|
|
with ThreadPoolExecutor(max_workers=args.num_workers) as executor:
|
|
futures = []
|
|
_annotate = partial(_worker, test_case_field=args.test_case_field)
|
|
for _input in pbar:
|
|
future = executor.submit(_annotate, _input)
|
|
futures.append(future)
|
|
pbar.update()
|
|
|
|
for future in tqdm(as_completed(futures), total=len(futures), desc="Collecting results"):
|
|
outputs.append(future.result())
|
|
|
|
large_mem = 0
|
|
for item in outputs:
|
|
if "res" in item and item["res"]:
|
|
if item["res"][0] is True:
|
|
corr += 1
|
|
if any(item["res"]):
|
|
corr_at_k += 1
|
|
|
|
try:
|
|
size_in_bytes = asizeof.asizeof(item["outputs"])
|
|
if size_in_bytes / (1024 ** 2) > 10: # 10MB
|
|
if "res" in item:
|
|
item.pop("res")
|
|
if "full_res" in item:
|
|
item.pop("full_res")
|
|
if "outputs" in item:
|
|
item.pop("outputs")
|
|
if "errors" in item:
|
|
item.pop("errors")
|
|
large_mem += 1
|
|
except:
|
|
print("failed to compute size. Still abandon the outputs.")
|
|
if "res" in item:
|
|
item.pop("res")
|
|
if "full_res" in item:
|
|
item.pop("full_res")
|
|
if "outputs" in item:
|
|
item.pop("outputs")
|
|
if "errors" in item:
|
|
item.pop("errors")
|
|
large_mem += 1
|
|
else:
|
|
missing += 1
|
|
|
|
new_outputs = []
|
|
for item in tqdm(outputs):
|
|
tmp = json.dumps(item, ensure_ascii=False)
|
|
if has_surrogate_characters(tmp):
|
|
print(f"Surrogate characters found in {item[args.id_field]}")
|
|
continue
|
|
new_outputs.append(item)
|
|
|
|
outputs = new_outputs
|
|
print(f"Missing: {missing / len(outputs)}")
|
|
print(f"Large memory: {large_mem / len(outputs)}")
|
|
print(f"Correct: {corr / len(outputs)}")
|
|
print(f"Correct at k: {corr_at_k / len(outputs)}")
|
|
json.dump(outputs, open(args.output_file, "w", encoding="utf-8"), ensure_ascii=False)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|
|
|
|
"""
|
|
>>> python scripts/apps/solution_run_outputs_local.py --completion_file ../msranlpintern/reward_modeling/experiments/deepseek-coder-v1.5-ins.7b.apps.r2c.gpt4o.distil.A100.w8.v3.0.s42/apps/checkpoint-400/train.0shot.tem1.0.n10.v1.1.s43.json --output_file ../msranlpintern/reward_modeling/experiments/deepseek-coder-v1.5-ins.7b.apps.r2c.gpt4o.distil.A100.w8.v3.0.s42/apps/checkpoint-400/train.0shot.tem1.0.n10.v1.1.s43.run_outputs.json --num_workers 24
|
|
|
|
>>> python scripts/apps/solution_run_outputs_local.py --completion_file ../msranlpintern/reward_modeling/experiments/deepseek-coder-v1.5-ins.7b.apps.r2c.sft_ps_test_case.process-dpo.V100.tp8dp16.v4.9.s42/oss-instruct-apps-train/checkpoint-700/train.0shot.tem1.0.n10.?-of-8.v2.0.json --output_file ../msranlpintern/reward_modeling/experiments/deepseek-coder-v1.5-ins.7b.apps.r2c.sft_ps_test_case.process-dpo.V100.tp8dp16.v4.9.s42/oss-instruct-apps-train/checkpoint-700/train.0shot.tem1.0.n10.v2.0.run_outputs.json --num_workers 24 --id_field "problem_id" --test_case_field "input_output"
|
|
|
|
"""
|