Add parallel requests option (--parallel flag and config)

2026-08-07 14:08:58 +02:00
parent fd40133142
commit d731958ced
3 changed files with 60 additions and 51 deletions
@@ -15,6 +15,7 @@ import json
 import re
 import sys
 import time
+from concurrent.futures import ThreadPoolExecutor, as_completed
 from datetime import datetime
 from pathlib import Path

@@ -251,11 +252,15 @@ def print_summary(report: dict, output_path: Path):
    print(f"{'='*60}\n")


-def run_benchmark(config_path: str, resume: bool = False, fresh: bool = False):
+def run_benchmark(config_path: str, resume: bool = False, fresh: bool = False, parallel: int | None = None):
    config = load_config(config_path)
    model_cfg = config['model']
    bench_cfg = config['benchmark']

+    parallelism = parallel if parallel is not None else bench_cfg.get('parallel_requests', 1)
+    if parallelism < 1:
+        parallelism = 1
+
    output_dir = Path(config_path).parent / bench_cfg.get('output_dir', 'runs')
    output_dir.mkdir(exist_ok=True)
    safe_name = sanitize_model_name(model_cfg['name'])
@@ -326,17 +331,13 @@ def run_benchmark(config_path: str, resume: bool = False, fresh: bool = False):

    global_idx = resume_from

-    for i, q in enumerate(questions, 1):
-        qid = q.get('id', f'unknown_{i}')
+    def process_question(idx: int, q: dict) -> dict:
+        qid = q.get('id', f'unknown_{idx}')
        question_text = q.get('text', '?')
        golden_key = q.get('golden_key', '?')
        options = q.get('options', {})

-        global_idx += 1
-        display_q = question_text[:80] + ('...' if len(question_text) > 80 else '')
-
        prompt = build_prompt(q)
-
        response = None
        timed_out = False

@@ -347,56 +348,54 @@ def run_benchmark(config_path: str, resume: bool = False, fresh: bool = False):
                continue
            break

-        is_correct = False
-        model_choice = ''
-
        if timed_out:
-            results['timeouts'].append({
-                'id': qid,
-                'question': question_text,
-                'expected_key': golden_key,
-                'options': options,
-            })
-        elif answers_match(golden_key, response or '', options):
-            is_correct = True
-            results['correct'].append({
-                'id': qid,
-                'question': question_text,
-                'expected_key': golden_key,
-                'model_response': (response or '')[:200],
-            })
+            return {'type': 'timeout', 'id': qid, 'question': question_text, 'expected_key': golden_key, 'options': options}
+
+        if answers_match(golden_key, response or '', options):
+            return {'type': 'correct', 'id': qid, 'question': question_text, 'expected_key': golden_key, 'model_response': (response or '')[:200]}
+
+        model_choice = extract_choice(response or '')
+        return {'type': 'incorrect', 'id': qid, 'question': question_text, 'expected_key': golden_key, 'model_response': (response or '')[:500], 'model_choice': model_choice, 'options': options}
+
+    total_q = len(questions) + resume_from
+
+    with ThreadPoolExecutor(max_workers=parallelism) as executor:
+        futures = {executor.submit(process_question, i, q): (i, q) for i, q in enumerate(questions, 1)}
+
+        for future in as_completed(futures):
+            i, q = futures[future]
+            qid = q.get('id', f'unknown_{i}')
+            question_text = q.get('text', '?')
+            golden_key = q.get('golden_key', '?')
+
+            global_idx += 1
+            display_q = question_text[:80] + ('...' if len(question_text) > 80 else '')
+
+            try:
+                res = future.result()
+            except Exception as e:
+                res = {'type': 'error', 'id': qid, 'error': str(e)}
+
+            if res['type'] == 'timeout':
+                results['timeouts'].append({'id': res['id'], 'question': res['question'], 'expected_key': res['expected_key'], 'options': res.get('options', {})})
+                label = 'TIMEOUT'
+            elif res['type'] == 'correct':
+                results['correct'].append({'id': res['id'], 'question': res['question'], 'expected_key': res['expected_key'], 'model_response': res['model_response']})
+                label = 'CORRECT'
+            elif res['type'] == 'incorrect':
+                results['incorrect'].append({'id': res['id'], 'question': res['question'], 'expected_key': res['expected_key'], 'model_response': res['model_response'], 'model_choice': res['model_choice'], 'options': res.get('options', {})})
+                label = f"WRONG (expected: {golden_key}, got: {res.get('model_choice', '?')})"
            else:
-            model_choice = extract_choice(response or '')
-            results['incorrect'].append({
-                'id': qid,
-                'question': question_text,
-                'expected_key': golden_key,
-                'model_response': (response or '')[:500],
-                'model_choice': model_choice,
-                'options': options,
-            })
+                label = f"ERROR: {res.get('error', 'unknown')}"

-        # Running accuracy
            answered = len(results['correct']) + len(results['incorrect']) + len(results['timeouts'])
            acc = (len(results['correct']) / answered * 100) if answered > 0 else 0.0

-        if timed_out:
-            label = 'TIMEOUT'
-        elif is_correct:
-            label = 'CORRECT'
-        else:
-            label = f"WRONG (expected: {golden_key}, got: {model_choice or '?'})"
+            print(f"  [{global_idx}/{total_q}] {display_q}  {label}  [{acc:.1f}%]")

-        status = f"  [{global_idx}/{len(questions) + resume_from}] {display_q}  {label}  [{acc:.1f}%]"
-        print(f"\r{status}")
-
-        # Save checkpoint after each question
            completed_ids.add(qid)
            save_checkpoint(checkpoint_path, results, list(completed_ids), start_time)

-        if bench_cfg.get('delay_between_requests', 0) > 0:
-            time.sleep(bench_cfg['delay_between_requests'])
-
    # Build and save final report
    report = build_report(config, results, start_time)

@@ -424,6 +423,8 @@ def main():
                        help='Resume from the latest checkpoint')
    parser.add_argument('--fresh', '-f', action='store_true',
                        help='Start a new run, ignoring existing checkpoint')
+    parser.add_argument('--parallel', '-p', type=int, default=None,
+                        help='Number of parallel requests (1 = sequential)')
    args = parser.parse_args()

    config_path = Path(args.config)
@@ -432,7 +433,7 @@ def main():
        print(f"   Copy config.example.yaml to config.yaml and fill in your values.", file=sys.stderr)
        sys.exit(1)

-    run_benchmark(str(config_path), resume=args.resume, fresh=args.fresh)
+    run_benchmark(str(config_path), resume=args.resume, fresh=args.fresh, parallel=args.parallel)


 if __name__ == '__main__':
@@ -24,6 +24,9 @@ benchmark:
  # How many times to retry a failed request
  max_retries: 1

+  # Number of parallel requests (1 = sequential)
+  parallel_requests: 1
+
  # Delay in seconds between requests (0 = none)
  delay_between_requests: 0

@@ -0,0 +1,5 @@
+fastparquet==2026.5.0
+pandas==3.0.5
+pyarrow==25.0.0
+pyyml==0.0.2
+requests==2.34.2