diff --git a/ais_bench/benchmark/cli/argument_parser.py b/ais_bench/benchmark/cli/argument_parser.py index b24bd706..11233934 100644 --- a/ais_bench/benchmark/cli/argument_parser.py +++ b/ais_bench/benchmark/cli/argument_parser.py @@ -2,11 +2,15 @@ import json from ais_bench.benchmark.cli.utils import ( get_current_time_str, + validate_batch_size, validate_max_workers, validate_max_workers_per_gpu, validate_num_prompts, validate_num_warmups, - validate_pressure_time + validate_prefix_len, + validate_pressure_time, + validate_request_count, + validate_request_size, ) from ais_bench.benchmark.cli.utils import DEFAULT_PRESSURE_TIME @@ -21,6 +25,7 @@ def __init__(self): self._custom_dataset_parser() self._agent_parser() self._api_model_parser() + self._synthetic_dataset_parser() def parse_args(self): args = self.parser.parse_args() @@ -115,8 +120,11 @@ def _base_parser(self): parser.add_argument( '--num-prompts', help='Num Prompts, Specify the number of prompts to infer and evaluate. ' - 'Must be integer >= 1, If not provided, all prompts will be inferred and evaluated. ', + 'Must be integer >= 1, If not provided, all prompts will be inferred and evaluated. ' + 'Accepts 1..N integers; with multiple values each one pairs with the ' + 'corresponding --batch-size value (or a single --batch-size broadcasts).', type=validate_num_prompts, + nargs='+', default=None ) parser.add_argument( @@ -424,8 +432,11 @@ def _api_model_parser(self): help='Endpoint url override') parser.add_argument('--max-out-len', type=int, default=None, help='Max output length override') - parser.add_argument('--batch-size', type=int, default=None, - help='Batch size override in the range (0, 100000]') + parser.add_argument('--batch-size', type=validate_batch_size, nargs='+', + default=None, + help='Batch size override in the range (0, 100000]. ' + 'Accepts 1..N integers; multiple values expand into one ' + 'serial sweep combination per concurrency level.') parser.add_argument( '--trust-remote-code', action=argparse.BooleanOptionalAction, @@ -434,4 +445,42 @@ def _api_model_parser(self): ) parser.add_argument('--generation-kwargs', type=json.loads, default=None, help='Generation kwargs override as a JSON object, ' - 'e.g. \'{"temperature": 0.01, "ignore_eos": false}\'') + 'e.g. \'{"temperature": 0.01, "ignore_eos": false}\'. ' + 'The `stream` key is not allowed here, use --stream instead.') + parser.add_argument('--model-abbr', type=str, default=None, + help='Model abbr override; each sweep combination appends -bs{N}') + parser.add_argument( + '--stream', + action=argparse.BooleanOptionalAction, + default=None, + help='Stream override (--stream / --no-stream); only streaming can measure TTFT/TPOT' + ) + parser.add_argument( + '--use-timestamp', + action=argparse.BooleanOptionalAction, + default=None, + help='use_timestamp override (--use-timestamp / --no-use-timestamp); ' + 'request_rate is ignored when enabled' + ) + + def _synthetic_dataset_parser(self): + """这些参数用于覆盖随机合成数据集配置，均为单值，1:1 映射配置字段。""" + parser = self.parser.add_argument_group('synthetic_dataset_args') + parser.add_argument('--request-count', type=validate_request_count, default=None, + help='Override RequestCount of the single random synthetic dataset ' + '(sample pool size, 1 <= v <= 2^20)') + parser.add_argument('--tokenid-config-request-size', type=validate_request_size, + default=None, + help='Override TokenIdConfig.RequestSize of the synthetic dataset ' + '(total input length in tokens, 1 <= v <= 2^20)') + parser.add_argument('--tokenid-config-prefix-len', type=validate_prefix_len, + default=None, + help='Override TokenIdConfig.PrefixLen of the synthetic dataset ' + '(absolute token count, 0 <= v <= RequestSize)') + parser.add_argument('--string-config-input', type=json.loads, default=None, + help='Override StringConfig.Input as a JSON object, ' + 'e.g. \'{"Method": "uniform", "Params": {"MinValue": 1, "MaxValue": 1024}}\'') + parser.add_argument('--string-config-output', type=json.loads, default=None, + help='Override StringConfig.Output as a JSON object') + parser.add_argument('--dataset-abbr', type=str, default=None, + help='Dataset abbr override (no suffix is appended)') diff --git a/ais_bench/benchmark/cli/config_manager.py b/ais_bench/benchmark/cli/config_manager.py index f28c3a33..e7d3684a 100644 --- a/ais_bench/benchmark/cli/config_manager.py +++ b/ais_bench/benchmark/cli/config_manager.py @@ -1,6 +1,7 @@ import os import os.path as osp import inspect +import json import tabulate from mmengine.config import Config @@ -11,6 +12,11 @@ from ais_bench.benchmark.utils.logging.exceptions import CommandError, AISBenchConfigError from ais_bench.benchmark.cli.utils import fill_model_path_if_datasets_need, fill_test_range_use_num_prompts, recur_convert_config_type from ais_bench.benchmark.utils.response_anomaly import ResponseAnomalyCoordinator +from ais_bench.benchmark.cli.sweep import ( + MANIFEST_FILENAME, + _normalize_sweep_arg, + expand_sweep_combinations, +) RESPONSE_ANOMALY_TOP_LOGPROBS = 20 @@ -113,12 +119,54 @@ def load_config(self, workflow): self.cfg = self._get_config_from_arg() self._init_response_anomaly_config() self._update_and_init_work_dir() + self._expand_sweep_configs() self._inject_response_anomaly_payload_storage() self._fill_dataset_configs() self._update_cfg_of_workflow(workflow) self._dump_and_reload_config() return self.cfg + def _expand_sweep_configs(self): + """按 `--batch-size`/`--num-prompts` 展开多组合配置并落盘 manifest。""" + cli_args = self.cfg.get('cli_args') or {} + result = expand_sweep_combinations( + self.cfg.get('models') or [], + self.cfg.get('datasets') or [], + cli_args, + ) + self.cfg['models'] = result.models + self.cfg['datasets'] = result.datasets + if result.combinations: + self.cfg['model_dataset_combinations'] = result.combinations + if result.sweep_cooldown_s: + self.cfg['sweep_cooldown_s'] = result.sweep_cooldown_s + if result.manifest is not None: + self._write_manifest(result.manifest) + + def _write_manifest(self, manifest): + """原子写入 `/combos_manifest.json`；失败仅告警不阻断。""" + work_dir = self.cfg.get('work_dir') + if not work_dir: + return + path = osp.join(work_dir, MANIFEST_FILENAME) + tmp_path = f'{path}.tmp{os.getpid()}' + try: + os.makedirs(work_dir, exist_ok=True) + with open(tmp_path, 'w', encoding='utf-8') as f: + json.dump(manifest, f, ensure_ascii=False, indent=2) + f.flush() + os.fsync(f.fileno()) + os.replace(tmp_path, path) + except (OSError, TypeError) as exc: + try: + if os.path.exists(tmp_path): + os.remove(tmp_path) + except OSError: + pass + self.logger.warning(f'Failed to write sweep manifest {path}: {exc}') + return + self.logger.info(f'Sweep manifest written to {path}') + def _init_response_anomaly_config(self): """Normalize the optional response anomaly detection configuration.""" global_cfg, configured_top_logprobs = ( @@ -600,7 +648,9 @@ def _fill_dataset_configs(self): for dataset_cfg in self.cfg["datasets"]: if dataset_cfg.get("infer_cfg", None) is None: continue - fill_test_range_use_num_prompts(self.cfg["cli_args"].get("num_prompts"), dataset_cfg) + fill_test_range_use_num_prompts( + _normalize_sweep_arg(self.cfg["cli_args"].get("num_prompts")), dataset_cfg + ) fill_model_path_if_datasets_need(self.cfg["models"][0], dataset_cfg) retriever_cfg = dataset_cfg["infer_cfg"]["retriever"] infer_cfg = dataset_cfg["infer_cfg"] @@ -829,6 +879,12 @@ def _apply_cli_api_model_overrides(self, models): cli_val = getattr(self.args, field, None) if cli_val is None or field not in model_cfg: continue + if field == "batch_size": + # 多档 sweep 时 batch_size 是列表，由组合展开器逐副本写入； + # 单值（含长度 1 列表）在此归一成 int 覆盖既有键。 + cli_val = _normalize_sweep_arg(cli_val) + if not isinstance(cli_val, int): + continue model_cfg[field] = cli_val def _load_summarizers_config(self): @@ -918,9 +974,12 @@ def _dump_and_reload_config(self): recur_convert_config_type(self.cfg) self.cfg.dump(output_config_path) # eval nums set - if (self.args.num_prompts and self.args.num_prompts < 0) or self.args.num_prompts == 0: + num_prompts = _normalize_sweep_arg(self.args.num_prompts) + if isinstance(num_prompts, int) and num_prompts <= 0: raise CommandError(TMAN_CODES.INVALID_ARG_VALUE_IN_CMD, "'--num-prompts' must be a positive integer greater than 0.") - self.cfg['num_prompts'] = self.args.num_prompts + # 多档 sweep 时 num_prompts 已逐组合写入数据集副本的 test_range， + # 顶层字段只承载单值语义，列表形态记 None。 + self.cfg['num_prompts'] = num_prompts if isinstance(num_prompts, int) else None # Config is intentally reloaded here to avoid initialized # types cannot be serialized try: diff --git a/ais_bench/benchmark/cli/sweep.py b/ais_bench/benchmark/cli/sweep.py new file mode 100644 index 00000000..ffd11587 --- /dev/null +++ b/ais_bench/benchmark/cli/sweep.py @@ -0,0 +1,692 @@ +import os +import copy +from abc import ABC, abstractmethod +from dataclasses import dataclass, field +from datetime import datetime +from typing import Any, List, Optional, Tuple, Union + +from ais_bench.benchmark import __version__ +from ais_bench.benchmark.utils.logging.logger import AISLogger +from ais_bench.benchmark.utils.logging.error_codes import TMAN_CODES +from ais_bench.benchmark.utils.logging.exceptions import AISBenchConfigError +from ais_bench.benchmark.cli.utils import ( + DATASETS_NEED_MODELS, + MAX_BATCH_SIZE, + MAX_REQUEST_COUNT, + MAX_REQUEST_SIZE, + get_config_type, +) + +logger = AISLogger() + +# 多组合 sweep 的内部调度字段（非 CLI、不进 manifest） +SWEEP_COOLDOWN_SECONDS = 5 +MANIFEST_FILENAME = "combos_manifest.json" +MANIFEST_SCHEMA_VERSION = 1 + + +@dataclass +class CombinationSpec: + """一个并发档组合的生效参数快照。""" + index: int + batch_size: int + num_prompts: Optional[int] + request_count: Optional[int] + request_size: Optional[int] + prefix_len: Optional[int] + model_abbr: str + dataset_abbr: str + model_name: Optional[str] = None + base_model_abbr: Optional[str] = None + dataset_type: Optional[str] = None + stream: bool = False + ttft_valid: bool = False + use_timestamp: bool = False + max_out_len: Optional[int] = None + + +@dataclass +class SweepResult: + """组合展开结果。""" + models: List[dict] = field(default_factory=list) + datasets: List[dict] = field(default_factory=list) + combinations: List[dict] = field(default_factory=list) + manifest: Optional[dict] = None + sweep_cooldown_s: int = 0 + + +class PairingStrategy(ABC): + """配对策略接口。""" + + @abstractmethod + def pair(self, left, right) -> List[Tuple[Any, Any]]: + pass + + +class PairedStrategy(PairingStrategy): + """双列表等长：按下标一一配对。""" + + def pair(self, left, right) -> List[Tuple[Any, Any]]: + return list(zip(left, right)) + + +class BroadcastStrategy(PairingStrategy): + """一列表一标量：标量广播到列表每一项。""" + + def pair(self, left, right) -> List[Tuple[Any, Any]]: + if isinstance(left, list) and not isinstance(right, list): + return [(item, right) for item in left] + if isinstance(right, list) and not isinstance(left, list): + return [(left, item) for item in right] + return [(left, right)] + + +def _normalize_sweep_arg(value) -> Union[int, List[int], None]: + """长度 1 的列表解包为 int；多元素列表原样返回；空列表报错。""" + if value is None: + return None + if isinstance(value, list): + if not value: + raise AISBenchConfigError( + TMAN_CODES.INVALID_ARG_VALUE_IN_CMD, + "`--batch-size`/`--num-prompts` must not be an empty list.", + ) + if len(value) == 1: + return int(value[0]) + return [int(item) for item in value] + return int(value) + + +def validate_pairing(left: Union[int, List[int], None], + right: Union[int, List[int], None], + request_count: Optional[int] = None, + test_range_set: bool = False + ) -> Tuple[List[Tuple[Optional[int], Optional[int]]], List[str]]: + """`--batch-size` ↔ `--num-prompts` 配对校验。 + + - 双列表等长 → PairedStrategy 按下标配对 + - 一列表一标量 → BroadcastStrategy 标量广播 + - 双列表不等长 → 抛 AISBenchConfigError（不静默截断） + - 双标量（含单值列表归一后的单档）同样走并发窗口检查，不做特例跳过 + - `batch_size` 是客户端最大并发请求数（并发窗口），请求完成后立即补位、 + 不存在固定波次，因此这里不做整除性检查；仅在并发窗口大于生效请求数 + （窗口跑不满、实际并发达不到设定值）时 WARNING，不阻断 + - 生效请求数：指定 `num_prompts` 时为 min(num_prompts, request_count)； + 未指定时默认对全量数据集评测，取 request_count；两者均未知或配置已显式 + 设置 test_range 时跳过该检查 + """ + warnings: List[str] = [] + if isinstance(left, list) and isinstance(right, list): + if len(left) != len(right): + raise AISBenchConfigError( + TMAN_CODES.INVALID_ARG_VALUE_IN_CMD, + f"`--batch-size` and `--num-prompts` must have the same length when " + f"both are lists (got {len(left)} vs {len(right)}); make them the same " + f"length or pass a single value on one side to broadcast.", + ) + strategy: PairingStrategy = PairedStrategy() + else: + strategy = BroadcastStrategy() + + pairs = strategy.pair(left, right) + if test_range_set: + return pairs, warnings + + for batch_size, num_prompts in pairs: + if not isinstance(batch_size, int): + continue + if isinstance(num_prompts, int): + effective = ( + num_prompts if request_count is None + else min(num_prompts, request_count) + ) + source = ( + f"min(`--num-prompts`={num_prompts}, `--request-count`={request_count}) " + f"= {effective}" + if effective != num_prompts + else f"`--num-prompts`={num_prompts}" + ) + else: + # 未指定 --num-prompts 时默认对全量数据集评测 + effective = request_count + source = f"all {effective} dataset samples (no `--num-prompts`)" + if isinstance(effective, int) and batch_size > effective: + warnings.append( + f"`--batch-size`={batch_size} is greater than the effective request count " + f"({source}): the concurrency window cannot be filled and the actual " + f"concurrency never exceeds {effective}, so the concurrency gain measured " + f"at this level is underestimated. Keep `--num-prompts` (or " + f"`--request-count`) not smaller than `--batch-size`." + ) + return pairs, warnings + + +def _config_type_name(cfg: Optional[dict]) -> str: + if not isinstance(cfg, dict): + return "" + return get_config_type(cfg.get("type")) or "" + + +def _is_synthetic_dataset(dataset_cfg: dict) -> bool: + return _config_type_name(dataset_cfg).endswith("SyntheticDataset") + + +def _is_dataset_need_models(dataset_cfg: dict) -> bool: + return _config_type_name(dataset_cfg) in DATASETS_NEED_MODELS + + +def _synthetic_config(dataset_cfg: dict) -> Optional[dict]: + synthetic_cfg = dataset_cfg.get("config") + if isinstance(synthetic_cfg, dict): + return synthetic_cfg + return None + + +def _dataset_request_count(datasets: List[dict], cli_args: dict) -> Optional[int]: + cli_value = cli_args.get("request_count") + if cli_value is not None: + return cli_value + for dataset_cfg in datasets: + synthetic_cfg = _synthetic_config(dataset_cfg) + if synthetic_cfg is not None and synthetic_cfg.get("RequestCount") is not None: + return synthetic_cfg.get("RequestCount") + return None + + +def _apply_model_sweep_overrides(model_cfg: dict, cli_args: dict) -> None: + """新增模型侧字段覆盖：不要求配置已存在该键。""" + if cli_args.get("model_abbr") is not None: + model_cfg["abbr"] = cli_args["model_abbr"] + if cli_args.get("stream") is not None: + model_cfg["stream"] = cli_args["stream"] + if cli_args.get("use_timestamp") is not None: + model_cfg["use_timestamp"] = cli_args["use_timestamp"] + + +def _apply_dataset_sweep_overrides(dataset_cfg: dict, cli_args: dict) -> None: + """新增数据集侧字段覆盖（均为单值，1:1 映射配置字段）。""" + if cli_args.get("dataset_abbr") is not None: + dataset_cfg["abbr"] = cli_args["dataset_abbr"] + + synthetic_cfg = _synthetic_config(dataset_cfg) + if synthetic_cfg is None: + return + + if cli_args.get("request_count") is not None: + synthetic_cfg["RequestCount"] = cli_args["request_count"] + + config_type = str(synthetic_cfg.get("Type", "")).lower() + if config_type == "tokenid": + tokenid_cfg = synthetic_cfg.get("TokenIdConfig") + if not isinstance(tokenid_cfg, dict): + tokenid_cfg = {} + synthetic_cfg["TokenIdConfig"] = tokenid_cfg + if cli_args.get("tokenid_config_request_size") is not None: + tokenid_cfg["RequestSize"] = cli_args["tokenid_config_request_size"] + if cli_args.get("tokenid_config_prefix_len") is not None: + tokenid_cfg["PrefixLen"] = cli_args["tokenid_config_prefix_len"] + elif config_type == "string": + string_cfg = synthetic_cfg.get("StringConfig") + if not isinstance(string_cfg, dict): + string_cfg = {} + synthetic_cfg["StringConfig"] = string_cfg + if cli_args.get("string_config_input") is not None: + string_cfg["Input"] = cli_args["string_config_input"] + if cli_args.get("string_config_output") is not None: + string_cfg["Output"] = cli_args["string_config_output"] + + +def _check_abbr_unique(items: List[dict], label: str, errors: List[str]) -> None: + seen = {} + for index, item in enumerate(items): + abbr = item.get("abbr") if isinstance(item, dict) else None + if abbr is None: + continue + if abbr in seen: + errors.append( + f"Duplicate {label} abbr '{abbr}': entry #{seen[abbr] + 1} conflicts " + f"with entry #{index + 1}" + ) + else: + seen[abbr] = index + + +def validate_config_fields(models: List[dict], datasets: List[dict], cli_args: dict, + pairing_warnings: Optional[List[str]] = None) -> List[str]: + """启动前校验 V01~V09（V01 的配对结果由调用方传入）。 + + 业务错误一次性汇总后抛 AISBenchConfigError；WARNING 项返回给调用方输出后继续。 + """ + errors: List[str] = [] + warnings: List[str] = list(pairing_warnings or []) + mode = cli_args.get("mode") + + # V02：batch_size / num_prompts 逐元素范围 + for arg_name, value in (("--batch-size", cli_args.get("batch_size")), + ("--num-prompts", cli_args.get("num_prompts"))): + if isinstance(value, list): + for index, item in enumerate(value, start=1): + if not isinstance(item, int) or isinstance(item, bool): + errors.append(f"`{arg_name}` value #{index} {item!r} is not an integer") + elif item < 1: + errors.append(f"`{arg_name}` value #{index} {item} is invalid (must be >= 1)") + elif arg_name == "--batch-size" and item > MAX_BATCH_SIZE: + errors.append( + f"`{arg_name}` value #{index} {item} is invalid " + f"(must be <= {MAX_BATCH_SIZE})" + ) + elif value is not None: + if not isinstance(value, int) or isinstance(value, bool): + errors.append(f"`{arg_name}` value {value!r} is not an integer") + elif value < 1: + errors.append(f"`{arg_name}` value {value} is invalid (must be >= 1)") + elif arg_name == "--batch-size" and value > MAX_BATCH_SIZE: + errors.append( + f"`{arg_name}` value {value} is invalid (must be <= {MAX_BATCH_SIZE})" + ) + + request_count = cli_args.get("request_count") + if request_count is not None and not (1 <= request_count <= MAX_REQUEST_COUNT): + errors.append( + f"`--request-count` value {request_count} is invalid " + f"(must satisfy 1 <= v <= {MAX_REQUEST_COUNT})" + ) + request_size_arg = cli_args.get("tokenid_config_request_size") + if request_size_arg is not None and not (1 <= request_size_arg <= MAX_REQUEST_SIZE): + errors.append( + f"`--tokenid-config-request-size` value {request_size_arg} is invalid " + f"(must satisfy 1 <= v <= {MAX_REQUEST_SIZE})" + ) + prefix_len_arg = cli_args.get("tokenid_config_prefix_len") + # V08：必须是整数 token 数 + if prefix_len_arg is not None: + if not isinstance(prefix_len_arg, int) or isinstance(prefix_len_arg, bool): + errors.append( + f"`--tokenid-config-prefix-len` must be an integer token count, " + f"got {prefix_len_arg!r}; fractions/ratios are not supported, use an " + f"absolute token count (e.g. --tokenid-config-prefix-len 6554)" + ) + elif prefix_len_arg < 0: + errors.append( + f"`--tokenid-config-prefix-len` value {prefix_len_arg} is invalid (must be >= 0)" + ) + + # V03：abbr 全局唯一（模型、数据集分别校验） + _check_abbr_unique(models, "model", errors) + _check_abbr_unique(datasets, "dataset", errors) + + synthetic_datasets = [d for d in datasets if _is_synthetic_dataset(d)] + need_model_path = any(_is_dataset_need_models(d) for d in datasets) + + # V04：path 非空且目录存在 + if need_model_path: + for model_cfg in models: + path = model_cfg.get("path") + if not path: + errors.append( + "The dataset requires a model path, `--path` must not be empty " + "(replaces runtime UTILS-CFG-001)" + ) + break + if not os.path.isdir(os.path.expanduser(str(path))): + errors.append( + f"`--path` directory does not exist: {path} " + "(replaces runtime 'ModelPath does not exist')" + ) + break + + # V07：一次运行只支持一个随机合成数据集 + if len(synthetic_datasets) > 1: + errors.append( + f"Only one random synthetic dataset is supported per run, but " + f"{len(synthetic_datasets)} were detected (multiple ones are merged into a " + f"single task by _merge_datasets and the results become indistinguishable; " + f"use an outer loop for multiple specs)" + ) + + tokenid_cli_used = any( + cli_args.get(key) is not None + for key in ("tokenid_config_request_size", "tokenid_config_prefix_len") + ) + string_cli_used = any( + cli_args.get(key) is not None + for key in ("string_config_input", "string_config_output") + ) + for dataset_cfg in synthetic_datasets: + synthetic_cfg = _synthetic_config(dataset_cfg) or {} + config_type = str(synthetic_cfg.get("Type", "")).lower() + # V09：参数与 config.Type 匹配（tokenid 组 / string 组不交叉） + if tokenid_cli_used and config_type != "tokenid": + errors.append( + "`--tokenid-config-request-size`/`--tokenid-config-prefix-len` only apply " + f"to datasets with config.Type=tokenid, current Type={synthetic_cfg.get('Type')!r}" + ) + if string_cli_used and config_type != "string": + errors.append( + "`--string-config-input`/`--string-config-output` only apply to datasets " + f"with config.Type=string, current Type={synthetic_cfg.get('Type')!r}" + ) + if config_type == "tokenid": + tokenid_cfg = synthetic_cfg.get("TokenIdConfig") + if isinstance(tokenid_cfg, dict): + effective_request_size = tokenid_cfg.get("RequestSize") + effective_prefix_len = tokenid_cfg.get("PrefixLen", 0) + # V05：PrefixLen ≤ RequestSize + if (isinstance(effective_request_size, int) + and isinstance(effective_prefix_len, int) + and effective_prefix_len > effective_request_size): + errors.append( + f"Dataset PrefixLen={effective_prefix_len} is greater than " + f"RequestSize={effective_request_size}; valid range is " + f"0 <= PrefixLen <= RequestSize" + ) + elif config_type == "string": + # V09：复用 SyntheticDataset 的 string 子配置校验 + try: + from ais_bench.benchmark.datasets.synthetic import SyntheticDataset + string_cfg = synthetic_cfg.get("StringConfig") or {} + SyntheticDataset.check_synthetic_string_config( + { + "Input": string_cfg.get("Input"), + "Output": string_cfg.get("Output"), + } + ) + except Exception as exc: + errors.append( + f"Invalid StringConfig: {exc}; expected " + '{"Input": {"Method": ..., "Params": {...}}, ' + '"Output": {"Method": ..., "Params": {...}}}' + ) + # V09：随机合成数据集无 timestamp 字段 + if cli_args.get("use_timestamp") is True: + errors.append( + "`--use-timestamp` cannot be used with a random synthetic dataset: it has " + "no `timestamp` field and would raise NO_TIMESTAMPS_ERROR at runtime; " + "use `--request-rate` instead" + ) + break + + # V02：num_prompts > request_count（样本池不足）→ WARNING + effective_request_count = _dataset_request_count(datasets, cli_args) + for num_prompts in _as_list(cli_args.get("num_prompts")): + if (isinstance(num_prompts, int) and isinstance(effective_request_count, int) + and num_prompts > effective_request_count): + warnings.append( + f"`--num-prompts`={num_prompts} is greater than `--request-count`=" + f"{effective_request_count}; the actual request count falls back to " + f"{effective_request_count} and the concurrency window may not be filled " + "(V01 warns about it as well). Increase `--request-count` accordingly." + ) + + # V06：stream=False + perf → WARNING 并标注 ttft_valid=false + if mode == "perf": + for model_cfg in models: + if not model_cfg.get("stream", False): + warnings.append( + f"Model [{model_cfg.get('abbr')}] is non-streaming (stream=false), " + "TTFT/TPOT cannot be measured; this combo is marked ttft_valid=false" + ) + break + + # 附加：--use-timestamp 与多并发档语义冲突 + batch_size_arg = cli_args.get("batch_size") + if cli_args.get("use_timestamp") is True and isinstance(batch_size_arg, list) \ + and len(batch_size_arg) > 1: + warnings.append( + "`--use-timestamp` is enabled but `--batch-size` provides multiple " + "concurrency levels: with timestamp scheduling batch_size no longer limits " + "in-flight requests, so the swept levels are not comparable" + ) + + if errors: + raise AISBenchConfigError( + TMAN_CODES.INVALID_ARG_VALUE_IN_CMD, + "Sweep argument validation failed:\n" + "\n".join(f" - {error}" for error in errors), + ) + return warnings + + +def _as_list(value) -> List: + if value is None: + return [] + if isinstance(value, list): + return value + return [value] + + +def _append_batch_suffix(abbr: str, batch_size: int) -> str: + """幂等追加 `-bs{N}` 后缀。""" + suffix = f"-bs{batch_size}" + if abbr.endswith(suffix): + return abbr + return abbr + suffix + + +def _manifest_cli_args(cli_args: dict) -> dict: + """manifest 中记录的用户原始参数（脱敏，不含 api_key/url/path）。""" + mapping = { + "max_out_len": "max_out_len", + "num_warmups": "num_warmups", + "generation_kwargs": "generation_kwargs", + "batch_size": "batch_size", + "num_prompts": "num_prompts", + "request_count": "request_count", + "request_size": "tokenid_config_request_size", + "prefix_len": "tokenid_config_prefix_len", + "string_config_input": "string_config_input", + "string_config_output": "string_config_output", + "model_abbr": "model_abbr", + "dataset_abbr": "dataset_abbr", + "stream": "stream", + "use_timestamp": "use_timestamp", + } + result = {} + for manifest_key, cli_key in mapping.items(): + value = cli_args.get(cli_key) + if value is not None: + result[manifest_key] = value + return result + + +def _build_manifest(specs: List[CombinationSpec], cli_args: dict) -> dict: + """组合参数快照（一次写入不再修改，不含敏感字段）。""" + combos = [] + for spec in specs: + combos.append( + { + "index": spec.index, + "model_abbr": spec.model_abbr, + "dataset_abbr": spec.dataset_abbr, + "batch_size": spec.batch_size, + "num_prompts": spec.num_prompts, + "request_count": spec.request_count, + "model_config": { + "abbr": spec.base_model_abbr, + "model_name": spec.model_name, + "stream": spec.stream, + "ttft_valid": spec.ttft_valid, + "use_timestamp": spec.use_timestamp, + "max_out_len": spec.max_out_len, + }, + "dataset_config": { + "Type": spec.dataset_type, + "RequestSize": spec.request_size, + "PrefixLen": spec.prefix_len, + }, + } + ) + return { + "schema_version": MANIFEST_SCHEMA_VERSION, + "ais_bench_version": __version__, + "created_at": datetime.now().isoformat(timespec="seconds"), + "cli_args": _manifest_cli_args(cli_args), + "combos": combos, + } + + +def expand_sweep_combinations(models: List[dict], datasets: List[dict], + cli_args: dict) -> SweepResult: + """组合展开（详见设计 4.4）。 + + - 长度 1 列表先归一为 int，单值路径只做新增字段覆盖，不产生副本/manifest/冷却字段 + - 多档时按组合深拷贝模型与数据集，模型 abbr 追加 `-bs{N}` 保证逐组合唯一 + - 校验 V01~V09，业务错误一次性汇总后抛错，WARNING 汇总输出后继续 + """ + if not models or not datasets: + return SweepResult(models=list(models), datasets=list(datasets)) + + batch_arg = _normalize_sweep_arg(cli_args.get("batch_size")) + num_arg = _normalize_sweep_arg(cli_args.get("num_prompts")) + + test_range_set = any( + isinstance(dataset_cfg.get("reader_cfg"), dict) + and "test_range" in dataset_cfg["reader_cfg"] + for dataset_cfg in datasets + ) + request_count = _dataset_request_count(datasets, cli_args) + pairs, pairing_warnings = validate_pairing( + batch_arg, num_arg, + request_count=request_count, + test_range_set=test_range_set, + ) + if test_range_set and num_arg is not None: + pairing_warnings.append( + "`test_range` has been set, `--num-prompts` will be ignored " + "(the config test_range is kept and manifest num_prompts is recorded as null)" + ) + + base_models = [copy.deepcopy(model_cfg) for model_cfg in models] + base_datasets = [copy.deepcopy(dataset_cfg) for dataset_cfg in datasets] + for model_cfg in base_models: + _apply_model_sweep_overrides(model_cfg, cli_args) + for dataset_cfg in base_datasets: + _apply_dataset_sweep_overrides(dataset_cfg, cli_args) + + warnings = validate_config_fields( + base_models, base_datasets, cli_args, + pairing_warnings=pairing_warnings, + ) + + is_multi = len(pairs) > 1 + + specs: List[CombinationSpec] = [] + combo_models: List[dict] = [] + combo_datasets: List[dict] = [] + for batch_size, num_prompts in pairs: + for model_cfg in base_models: + for dataset_cfg in base_datasets: + model_copy = copy.deepcopy(model_cfg) + if batch_size is not None: + model_copy["batch_size"] = batch_size + base_model_abbr = model_copy.get("abbr") + if is_multi and base_model_abbr: + model_copy["abbr"] = _append_batch_suffix(base_model_abbr, batch_size) + + dataset_copy = copy.deepcopy(dataset_cfg) + if is_multi and num_prompts is not None and not test_range_set: + reader_cfg = dataset_copy.get("reader_cfg") + if not isinstance(reader_cfg, dict): + reader_cfg = {} + dataset_copy["reader_cfg"] = reader_cfg + reader_cfg["test_range"] = f"[:{num_prompts}]" + if _is_dataset_need_models(dataset_copy): + dataset_copy["model_path"] = model_copy.get("path") + dataset_copy["trust_remote_code"] = model_copy.get( + "trust_remote_code", False + ) + + synthetic_cfg = _synthetic_config(dataset_copy) or {} + tokenid_cfg = synthetic_cfg.get("TokenIdConfig") or {} + effective_batch_size = model_copy.get("batch_size", 1) + # manifest / 报告记录生效请求数：未指定 --num-prompts 时默认对全量 + # 数据集评测，取数据集全部样本；指定时受样本池上限约束 + if test_range_set: + effective_num_prompts = None + elif isinstance(num_prompts, int): + effective_num_prompts = ( + num_prompts if request_count is None + else min(num_prompts, request_count) + ) + else: + effective_num_prompts = request_count + + stream = bool(model_copy.get("stream", False)) + specs.append( + CombinationSpec( + index=len(specs), + batch_size=effective_batch_size, + num_prompts=effective_num_prompts, + request_count=request_count, + request_size=tokenid_cfg.get("RequestSize"), + prefix_len=tokenid_cfg.get("PrefixLen"), + model_abbr=model_copy.get("abbr"), + dataset_abbr=dataset_copy.get("abbr"), + model_name=model_copy.get("model") + or model_copy.get("model_name"), + base_model_abbr=base_model_abbr, + dataset_type=synthetic_cfg.get("Type"), + stream=stream, + ttft_valid=stream, + use_timestamp=bool(model_copy.get("use_timestamp", False)), + max_out_len=model_copy.get("max_out_len"), + ) + ) + combo_models.append(model_copy) + combo_datasets.append(dataset_copy) + + # 展开后仍需保证每个组合的输出路径唯一：同一 (模型 abbr, 数据集 abbr) 只允许 + # 出现一次，否则 performances//.json 会互相覆盖。 + # 多模型/多数据集笛卡尔积下同一 abbr 会在不同组合中重复出现，属正常情况。 + post_errors: List[str] = [] + seen_outputs = {} + for combo_index, (model_copy, dataset_copy) in enumerate( + zip(combo_models, combo_datasets)): + output_key = (model_copy.get("abbr"), dataset_copy.get("abbr")) + if output_key in seen_outputs: + post_errors.append( + f"Duplicate combo output path: model abbr '{output_key[0]}' and dataset " + f"abbr '{output_key[1]}' appear in both combo " + f"#{seen_outputs[output_key] + 1} and combo #{combo_index + 1}" + ) + else: + seen_outputs[output_key] = combo_index + if post_errors: + raise AISBenchConfigError( + TMAN_CODES.INVALID_ARG_VALUE_IN_CMD, + "Sweep combo abbr conflict after expansion:\n" + + "\n".join(f" - {error}" for error in post_errors), + ) + + for warning in warnings: + logger.warning(warning) + + if is_multi and cli_args.get("max_num_workers", 1) > 1 and cli_args.get("mode") == "perf": + logger.warning( + "Detected %d sweep combos with --max-num-workers=%s > 1: multiple tasks will " + "hit the same service concurrently and the perf data may pollute each other; " + "keep it serial (--max-num-workers 1)", + len(specs), + cli_args.get("max_num_workers"), + ) + + if not is_multi: + # 单值等价路径：不产生 model_dataset_combinations / manifest / 冷却字段 + return SweepResult( + models=combo_models, + datasets=combo_datasets, + combinations=[], + manifest=None, + sweep_cooldown_s=0, + ) + + combinations = [ + {"models": [model_copy], "datasets": [dataset_copy]} + for model_copy, dataset_copy in zip(combo_models, combo_datasets) + ] + return SweepResult( + models=combo_models, + datasets=combo_datasets, + combinations=combinations, + manifest=_build_manifest(specs, cli_args), + sweep_cooldown_s=SWEEP_COOLDOWN_SECONDS, + ) diff --git a/ais_bench/benchmark/cli/utils.py b/ais_bench/benchmark/cli/utils.py index 8fa6f6b7..58fffa92 100644 --- a/ais_bench/benchmark/cli/utils.py +++ b/ais_bench/benchmark/cli/utils.py @@ -13,6 +13,9 @@ MAX_NUM_WORKERS = int(os.cpu_count() * 0.8) DEFAULT_PRESSURE_TIME = 15 MAX_PRESSURE_TIME = 60 * 60 * 24 # 24 hours +MAX_BATCH_SIZE = 100000 +MAX_REQUEST_COUNT = 2 ** 20 +MAX_REQUEST_SIZE = 2 ** 20 logger = AISLogger() @@ -63,9 +66,13 @@ def fill_model_path_if_datasets_need(model_cfg, dataset_cfg): dataset_cfg.update({"model_path": model_path, "trust_remote_code": trust_remote_code}) -def fill_test_range_use_num_prompts(num_prompts: int, dataset_cfg: dict): +def fill_test_range_use_num_prompts(num_prompts, dataset_cfg: dict): if not num_prompts: return + # 列表形态说明本次运行是多组合 sweep，test_range 由展开器逐副本写入， + # 这里必须跳过，避免把列表拼进 test_range 字符串。 + if isinstance(num_prompts, (list, tuple)): + return reader_cfg = dataset_cfg["reader_cfg"] if "test_range" in reader_cfg: if isinstance(num_prompts, int): @@ -186,3 +193,27 @@ def validator(value): max_value=MAX_PRESSURE_TIME, error_message_suffix=f"(maximum: {MAX_PRESSURE_TIME} seconds, which is 24 hours)" ) + +validate_batch_size = create_int_validator( + param_name="--batch-size", + min_value=1, + max_value=MAX_BATCH_SIZE, +) + +validate_request_count = create_int_validator( + param_name="--request-count", + min_value=1, + max_value=MAX_REQUEST_COUNT, +) + +validate_request_size = create_int_validator( + param_name="--tokenid-config-request-size", + min_value=1, + max_value=MAX_REQUEST_SIZE, +) + +validate_prefix_len = create_int_validator( + param_name="--tokenid-config-prefix-len", + min_value=0, + error_message_suffix="(an absolute token count is required; fractions/ratios are not supported)", +) diff --git a/ais_bench/benchmark/cli/workers.py b/ais_bench/benchmark/cli/workers.py index 55af6a78..1ea83887 100644 --- a/ais_bench/benchmark/cli/workers.py +++ b/ais_bench/benchmark/cli/workers.py @@ -102,6 +102,7 @@ def update_new_infer_cfg(new_cfg: ConfigDict) -> None: runner_cfg['max_num_workers'] = self.args.max_num_workers runner_cfg['max_workers_per_gpu'] = self.args.max_workers_per_gpu runner_cfg['debug'] = self.args.debug or cfg.cli_args.debug + runner_cfg['sweep_cooldown_s'] = cfg.get('sweep_cooldown_s', 0) if cfg.get('infer'): new_cfg = dict(infer=cfg.infer) diff --git a/ais_bench/benchmark/runners/local.py b/ais_bench/benchmark/runners/local.py index 36a23b4c..e94dac28 100644 --- a/ais_bench/benchmark/runners/local.py +++ b/ais_bench/benchmark/runners/local.py @@ -57,11 +57,13 @@ def __init__(self, debug: bool = False, max_workers_per_gpu: int = 1, keep_tmp_file: bool = False, + sweep_cooldown_s: int = 0, **kwargs): super().__init__(task=task, debug=debug) self.max_num_workers = max_num_workers self.max_workers_per_gpu = max_workers_per_gpu self.keep_tmp_file = keep_tmp_file + self.sweep_cooldown_s = sweep_cooldown_s for k, v in kwargs.items(): self.logger.warning(f'Ignored argument in {self.__module__}: {k}={v}') @@ -206,6 +208,15 @@ def submit(task, index): lock.release() time.sleep(1) + # 多组合串行压测时的档间冷却：仅串行生效，避免上一档的 in-flight + # 请求/连接残留污染下一档性能数据。 + if index > 0 and self.sweep_cooldown_s and self.max_num_workers == 1: + self.logger.info( + f"Cooldown {self.sweep_cooldown_s}s before launching task " + f"{index + 1}/{len(tasks)} ({task.name})" + ) + time.sleep(self.sweep_cooldown_s) + res = self._launch(task, gpu_ids, index) with lock: diff --git a/ais_bench/benchmark/summarizers/default_perf.py b/ais_bench/benchmark/summarizers/default_perf.py index 80c596de..faab4203 100644 --- a/ais_bench/benchmark/summarizers/default_perf.py +++ b/ais_bench/benchmark/summarizers/default_perf.py @@ -30,6 +30,7 @@ from ais_bench.benchmark.utils.logging.error_codes import SUMM_CODES from ais_bench.benchmark.utils.file.load_tokenizer import load_tokenizer, AISTokenizer from ais_bench.benchmark.utils.visualization.rps_distribution_plot import add_actual_rps_to_chart +from ais_bench.benchmark.summarizers.perf_batch import aggregate_batch_perf def model_abbr_from_cfg_used_in_summarizer(model): @@ -77,7 +78,16 @@ def __init__(self, config: ConfigDict, calculator: ConfigDict) -> None: str(dataset_cfg['type']) + "_" # same dataset type + str(dataset_cfg["infer_cfg"]["inferencer"]) # same inferencer with the same args ) - dataset_groups[data_key].append(dataset_cfg) + group = dataset_groups[data_key] + # 多组合 sweep 会为每个并发档复制一份数据集配置（仅 test_range 不同， + # abbr 相同）。同组内按 abbr 去重，避免同一份 details 被加载多次导致 + # 指标翻倍，也保证文件路径仍按数据集 abbr 定位。 + abbr = dataset_cfg.get("abbr") + if abbr is not None and any( + item.get("abbr") == abbr for item in group + ): + continue + group.append(dataset_cfg) self.dataset_groups = dataset_groups self.work_dir = self.cfg["work_dir"] @@ -485,3 +495,10 @@ def summarize(self): # Output to screen self._output_to_screen(perf_tables) + + # 多组合并发批量压测：聚合各组合结果，产出 CSV 与 HTML 报告 + # （manifest 缺失或单组合时汇总器内部直接跳过，零开销） + try: + aggregate_batch_perf(self.work_dir) + except Exception as exc: # 汇总失败不影响性能结果与退出码 + self.logger.warning(f"Batch perf aggregation skipped: {exc}") diff --git a/ais_bench/benchmark/summarizers/perf_batch.py b/ais_bench/benchmark/summarizers/perf_batch.py new file mode 100644 index 00000000..8e1ffb92 --- /dev/null +++ b/ais_bench/benchmark/summarizers/perf_batch.py @@ -0,0 +1,1201 @@ +"""多组合并发批量压测的汇总与报告渲染。 + +`aggregate_batch_perf` 读取配置加载阶段落盘的 `combos_manifest.json`，逐组合 +汇总 `performances//.json|.csv`，先落盘 +`batch_perf_summary.csv`，再以同一份 rows 渲染 `batch_perf_report.html`。 + +单组合（manifest 缺失或组合数 ≤ 1）时直接跳过，零开销；个别组合结果缺失时 +该行标记 fail，不阻塞其余行；HTML 渲染失败只 WARNING，不影响 CSV 与退出码。 +""" + +import csv +import json +import os +import re +import sqlite3 +from collections import defaultdict +from dataclasses import dataclass +from datetime import datetime +from pathlib import Path +from typing import Dict, List, Optional, Tuple + +import numpy as np + +from ais_bench.benchmark import __version__ +from ais_bench.benchmark.cli.sweep import MANIFEST_FILENAME +from ais_bench.benchmark.utils.logging.logger import AISLogger + +logger = AISLogger() + +SUMMARY_CSV_NAME = "batch_perf_summary.csv" +REPORT_HTML_NAME = "batch_perf_report.html" +REPORT_MARKER = "" +TEMPLATE_NAME = "batch_perf_report.html.j2" +# 与 EvalScope PLOTLY_CDN_URL 对齐：单文件报告只引一次 CDN，不内联 JS +PLOTLY_CDN_URL = "https://cdn.plot.ly/plotly-2.27.0.min.js" +# 图表文字色取中间灰：报告默认深色主题，但切到浅色主题后仍需可读 +CHART_TEXT_COLOR = "#7f8a9e" + +# HTML 概览宽表：一个组合一行，列口径对齐 EvalScope 的 Detailed Performance Metrics +OVERVIEW_COLUMNS = [ + ("Batch Size", "batch_size"), + ("Num Prompts", "num_prompts"), + ("RPS", "request_throughput"), + ("Avg Lat.(s)", "e2el_avg_s"), + ("P99 Lat.(s)", "e2el_p99_s"), + ("Avg TTFT(ms)", "ttft_avg"), + ("P99 TTFT(ms)", "ttft_p99"), + ("Avg TPOT(ms)", "tpot_avg"), + ("P99 TPOT(ms)", "tpot_p99"), + ("Gen. tok/s", "throughput_tok_s"), + ("Success Rate", "success_rate_text"), + ("Task Status", "status"), +] +OVERVIEW_HEADERS = [label for label, _ in OVERVIEW_COLUMNS] + +# batch_perf_summary.csv 与单任务 synthetic.csv 对齐：指标逐行展开， +# 身份列逐行重复填充，保证可排序、可筛选、可直接透视 +# CSV 表头使用可读名称，右侧为内部字段名 +IDENTITY_COLUMNS = [ + ("Model Abbr", "model_abbr"), + ("Dataset Abbr", "dataset_abbr"), + ("Batch Size", "batch_size"), + ("Num Prompts", "num_prompts"), + ("Task Status", "status"), +] +IDENTITY_HEADERS = [label for label, _ in IDENTITY_COLUMNS] + +PERF_METRIC_HEADERS = [ + "Performance Parameters", + "Stage", + "Average", + "Min", + "Max", + "Median", + "P75", + "P90", + "P99", + "N", +] + +CSV_HEADERS = IDENTITY_HEADERS + PERF_METRIC_HEADERS + +# CSV 汇总区（synthetic.json 的 common res）：一个组合一行，指标名作表头、取值保留原单位 +RUN_SUMMARY_METRICS = [ + "Benchmark Duration", + "Total Requests", + "Failed Requests", + "Success Requests", + "Concurrency", + "Max Concurrency", + "Request Throughput", + "Total Input Tokens", + "Total Generated Tokens", + "Input Token Throughput", + "Output Token Throughput", + "Total Token Throughput", +] +RUN_SUMMARY_HEADERS = IDENTITY_HEADERS + RUN_SUMMARY_METRICS + +# 指标统计折线图：横轴为百分位点 Min→Max（不含 Average/Median） +PERCENTILE_STATS = ["Min", "P75", "P90", "P99", "Max"] +# 指标行内统计列下标（与 PERF_METRIC_HEADERS 一一对应） +PERF_STAT_INDEX = {label: index for index, label in enumerate(PERF_METRIC_HEADERS)} + +# 指标统计图：(中文标题, 英文标题, y 轴标题, [(指标名, 图例名, 颜色)]) +METRIC_CHART_SPECS = [ + ("时延", "E2EL", "Latency (ms)", + [("E2EL", "E2EL", "#63b3ed")]), + ("首Token/每Token/Token间时延", "TTFT/TPOT/ITL", "Latency (ms)", + [("TTFT", "TTFT", "#63b3ed"), + ("TPOT", "TPOT", "#a78bfa"), + ("ITL", "ITL", "#34d399")]), + ("输入/输出Token数", "Input/Output Tokens", "Tokens", + [("InputTokens", "Input Tokens", "#63b3ed"), + ("OutputTokens", "Output Tokens", "#34d399")]), + ("输入/输出Token吞吐", "Prefill/Output Token Throughput", "Tokens/s", + [("OutputTokenThroughput", "Output Token Throughput", "#34d399"), + ("PrefillTokenThroughput", "Prefill Token Throughput", "#a78bfa")]), +] + +# 逐请求曲线超过该点数时只画线不画点，避免密集遮挡 +DETAIL_MARKER_LIMIT = 200 + +_NUMBER_PATTERN = re.compile(r"^\s*(-?\d+(?:\.\d+)?(?:[eE][-+]?\d+)?)") + + +@dataclass +class BatchPerfArtifacts: + """汇总产物路径；跳过或失败时为 None。""" + csv_path: Optional[Path] = None + html_path: Optional[Path] = None + + +def _parse_number(value) -> Optional[float]: + """把 `"123.4 ms"` / `"1.2e+03"` / 数字统一解析为 float。""" + if value is None: + return None + if isinstance(value, (int, float)) and not isinstance(value, bool): + return float(value) + match = _NUMBER_PATTERN.match(str(value)) + if not match: + return None + try: + return float(match.group(1)) + except ValueError: + return None + + +def _load_perf_json(path: Path) -> Optional[dict]: + """读取 common res（`{metric: {stage: "value unit"}}`）。""" + if not path.exists(): + return None + try: + with open(path, "r", encoding="utf-8") as file: + data = json.load(file) + except (OSError, ValueError) as exc: + logger.warning(f"Failed to load perf result {path}: {exc}") + return None + return data if isinstance(data, dict) else None + + +def _common_metric(common: dict, name: str) -> Optional[float]: + """取 common res 指标的 total 阶段值（无 total 时取第一个阶段）。""" + stage_map = common.get(name) + if not isinstance(stage_map, dict) or not stage_map: + return None + if "total" in stage_map: + return _parse_number(stage_map["total"]) + return _parse_number(next(iter(stage_map.values()))) + + +def _load_perf_csv(path: Path) -> Dict[Tuple[str, str], Dict[str, Optional[float]]]: + """读取逐请求指标统计（`[param, stage, stat...]`）。""" + table: Dict[Tuple[str, str], Dict[str, Optional[float]]] = {} + if not path.exists(): + return table + try: + with open(path, "r", encoding="utf-8", newline="") as file: + reader = csv.reader(file) + try: + headers = next(reader) + except StopIteration: + return table + stats = headers[2:] + for raw_row in reader: + if len(raw_row) < 2: + continue + param, stage = raw_row[0], raw_row[1] + values = raw_row[2:] + table[(param, stage)] = { + stat: _parse_number(values[index] if index < len(values) else None) + for index, stat in enumerate(stats) + } + except OSError as exc: + logger.warning(f"Failed to load perf csv {path}: {exc}") + return table + + +def _csv_stat(table, param: str, stat: str, + stage: str = "total") -> Optional[float]: + stage_map = table.get((param, stage)) + if not stage_map: + return None + return stage_map.get(stat) + + +def _manifest_path(work_dir: Path) -> Path: + return work_dir / MANIFEST_FILENAME + + +def _load_manifest(work_dir: Path) -> Optional[dict]: + path = _manifest_path(work_dir) + if not path.exists(): + return None + try: + with open(path, "r", encoding="utf-8") as file: + manifest = json.load(file) + except (OSError, ValueError) as exc: + logger.warning(f"Failed to load sweep manifest {path}: {exc}") + return None + return manifest if isinstance(manifest, dict) else None + + +def _is_truthy(value) -> bool: + if isinstance(value, str): + return value.strip().lower() in ("1", "true", "yes") + return bool(value) + + +def _build_row(work_dir: Path, combo: dict) -> dict: + """单个组合的汇总行（数值态，格式化在落盘/渲染时统一做）。""" + model_abbr = combo.get("model_abbr") + dataset_abbr = combo.get("dataset_abbr") + model_cfg = combo.get("model_config") or {} + dataset_cfg = combo.get("dataset_config") or {} + stream = _is_truthy(model_cfg.get("stream", False)) + ttft_valid = _is_truthy(model_cfg.get("ttft_valid", False)) + + row = { + "index": combo.get("index"), + "model_abbr": model_abbr, + "dataset_abbr": dataset_abbr, + "batch_size": combo.get("batch_size"), + "num_prompts": combo.get("num_prompts"), + "request_count": combo.get("request_count"), + "stream": stream, + "ttft_valid": ttft_valid, + "use_timestamp": _is_truthy(model_cfg.get("use_timestamp", False)), + "prefix_len": dataset_cfg.get("PrefixLen"), + "request_size": dataset_cfg.get("RequestSize"), + "output_len": model_cfg.get("max_out_len"), + "dataset_type": dataset_cfg.get("Type"), + "status": "fail", + "throughput_tok_s": None, + "total_throughput_tok_s": None, + "input_throughput_tok_s": None, + "request_throughput": None, + "e2el_avg": None, + "e2el_p99": None, + "e2el_avg_s": None, + "e2el_p99_s": None, + "ttft_avg": None, + "ttft_p90": None, + "ttft_p99": None, + "tpot_avg": None, + "tpot_p99": None, + "itl_avg": None, + "itl_p99": None, + "success_rate": None, + "success_rate_text": "", + "completed_requests": None, + "total_requests": None, + "failed_requests": None, + "concurrency": None, + "max_concurrency": None, + "total_input_tokens": None, + "total_output_tokens": None, + "avg_input_tokens": None, + "avg_output_tokens": None, + "duration_s": None, + } + + perf_dir = work_dir / "performances" / str(model_abbr) + common = _load_perf_json(perf_dir / f"{dataset_abbr}.json") + if common is None: + return row + metrics = _load_perf_csv(perf_dir / f"{dataset_abbr}.csv") + + row["status"] = "success" + row["throughput_tok_s"] = _common_metric(common, "Output Token Throughput") + row["total_throughput_tok_s"] = _common_metric(common, "Total Token Throughput") + row["input_throughput_tok_s"] = _common_metric(common, "Input Token Throughput") + row["request_throughput"] = _common_metric(common, "Request Throughput") + row["concurrency"] = _common_metric(common, "Concurrency") + row["max_concurrency"] = _common_metric(common, "Max Concurrency") + row["total_input_tokens"] = _common_metric(common, "Total Input Tokens") + row["total_output_tokens"] = _common_metric(common, "Total Generated Tokens") + + total_requests = _common_metric(common, "Total Requests") + success_requests = _common_metric(common, "Success Requests") + failed_requests = _common_metric(common, "Failed Requests") + if total_requests is not None: + row["total_requests"] = int(total_requests) + if success_requests is not None: + row["completed_requests"] = int(success_requests) + if failed_requests is not None: + row["failed_requests"] = int(failed_requests) + if total_requests: + row["success_rate"] = round(success_requests / total_requests * 100, 2) \ + if success_requests is not None else None + if row["success_rate"] is not None: + row["success_rate_text"] = f"{_format_value(row['success_rate'])}%" + duration_ms = _common_metric(common, "Benchmark Duration") + if duration_ms is not None: + row["duration_s"] = round(duration_ms / 1000, 4) + + row["e2el_avg"] = _csv_stat(metrics, "E2EL", "Average") + row["e2el_p99"] = _csv_stat(metrics, "E2EL", "P99") + if row["e2el_avg"] is not None: + row["e2el_avg_s"] = round(row["e2el_avg"] / 1000, 3) + if row["e2el_p99"] is not None: + row["e2el_p99_s"] = round(row["e2el_p99"] / 1000, 3) + row["avg_input_tokens"] = _csv_stat(metrics, "InputTokens", "Average") + row["avg_output_tokens"] = _csv_stat(metrics, "OutputTokens", "Average") + if ttft_valid: + row["ttft_avg"] = _csv_stat(metrics, "TTFT", "Average") + row["ttft_p90"] = _csv_stat(metrics, "TTFT", "P90") + row["ttft_p99"] = _csv_stat(metrics, "TTFT", "P99") + row["tpot_avg"] = _csv_stat(metrics, "TPOT", "Average") + row["tpot_p99"] = _csv_stat(metrics, "TPOT", "P99") + row["itl_avg"] = _csv_stat(metrics, "ITL", "Average") + row["itl_p99"] = _csv_stat(metrics, "ITL", "P99") + return row + + +def _format_value(value) -> str: + """统一格式化（CSV 与 HTML 共用，保证数值逐单元格一致）。""" + if value is None: + return "" + if isinstance(value, bool): + return "true" if value else "false" + if isinstance(value, float): + return f"{value:.4f}".rstrip("0").rstrip(".") + return str(value) + + +def _escape_csv_text(value) -> str: + """CSV 公式注入转义：仅作用于文本列。""" + text = "" if value is None else str(value) + if text[:1] in ("=", "+", "-", "@", "\t", "\r"): + return "'" + text + return text + + +def _format_row(row: dict) -> dict: + """HTML 展示用的格式化行（自动转义由模板负责，这里只做数值格式化）。""" + return {key: _format_value(value) for key, value in row.items()} + + +def _identity_values(row: dict) -> List[str]: + """汇总 CSV 的身份列取值（文本列做公式注入转义）。""" + values = [] + for _, key in IDENTITY_COLUMNS: + if key in ("model_abbr", "dataset_abbr", "status"): + values.append(_escape_csv_text(row.get(key))) + else: + values.append(_format_value(row.get(key))) + return values + + +def _read_metric_rows(path: Path) -> Tuple[List[str], List[List[str]]]: + """原样读取单组合 `synthetic.csv`：返回 (表头, 指标行)，值不做二次格式化。""" + if not path.exists(): + return [], [] + try: + with open(path, "r", encoding="utf-8", newline="") as file: + reader = csv.reader(file) + try: + header = next(reader) + except StopIteration: + return [], [] + raw_rows = [row for row in reader if len(row) >= 2] + except OSError as exc: + logger.warning(f"Failed to load perf csv {path}: {exc}") + return [], [] + width = len(header) + return header, [(row + [""] * width)[:width] for row in raw_rows] + + +def _metric_rows_for_row(row: dict, work_dir: Path) -> Tuple[List[str], List[List[str]]]: + """读取单组合的指标行；非流式组合的逐 token 时延口径置空。""" + if row.get("status") != "success": + return [], [] + perf_dir = work_dir / "performances" / str(row.get("model_abbr")) + header, metric_rows = _read_metric_rows( + perf_dir / f"{row.get('dataset_abbr')}.csv") + normalized: List[List[str]] = [] + for metric_row in metric_rows: + if not row.get("ttft_valid") and metric_row[0] in ("TTFT", "TPOT", "ITL"): + # 非流式组合的逐 token 时延口径无效：保留指标名、清空统计值 + metric_row = metric_row[:2] + [""] * (len(header) - 2) + normalized.append(metric_row) + return header, normalized + + +def _expand_summary_rows(rows: List[dict], work_dir: Path) -> List[List[str]]: + """把"一组合一行"展开为「身份列 + 指标行」，身份列每行重复填充。""" + expanded: List[List[str]] = [] + for row in rows: + identity = _identity_values(row) + _, metric_rows = _metric_rows_for_row(row, work_dir) + if not metric_rows: + expanded.append(identity + [""] * len(PERF_METRIC_HEADERS)) + continue + for metric_row in metric_rows: + # CSV 表头固定为 PERF_METRIC_HEADERS，逐行对齐列宽 + metric_row = (metric_row + [""] * len(PERF_METRIC_HEADERS))[ + :len(PERF_METRIC_HEADERS)] + expanded.append(identity + metric_row) + return expanded + + +def _build_run_summary_row(row: dict, work_dir: Path) -> List[str]: + """汇总区一行：身份列 + synthetic.json 原文取值（含单位），缺失的组合留空。""" + perf_dir = work_dir / "performances" / str(row.get("model_abbr")) + common = _load_perf_json( + perf_dir / f"{row.get('dataset_abbr')}.json") or {} + values: List[str] = [] + for name in RUN_SUMMARY_METRICS: + stage_map = common.get(name) + if isinstance(stage_map, dict) and stage_map: + stage = "total" if "total" in stage_map else next(iter(stage_map)) + values.append(_format_value(stage_map.get(stage))) + else: + values.append("") + return _identity_values(row) + values + + +def _write_summary_csv(path: Path, rows: List[dict], work_dir: Path) -> None: + """写汇总 CSV：先按组合展开指标行，再空一行追加 synthetic.json 汇总区。""" + expanded = _expand_summary_rows(rows, work_dir) + run_rows = [_build_run_summary_row(row, work_dir) for row in rows] + tmp_path = path.with_name(f"{path.name}.tmp{os.getpid()}") + try: + with open(tmp_path, "w", encoding="utf-8", newline="") as file: + writer = csv.writer(file) + writer.writerow(CSV_HEADERS) + writer.writerows(expanded) + if run_rows: + # 空一行后是独立表头 + 每组合一行的单值汇总区 + writer.writerow([]) + writer.writerow(RUN_SUMMARY_HEADERS) + writer.writerows(run_rows) + file.flush() + os.fsync(file.fileno()) + os.replace(tmp_path, path) + except OSError: + try: + if tmp_path.exists(): + os.remove(tmp_path) + except OSError: + pass + raise + + +def _comparability_key(row: dict) -> Tuple: + """可比性分组：不同输入规格/口径不混画。""" + return ( + row.get("stream"), + row.get("ttft_valid"), + row.get("use_timestamp"), + row.get("request_size"), + row.get("prefix_len"), + ) + + +def _sorted_groups(rows: List[dict]) -> List[Tuple[Tuple, List[dict]]]: + groups: Dict[Tuple, List[dict]] = defaultdict(list) + for row in rows: + if row.get("status") != "success": + continue + groups[_comparability_key(row)].append(row) + for group_rows in groups.values(): + group_rows.sort(key=lambda item: (item.get("batch_size") or 0)) + return sorted(groups.items(), key=lambda item: str(item[0])) + + +def _figure_html(figure) -> str: + return figure.to_html( + full_html=False, + include_plotlyjs=False, + config={"displayModeBar": False, "responsive": True}, + ) + + +def _style_figure(figure, x_title: str, y_title: str, height: int = 380) -> None: + """统一图表样式：透明背景、横向图例、宽度随容器自适应。""" + figure.update_layout( + font={"color": CHART_TEXT_COLOR, "size": 12}, + margin={"l": 60, "r": 20, "t": 40, "b": 50}, + height=height, + legend={"orientation": "h", "yanchor": "bottom", "y": 1.02, + "xanchor": "right", "x": 1, + "font": {"color": CHART_TEXT_COLOR}}, + hovermode="x unified", + hoverlabel={"bgcolor": "#16163a", "bordercolor": "rgba(129,109,248,0.4)", + "font": {"color": "#e2e8f0"}}, + paper_bgcolor="rgba(0,0,0,0)", + plot_bgcolor="rgba(0,0,0,0)", + xaxis={"title": {"text": x_title}, + "tickfont": {"color": CHART_TEXT_COLOR}, + "gridcolor": "rgba(129,109,248,0.12)", + "zerolinecolor": "rgba(129,109,248,0.2)"}, + yaxis={"title": {"text": y_title}, + "tickfont": {"color": CHART_TEXT_COLOR}, + "gridcolor": "rgba(129,109,248,0.12)", + "zerolinecolor": "rgba(129,109,248,0.2)"}, + ) + + +def _stat_value(metric_rows: List[List[str]], metric: str, + stat: str) -> Optional[float]: + """从指标行取某指标的统计值；同一指标有多个 stage 时优先 total。""" + index = PERF_STAT_INDEX.get(stat) + if index is None: + return None + fallback = None + for metric_row in metric_rows: + if len(metric_row) <= index or metric_row[0] != metric: + continue + value = _parse_number(metric_row[index]) + if value is None: + continue + if len(metric_row) > 1 and metric_row[1] == "total": + return value + if fallback is None: + fallback = value + return fallback + + +def _build_metric_charts(metric_rows: List[List[str]]) -> List[dict]: + """指标统计折线图：横轴为百分位点 Min→P99→Max，仅画连线。""" + try: + import plotly.graph_objects as go + except Exception as exc: # plotly 缺失时降级为无图表 + logger.warning(f"plotly is unavailable, skip metric percentile charts: {exc}") + return [] + + charts: List[dict] = [] + for title_zh, title_en, y_title, series in METRIC_CHART_SPECS: + figure = go.Figure() + has_data = False + for metric, label, color in series: + line_points = [(stat, _stat_value(metric_rows, metric, stat)) + for stat in PERCENTILE_STATS] + line_points = [(stat, value) for stat, value in line_points + if value is not None] + if not line_points: + continue + has_data = True + figure.add_trace( + go.Scatter( + x=[stat for stat, _ in line_points], + y=[value for _, value in line_points], + mode="lines+markers", + name=label, + line={"color": color, "width": 2}, + marker={"size": 7}, + hovertemplate="%{x}: %{y}" + label + "", + ) + ) + if not has_data: + continue + _style_figure(figure, "Percentile", y_title) + figure.update_xaxes(type="category", categoryorder="array", + categoryarray=PERCENTILE_STATS) + charts.append({"title_zh": title_zh, "title_en": title_en, + "html": _figure_html(figure)}) + return charts + + +def _build_request_charts(detail_rows: List[dict]) -> List[dict]: + """请求明细折线图：延迟 / TTFT-TPOT-ITL / 输入输出 token / 成功失败。""" + try: + import plotly.graph_objects as go + except Exception as exc: # plotly 缺失时降级为无图表 + logger.warning(f"plotly is unavailable, skip request charts: {exc}") + return [] + + indexes = list(range(1, len(detail_rows) + 1)) + charts: List[dict] = [] + + def add(title_zh: str, title_en: str, y_title: str, series) -> None: + figure = go.Figure() + has_data = False + for field, label, color in series: + # 明细字段可能是 list（多模态输入）或缺失，统一走 _parse_number 只保留数值点 + points = [(index, _parse_number(item.get(field))) + for index, item in zip(indexes, detail_rows)] + points = [(index, value) for index, value in points + if value is not None] + if not points: + continue + has_data = True + figure.add_trace( + go.Scatter( + x=[point[0] for point in points], + y=[point[1] for point in points], + mode="lines+markers" if len(points) <= DETAIL_MARKER_LIMIT else "lines", + name=label, + line={"color": color, "width": 2}, + marker={"size": 5}, + ) + ) + if not has_data: + return + _style_figure(figure, "Request Index", y_title) + charts.append({"title_zh": title_zh, "title_en": title_en, + "html": _figure_html(figure)}) + + add("时延", "Latency", "Latency (ms)", + [("latency", "Latency", "#63b3ed")]) + add("首Token/每Token/Token间时延", "TTFT/TPOT/ITL", "Latency (ms)", + [("ttft", "TTFT", "#63b3ed"), + ("tpot", "TPOT", "#a78bfa"), + ("itl", "ITL", "#34d399")]) + add("输入/输出Token数", "Input/Output Tokens", "Tokens", + [("input_len", "Input Tokens", "#63b3ed"), + ("output_len", "Output Tokens", "#34d399")]) + + status_points = [(index, 1 if item.get("success") else 0) + for index, item in zip(indexes, detail_rows)] + if status_points: + figure = go.Figure() + figure.add_trace( + go.Scatter( + x=[point[0] for point in status_points], + y=[point[1] for point in status_points], + mode="lines+markers", + name="Status", + line={"shape": "hv", "width": 1.5, "color": CHART_TEXT_COLOR}, + marker={ + "size": 7, + "color": ["#34d399" if point[1] else "#f87171" + for point in status_points], + }, + hovertemplate="Request %{x}: %{y}", + ) + ) + _style_figure(figure, "Request Index", "Status") + figure.update_yaxes(tickvals=[0, 1], ticktext=["fail", "success"], + range=[-0.2, 1.2]) + charts.append({"title_zh": "请求状态", "title_en": "Request Status", + "html": _figure_html(figure)}) + return charts + + +def _build_charts(rows: List[dict]) -> Dict[str, List[dict]]: + """容量曲线：延迟分析（Latency/TTFT/TPOT）与吞吐分析（RPS/Token/成功率）。""" + try: + import plotly.graph_objects as go + except Exception as exc: # plotly 缺失时降级为空图表 + logger.warning(f"plotly is unavailable, skip capacity charts: {exc}") + return {} + + charts: Dict[str, List[dict]] = {"latency": [], "throughput": []} + groups = _sorted_groups(rows) + multi_group = len(groups) > 1 + + def group_prefix(row: dict) -> str: + if not multi_group: + return "" + return (f"{row.get('model_abbr')}/{row.get('dataset_abbr')} " + f"(size={row.get('request_size')}, prefix={row.get('prefix_len')}) ") + + def add(category: str, title_zh: str, title_en: str, + traces: List[dict], y_title: str) -> None: + traces = [trace for trace in traces + if any(point[1] is not None for point in trace["points"])] + if not traces: + return + figure = go.Figure() + for trace in traces: + points = [point for point in trace["points"] if point[1] is not None] + figure.add_trace( + go.Scatter( + x=[point[0] for point in points], + y=[point[1] for point in points], + mode="lines+markers", + name=trace["name"], + line={ + "color": trace["color"], + "width": 2, + "dash": trace.get("dash"), + }, + marker={"size": 7}, + fill="tozeroy" if trace.get("fill") else None, + fillcolor=trace.get("fillcolor"), + ) + ) + _style_figure(figure, "Concurrency (batch_size)", y_title) + # 图内不再重复画标题：Tab 名称与纵坐标标题已表达 + charts[category].append({"title_zh": title_zh, + "title_en": title_en, + "html": _figure_html(figure)}) + + latency_pairs = [ + ("Avg Latency", "e2el_avg_s", "P99 Latency", "e2el_p99_s", + "时延", "Latency", "Latency (s)"), + ("Avg TTFT", "ttft_avg", "P99 TTFT", "ttft_p99", + "首Token时延", "TTFT", "TTFT (ms)"), + ("Avg TPOT", "tpot_avg", "P99 TPOT", "tpot_p99", + "每Token时延", "TPOT", "TPOT (ms)"), + ] + for avg_label, avg_key, p99_label, p99_key, title_zh, title_en, y_title in latency_pairs: + traces: List[dict] = [] + for _, group_rows in groups: + if avg_key != "e2el_avg_s" and not group_rows[0].get("ttft_valid"): + continue + prefix = group_prefix(group_rows[0]) + traces.append( + { + "name": f"{prefix}{avg_label}", + "color": "#63b3ed", + "points": [(row.get("batch_size"), row.get(avg_key)) + for row in group_rows], + } + ) + traces.append( + { + "name": f"{prefix}{p99_label}", + "color": "#a78bfa", + "dash": "dash", + "points": [(row.get("batch_size"), row.get(p99_key)) + for row in group_rows], + } + ) + add("latency", title_zh, title_en, traces, y_title) + + rps_traces: List[dict] = [] + for _, group_rows in groups: + rps_traces.append( + { + "name": f"{group_prefix(group_rows[0])}RPS", + "color": "#63b3ed", + "fill": True, + "fillcolor": "rgba(99,179,237,0.08)", + "points": [(row.get("batch_size"), row.get("request_throughput")) + for row in group_rows], + } + ) + add("throughput", "请求吞吐量", "Request Throughput", rps_traces, + "requests/s") + + token_traces: List[dict] = [] + for _, group_rows in groups: + prefix = group_prefix(group_rows[0]) + token_traces.append( + { + "name": f"{prefix}Output tok/s", + "color": "#34d399", + "fill": True, + "fillcolor": "rgba(52,211,153,0.08)", + "points": [(row.get("batch_size"), row.get("throughput_tok_s")) + for row in group_rows], + } + ) + token_traces.append( + { + "name": f"{prefix}Total tok/s", + "color": "#a78bfa", + "dash": "dash", + "points": [(row.get("batch_size"), + row.get("total_throughput_tok_s")) + for row in group_rows], + } + ) + add("throughput", "Token吞吐量", "Token Throughput", token_traces, + "tokens/s") + + success_traces: List[dict] = [] + for _, group_rows in groups: + success_traces.append( + { + "name": f"{group_prefix(group_rows[0])}Success Rate", + "color": "#fbbf24", + "fill": True, + "fillcolor": "rgba(251,191,36,0.08)", + "points": [(row.get("batch_size"), row.get("success_rate")) + for row in group_rows], + } + ) + add("throughput", "成功率", "Success Rate", success_traces, + "Success Rate (%)") + + return charts + + +def _resolve_db_data(perf_dir: Path, db_name: str): + """按 PerfViz 同规则定位并读取 numpy_store。""" + from ais_bench.benchmark.openicl.icl_inferencer.output_handler.db_utils import ( + init_db, + load_all_numpy_from_db, + ) + + for candidate in ( + perf_dir / "db_data" / db_name, + perf_dir / "tmp" / f"tmp_{db_name}", + ): + if not candidate.exists(): + continue + try: + conn = init_db(str(candidate)) + try: + return load_all_numpy_from_db(conn) + finally: + conn.close() + except sqlite3.Error as exc: + logger.warning(f"Failed to read perf db {candidate}: {exc}") + return {} + + +def _load_detail_rows(perf_dir: Path, dataset_abbr: str) -> List[dict]: + """读取逐请求明细并还原 time_points，用于逐组合的请求明细表。""" + path = perf_dir / f"{dataset_abbr}_details.jsonl" + if not path.exists(): + return [] + grouped = defaultdict(list) + try: + with open(path, "rb") as file: + for line in file: + line = line.strip() + if not line: + continue + try: + record = json.loads(line) + except ValueError: + continue + db_name = record.get("db_name") + if db_name: + grouped[db_name].append(record) + except OSError as exc: + logger.warning(f"Failed to load detail rows {path}: {exc}") + return [] + + rows: List[dict] = [] + for db_name, records in grouped.items(): + numpy_data = _resolve_db_data(perf_dir, db_name) + for record in records: + input_tokens = record.get("input_tokens") + if not record.get("success", True): + rows.append({"success": False, "latency": None, "ttft": None, + "tpot": None, "itl": None, "input_len": input_tokens, + "output_len": None}) + continue + time_points = record.get("time_points") + if isinstance(time_points, dict) and "__db_ref__" in time_points: + time_points = numpy_data.get(time_points["__db_ref__"]) + if time_points is None: + rows.append({"success": False, "latency": None, "ttft": None, + "tpot": None, "itl": None, "input_len": input_tokens, + "output_len": None}) + continue + time_points = np.asarray(time_points, dtype=float).ravel() + if time_points.size < 2: + rows.append({"success": False, "latency": None, "ttft": None, + "tpot": None, "itl": None, "input_len": input_tokens, + "output_len": None}) + continue + latency = float(time_points[-1] - time_points[0]) + ttft = float(time_points[1] - time_points[0]) + output_tokens = record.get("output_tokens") + tpot = None + if isinstance(output_tokens, int) and output_tokens > 1: + tpot = (latency - ttft) / (output_tokens - 1) * 1000 + # 平均 ITL：首个 token 之后相邻 token 的间隔均值（与 EvalScope 的 avg inter-token latency 同口径） + itl = None + gaps = np.diff(time_points[1:]) + if gaps.size: + itl = float(gaps.mean()) * 1000 + rows.append( + { + "success": True, + "latency": latency * 1000, + "ttft": ttft * 1000, + "tpot": tpot, + "itl": itl, + "input_len": input_tokens, + "output_len": output_tokens, + } + ) + return rows + + +def _load_template(): + """惰性导入 jinja2 并加载模板（依赖缺失返回 None）。""" + try: + from jinja2 import Environment, FileSystemLoader, select_autoescape + except Exception as exc: + logger.warning(f"jinja2 is unavailable, skip HTML report: {exc}") + return None + template_dir = Path(__file__).resolve().parent / "templates" + if not (template_dir / TEMPLATE_NAME).exists(): + logger.warning(f"Report template not found: {template_dir / TEMPLATE_NAME}") + return None + env = Environment( + loader=FileSystemLoader(str(template_dir)), + autoescape=select_autoescape(["html", "htm", "j2"]), + trim_blocks=True, + lstrip_blocks=True, + ) + return env.get_template(TEMPLATE_NAME) + + +def _best_configs(rows: List[dict]) -> dict: + """最佳配置：最高吞吐与最低平均 TTFT 两个口径分别标注。""" + ok_rows = [row for row in rows if row.get("status") == "success"] + best_throughput = None + for row in ok_rows: + value = row.get("throughput_tok_s") + if value is None: + continue + if best_throughput is None or value > best_throughput["throughput_tok_s"]: + best_throughput = row + best_ttft = None + for row in ok_rows: + if not row.get("ttft_valid"): + continue + value = row.get("ttft_avg") + if value is None: + continue + if best_ttft is None or value < best_ttft["ttft_avg"]: + best_ttft = row + return {"throughput": best_throughput, "ttft": best_ttft} + + +def _saturation_note(rows: List[dict]) -> Tuple[str, str]: + """按吞吐-并发曲线给出 未饱和/接近饱和/已过载 的粗判（中英双语）。""" + ok_rows = [row for row in rows + if row.get("status") == "success" and row.get("throughput_tok_s") is not None] + if len(ok_rows) < 2: + return ("可用任务不足 2 个，无法判定吞吐-并发曲线形态。", + "Fewer than 2 usable tasks; cannot judge the throughput-concurrency curve.") + ok_rows.sort(key=lambda item: item.get("batch_size") or 0) + last, prev = ok_rows[-1], ok_rows[-2] + last_throughput = last.get("throughput_tok_s") or 0.0 + prev_throughput = prev.get("throughput_tok_s") or 0.0 + last_success = last.get("success_rate") + if last_success is not None and last_success < 99: + return ("已过载：最大并发档成功率下降（{}%），吞吐指标不可直接比较。".format( + _format_value(last_success)), + "Overloaded: success rate drops at the largest batch size ({}%); " + "throughput metrics are not directly comparable.".format( + _format_value(last_success))) + if prev_throughput > 0 and last_throughput <= prev_throughput: + return ("已饱和：最大并发档吞吐不再随并发上升，继续加大 batch_size 收益有限。", + "Saturated: throughput no longer grows with concurrency; " + "further increasing batch_size yields limited gain.") + if prev_throughput > 0 and (last_throughput - prev_throughput) / prev_throughput < 0.05: + return ("接近饱和：最大并发档吞吐增幅不足 5%，已接近服务上限。", + "Near saturation: throughput gain at the largest batch size is under 5%; " + "the service limit is close.") + return ("未饱和：吞吐仍随并发上升，可继续增大 batch_size 观察拐点。", + "Not saturated: throughput still grows with concurrency; " + "keep increasing batch_size to locate the knee.") + + +def _display(value) -> str: + text = _format_value(value) + return text if text else "—" + + +def _run_summary(row: dict) -> List[Tuple[str, str]]: + """逐组合 Run Summary 的键值对（对齐 EvalScope 的 stat-grid）。""" + fields = [ + ("Total Requests", "total_requests"), + ("Success Requests", "completed_requests"), + ("Failed Requests", "failed_requests"), + ("Concurrency", "concurrency"), + ("Max Concurrency", "max_concurrency"), + ("Time Taken (s)", "duration_s"), + ("Request Throughput (req/s)", "request_throughput"), + ("Avg Latency (s)", "e2el_avg_s"), + ("Output Tok Throughput (tok/s)", "throughput_tok_s"), + ("Total Tok Throughput (tok/s)", "total_throughput_tok_s"), + ("Avg TTFT (ms)", "ttft_avg"), + ("Avg TPOT (ms)", "tpot_avg"), + ("Avg ITL (ms)", "itl_avg"), + ("Avg Input Tokens", "avg_input_tokens"), + ("Avg Output Tokens", "avg_output_tokens"), + ("Success Rate", "success_rate_text"), + ] + return [(label, _display(row.get(key))) for label, key in fields] + + +def _basic_info(rows: List[dict], first_combo: dict) -> List[Tuple[str, str, str]]: + """报告头部的 Basic Information（manifest + 汇总合计），返回 (中文, English, 值)。""" + model_cfg = first_combo.get("model_config") or {} + ok_rows = [row for row in rows if row.get("status") == "success"] + + def total(key: str): + values = [row.get(key) for row in ok_rows if row.get(key) is not None] + return round(sum(values), 4) if values else None + + fields = [ + ("模型", "Model", + model_cfg.get("model_name")), + ("模型任务", "Model Task", + model_cfg.get("abbr") or first_combo.get("model_abbr")), + ("数据集", "Dataset", first_combo.get("dataset_abbr")), + ("总请求数", "Total Requests", total("total_requests")), + ("成功请求数", "Success Requests", total("completed_requests")), + ("总测试时长 (s)", "Total Test Time (s)", total("duration_s")), + ("输出 Token 总数", "Total Output Tokens", total("total_output_tokens")), + ] + return [(zh, en, _display(value)) for zh, en, value in fields] + + +def _recommendations(rows: List[dict], best: dict) -> List[Tuple[str, str]]: + """性能建议：饱和判定 + 最佳档位 + 延迟增长 + 失败组合提示（中英双语）。""" + notes = [_saturation_note(rows)] + best_throughput = best.get("throughput") + if best_throughput: + batch_size = _display(best_throughput.get("batch_size")) + tok_s = _display(best_throughput.get("throughput_tok_s")) + rps = _display(best_throughput.get("request_throughput")) + success = _display(best_throughput.get("success_rate_text")) + notes.append( + ("建议并发档：batch_size={}，输出吞吐 {} token/s，RPS {}，成功率 {}。".format( + batch_size, tok_s, rps, success), + "Recommended concurrency: batch_size={}, output throughput {} token/s, " + "RPS {}, success rate {}.".format(batch_size, tok_s, rps, success)) + ) + latencies = sorted( + (row.get("batch_size") or 0, row.get("e2el_avg_s")) + for row in rows + if row.get("status") == "success" and row.get("e2el_avg_s") is not None + ) + if len(latencies) >= 2: + first_batch, first_latency = latencies[0] + last_batch, last_latency = latencies[-1] + if first_latency: + ratio = round(last_latency / first_latency, 2) + notes.append( + ("端到端平均时延从 batch_size={} 的 {} s 增长到 batch_size={} 的 {} s（{}×）。".format( + first_batch, _display(first_latency), last_batch, + _display(last_latency), ratio), + "Average end-to-end latency grows from {} s at batch_size={} " + "to {} s at batch_size={} ({}x).".format( + _display(first_latency), first_batch, + _display(last_latency), last_batch, ratio)) + ) + failed = [row for row in rows if row.get("status") == "fail"] + if failed: + notes.append( + ("{} 个任务未产出结果（Task Status=fail），不参与曲线与最佳配置比较。".format( + len(failed)), + "{} task(s) produced no result (Task Status=fail) and are excluded " + "from curves and best-configuration comparison.".format(len(failed))) + ) + return notes + + +def render_batch_perf_report(rows: List[dict], manifest: dict, + warnings: List[str], out_dir: Path) -> Optional[Path]: + """以与 CSV 同源的 rows 渲染 `batch_perf_report.html`；异常只告警。""" + template = _load_template() + if template is None: + return None + try: + charts = _build_charts(rows) + report_rows = [] + for row in rows: + formatted = _format_row(row) + report_rows.append( + { + "status": row.get("status"), + "cells": [formatted.get(key) or "—" + for _, key in OVERVIEW_COLUMNS], + } + ) + details = [] + for row in rows: + if row.get("status") != "success": + continue + perf_dir = out_dir / "performances" / str(row.get("model_abbr")) + detail_rows = _load_detail_rows(perf_dir, str(row.get("dataset_abbr"))) + metric_headers, metric_rows = _metric_rows_for_row(row, out_dir) + details.append( + { + "label": "BatchSize {} · NumPrompts {}".format( + _display(row.get("batch_size")), + _display(row.get("num_prompts") + if row.get("num_prompts") is not None + else row.get("total_requests")), + ), + "badge_zh": "{} 个请求".format( + _display(row.get("total_requests"))), + "badge_en": "{} requests".format( + _display(row.get("total_requests"))), + "stats": _run_summary(row), + "metric_headers": metric_headers, + "metric_table": metric_rows, + "metric_charts": _build_metric_charts(metric_rows), + "request_charts": _build_request_charts(detail_rows), + "request_total": len(detail_rows), + } + ) + + combos = manifest.get("combos") or [] + first_combo = combos[0] if combos else {} + first_model_cfg = first_combo.get("model_config") or {} + best = { + key: (_format_row(value) if value else None) + for key, value in _best_configs(rows).items() + } + context = { + "marker": REPORT_MARKER, + "plotly_cdn": PLOTLY_CDN_URL, + "generated_at": datetime.now().isoformat(timespec="seconds"), + "ais_bench_version": __version__, + "schema_version": manifest.get("schema_version"), + "combo_count": len(combos), + "model_label": _display( + first_model_cfg.get("model_name") + or first_model_cfg.get("abbr") + or first_combo.get("model_abbr")), + "dataset_label": _display(first_combo.get("dataset_abbr")), + "basic_info": _basic_info(rows, first_combo), + "overview_headers": OVERVIEW_HEADERS, + "report_rows": report_rows, + "charts": charts, + "best": best, + "saturation": _saturation_note(rows)[0], + "recommendations": _recommendations(rows, best), + "details": details, + "warnings": warnings or [], + } + html = template.render(**context) + if not html.startswith(""): + html = "\n" + html + path = out_dir / REPORT_HTML_NAME + tmp_path = path.with_name(f"{path.name}.tmp{os.getpid()}") + try: + with open(tmp_path, "w", encoding="utf-8") as file: + file.write(html) + file.flush() + os.fsync(file.fileno()) + os.replace(tmp_path, path) + except OSError: + try: + if tmp_path.exists(): + os.remove(tmp_path) + except OSError: + pass + raise + return path + except Exception as exc: + logger.warning(f"Failed to render batch perf report: {exc}") + return None + + +def aggregate_batch_perf(work_dir, manifest: Optional[dict] = None, + warnings: Optional[List[str]] = None) -> BatchPerfArtifacts: + """聚合 manifest 内各组合的性能结果，产出 CSV 与 HTML 报告。 + + manifest 不存在或组合数 ≤ 1 时跳过（单组合零开销）；个别组合结果缺失时该行 + 标记 fail，不阻塞其余行；HTML 渲染失败只告警，CSV 正常返回。 + """ + work_path = Path(work_dir) + manifest = manifest if manifest is not None else _load_manifest(work_path) + if not manifest: + logger.debug("No sweep manifest found, skip batch perf aggregation.") + return BatchPerfArtifacts() + combos = manifest.get("combos") or [] + if len(combos) <= 1: + logger.debug("Single combo run, skip batch perf aggregation.") + return BatchPerfArtifacts() + + rows = [_build_row(work_path, combo) for combo in combos] + # 行序按 batch_size 升序，"并发-吞吐曲线"可直接从表读出 + rows.sort(key=lambda row: (row.get("batch_size") or 0)) + failed = [row.get("model_abbr") for row in rows if row.get("status") == "fail"] + if failed: + logger.warning( + "Batch perf results missing for combo(s): %s (marked as fail rows).", + ", ".join(str(item) for item in failed), + ) + + csv_path = work_path / SUMMARY_CSV_NAME + try: + _write_summary_csv(csv_path, rows, work_path) + except OSError as exc: + logger.warning(f"Failed to write {csv_path}: {exc}") + return BatchPerfArtifacts() + logger.info(f"Batch perf summary written to {csv_path}") + + html_path = render_batch_perf_report(rows, manifest, list(warnings or []), work_path) + if html_path is not None: + logger.info(f"Batch perf report written to {html_path}") + return BatchPerfArtifacts(csv_path=csv_path, html_path=html_path) diff --git a/ais_bench/benchmark/summarizers/templates/batch_perf_report.html.j2 b/ais_bench/benchmark/summarizers/templates/batch_perf_report.html.j2 new file mode 100644 index 00000000..67f10c35 --- /dev/null +++ b/ais_bench/benchmark/summarizers/templates/batch_perf_report.html.j2 @@ -0,0 +1,706 @@ + +{{ marker | safe }} +{% macro bi(zh, en) -%} +{{ zh }} +{%- endmacro %} +{% macro chart_tabs(charts, prefix) -%} +
+
+ {% for chart in charts %} + + {% endfor %} +
+ {% for chart in charts %} +
+
{{ chart.html | safe }}
+
+ {% endfor %} +
+{%- endmacro %} + + + + +AISBench 性能测试报告 + + + + + + +
+ +
+
+ {{ bi('AISBench · 批量压测', 'AISBench · Batch Sweep') }} + + + + +
+

{{ bi('AISBench 性能测试报告', 'AISBench Performance Test Report') }}

+
+
+ {{ bi('模型', 'Model') }} + {{ model_label }} +
+
+ {{ bi('数据集', 'Dataset') }} + {{ dataset_label }} +
+
+ {{ bi('任务数', 'Tasks') }} + {{ combo_count }} +
+
+ {{ bi('生成时间', 'Generated at') }} + {{ generated_at }} +
+
+ {{ bi('版本', 'ais_bench') }} + {{ ais_bench_version }} +
+
+
+ + + + +
+
+ + + + + + + + +

{{ bi('总览', 'Overview') }}

+
+ +
+
{{ bi('基础信息', 'Basic Information') }}
+
+ + + {% for zh, en, value in basic_info %} + + {% endfor %} + +
{{ bi(zh, en) }}{{ value }}
+
+
+ +
+
+ {{ bi('详细性能指标', 'Detailed Performance Metrics') }} + {{ combo_count }} {{ bi('个任务', 'tasks') }} +
+
+ + + {% for header in overview_headers %}{% endfor %} + + + {% for row in report_rows %} + + {% for cell in row.cells %} + {% if loop.last %} + + {% else %} + + {% endif %} + {% endfor %} + + {% endfor %} + +
{{ header }}
{{ cell }}{{ cell }}
+
+
+ +
+
{{ bi('最佳性能配置', 'Best Performance Configuration') }}
+
+ + + + + + + + + + + +
{{ bi('最高输出吞吐', 'Highest output throughput') }} + {% if best.throughput %} + {{ best.throughput.throughput_tok_s }} token/s · Batch Size {{ best.throughput.batch_size }} + · RPS {{ best.throughput.request_throughput }} · {{ bi('成功率', 'Success rate') }} {{ best.throughput.success_rate_text }} + {% else %}—{% endif %} +
{{ bi('最低平均 TTFT', 'Lowest average TTFT') }} + {% if best.ttft %} + {{ best.ttft.ttft_avg }} ms · Batch Size {{ best.ttft.batch_size }} + {% else %}—{% endif %} +
+
+
+ +
+
{{ bi('性能建议', 'Performance Recommendations') }}
+
+
    + {% for zh, en in recommendations %}
  • {{ bi(zh, en) }}
    • {% endfor %} +
+
+
+
+ + +
+
+ + + + + + +

{{ bi('时延分析', 'Latency Analysis') }}

+
+ {% if charts.latency %} +
+
+
+ {% for chart in charts.latency %} + + {% endfor %} +
+ {% for chart in charts.latency %} +
+
{{ chart.html | safe }}
+
+ {% endfor %} +
+
+ {% else %} +
{{ bi('暂无时延曲线（plotly 缺失或无非流式可用结果）。', 'No latency curves available (plotly missing or no usable streaming results).') }}
+ {% endif %} +
+ + +
+
+ + + + + +

{{ bi('吞吐分析', 'Throughput Analysis') }}

+
+ {% if charts.throughput %} +
+
+
+ {% for chart in charts.throughput %} + + {% endfor %} +
+ {% for chart in charts.throughput %} +
+
{{ chart.html | safe }}
+
+ {% endfor %} +
+
+ {% else %} +
{{ bi('暂无吞吐曲线（plotly 缺失或无可用结果）。', 'No throughput curves available (plotly missing or no usable results).') }}
+ {% endif %} +
+ + +
+
+ + + + + + +

{{ bi('逐任务详情', 'Task Details') }}

+
+ {% if details %} + {% for detail in details %} +
+ + + + + + + + {{ detail.label }} + + {{ bi(detail.badge_zh, detail.badge_en) }} + +
+
+
{{ bi('运行摘要', 'Run Summary') }}
+
+ {% for key, value in detail.stats %} +
+ {{ key }} + {{ value }} +
+ {% endfor %} +
+
+ +
+
{{ bi('指标统计', 'Metric Statistics') }}
+ {% if detail.metric_table %} +
+ + + {% for col in detail.metric_headers %}{% endfor %} + + + {% for row in detail.metric_table %} + {% for val in row %}{% endfor %} + {% endfor %} + +
{{ col }}
{{ val if val != '' else '—' }}
+
+ {% else %} +
{{ bi('未找到该任务的指标统计（synthetic.csv 缺失或不可解析）。', 'No metric statistics found for this task (synthetic.csv missing or unparsable).') }}
+ {% endif %} +
+ +
+
{{ bi('百分位曲线', 'Percentile Curves') }}
+ {% if detail.metric_charts %} + {{ chart_tabs(detail.metric_charts, 'metric-run-' ~ loop.index0) }} + {% else %} +
{{ bi('未找到该任务的百分位曲线（perf CSV 缺失或指标不可用）。', 'No percentile curves found for this task (perf CSV missing or metrics unavailable).') }}
+ {% endif %} +
+ +
+
+ {{ bi('请求明细（共 %d 条）' % detail.request_total, + 'Request Details (%d requests)' % detail.request_total) }} +
+ {% if detail.request_charts %} + {{ chart_tabs(detail.request_charts, 'request-run-' ~ loop.index0) }} + {% else %} +
{{ bi('未找到该任务的请求明细（details jsonl 缺失或不可解析）。', 'No request details found for this task (details jsonl missing or unparsable).') }}
+ {% endif %} +
+
+
+ {% endfor %} + {% else %} +
{{ bi('无可用结果。', 'No usable results.') }}
+ {% endif %} +
+ + +
+
+ + + + + + + +

{{ bi('口径与说明', 'Notes') }}

+
+
+
+

{{ bi('指标定义', 'Metric Definitions') }}

+
    +
  • {{ bi('Gen. tok/s / Output Tok Throughput：输出 token 吞吐，来源 performances 聚合结果。', + 'Gen. tok/s / Output Tok Throughput: output token throughput, taken from the aggregated performances results.') }}
    • +
  • {{ bi('RPS：请求吞吐（req/s）；Total Tok Throughput = 输入 + 输出 token 吞吐。', + 'RPS: request throughput (req/s); Total Tok Throughput = input + output token throughput.') }}
    • +
  • {{ bi('Avg/P99 Lat.(s)：端到端时延 E2EL 的平均值与 P99，单位秒。', + 'Avg/P99 Lat.(s): mean and P99 of end-to-end latency (E2EL), in seconds.') }}
    • +
  • {{ bi('TTFT/TPOT/ITL 单位 ms；仅 stream=true 且 ttft_valid=true 的任务有意义，否则置空（不填 0）。', + 'TTFT/TPOT/ITL are in ms; only meaningful when stream=true and ttft_valid=true, otherwise left empty (never filled with 0).') }}
    • +
  • {{ bi('Success Rate：成功请求数 / 总请求数；Batch Size 为客户端最大并发请求数（并发窗口），请求完成后立即补位，不存在固定波次。', + 'Success Rate: successful requests / total requests; Batch Size is the client-side maximum concurrent requests (concurrency window) with immediate backfilling, so there are no fixed waves.') }}
    • +
+ +

{{ bi('缺失说明', 'Missing Data') }}

+

{{ bi('Task Status=fail 的任务在汇总表灰显，且不进入任何曲线与最佳配置比较；全部为 fail 时仍产出本报告，以空态提示无可用结果。', + 'Tasks with Task Status=fail are greyed out in the summary table and excluded from all curves and best-configuration comparison; the report is still produced when every task fails, showing an empty state.') }}

+ +

{{ bi('运行期告警', 'Runtime Warnings') }}

+ {% if warnings %} +
    {% for warning in warnings %}
  • {{ warning }}
    • {% endfor %}
+ {% else %} +

{{ bi('无。', 'None.') }}

+ {% endif %} +
+
+
+ +
+ + + diff --git a/requirements/runtime.txt b/requirements/runtime.txt index 20136abc..9bb9c2c3 100644 --- a/requirements/runtime.txt +++ b/requirements/runtime.txt @@ -11,6 +11,7 @@ gradio-client httpx immutabledict importlib-metadata +jinja2 jieba jsonlines mmengine-lite diff --git a/tests/UT/cli/test_argument_parser.py b/tests/UT/cli/test_argument_parser.py index bc06afc8..e1d58276 100644 --- a/tests/UT/cli/test_argument_parser.py +++ b/tests/UT/cli/test_argument_parser.py @@ -118,7 +118,7 @@ def test_parse_args_base_options(self, mock_get_current_time_str): self.assertEqual(args.config_dir, 'custom_configs') self.assertEqual(args.max_num_workers, 4) self.assertEqual(args.max_workers_per_gpu, 2) - self.assertEqual(args.num_prompts, 10) + self.assertEqual(args.num_prompts, [10]) self.assertEqual(args.num_warmups, 3) @patch('ais_bench.benchmark.cli.argument_parser.get_current_time_str') @@ -315,7 +315,7 @@ def test_parse_args_api_model_options(self, mock_get_current_time_str): self.assertEqual(args.host_port, 8000) self.assertEqual(args.url, 'http://example.com/v1') self.assertEqual(args.max_out_len, 256) - self.assertEqual(args.batch_size, 4) + self.assertEqual(args.batch_size, [4]) self.assertTrue(args.trust_remote_code) self.assertEqual(args.generation_kwargs, {'temperature': 0.5, 'ignore_eos': False}) @@ -368,5 +368,62 @@ def test_init_method_creates_parser(self): self.assertTrue(hasattr(args, 'host_port')) # api_model_args self.assertTrue(hasattr(args, 'generation_kwargs')) # api_model_args self.assertTrue(hasattr(args, 'trust_remote_code')) # api_model_args + + @patch('ais_bench.benchmark.cli.argument_parser.get_current_time_str') + def test_parse_args_sweep_list_options(self, mock_get_current_time_str): + """--batch-size / --num-prompts 列表化解析""" + mock_get_current_time_str.return_value = "20230516_144254" + sys.argv = ['benchmark.py', '--batch-size', '1', '8', '16', + '--num-prompts', '1', '8', '16'] + args = ArgumentParser().parse_args() + self.assertEqual(args.batch_size, [1, 8, 16]) + self.assertEqual(args.num_prompts, [1, 8, 16]) + + @patch('ais_bench.benchmark.cli.argument_parser.get_current_time_str') + def test_parse_args_synthetic_dataset_parser_options(self, mock_get_current_time_str): + """随机合成数据集参数与模型通用覆盖参数解析""" + mock_get_current_time_str.return_value = "20230516_144254" + sys.argv = [ + 'benchmark.py', + '--request-count', '64', + '--tokenid-config-request-size', '8192', + '--tokenid-config-prefix-len', '6554', + '--string-config-input', '{"Method": "uniform", "Params": {"MinValue": 1}}', + '--string-config-output', '{"Method": "uniform", "Params": {"MinValue": 1}}', + '--model-abbr', 'qwen', + '--dataset-abbr', 'synthetic', + '--stream', + '--no-use-timestamp', + ] + args = ArgumentParser().parse_args() + self.assertEqual(args.request_count, 64) + self.assertEqual(args.tokenid_config_request_size, 8192) + self.assertEqual(args.tokenid_config_prefix_len, 6554) + self.assertEqual(args.string_config_input, + {'Method': 'uniform', 'Params': {'MinValue': 1}}) + self.assertEqual(args.model_abbr, 'qwen') + self.assertEqual(args.dataset_abbr, 'synthetic') + self.assertTrue(args.stream) + self.assertFalse(args.use_timestamp) + + @patch('ais_bench.benchmark.cli.argument_parser.get_current_time_str') + def test_parse_args_sweep_defaults_none(self, mock_get_current_time_str): + """新增覆盖参数未显式指定时默认均为 None（不覆盖配置）""" + mock_get_current_time_str.return_value = "20230516_144254" + sys.argv = ['benchmark.py'] + args = ArgumentParser().parse_args() + for attr in ('request_count', 'tokenid_config_request_size', + 'tokenid_config_prefix_len', 'string_config_input', + 'string_config_output', 'model_abbr', 'dataset_abbr', + 'stream', 'use_timestamp'): + self.assertIsNone(getattr(args, attr)) + + def test_parse_args_tokenid_prefix_len_rejects_fraction(self): + """--tokenid-config-prefix-len 不接受小数/比例写法""" + sys.argv = ['benchmark.py', '--tokenid-config-prefix-len', '0.8'] + with self.assertRaises(BaseException): + ArgumentParser().parse_args() + + if __name__ == '__main__': unittest.main() diff --git a/tests/UT/cli/test_config_manager.py b/tests/UT/cli/test_config_manager.py index 297f0f3a..1e5019ef 100644 --- a/tests/UT/cli/test_config_manager.py +++ b/tests/UT/cli/test_config_manager.py @@ -1,10 +1,19 @@ import os +import json import unittest from unittest import mock import tempfile import shutil -from ais_bench.benchmark.cli.config_manager import CustomConfigChecker, ConfigManager +from ais_bench.benchmark.cli.config_manager import ( + CustomConfigChecker, + ConfigManager, +) +from ais_bench.benchmark.cli.sweep import ( + expand_sweep_combinations, + validate_pairing, + _normalize_sweep_arg, +) from ais_bench.benchmark.models import VLLMCustomAPI, VLLMCustomAPIChat from ais_bench.benchmark.models import TritonCustomAPI, MindieStreamApi, TGICustomAPI from ais_bench.benchmark.utils.logging.exceptions import CommandError, AISBenchConfigError @@ -1432,5 +1441,375 @@ def test_response_anomaly_rejects_nonexistent_msprobe_paths(self): self.assertIn('/workspace/missing/mtype_config.json', message) self.assertIn('/workspace/missing/token2category', message) +class TestSweepPairing(unittest.TestCase): + """--batch-size ↔ --num-prompts 配对/广播与并发窗口校验。""" + + def test_normalize_sweep_arg(self): + self.assertIsNone(_normalize_sweep_arg(None)) + self.assertEqual(_normalize_sweep_arg([8]), 8) + self.assertEqual(_normalize_sweep_arg([8, 16]), [8, 16]) + self.assertEqual(_normalize_sweep_arg("8"), 8) + + def test_normalize_empty_list_raises(self): + with self.assertRaises(AISBenchConfigError): + _normalize_sweep_arg([]) + + def test_paired_strategy(self): + pairs, warnings = validate_pairing([1, 8], [1, 8]) + self.assertEqual(pairs, [(1, 1), (8, 8)]) + self.assertEqual(warnings, []) + + def test_broadcast_strategy(self): + pairs, _ = validate_pairing([1, 8, 16], 16) + self.assertEqual(pairs, [(1, 16), (8, 16), (16, 16)]) + pairs, _ = validate_pairing(16, [1, 8, 16]) + self.assertEqual(pairs, [(16, 1), (16, 8), (16, 16)]) + + def test_unequal_list_raises(self): + with self.assertRaises(AISBenchConfigError) as cm: + validate_pairing([1, 8], [1, 8, 16]) + self.assertEqual( + cm.exception.error_code_str, + TMAN_CODES.INVALID_ARG_VALUE_IN_CMD.full_code, + ) + + def test_window_not_saturated_warning(self): + """batch_size 大于生效请求数时，并发窗口跑不满 → WARNING""" + _, warnings = validate_pairing([16], 8, request_count=8) + self.assertTrue(any("concurrency window" in warning for warning in warnings)) + + def test_scalar_pairing_warns_window_not_filled(self): + """双标量（单档归一后的形态）同样要做并发窗口检查，不能跳过""" + pairs, warnings = validate_pairing(64, 16, request_count=64) + self.assertEqual(pairs, [(64, 16)]) + self.assertTrue(any("concurrency window" in warning for warning in warnings)) + + def test_scalar_pairing_window_saturated_no_warning(self): + """双标量且窗口能被填满时不告警""" + _, warnings = validate_pairing(8, 16, request_count=16) + self.assertEqual(warnings, []) + + def test_window_saturated_no_warning(self): + """并发窗口能被请求数填满时不告警（含 num_prompts 大于 batch_size 的常见组合）""" + _, warnings = validate_pairing([8], 10, request_count=10) + self.assertEqual(warnings, []) + + def test_request_count_limits_effective_requests(self): + """生效请求数取 min(num_prompts, request_count)""" + _, warnings = validate_pairing([8], 16, request_count=4) + self.assertTrue(any("concurrency window" in warning for warning in warnings)) + + def test_test_range_set_skips_window_check(self): + pairs, warnings = validate_pairing([16], 8, test_range_set=True) + self.assertEqual(pairs, [(16, 8)]) + self.assertEqual(warnings, []) + + def test_missing_num_prompts_falls_back_to_full_dataset(self): + """未配置 --num-prompts 时默认全量评测，按数据集全部样本判断并发窗口""" + pairs, warnings = validate_pairing([64], None, request_count=16) + self.assertEqual(pairs, [(64, None)]) + self.assertTrue(any("dataset samples" in warning for warning in warnings)) + + def test_missing_num_prompts_and_request_count_skips_check(self): + """两者都未配置时无法确定请求数，不告警""" + pairs, warnings = validate_pairing([64], None) + self.assertEqual(pairs, [(64, None)]) + self.assertEqual(warnings, []) + + +class TestExpandSweepCombinations(unittest.TestCase): + """expand_sweep_combinations 的组合展开、校验与 manifest。""" + + SYNTHETIC_TYPE = "ais_bench.benchmark.datasets.synthetic.SyntheticDataset" + + def setUp(self): + self.model_dir = tempfile.mkdtemp() + self.addCleanup(shutil.rmtree, self.model_dir) + + def _model(self, abbr="qwen", stream=True, **overrides): + model = { + "type": "ais_bench.benchmark.models.VLLMCustomAPIChat", + "abbr": abbr, + "attr": "service", + "path": self.model_dir, + "stream": stream, + "batch_size": 1, + "max_out_len": 128, + "generation_kwargs": {}, + } + model.update(overrides) + return model + + def _dataset(self, abbr="synthetic", request_count=64, test_range=None, + config_type="tokenid", request_size=1024, prefix_len=0): + reader_cfg = {} + if test_range is not None: + reader_cfg["test_range"] = test_range + return { + "type": self.SYNTHETIC_TYPE, + "abbr": abbr, + "reader_cfg": reader_cfg, + "infer_cfg": {"retriever": {}}, + "config": { + "Type": config_type, + "RequestCount": request_count, + "TokenIdConfig": { + "RequestSize": request_size, + "PrefixLen": prefix_len, + }, + }, + } + + def _args(self, **overrides): + args = { + "mode": "perf", + "batch_size": None, + "num_prompts": None, + "request_count": None, + "tokenid_config_request_size": None, + "tokenid_config_prefix_len": None, + "string_config_input": None, + "string_config_output": None, + "model_abbr": None, + "dataset_abbr": None, + "stream": None, + "use_timestamp": None, + "generation_kwargs": None, + "max_num_workers": 1, + "num_warmups": 1, + "max_out_len": None, + } + args.update(overrides) + return args + + def test_multi_combinations(self): + result = expand_sweep_combinations( + [self._model()], [self._dataset()], + self._args(batch_size=[1, 8], num_prompts=[1, 8]), + ) + + self.assertEqual(len(result.models), 2) + self.assertEqual([m["abbr"] for m in result.models], ["qwen-bs1", "qwen-bs8"]) + self.assertEqual([m["batch_size"] for m in result.models], [1, 8]) + self.assertEqual( + [d["reader_cfg"]["test_range"] for d in result.datasets], + ["[:1]", "[:8]"], + ) + self.assertEqual(len(result.combinations), 2) + for combo in result.combinations: + self.assertIn(combo["models"][0], result.models) + self.assertIn(combo["datasets"][0], result.datasets) + self.assertEqual(result.sweep_cooldown_s, 5) + self.assertIsNotNone(result.manifest) + manifest_batch_sizes = [ + combo["batch_size"] for combo in result.manifest["combos"] + ] + self.assertEqual(manifest_batch_sizes, [1, 8]) + + def test_missing_num_prompts_uses_full_dataset(self): + """未指定 --num-prompts 时按数据集全部样本请求，且不写 test_range""" + result = expand_sweep_combinations( + [self._model()], [self._dataset(request_count=32)], + self._args(batch_size=[1, 8]), + ) + + self.assertEqual(len(result.combinations), 2) + self.assertEqual( + [combo["num_prompts"] for combo in result.manifest["combos"]], + [32, 32], + ) + self.assertTrue( + all("test_range" not in d["reader_cfg"] for d in result.datasets) + ) + + def test_single_value_equivalent_path(self): + result = expand_sweep_combinations( + [self._model()], [self._dataset()], + self._args(num_prompts=10), + ) + + self.assertEqual(result.combinations, []) + self.assertIsNone(result.manifest) + self.assertEqual(result.sweep_cooldown_s, 0) + self.assertEqual(result.models[0]["abbr"], "qwen") + # 单值时 test_range 仍由 _fill_dataset_configs 写入 + self.assertNotIn("test_range", result.datasets[0]["reader_cfg"]) + + def test_existing_test_range_is_kept(self): + result = expand_sweep_combinations( + [self._model()], [self._dataset(test_range="[:5]")], + self._args(batch_size=[1, 8], num_prompts=[1, 8]), + ) + + for dataset in result.datasets: + self.assertEqual(dataset["reader_cfg"]["test_range"], "[:5]") + combos = result.manifest["combos"] + self.assertTrue(all(combo["num_prompts"] is None for combo in combos)) + + def test_abbr_suffix_is_idempotent(self): + result = expand_sweep_combinations( + [self._model(abbr="qwen-bs8")], [self._dataset()], + self._args(batch_size=[8, 16], num_prompts=[8, 16]), + ) + self.assertEqual([m["abbr"] for m in result.models], ["qwen-bs8", "qwen-bs8-bs16"]) + + def test_multi_model_combinations_do_not_falsely_conflict(self): + """多模型 × 多并发档下同一数据集 abbr 会重复出现，不应误判为冲突""" + result = expand_sweep_combinations( + [self._model(abbr="qwen"), self._model(abbr="qwen2")], + [self._dataset()], + self._args(batch_size=[1, 8], num_prompts=[1, 8]), + ) + + self.assertEqual(len(result.models), 4) + self.assertEqual(len(result.combinations), 4) + self.assertEqual( + [m["abbr"] for m in result.models], + ["qwen-bs1", "qwen2-bs1", "qwen-bs8", "qwen2-bs8"], + ) + output_pairs = { + (m["abbr"], d["abbr"]) + for m, d in zip(result.models, result.datasets) + } + self.assertEqual(len(output_pairs), 4) + + def test_manifest_is_desensitized(self): + manifest = expand_sweep_combinations( + [self._model(api_key="sk-secret")], [self._dataset()], + self._args(batch_size=[1, 8], num_prompts=[1, 8]), + ).manifest + + dumped = json.dumps(manifest, ensure_ascii=False) + self.assertNotIn("sk-secret", dumped) + self.assertNotIn(self.model_dir, dumped) + self.assertEqual(manifest["schema_version"], 1) + self.assertIn("ais_bench_version", manifest) + self.assertEqual(len(manifest["combos"]), 2) + + def test_manifest_records_model_name(self): + """manifest 记录服务端模型名（`--model-name` 覆盖后的 `model` 字段），供 HTML 报告展示""" + manifest = expand_sweep_combinations( + [self._model(model="Qwen3.8-27B-w8a8")], [self._dataset()], + self._args(batch_size=[1, 8], num_prompts=[1, 8]), + ).manifest + + self.assertEqual( + [combo["model_config"]["model_name"] for combo in manifest["combos"]], + ["Qwen3.8-27B-w8a8", "Qwen3.8-27B-w8a8"], + ) + + def test_multiple_synthetic_datasets_rejected(self): + with self.assertRaises(AISBenchConfigError) as cm: + expand_sweep_combinations( + [self._model()], [self._dataset(), self._dataset(abbr="synthetic2")], + self._args(batch_size=[1, 8]), + ) + self.assertIn("Only one random synthetic dataset", str(cm.exception)) + + def test_invalid_batch_size_reported(self): + with self.assertRaises(AISBenchConfigError) as cm: + expand_sweep_combinations( + [self._model()], [self._dataset()], + self._args(batch_size=[1, 0]), + ) + self.assertIn("value #2", str(cm.exception)) + + def test_prefix_len_exceeds_request_size_rejected(self): + with self.assertRaises(AISBenchConfigError) as cm: + expand_sweep_combinations( + [self._model()], + [self._dataset(request_size=1024, prefix_len=2048)], + self._args(batch_size=[1, 8]), + ) + self.assertIn("PrefixLen", str(cm.exception)) + + def test_tokenid_arg_with_string_dataset_rejected(self): + dataset = self._dataset(config_type="string") + dataset["config"]["StringConfig"] = { + "Input": {"Method": "uniform", "Params": {"MinValue": 1, "MaxValue": 8}}, + "Output": {"Method": "uniform", "Params": {"MinValue": 1, "MaxValue": 8}}, + } + with self.assertRaises(AISBenchConfigError) as cm: + expand_sweep_combinations( + [self._model()], [dataset], + self._args(batch_size=[1, 8], tokenid_config_request_size=512), + ) + self.assertIn("tokenid", str(cm.exception)) + + def test_use_timestamp_with_synthetic_rejected(self): + with self.assertRaises(AISBenchConfigError) as cm: + expand_sweep_combinations( + [self._model()], [self._dataset()], + self._args(batch_size=[1, 8], use_timestamp=True), + ) + self.assertIn("use-timestamp", str(cm.exception)) + + def test_non_stream_warning_keeps_ttft_invalid(self): + result = expand_sweep_combinations( + [self._model(stream=False)], [self._dataset()], + self._args(batch_size=[1, 8], num_prompts=[1, 8]), + ) + for combo in result.manifest["combos"]: + self.assertFalse(combo["model_config"]["ttft_valid"]) + + def test_max_num_workers_parallel_warning(self): + with mock.patch( + "ais_bench.benchmark.cli.sweep.logger" + ) as mock_logger: + expand_sweep_combinations( + [self._model()], [self._dataset()], + self._args(batch_size=[1, 8], num_prompts=[1, 8], + max_num_workers=4), + ) + messages = [str(call) for call in mock_logger.warning.call_args_list] + self.assertTrue(any("max-num-workers" in message for message in messages)) + + +class TestExpandSweepConfigsWritesManifest(unittest.TestCase): + """_expand_sweep_configs 落盘 combos_manifest.json。""" + + def test_manifest_written(self): + temp_dir = tempfile.mkdtemp() + self.addCleanup(shutil.rmtree, temp_dir) + model_dir = tempfile.mkdtemp() + self.addCleanup(shutil.rmtree, model_dir) + + args = mock.MagicMock() + config_manager = ConfigManager(args) + config_manager.cfg = { + "work_dir": temp_dir, + "models": [{ + "type": "ais_bench.benchmark.models.VLLMCustomAPIChat", + "abbr": "qwen", + "attr": "service", + "path": model_dir, + "stream": True, + "batch_size": 1, + }], + "datasets": [{ + "type": TestExpandSweepCombinations.SYNTHETIC_TYPE, + "abbr": "synthetic", + "reader_cfg": {}, + "config": {"Type": "tokenid", "RequestCount": 64}, + }], + "cli_args": { + "mode": "perf", + "batch_size": [1, 8], + "num_prompts": [1, 8], + "max_num_workers": 1, + }, + } + + config_manager._expand_sweep_configs() + + manifest_path = os.path.join(temp_dir, "combos_manifest.json") + self.assertTrue(os.path.exists(manifest_path)) + with open(manifest_path, "r", encoding="utf-8") as file: + manifest = json.load(file) + self.assertEqual(len(manifest["combos"]), 2) + self.assertEqual(len(config_manager.cfg["model_dataset_combinations"]), 2) + self.assertEqual(config_manager.cfg["sweep_cooldown_s"], 5) + + if __name__ == '__main__': unittest.main() diff --git a/tests/UT/cli/test_utils.py b/tests/UT/cli/test_utils.py index eeddf72b..e86fc972 100644 --- a/tests/UT/cli/test_utils.py +++ b/tests/UT/cli/test_utils.py @@ -10,6 +10,10 @@ validate_num_prompts, validate_num_warmups, validate_pressure_time, + validate_batch_size, + validate_request_count, + validate_request_size, + validate_prefix_len, MAX_NUM_WORKERS, DEFAULT_PRESSURE_TIME ) @@ -212,6 +216,20 @@ def test_fill_test_range_use_num_prompts_with_string_num_prompts(self, mock_logg self.assertEqual(dataset_cfg["reader_cfg"].get("test_range"), "[:10]") mock_logger.info.assert_called_once_with("Keeping the first 10 prompts for dataset [test_dataset]") + @patch('ais_bench.benchmark.cli.utils.logger') + def test_fill_test_range_use_num_prompts_with_list_skips(self, mock_logger): + """多组合 sweep 时 num_prompts 为列表，test_range 由展开器逐副本写入""" + for num_prompts in ([10, 32], [10], (10, 32)): + with self.subTest(num_prompts=num_prompts): + dataset_cfg = {"reader_cfg": {}, "abbr": "test_dataset"} + + fill_test_range_use_num_prompts(num_prompts, dataset_cfg) + + self.assertNotIn("test_range", dataset_cfg["reader_cfg"]) + mock_logger.info.assert_not_called() + mock_logger.warning.assert_not_called() + mock_logger.reset_mock() + class TestValidateMaxWorkers(unittest.TestCase): """测试 validate_max_workers 函数""" @@ -381,5 +399,50 @@ def test_validate_pressure_time_negative(self): self.assertIn("must be >=", str(context.exception)) +class TestValidateSweepScalarArgs(unittest.TestCase): + """测试 sweep 新增单值参数的校验器""" + + def test_validate_batch_size_range(self): + self.assertEqual(validate_batch_size("1"), 1) + self.assertEqual(validate_batch_size("100000"), 100000) + with self.assertRaises(AISBenchConfigError) as context: + validate_batch_size("0") + self.assertIn(UTILS_CODES.ARGUMENT_TOO_SMALL.full_code, str(context.exception)) + with self.assertRaises(AISBenchConfigError) as context: + validate_batch_size("100001") + self.assertIn(UTILS_CODES.ARGUMENT_TOO_LARGE.full_code, str(context.exception)) + + def test_validate_request_count_range(self): + self.assertEqual(validate_request_count("1"), 1) + self.assertEqual(validate_request_count(str(2 ** 20)), 2 ** 20) + with self.assertRaises(AISBenchConfigError) as context: + validate_request_count("0") + self.assertIn(UTILS_CODES.ARGUMENT_TOO_SMALL.full_code, str(context.exception)) + with self.assertRaises(AISBenchConfigError) as context: + validate_request_count(str(2 ** 20 + 1)) + self.assertIn(UTILS_CODES.ARGUMENT_TOO_LARGE.full_code, str(context.exception)) + + def test_validate_request_size_range(self): + self.assertEqual(validate_request_size("1"), 1) + with self.assertRaises(AISBenchConfigError) as context: + validate_request_size("0") + self.assertIn(UTILS_CODES.ARGUMENT_TOO_SMALL.full_code, str(context.exception)) + with self.assertRaises(AISBenchConfigError) as context: + validate_request_size(str(2 ** 20 + 1)) + self.assertIn(UTILS_CODES.ARGUMENT_TOO_LARGE.full_code, str(context.exception)) + + def test_validate_prefix_len_absolute_tokens_only(self): + """prefix_len 接受 0 与正整数，拒绝小数/比例并提示需绝对 token 数""" + self.assertEqual(validate_prefix_len("0"), 0) + self.assertEqual(validate_prefix_len("512"), 512) + with self.assertRaises(AISBenchConfigError) as context: + validate_prefix_len("-1") + self.assertIn(UTILS_CODES.ARGUMENT_TOO_SMALL.full_code, str(context.exception)) + with self.assertRaises(AISBenchConfigError) as context: + validate_prefix_len("0.5") + self.assertIn(UTILS_CODES.INVALID_INTEGER_TYPE.full_code, str(context.exception)) + self.assertIn("absolute token count", str(context.exception)) + + if __name__ == '__main__': unittest.main() \ No newline at end of file diff --git a/tests/UT/summarizers/test_default_perf.py b/tests/UT/summarizers/test_default_perf.py index c8f67016..d6586a15 100644 --- a/tests/UT/summarizers/test_default_perf.py +++ b/tests/UT/summarizers/test_default_perf.py @@ -84,6 +84,28 @@ def test_get_dataset_abbr_multiple(self): result = summarizer._get_dataset_abbr([self.dataset_cfg, dataset_cfg2]) self.assertEqual(result, "testdataset") + def test_dataset_groups_dedupe_same_abbr(self): + """sweep 展开出的同 abbr 数据集副本不应重复计入分组（否则指标翻倍）""" + config = ConfigDict({ + "models": [self.model_cfg], + "datasets": [ + self.dataset_cfg, + { + "type": "TestDataset", + "abbr": "test_ds", + "infer_cfg": {"inferencer": "test_inferencer"}, + "reader_cfg": {"test_range": "[:8]"}, + }, + ], + "work_dir": "/tmp/test_work_dir", + "cli_args": {"merge_ds": False}, + }) + summarizer = DefaultPerfSummarizer(config, self.calculator_cfg) + groups = list(summarizer.dataset_groups.values()) + self.assertEqual(len(groups), 1) + self.assertEqual(len(groups[0]), 1) + self.assertEqual(summarizer._get_dataset_abbr(groups[0]), "test_ds") + @patch('ais_bench.benchmark.summarizers.default_perf.build_model_from_cfg') @patch('ais_bench.benchmark.summarizers.default_perf.init_db') @patch('ais_bench.benchmark.summarizers.default_perf.load_all_numpy_from_db') diff --git a/tests/UT/summarizers/test_perf_batch.py b/tests/UT/summarizers/test_perf_batch.py new file mode 100644 index 00000000..4f2b5f5f --- /dev/null +++ b/tests/UT/summarizers/test_perf_batch.py @@ -0,0 +1,358 @@ +import csv +import json +import unittest +from pathlib import Path +from tempfile import TemporaryDirectory +from unittest import mock + +from ais_bench.benchmark.summarizers import perf_batch +from ais_bench.benchmark.summarizers.perf_batch import ( + CSV_HEADERS, + REPORT_HTML_NAME, + REPORT_MARKER, + SUMMARY_CSV_NAME, + _escape_csv_text, + _format_value, + _parse_number, + _saturation_note, + aggregate_batch_perf, + render_batch_perf_report, +) + +SYNTHETIC_TYPE = "ais_bench.benchmark.datasets.synthetic.SyntheticDataset" + + +def _make_combo(index, batch_size, model_abbr, dataset_abbr="synthetic", + stream=True, ttft_valid=True, request_size=1024, + prefix_len=0, num_prompts=64): + return { + "index": index, + "model_abbr": model_abbr, + "dataset_abbr": dataset_abbr, + "batch_size": batch_size, + "num_prompts": num_prompts, + "request_count": 128, + "model_config": { + "abbr": model_abbr, + "stream": stream, + "ttft_valid": ttft_valid, + "use_timestamp": False, + "max_out_len": 512, + }, + "dataset_config": { + "Type": SYNTHETIC_TYPE, + "RequestSize": request_size, + "PrefixLen": prefix_len, + }, + } + + +def _make_manifest(combos): + return { + "schema_version": 1, + "ais_bench_version": "test", + "created_at": "2026-01-01T00:00:00", + "cli_args": {"batch_size": [c["batch_size"] for c in combos]}, + "combos": combos, + } + + +def _write_perf(work_dir, model_abbr, dataset_abbr, throughput=1000.0, + total=10, success=10, duration_ms=2000.0, + ttft_average=12.5, tpot_average=3.5, itl_p99=4.2): + perf_dir = Path(work_dir) / "performances" / model_abbr + perf_dir.mkdir(parents=True, exist_ok=True) + common = { + "Output Token Throughput": {"total": f"{throughput} token/s"}, + "Total Requests": {"total": str(total)}, + "Success Requests": {"total": str(success)}, + # 真实 synthetic.json 必含该指标，汇总区原样取值，缺失时留空 + "Failed Requests": {"total": str(total - success)}, + "Benchmark Duration": {"total": f"{duration_ms} ms"}, + } + with open(perf_dir / f"{dataset_abbr}.json", "w", encoding="utf-8") as file: + json.dump(common, file) + rows = [ + ["Performance Parameters", "Stage", "Average", "Min", "Max", + "Median", "P75", "P90", "P99"], + ["TTFT", "total", str(ttft_average), "1", "30", "10", "15", "20", "25"], + ["TPOT", "total", str(tpot_average), "1", "5", "3", "3.4", "4", "4.5"], + ["ITL", "total", "3.0", "1", "5", "3", "3.2", "3.8", str(itl_p99)], + ] + with open(perf_dir / f"{dataset_abbr}.csv", "w", encoding="utf-8", newline="") as file: + csv.writer(file).writerows(rows) + + +def _write_manifest(work_dir, combos): + manifest = _make_manifest(combos) + with open(Path(work_dir) / "combos_manifest.json", "w", encoding="utf-8") as file: + json.dump(manifest, file) + return manifest + + +def _read_sections(work_dir): + """按空行切分汇总 CSV：返回 (指标区, 汇总区)；汇总区自带独立表头。""" + with open(Path(work_dir) / SUMMARY_CSV_NAME, "r", encoding="utf-8", newline="") as file: + rows = list(csv.reader(file)) + header, body = rows[0], rows[1:] + if [] not in body: + return [dict(zip(header, row)) for row in body], [] + split_at = body.index([]) + run_header = body[split_at + 1] + return ([dict(zip(header, row)) for row in body[:split_at]], + [dict(zip(run_header, row)) for row in body[split_at + 2:]]) + + +def _read_summary(work_dir): + return _read_sections(work_dir)[0] + + +class TestAggregateBatchPerf(unittest.TestCase): + def setUp(self): + self._tmp = TemporaryDirectory() + self.work_dir = Path(self._tmp.name) + + def tearDown(self): + self._tmp.cleanup() + + def test_skip_without_manifest(self): + """无 manifest 时零开销跳过，不落盘任何产物""" + artifacts = aggregate_batch_perf(self.work_dir) + self.assertIsNone(artifacts.csv_path) + self.assertIsNone(artifacts.html_path) + self.assertFalse((self.work_dir / SUMMARY_CSV_NAME).exists()) + + def test_skip_with_single_combo(self): + """单组合不聚合（单值路径零开销）""" + _write_manifest(self.work_dir, [_make_combo(0, 4, "m-bs4")]) + artifacts = aggregate_batch_perf(self.work_dir) + self.assertIsNone(artifacts.csv_path) + self.assertFalse((self.work_dir / SUMMARY_CSV_NAME).exists()) + + def test_missing_combo_marked_and_others_kept(self): + """个别组合结果缺失时标记 fail，不阻塞其余行""" + _write_manifest( + self.work_dir, + [_make_combo(0, 4, "m-bs4"), _make_combo(1, 16, "m-bs16")], + ) + _write_perf(self.work_dir, "m-bs4", "synthetic") + + artifacts = aggregate_batch_perf(self.work_dir) + self.assertEqual(artifacts.csv_path, self.work_dir / SUMMARY_CSV_NAME) + rows = _read_summary(self.work_dir) + # 正常组合展开为 3 个指标行（身份列逐行填充），缺失组合只保留一行标识 + self.assertEqual(len(rows), 4) + self.assertEqual( + [row["Batch Size"] for row in rows], ["4", "4", "4", "16"] + ) + self.assertEqual(rows[0]["Task Status"], "success") + self.assertEqual(rows[-1]["Task Status"], "fail") + self.assertEqual(rows[-1]["Performance Parameters"], "") + + def test_csv_values_and_header(self): + """CSV 与单任务 synthetic.csv 对齐：每个指标一行，身份列逐行填充""" + _write_perf(self.work_dir, "m-bs4", "synthetic") + # 单组合会跳过聚合，这里直接走写表路径验证列口径 + perf_batch._write_summary_csv( + self.work_dir / SUMMARY_CSV_NAME, + [perf_batch._build_row(self.work_dir, _make_combo(0, 4, "m-bs4"))], + self.work_dir, + ) + with open(self.work_dir / SUMMARY_CSV_NAME, "r", encoding="utf-8") as file: + header = next(csv.reader(file)) + self.assertEqual(header, CSV_HEADERS) + + rows = _read_summary(self.work_dir) + self.assertEqual( + [row["Performance Parameters"] for row in rows], + ["TTFT", "TPOT", "ITL"], + ) + self.assertEqual(rows[0]["Model Abbr"], "m-bs4") + self.assertEqual(rows[0]["Batch Size"], "4") + self.assertEqual([row["Model Abbr"] for row in rows[1:]], + ["m-bs4", "m-bs4"]) + self.assertEqual(rows[0]["Average"], "12.5") + self.assertEqual(rows[0]["P90"], "20") + self.assertEqual(rows[0]["P99"], "25") + self.assertEqual(rows[1]["Average"], "3.5") + self.assertEqual(rows[2]["P99"], "4.2") + + def test_non_stream_blanks_ttft_tpot_itl(self): + """非流式（ttft_valid=false）时延迟指标只保留指标名，统计值置空""" + combo = _make_combo(0, 4, "m-bs4", stream=False, ttft_valid=False) + _write_manifest(self.work_dir, [combo, _make_combo(1, 16, "m-bs16")]) + _write_perf(self.work_dir, "m-bs4", "synthetic") + + aggregate_batch_perf(self.work_dir) + rows = _read_summary(self.work_dir) + for row in rows[:3]: + for key in ("Average", "Min", "Max", "Median", "P75", "P90", "P99"): + self.assertEqual(row[key], "", f"{key} should be empty") + + def test_csv_formula_injection_escaped(self): + """文本列以 = + - @ 开头时加前缀引号，避免 CSV 公式注入""" + combo = _make_combo(0, 4, "=cmd|'/c calc'!A1") + _write_manifest(self.work_dir, [combo, _make_combo(1, 16, "m-bs16")]) + _write_perf(self.work_dir, "=cmd|'/c calc'!A1", "synthetic") + + aggregate_batch_perf(self.work_dir) + rows = _read_summary(self.work_dir) + self.assertEqual(rows[0]["Model Abbr"], "'=cmd|'/c calc'!A1") + + def test_rows_sorted_by_batch_size(self): + """行序按 batch_size 升序，便于直接读出并发-吞吐曲线""" + _write_manifest( + self.work_dir, + [ + _make_combo(0, 32, "m-bs32"), + _make_combo(1, 4, "m-bs4"), + _make_combo(2, 16, "m-bs16"), + ], + ) + for batch_size, abbr in ((32, "m-bs32"), (4, "m-bs4"), (16, "m-bs16")): + _write_perf(self.work_dir, abbr, "synthetic", throughput=batch_size * 10) + aggregate_batch_perf(self.work_dir) + rows = _read_summary(self.work_dir) + # 身份列逐行填充，故按首次出现顺序取组合档位 + batch_sizes = [] + for row in rows: + if not batch_sizes or batch_sizes[-1] != row["Batch Size"]: + batch_sizes.append(row["Batch Size"]) + self.assertEqual(batch_sizes, ["4", "16", "32"]) + + def test_run_summary_appended_after_blank_line(self): + """synthetic.json 汇总区：空行 + 独立表头 + 每组合一行，身份列不变""" + _write_manifest( + self.work_dir, + [_make_combo(0, 4, "m-bs4"), _make_combo(1, 16, "m-bs16")], + ) + _write_perf(self.work_dir, "m-bs4", "synthetic") + + aggregate_batch_perf(self.work_dir) + metric_rows, run_rows = _read_sections(self.work_dir) + # 指标区仍是 CSV 的 3 行 + 缺失组合的 1 行标识 + self.assertEqual(len(metric_rows), 4) + self.assertEqual(len(run_rows), 2) + self.assertEqual( + list(run_rows[0])[:5], + ["Model Abbr", "Dataset Abbr", "Batch Size", "Num Prompts", "Task Status"], + ) + self.assertEqual(run_rows[0]["Model Abbr"], "m-bs4") + self.assertEqual(run_rows[0]["Batch Size"], "4") + self.assertEqual(run_rows[0]["Task Status"], "success") + # 表头只写指标名，取值原样保留单位 + self.assertEqual(run_rows[0]["Benchmark Duration"], "2000.0 ms") + self.assertEqual(run_rows[0]["Total Requests"], "10") + self.assertEqual(run_rows[0]["Failed Requests"], "0") + self.assertEqual(run_rows[0]["Success Requests"], "10") + self.assertEqual(run_rows[0]["Output Token Throughput"], "1000.0 token/s") + # 结果缺失的组合只保留身份列，指标列留空 + self.assertEqual(run_rows[1]["Batch Size"], "16") + self.assertEqual(run_rows[1]["Task Status"], "fail") + self.assertEqual(run_rows[1]["Benchmark Duration"], "") + + +class TestRenderBatchPerfReport(unittest.TestCase): + def setUp(self): + self._tmp = TemporaryDirectory() + self.work_dir = Path(self._tmp.name) + self.combos = [ + _make_combo(0, 4, "m-bs4"), + _make_combo(1, 16, "m-bs16"), + ] + _write_perf(self.work_dir, "m-bs4", "synthetic", throughput=1000.0) + _write_perf(self.work_dir, "m-bs16", "synthetic", throughput=1200.0) + self.rows = [ + perf_batch._build_row(self.work_dir, combo) for combo in self.combos + ] + + def tearDown(self): + self._tmp.cleanup() + + def test_marker_line_and_escape(self): + """HTML 首两行为 DOCTYPE 与渲染标记；模板变量被转义""" + if perf_batch._load_template() is None: + self.skipTest("jinja2 or report template is unavailable") + warnings = [""] + path = render_batch_perf_report( + self.rows, _make_manifest(self.combos), warnings, self.work_dir + ) + self.assertIsNotNone(path) + self.assertEqual(path.name, REPORT_HTML_NAME) + with open(path, "r", encoding="utf-8") as file: + lines = file.read().splitlines() + self.assertEqual(lines[0].strip(), "") + self.assertEqual(lines[1].strip(), REPORT_MARKER) + content = "\n".join(lines) + self.assertIn("<script>", content) + self.assertNotIn("", content) + self.assertIn("AISBench 性能测试报告", content) + + def test_render_failure_degrades_to_none(self): + """渲染异常只告警，返回 None，不影响已落盘的 CSV""" + _write_manifest(self.work_dir, self.combos) + with mock.patch.object( + perf_batch, "_build_charts", side_effect=RuntimeError("boom") + ): + artifacts = aggregate_batch_perf(self.work_dir) + self.assertIsNotNone(artifacts.csv_path) + self.assertIsNone(artifacts.html_path) + self.assertTrue((self.work_dir / SUMMARY_CSV_NAME).exists()) + self.assertFalse((self.work_dir / REPORT_HTML_NAME).exists()) + + def test_missing_combo_not_in_details(self): + """fail 组合只进汇总表，不进入逐组合详情与曲线""" + _write_manifest(self.work_dir, self.combos) + missing = _make_combo(2, 64, "m-bs64") + row = perf_batch._build_row(self.work_dir, missing) + self.assertEqual(row["status"], "fail") + groups = perf_batch._sorted_groups(self.rows + [row]) + self.assertTrue(all( + all(item.get("status") == "success" for item in group_rows) + for _, group_rows in groups + )) + + +class TestPerfBatchHelpers(unittest.TestCase): + def test_parse_number(self): + self.assertEqual(_parse_number("123.4 ms"), 123.4) + self.assertEqual(_parse_number(12), 12.0) + self.assertEqual(_parse_number("1.2e+03"), 1200.0) + self.assertIsNone(_parse_number("")) + self.assertIsNone(_parse_number("abc")) + self.assertIsNone(_parse_number(None)) + + def test_format_value(self): + self.assertEqual(_format_value(None), "") + self.assertEqual(_format_value(1.0), "1") + self.assertEqual(_format_value(1.25), "1.25") + self.assertEqual(_format_value(True), "true") + self.assertEqual(_format_value(3), "3") + + def test_escape_csv_text(self): + self.assertEqual(_escape_csv_text("=1+1"), "'=1+1") + self.assertEqual(_escape_csv_text("+1"), "'+1") + self.assertEqual(_escape_csv_text("-1"), "'-1") + self.assertEqual(_escape_csv_text("@x"), "'@x") + self.assertEqual(_escape_csv_text("ok"), "ok") + self.assertEqual(_escape_csv_text(None), "") + + def test_saturation_note(self): + def row(batch_size, throughput, success_rate=100): + return { + "status": "success", + "batch_size": batch_size, + "throughput_tok_s": throughput, + "success_rate": success_rate, + } + + # 返回 (中文, English)，断言中文文案 + self.assertIn("不足 2 个", _saturation_note([row(4, 1000)])[0]) + self.assertIn("未饱和", _saturation_note([row(4, 1000), row(16, 1200)])[0]) + self.assertIn("接近饱和", _saturation_note([row(4, 1000), row(16, 1020)])[0]) + self.assertIn("已饱和", _saturation_note([row(4, 1000), row(16, 900)])[0]) + self.assertIn("已过载", _saturation_note([row(4, 1000), row(16, 1200, 90)])[0]) + + +if __name__ == "__main__": + unittest.main()