forked from bytedance/FullStackBench
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathevaluate.py
More file actions
147 lines (128 loc) · 4.58 KB
/
Copy pathevaluate.py
File metadata and controls
147 lines (128 loc) · 4.58 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
import argparse
import asyncio
import logging
from pathlib import Path
from sandbox_fusion import (
SubmitRequest,
TestConfig,
set_endpoint,
submit_async,
)
from tqdm.asyncio import tqdm_asyncio
from utils import read_jsonl, write_jsonl
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
async def evaluate_sample(
sample: dict, natural_language: str, semaphore: asyncio.Semaphore
) -> list[dict]:
async with semaphore:
completions: str | list[str] = sample.get('completion', [])
if isinstance(completions, str):
completions = [completions]
n = len(completions)
results = []
pass_at_k = False
all_k_correct = True
for completion in completions:
result = {}
eval_result = await submit_async(
SubmitRequest(
dataset='FullStackBench',
id=sample['id'],
completion=completion,
config=TestConfig(
compile_timeout=50,
run_timeout=50,
dataset_type='AutoEvalDataset',
provided_data=sample
)
)
)
passed = eval_result.accepted
result['accepted'] = passed
result['pass_at_1'] = 1 if passed else 0
result['task_id'] = sample['id']
result['n'] = n
if passed:
pass_at_k = True
else:
all_k_correct = False
result['natural_language'] = natural_language
if 'labels' in sample and 'programming_language' in sample['labels']:
result['programming_language'] = sample['labels']['programming_language']
result['category'] = sample['labels'].get('category', '')
result['difficulty'] = sample['labels'].get('difficulty', '')
results.append(result)
for result in results:
result['pass_at_k'] = 1 if pass_at_k else 0
result['all_k_correct'] = 1 if all_k_correct else 0
return results
async def main():
parser = argparse.ArgumentParser(
description='Evaluate FullStackBench predictions using SandboxFusion'
)
parser.add_argument(
'--inference-file',
type=str,
required=True,
help='Path to inference output file (JSONL format)'
)
parser.add_argument(
'--output-file',
type=str,
required=True,
help='Path to save evaluation results (JSONL format)'
)
parser.add_argument(
'--sandbox-endpoint',
type=str,
default='http://localhost:8080',
help='SandboxFusion server endpoint'
)
parser.add_argument(
'--language',
type=str,
required=True,
choices=['en', 'zh'],
help='Natural language of the dataset (en or zh)'
)
parser.add_argument(
'--num-workers',
type=int,
default=10,
help='Maximum number of concurrent evaluation requests (default: 10)'
)
args = parser.parse_args()
inference_file = Path(args.inference_file)
if not inference_file.exists():
logger.error(f"Inference file not found: {args.inference_file}")
return 1
set_endpoint(args.sandbox_endpoint)
logger.info(f"Using SandboxFusion endpoint: {args.sandbox_endpoint}")
logger.info(f"Reading inference results from: {args.inference_file}")
samples = read_jsonl(args.inference_file)
logger.info(f"Found {len(samples)} samples to evaluate")
if args.num_workers < 1:
logger.error(f"--num-workers must be at least 1, got {args.num_workers}")
return 1
logger.info("Starting evaluation...")
# Limit concurrent requests to avoid "Too many open files" error
semaphore = asyncio.Semaphore(args.num_workers)
tasks = [evaluate_sample(sample, args.language, semaphore) for sample in samples]
results = []
for result in await tqdm_asyncio.gather(*tasks, desc="Evaluating"):
for r in result:
results.append(r)
pass_count = sum([r['pass_at_1'] for r in results])
pass_rate = pass_count / len(results) if results else 0.0
logger.info("Evaluation complete!")
logger.info(f"Pass rate: {pass_rate:.4%} ({pass_count}/{len(results)})")
logger.info(f"Writing results to: {args.output_file}")
write_jsonl(args.output_file, results)
return 0
if __name__ == '__main__':
exit_code = asyncio.run(main())
exit(exit_code)