File size: 7,578 Bytes
62dca4c | 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 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 | """
Base class for benchmark implementations.
"""
import time
from abc import ABC, abstractmethod
from argparse import Namespace
from typing import Any, Callable, Dict, List, Optional, Tuple
from sglang import set_default_backend
from sglang.test.test_utils import select_sglang_backend
from .utils import compute_metrics
class Benchmarker(ABC):
"""
Base class for benchmark implementations.
Subclasses should implement:
- load_data(): Load and preprocess dataset
- create_sgl_function(): Create the SGL function for inference
Optional overrides:
- extract_answer(): Extract answer from model output (if needed)
- compute_accuracy(): Compute accuracy metric (if applicable)
- get_answer_keys(): Get list of answer keys for multi-turn conversations
Args:
num_samples: The number of samples to run the benchmark on. If not provided, all questions will be used.
subset: The subset of the dataset to run the benchmark on. If not provided, all subsets will be used.
"""
def __init__(
self, num_samples: Optional[int] = None, subset: Optional[List[str]] = None
):
self.num_samples = num_samples
self.subset = subset
@abstractmethod
def load_data(self) -> Tuple[List[Dict[str, Any]], List[Any]]:
"""
Load and preprocess the dataset.
Returns:
Tuple of (questions, labels) where:
- questions: List of question dicts for SGL function
- labels: List of ground truth labels (can be None if not applicable)
"""
raise NotImplementedError
@abstractmethod
def create_sgl_function(self) -> Callable:
"""
Create the SGL function for inference.
Returns:
SGL function decorated with @sgl.function
"""
raise NotImplementedError
def extract_answer(self, output: str, label: Optional[Any] = None) -> Optional[Any]:
"""
Extract answer from model output.
Args:
output: Raw model output string
label: Optional ground truth label for reference
Returns:
Extracted answer, or None if extraction fails
"""
return output
def compute_accuracy(
self, predictions: List[Any], labels: List[Any]
) -> Optional[float]:
"""
Compute accuracy metric.
Args:
predictions: List of predicted answers
labels: List of ground truth labels
Returns:
Accuracy score (0-1), or None if not applicable
"""
return None
def get_answer_keys(self) -> Optional[List[str]]:
"""
Get list of answer keys for multi-turn conversations.
Returns:
List of answer keys (e.g., ["answer_1", "answer_2"]), or None for single-turn
"""
return None
def get_max_new_tokens(self) -> int:
"""
Get maximum number of new tokens to generate.
Returns:
Maximum tokens (default: 2048)
"""
return 2048
def run(
self,
host: str,
port: int,
batch_size: int,
max_new_tokens: int = None,
num_runs: int = 1,
):
"""
Run the benchmark evaluation.
This method handles the common workflow:
1. Initialize backend
2. Load data
3. Create SGL function
4. Run inference loops
5. Compute metrics
6. Print results
Args:
host (str): The host of the SGLang server
port (int): The port of the SGLang server
batch_size (int): The number of prompts to process in parallel
num_samples (int): The number of samples to run the benchmark on. If not provided, all samples will be used.
max_new_tokens (int): Maximum number of new tokens to generate, default is 2048
num_runs (int): The number of times to run this benchmark, default is 1. You can set it to a larger number if you want to get more stable results.
"""
if not host.startswith(("http://", "https://")):
host = f"http://{host}"
# Initialize backend
sglang_args = Namespace(host=host, port=port, backend="srt-no-parallel")
set_default_backend(select_sglang_backend(sglang_args))
# Load data
questions, labels = self.load_data()
if len(questions) == 0:
print("No valid questions found. Please check the dataset format.")
return
# Create SGL function
sgl_function = self.create_sgl_function()
# Run evaluation loops
metrics_list = []
answer_keys = self.get_answer_keys()
max_new_tokens = max_new_tokens or self.get_max_new_tokens()
for _ in range(num_runs):
tic = time.perf_counter()
states = sgl_function.run_batch(
questions,
temperature=0,
max_new_tokens=max_new_tokens,
num_threads=batch_size,
progress_bar=True,
)
latency = time.perf_counter() - tic
# Extract predictions
predictions = []
primary_answer_key = answer_keys[0] if answer_keys else "answer"
for i in range(len(states)):
# Access answer from state object (states[i] supports dict-like access)
output = states[i][primary_answer_key]
if isinstance(output, str):
extracted = self.extract_answer(
output,
(labels[i] if labels and i < len(labels) else None),
)
else:
extracted = output
predictions.append(extracted)
# Compute accuracy if applicable
accuracy = None
# Check if we have a labels list (even if all labels are None)
has_labels_list = labels and len(labels) > 0
if has_labels_list:
# Always call compute_accuracy if we have a labels list
# This allows it to return None, which will be displayed in print_results
accuracy = self.compute_accuracy(predictions, labels)
if accuracy is not None:
valid_count = sum(1 for p in predictions if p is not None)
if valid_count < len(predictions):
print(
f"Warning: {len(predictions) - valid_count} predictions could not be extracted."
)
# Compute performance metrics
metrics = compute_metrics(
states,
latency,
answer_key=primary_answer_key,
additional_answer_keys=(
answer_keys[1:] if answer_keys and len(answer_keys) > 1 else None
),
)
# Always set accuracy if we have a labels list (even if compute_accuracy returns None)
# This allows print_results to show None when compute_accuracy returns None
if has_labels_list:
metrics.accuracy = (
accuracy # Can be None if compute_accuracy returns None
)
if accuracy is not None:
metrics.num_valid_predictions = sum(
1 for p in predictions if p is not None
)
metrics_list.append(metrics)
return metrics_list
|