Skip to content

vllm_mlx.bench_serve

Serving benchmark for vllm-mlx.

View the complete module source at #L1-L2638.

API details

Each callable below includes its exact signature, type annotations, inputs, defaults, return contract, documented exceptions, implementation source, and parsed docstring sections when the source provides them.

vllm_mlx.bench_serve

Serving benchmark for vllm-mlx.

Measures end-to-end HTTP performance of a running vllm-mlx server: - Time to First Token (TTFT) - Time Per Output Token (TPOT) - End-to-end latency - Generation and prompt throughput - Concurrent request handling - KV cache hit rates - Metal memory utilization

This module has no MLX dependency and can be imported on any platform. It is a pure HTTP client that talks to a running OpenAI-compatible server.

vllm_mlx.bench_serve._BUILTIN_DIR module-attribute 

_BUILTIN_DIR = Path(__file__).parent / 'bench_serve_prompts'

vllm_mlx.bench_serve._BUILTIN_NAMES module-attribute 

_BUILTIN_NAMES = {'short', 'medium', 'long', 'thinking'}

vllm_mlx.bench_serve._SQL_IDENTIFIER_RE module-attribute 

_SQL_IDENTIFIER_RE = re.compile('^[a-z_][a-z0-9_]*$')

vllm_mlx.bench_serve.SweepConfig module-attribute 

SweepConfig = tuple[str, int, Optional[bool], str, int]

vllm_mlx.bench_serve.RESULT_COLUMNS module-attribute 

RESULT_COLUMNS: list[str] = [f.name for f in _dataclasses.fields(BenchServeResult)]

vllm_mlx.bench_serve._TABLE_COLUMNS module-attribute 

_TABLE_COLUMNS = ['prompt_set', 'concurrency', 'prompt_tokens', 'ttft_ms', 'tpot_ms', 'gen_tps', 'prompt_tps', 'e2e_latency_ms', 'validated']

vllm_mlx.bench_serve._SQL_SCHEMA module-attribute 

_SQL_SCHEMA = 'run_id TEXT, timestamp TEXT, tag TEXT, chip TEXT, gpu_cores INTEGER, memory_gb REAL, bandwidth_gbs REAL, os_version TEXT, model_id TEXT, model_type TEXT, engine_type TEXT, mtp_enabled BOOLEAN, specprefill BOOLEAN, kv_quant TEXT, cache_type TEXT, prompt_set TEXT, concurrency INTEGER, max_tokens INTEGER, enable_thinking BOOLEAN, extra_body TEXT, repetition INTEGER, prompt_tokens INTEGER, ttft_ms REAL, tpot_ms REAL, e2e_latency_ms REAL, gen_tps REAL, prompt_tps REAL, throughput_tps REAL, requests_per_s REAL, metal_active_gb REAL, metal_peak_gb REAL, metal_cache_gb REAL, cache_hits INTEGER, cache_misses INTEGER, cache_hit_rate REAL, tokens_saved INTEGER, validated BOOLEAN'

vllm_mlx.bench_serve.WORKLOAD_RESULT_COLUMNS module-attribute 

WORKLOAD_RESULT_COLUMNS = ['run_id', 'timestamp', 'workload', 'case_id', 'repetition', 'tags', 'model_id', 'chip', 'memory_gb', 'os_version', 'engine_type', 'model_type', 'mtp_enabled', 'specprefill', 'kv_quant', 'cache_type', 'request_max_tokens', 'request_enable_thinking', 'request_extra_body', 'policy_timeout_ms', 'within_policy_timeout', 'ttft_ms', 'tpot_ms', 'e2e_latency_ms', 'gen_tps', 'prompt_tps', 'prompt_tokens', 'completion_tokens', 'cache_hits', 'cache_misses', 'tokens_saved', 'metal_active_gb', 'metal_peak_gb', 'metal_cache_gb', 'quality_ok', 'quality_issues', 'finish_reason', 'content_chars', 'content_preview']

vllm_mlx.bench_serve._WORKLOAD_TABLE_COLUMNS module-attribute 

_WORKLOAD_TABLE_COLUMNS = ['case_id', 'repetition', 'tags', 'quality_ok', 'within_policy_timeout', 'ttft_ms', 'gen_tps', 'e2e_latency_ms', 'cache_hits', 'tokens_saved', 'finish_reason']

vllm_mlx.bench_serve._WORKLOAD_SQL_SCHEMA module-attribute 

_WORKLOAD_SQL_SCHEMA = 'run_id TEXT, timestamp TEXT, workload TEXT, case_id TEXT, repetition INTEGER, tags TEXT, model_id TEXT, chip TEXT, memory_gb REAL, os_version TEXT, engine_type TEXT, model_type TEXT, mtp_enabled BOOLEAN, specprefill BOOLEAN, kv_quant TEXT, cache_type TEXT, request_max_tokens INTEGER, request_enable_thinking BOOLEAN, request_extra_body TEXT, policy_timeout_ms INTEGER, within_policy_timeout BOOLEAN, ttft_ms REAL, tpot_ms REAL, e2e_latency_ms REAL, gen_tps REAL, prompt_tps REAL, prompt_tokens INTEGER, completion_tokens INTEGER, cache_hits INTEGER, cache_misses INTEGER, tokens_saved INTEGER, metal_active_gb REAL, metal_peak_gb REAL, metal_cache_gb REAL, quality_ok BOOLEAN, quality_issues TEXT, finish_reason TEXT, content_chars INTEGER, content_preview TEXT'

vllm_mlx.bench_serve.logger module-attribute 

logger = logging.getLogger(__name__)

vllm_mlx.bench_serve.WorkloadCase dataclass 

WorkloadCase(case_id: str, messages: list[dict], request_path: Optional[str] = None, max_tokens: Optional[int] = None, enable_thinking: Optional[bool] = None, extra_body: Optional[dict] = None, policy_timeout_ms: Optional[int] = None, checks: Optional[dict] = None, tags: tuple[str, ...] = ())

One declarative benchmark case for contract-style serving tests.

vllm_mlx.bench_serve.WorkloadCase.case_id instance-attribute 

case_id: str

vllm_mlx.bench_serve.WorkloadCase.messages instance-attribute 

messages: list[dict]

vllm_mlx.bench_serve.WorkloadCase.request_path class-attribute instance-attribute 

request_path: Optional[str] = None

vllm_mlx.bench_serve.WorkloadCase.max_tokens class-attribute instance-attribute 

max_tokens: Optional[int] = None

vllm_mlx.bench_serve.WorkloadCase.enable_thinking class-attribute instance-attribute 

enable_thinking: Optional[bool] = None

vllm_mlx.bench_serve.WorkloadCase.extra_body class-attribute instance-attribute 

extra_body: Optional[dict] = None

vllm_mlx.bench_serve.WorkloadCase.policy_timeout_ms class-attribute instance-attribute 

policy_timeout_ms: Optional[int] = None

vllm_mlx.bench_serve.WorkloadCase.checks class-attribute instance-attribute 

checks: Optional[dict] = None

vllm_mlx.bench_serve.WorkloadCase.tags class-attribute instance-attribute 

tags: tuple[str, ...] = ()

vllm_mlx.bench_serve.Workload dataclass 

Workload(name: str, description: str, defaults: dict, cases: list[WorkloadCase])

Normalized bench-serve workload manifest.

vllm_mlx.bench_serve.Workload.name instance-attribute 

name: str

vllm_mlx.bench_serve.Workload.description instance-attribute 

description: str

vllm_mlx.bench_serve.Workload.defaults instance-attribute 

defaults: dict

vllm_mlx.bench_serve.Workload.cases instance-attribute 

cases: list[WorkloadCase]

vllm_mlx.bench_serve.BenchServeResult dataclass 

BenchServeResult(run_id: str = '', timestamp: str = '', tag: str = '', chip: str = '', gpu_cores: int = 0, memory_gb: float = 0.0, bandwidth_gbs: float = 0.0, os_version: str = '', model_id: str = '', model_type: str = '', engine_type: str = '', mtp_enabled: bool = False, specprefill: bool = False, kv_quant: str = '', cache_type: str = '', prompt_set: str = '', concurrency: int = 1, max_tokens: int = 256, enable_thinking: Optional[bool] = None, extra_body: str = '', repetition: int = 0, prompt_tokens: int = 0, ttft_ms: float = 0.0, tpot_ms: float = 0.0, e2e_latency_ms: float = 0.0, gen_tps: float = 0.0, prompt_tps: float = 0.0, throughput_tps: float = 0.0, requests_per_s: float = 0.0, metal_active_gb: float = 0.0, metal_peak_gb: float = 0.0, metal_cache_gb: float = 0.0, cache_hits: int = 0, cache_misses: int = 0, cache_hit_rate: float = 0.0, tokens_saved: int = 0, validated: bool = True)

Aggregated results from a single bench-serve run configuration.

vllm_mlx.bench_serve.BenchServeResult.run_id class-attribute instance-attribute 

run_id: str = ''

vllm_mlx.bench_serve.BenchServeResult.timestamp class-attribute instance-attribute 

timestamp: str = ''

vllm_mlx.bench_serve.BenchServeResult.tag class-attribute instance-attribute 

tag: str = ''

vllm_mlx.bench_serve.BenchServeResult.chip class-attribute instance-attribute 

chip: str = ''

vllm_mlx.bench_serve.BenchServeResult.gpu_cores class-attribute instance-attribute 

gpu_cores: int = 0

vllm_mlx.bench_serve.BenchServeResult.memory_gb class-attribute instance-attribute 

memory_gb: float = 0.0

vllm_mlx.bench_serve.BenchServeResult.bandwidth_gbs class-attribute instance-attribute 

bandwidth_gbs: float = 0.0

vllm_mlx.bench_serve.BenchServeResult.os_version class-attribute instance-attribute 

os_version: str = ''

vllm_mlx.bench_serve.BenchServeResult.model_id class-attribute instance-attribute 

model_id: str = ''

vllm_mlx.bench_serve.BenchServeResult.model_type class-attribute instance-attribute 

model_type: str = ''

vllm_mlx.bench_serve.BenchServeResult.engine_type class-attribute instance-attribute 

engine_type: str = ''

vllm_mlx.bench_serve.BenchServeResult.mtp_enabled class-attribute instance-attribute 

mtp_enabled: bool = False

vllm_mlx.bench_serve.BenchServeResult.specprefill class-attribute instance-attribute 

specprefill: bool = False

vllm_mlx.bench_serve.BenchServeResult.kv_quant class-attribute instance-attribute 

kv_quant: str = ''

vllm_mlx.bench_serve.BenchServeResult.cache_type class-attribute instance-attribute 

cache_type: str = ''

vllm_mlx.bench_serve.BenchServeResult.prompt_set class-attribute instance-attribute 

prompt_set: str = ''

vllm_mlx.bench_serve.BenchServeResult.concurrency class-attribute instance-attribute 

concurrency: int = 1

vllm_mlx.bench_serve.BenchServeResult.max_tokens class-attribute instance-attribute 

max_tokens: int = 256

vllm_mlx.bench_serve.BenchServeResult.enable_thinking class-attribute instance-attribute 

enable_thinking: Optional[bool] = None

vllm_mlx.bench_serve.BenchServeResult.extra_body class-attribute instance-attribute 

extra_body: str = ''

vllm_mlx.bench_serve.BenchServeResult.repetition class-attribute instance-attribute 

repetition: int = 0

vllm_mlx.bench_serve.BenchServeResult.prompt_tokens class-attribute instance-attribute 

prompt_tokens: int = 0

vllm_mlx.bench_serve.BenchServeResult.ttft_ms class-attribute instance-attribute 

ttft_ms: float = 0.0

vllm_mlx.bench_serve.BenchServeResult.tpot_ms class-attribute instance-attribute 

tpot_ms: float = 0.0

vllm_mlx.bench_serve.BenchServeResult.e2e_latency_ms class-attribute instance-attribute 

e2e_latency_ms: float = 0.0

vllm_mlx.bench_serve.BenchServeResult.gen_tps class-attribute instance-attribute 

gen_tps: float = 0.0

vllm_mlx.bench_serve.BenchServeResult.prompt_tps class-attribute instance-attribute 

prompt_tps: float = 0.0

vllm_mlx.bench_serve.BenchServeResult.throughput_tps class-attribute instance-attribute 

throughput_tps: float = 0.0

vllm_mlx.bench_serve.BenchServeResult.requests_per_s class-attribute instance-attribute 

requests_per_s: float = 0.0

vllm_mlx.bench_serve.BenchServeResult.metal_active_gb class-attribute instance-attribute 

metal_active_gb: float = 0.0

vllm_mlx.bench_serve.BenchServeResult.metal_peak_gb class-attribute instance-attribute 

metal_peak_gb: float = 0.0

vllm_mlx.bench_serve.BenchServeResult.metal_cache_gb class-attribute instance-attribute 

metal_cache_gb: float = 0.0

vllm_mlx.bench_serve.BenchServeResult.cache_hits class-attribute instance-attribute 

cache_hits: int = 0

vllm_mlx.bench_serve.BenchServeResult.cache_misses class-attribute instance-attribute 

cache_misses: int = 0

vllm_mlx.bench_serve.BenchServeResult.cache_hit_rate class-attribute instance-attribute 

cache_hit_rate: float = 0.0

vllm_mlx.bench_serve.BenchServeResult.tokens_saved class-attribute instance-attribute 

tokens_saved: int = 0

vllm_mlx.bench_serve.BenchServeResult.validated class-attribute instance-attribute 

validated: bool = True

vllm_mlx.bench_serve.load_prompt_set 

load_prompt_set(name_or_path: str) -> list[list[dict]]

Load a prompt set by builtin name or file path.

Builtin sets (short, medium, long, thinking) are loaded from the bench_serve_prompts/ directory next to this module. Any other value is treated as a filesystem path and loaded directly.

Two file formats are accepted (detected automatically):

  1. Flat — list of single message dicts. Each dict becomes a single-message prompt. Backwards-compatible with the original format.

[{"role": "user", "content": "..."}, ...]

  1. Multi-message — list of message-dict lists. Each inner list is a full chat history (e.g. [system, user]). Use this format when you want to benchmark with system prompts that match an --warm-prompts warm-up, or to simulate multi-turn conversation.

[[{"role":"system","content":"..."}, {"role":"user","content":"..."}], ...]

Returns:

  • list[list[dict]] –

    A list of message-dict lists, i.e. every entry is a full chat history.

  • list[list[dict]] –

    Flat-format files are normalized to single-element lists.

Raises:

  • FileNotFoundError –

    If name_or_path is not a known builtin name and the path does not exist, or if a builtin name is requested but its JSON file is missing from the package.

  • ValueError –

    If the file shape is not recognised.

Source code in vllm_mlx/bench_serve.py 
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
def load_prompt_set(name_or_path: str) -> list[list[dict]]:
    """Load a prompt set by builtin name or file path.

    Builtin sets (``short``, ``medium``, ``long``, ``thinking``) are loaded
    from the ``bench_serve_prompts/`` directory next to this module.  Any
    other value is treated as a filesystem path and loaded directly.

    Two file formats are accepted (detected automatically):

    1. **Flat** — list of single message dicts. Each dict becomes a
       single-message prompt. Backwards-compatible with the original format.

       ``[{"role": "user", "content": "..."}, ...]``

    2. **Multi-message** — list of message-dict lists. Each inner list is a
       full chat history (e.g. ``[system, user]``). Use this format when you
       want to benchmark with system prompts that match an ``--warm-prompts``
       warm-up, or to simulate multi-turn conversation.

       ``[[{"role":"system","content":"..."}, {"role":"user","content":"..."}], ...]``

    Returns:
        A list of message-dict lists, i.e. every entry is a full chat history.
        Flat-format files are normalized to single-element lists.

    Raises:
        FileNotFoundError: If ``name_or_path`` is not a known builtin name and
            the path does not exist, or if a builtin name is requested but its
            JSON file is missing from the package.
        ValueError: If the file shape is not recognised.
    """
    if name_or_path in _BUILTIN_NAMES:
        target = _BUILTIN_DIR / f"{name_or_path}.json"
        if not target.exists():
            raise FileNotFoundError(
                f"Builtin prompt set '{name_or_path}' not found at {target}"
            )
        with target.open() as fh:
            raw = json.load(fh)
    else:
        path = Path(name_or_path).expanduser()
        if not path.exists():
            raise FileNotFoundError(
                f"Unknown prompt set name or missing file: '{name_or_path}'. "
                f"Builtin names are: {sorted(_BUILTIN_NAMES)}"
            )
        with path.open() as fh:
            raw = json.load(fh)

    if not isinstance(raw, list) or not raw:
        raise ValueError(f"Prompt file must be a non-empty JSON list: {name_or_path}")

    # Auto-detect format: dict entries = flat; list entries = multi-message.
    first = raw[0]
    if isinstance(first, dict):
        # Flat format: wrap each message in a single-element list.
        return [[msg] for msg in raw]
    if isinstance(first, list):
        return raw
    raise ValueError(
        f"Prompt entries must be dict or list, got {type(first).__name__} "
        f"in {name_or_path}"
    )

vllm_mlx.bench_serve._require_message_list 

_require_message_list(value: Any, *, label: str) -> list[dict]
Source code in vllm_mlx/bench_serve.py 
140
141
142
143
144
145
146
147
148
def _require_message_list(value: Any, *, label: str) -> list[dict]:
    if not isinstance(value, list) or not value:
        raise ValueError(f"{label}: messages must be a non-empty list")
    for idx, message in enumerate(value):
        if not isinstance(message, dict):
            raise ValueError(f"{label}: message {idx} must be an object")
        if "role" not in message or "content" not in message:
            raise ValueError(f"{label}: message {idx} must include role and content")
    return value

vllm_mlx.bench_serve._load_case_request 

_load_case_request(path: str, *, workload_path: Path, case_id: str) -> dict
Source code in vllm_mlx/bench_serve.py 
151
152
153
154
155
156
157
158
159
def _load_case_request(path: str, *, workload_path: Path, case_id: str) -> dict:
    request_path = Path(path).expanduser()
    if not request_path.is_absolute():
        request_path = workload_path.parent / request_path
    with request_path.open() as fh:
        request = json.load(fh)
    if not isinstance(request, dict):
        raise ValueError(f"{case_id}: request_path must point to a JSON object")
    return request

vllm_mlx.bench_serve._request_extra_body 

_request_extra_body(request: dict) -> dict
Source code in vllm_mlx/bench_serve.py 
162
163
164
165
166
167
168
169
170
171
def _request_extra_body(request: dict) -> dict:
    reserved = {
        "model",
        "messages",
        "max_tokens",
        "stream",
        "stream_options",
        "enable_thinking",
    }
    return {key: value for key, value in request.items() if key not in reserved}

vllm_mlx.bench_serve._first_not_none 

_first_not_none(*values: Any) -> Any
Source code in vllm_mlx/bench_serve.py 
174
175
176
177
178
def _first_not_none(*values: Any) -> Any:
    for value in values:
        if value is not None:
            return value
    return None

vllm_mlx.bench_serve._normalize_tags 

_normalize_tags(tags: Any, *, case_id: str) -> tuple[str, ...]

Coerce a workload case's tags field to a tuple of strings.

Accepts either a single string (treated as a one-element list) or a list. Any other type is rejected with a case-scoped ValueError.

Source code in vllm_mlx/bench_serve.py 
181
182
183
184
185
186
187
188
189
190
191
def _normalize_tags(tags: Any, *, case_id: str) -> tuple[str, ...]:
    """Coerce a workload case's ``tags`` field to a tuple of strings.

    Accepts either a single string (treated as a one-element list) or a
    list. Any other type is rejected with a case-scoped ``ValueError``.
    """
    if isinstance(tags, str):
        tags = [tags]
    if not isinstance(tags, list):
        raise ValueError(f"{case_id}: tags must be a list or string")
    return tuple(str(tag) for tag in tags)

vllm_mlx.bench_serve._merge_case_checks 

_merge_case_checks(default_checks: Any, case_checks: Any, *, case_id: str) -> Optional[dict]

Merge a case's checks over the workload defaults.

Most keys are overridden by the case-level value. The two regex list keys (required_regex, forbidden_regex) are list-concatenated with default patterns first and case patterns appended, so case-level patterns extend defaults rather than replace them. Returns None when neither source contributes any checks.

Rejects a non-dict case_checks with a ValueError named after the case so the operator gets a clear message instead of the AttributeError that the previous inline code raised on case_checks.items().

Source code in vllm_mlx/bench_serve.py 
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
def _merge_case_checks(
    default_checks: Any,
    case_checks: Any,
    *,
    case_id: str,
) -> Optional[dict]:
    """Merge a case's ``checks`` over the workload defaults.

    Most keys are overridden by the case-level value. The two regex list
    keys (``required_regex``, ``forbidden_regex``) are list-concatenated
    with default patterns first and case patterns appended, so
    case-level patterns extend defaults rather than replace them.
    Returns ``None`` when neither source contributes any checks.

    Rejects a non-dict ``case_checks`` with a ``ValueError`` named after
    the case so the operator gets a clear message instead of the
    ``AttributeError`` that the previous inline code raised on
    ``case_checks.items()``.
    """
    merged: dict = dict(default_checks or {})
    if not case_checks:
        return merged or None
    if not isinstance(case_checks, dict):
        raise ValueError(f"{case_id}: checks must be an object")
    for key, value in case_checks.items():
        if (
            key in ("required_regex", "forbidden_regex")
            and isinstance(value, list)
            and isinstance(merged.get(key), list)
        ):
            merged[key] = merged[key] + value
        else:
            merged[key] = value
    return merged or None

vllm_mlx.bench_serve._build_workload_case 

_build_workload_case(item: Any, idx: int, *, defaults: dict, workload_path: Path) -> WorkloadCase

Construct one WorkloadCase from a raw workload entry.

Validates the entry shape, loads request defaults from a sibling JSON file when request_path is provided, merges extra_body and checks against the workload defaults, and resolves scalar fields (max_tokens, enable_thinking, policy_timeout_ms) via _first_not_none priority: case-level beats request_path defaults beats workload defaults.

extra_body follows a different merge: it composes the request_path extras (base) with either the case-level extra_body if present, otherwise the workload-default extra_body. The case-vs-default fallback is a get-with-default, not a three-way merge.

Source code in vllm_mlx/bench_serve.py 
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
def _build_workload_case(
    item: Any,
    idx: int,
    *,
    defaults: dict,
    workload_path: Path,
) -> WorkloadCase:
    """Construct one ``WorkloadCase`` from a raw workload entry.

    Validates the entry shape, loads request defaults from a sibling JSON
    file when ``request_path`` is provided, merges ``extra_body`` and
    ``checks`` against the workload defaults, and resolves scalar fields
    (``max_tokens``, ``enable_thinking``, ``policy_timeout_ms``) via
    ``_first_not_none`` priority: case-level beats request_path defaults
    beats workload defaults.

    ``extra_body`` follows a different merge: it composes the
    ``request_path`` extras (base) with either the case-level
    ``extra_body`` if present, otherwise the workload-default
    ``extra_body``. The case-vs-default fallback is a get-with-default,
    not a three-way merge.
    """
    if not isinstance(item, dict):
        raise ValueError(f"case {idx}: case must be an object")
    case_id = str(item.get("id") or f"case_{idx + 1}")

    request_path = item.get("request_path")
    request_defaults: dict = {}
    if request_path is not None:
        request_defaults = _load_case_request(
            str(request_path), workload_path=workload_path, case_id=case_id
        )

    messages = _require_message_list(
        item.get("messages", request_defaults.get("messages")),
        label=case_id,
    )

    extra_body = item.get("extra_body", defaults.get("extra_body"))
    request_extra = _request_extra_body(request_defaults)
    if extra_body:
        request_extra.update(extra_body)
    extra_body = request_extra or None

    checks = _merge_case_checks(
        defaults.get("checks"), item.get("checks"), case_id=case_id
    )

    return WorkloadCase(
        case_id=case_id,
        messages=messages,
        request_path=str(request_path) if request_path is not None else None,
        max_tokens=_first_not_none(
            item.get("max_tokens"),
            request_defaults.get("max_tokens"),
            defaults.get("max_tokens"),
        ),
        enable_thinking=_first_not_none(
            item.get("enable_thinking"),
            request_defaults.get("enable_thinking"),
            defaults.get("enable_thinking"),
        ),
        extra_body=extra_body,
        policy_timeout_ms=_first_not_none(
            item.get("policy_timeout_ms"),
            request_defaults.get("policy_timeout_ms"),
            defaults.get("policy_timeout_ms"),
        ),
        checks=checks,
        tags=_normalize_tags(item.get("tags", []), case_id=case_id),
    )

vllm_mlx.bench_serve.load_workload 

load_workload(path: str | Path) -> Workload

Load a declarative serving benchmark workload.

Workloads are for product-like qualification where each case can carry request settings, comparison-only policy timeouts, and quality checks. Timeout fields are metadata unless the runner explicitly uses them as a transport limit; they are not treated as hardware capability claims.

Source code in vllm_mlx/bench_serve.py 
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
def load_workload(path: str | Path) -> Workload:
    """Load a declarative serving benchmark workload.

    Workloads are for product-like qualification where each case can carry
    request settings, comparison-only policy timeouts, and quality checks.
    Timeout fields are metadata unless the runner explicitly uses them as a
    transport limit; they are not treated as hardware capability claims.
    """
    workload_path = Path(path).expanduser()
    with workload_path.open() as fh:
        raw = json.load(fh)

    if not isinstance(raw, dict):
        raise ValueError("workload root must be a JSON object")
    raw_cases = raw.get("cases")
    if not isinstance(raw_cases, list) or not raw_cases:
        raise ValueError("workload must contain a non-empty cases list")

    defaults = raw.get("defaults") or {}
    if not isinstance(defaults, dict):
        raise ValueError("workload defaults must be an object")

    cases = [
        _build_workload_case(item, idx, defaults=defaults, workload_path=workload_path)
        for idx, item in enumerate(raw_cases)
    ]

    return Workload(
        name=str(raw.get("name") or workload_path.stem),
        description=str(raw.get("description") or ""),
        defaults=defaults,
        cases=cases,
    )

vllm_mlx.bench_serve.expand_sweep 

expand_sweep(prompt_sets: list[str], concurrencies: list[int], thinking_values: list[Optional[bool]], extra_bodies: list[str], repetitions: int) -> list[SweepConfig]

Expand sweep parameters into a flat list of configurations.

Performs the full Cartesian product of all input dimensions and then unfolds each combination across repetitions repetition indices (0-based).

Parameters:

  • prompt_sets (list[str]) –

    Names or paths of prompt sets to include.

  • concurrencies (list[int]) –

    Concurrency levels to test (e.g. [1, 4, 16]).

  • thinking_values (list[Optional[bool]]) –

    Values for enable_thinking (e.g. [None, True, False]).

  • extra_bodies (list[str]) –

    JSON strings (or empty string) to pass as extra body parameters on each request.

  • repetitions (int) –

    Number of times to repeat each unique combination. Each repeat gets a distinct 0-based repetition index.

Returns:

  • list[SweepConfig] –

    A list of :data:SweepConfig tuples in the order::

    (prompt_set, concurrency, thinking, extra_body, repetition_index)

Source code in vllm_mlx/bench_serve.py 
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
def expand_sweep(
    prompt_sets: list[str],
    concurrencies: list[int],
    thinking_values: list[Optional[bool]],
    extra_bodies: list[str],
    repetitions: int,
) -> list[SweepConfig]:
    """Expand sweep parameters into a flat list of configurations.

    Performs the full Cartesian product of all input dimensions and then
    unfolds each combination across ``repetitions`` repetition indices
    (0-based).

    Args:
        prompt_sets: Names or paths of prompt sets to include.
        concurrencies: Concurrency levels to test (e.g. ``[1, 4, 16]``).
        thinking_values: Values for ``enable_thinking`` (e.g.
            ``[None, True, False]``).
        extra_bodies: JSON strings (or empty string) to pass as extra body
            parameters on each request.
        repetitions: Number of times to repeat each unique combination.
            Each repeat gets a distinct 0-based repetition index.

    Returns:
        A list of :data:`SweepConfig` tuples in the order::

            (prompt_set, concurrency, thinking, extra_body, repetition_index)
    """
    configs: list[SweepConfig] = []
    for prompt_set, concurrency, thinking, extra_body, rep in itertools.product(
        prompt_sets, concurrencies, thinking_values, extra_bodies, range(repetitions)
    ):
        configs.append((prompt_set, concurrency, thinking, extra_body, rep))
    return configs

vllm_mlx.bench_serve.parse_health_response 

parse_health_response(data: dict) -> dict

Extract model identity fields from a GET /health response.

Parameters:

  • data (dict) –

    Parsed JSON body from the /health endpoint. Expected shape::

    {"status": "healthy", "model_loaded": True, "model_name": "...", "model_type": "llm"|"mllm"}

Returns:

  • dict –

    {"model_name": str, "model_type": str}

Source code in vllm_mlx/bench_serve.py 
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
def parse_health_response(data: dict) -> dict:
    """Extract model identity fields from a GET /health response.

    Args:
        data: Parsed JSON body from the /health endpoint.  Expected shape::

            {"status": "healthy", "model_loaded": True,
             "model_name": "...", "model_type": "llm"|"mllm"}

    Returns:
        ``{"model_name": str, "model_type": str}``
    """
    return {
        "model_name": data.get("model_name", ""),
        "model_type": data.get("model_type", ""),
    }

vllm_mlx.bench_serve.parse_status_response 

parse_status_response(data: dict) -> dict

Extract metal and cache info from a GET /v1/status response.

Parameters:

  • data (dict) –

    Parsed JSON body from the /v1/status endpoint. Metal info is expected under data["metal"] and cache info under data["cache"]. Missing keys are handled gracefully.

Returns:

  • dict –

    ``{"model": str, "metal_active_gb": float, "metal_peak_gb": float,

  • dict –

    "metal_cache_gb": float, "cache_type": str}``

Source code in vllm_mlx/bench_serve.py 
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
def parse_status_response(data: dict) -> dict:
    """Extract metal and cache info from a GET /v1/status response.

    Args:
        data: Parsed JSON body from the /v1/status endpoint.  Metal info is
            expected under ``data["metal"]`` and cache info under
            ``data["cache"]``.  Missing keys are handled gracefully.

    Returns:
        ``{"model": str, "metal_active_gb": float, "metal_peak_gb": float,
        "metal_cache_gb": float, "cache_type": str}``
    """
    metal = data.get("metal") or {}
    cache = data.get("cache") or {}
    return {
        "model": data.get("model", ""),
        "metal_active_gb": float(
            metal.get("active_memory_gb") or metal.get("active_gb") or 0.0
        ),
        "metal_peak_gb": float(
            metal.get("peak_memory_gb") or metal.get("peak_gb") or 0.0
        ),
        "metal_cache_gb": float(
            metal.get("cache_memory_gb") or metal.get("cache_gb") or 0.0
        ),
        "cache_type": cache.get("type", "") or "",
    }

vllm_mlx.bench_serve.parse_metrics_text 

parse_metrics_text(text: str) -> dict

Parse Prometheus text exposition format from GET /metrics.

Extracts the three prefix-cache counters used for bench reporting.

Parameters:

  • text (str) –

    Raw response body from the /metrics endpoint.

Returns:

  • dict –

    {"cache_hits": int, "cache_misses": int, "tokens_saved": int}

  • dict –

    — each value defaults to 0 when the metric line is absent.

Source code in vllm_mlx/bench_serve.py 
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
def parse_metrics_text(text: str) -> dict:
    """Parse Prometheus text exposition format from GET /metrics.

    Extracts the three prefix-cache counters used for bench reporting.

    Args:
        text: Raw response body from the /metrics endpoint.

    Returns:
        ``{"cache_hits": int, "cache_misses": int, "tokens_saved": int}``
        — each value defaults to ``0`` when the metric line is absent.
    """

    def _extract(metric_name: str) -> int:
        pattern = rf"^{re.escape(metric_name)}\s+(\d+)"
        m = re.search(pattern, text, re.MULTILINE)
        return int(m.group(1)) if m else 0

    return {
        "cache_hits": _extract("vllm_prefix_cache_hits_total"),
        "cache_misses": _extract("vllm_prefix_cache_misses_total"),
        "tokens_saved": _extract("vllm_prefix_cache_tokens_saved_total"),
    }

vllm_mlx.bench_serve.detect_hardware_fingerprint 

detect_hardware_fingerprint() -> dict

Return a hardware fingerprint dict for the current machine.

Tries to import :func:vllm_mlx.optimizations.detect_hardware (which requires MLX). Falls back to reading hw.memsize via sysctl when MLX is unavailable. os_version is always obtained from :func:platform.platform.

Returns:

  • dict –

    ``{"chip": str, "gpu_cores": int, "memory_gb": float,

  • dict –

    "bandwidth_gbs": float, "os_version": str}``

Source code in vllm_mlx/bench_serve.py 
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
def detect_hardware_fingerprint() -> dict:
    """Return a hardware fingerprint dict for the current machine.

    Tries to import :func:`vllm_mlx.optimizations.detect_hardware` (which
    requires MLX).  Falls back to reading ``hw.memsize`` via ``sysctl`` when
    MLX is unavailable.  ``os_version`` is always obtained from
    :func:`platform.platform`.

    Returns:
        ``{"chip": str, "gpu_cores": int, "memory_gb": float,
        "bandwidth_gbs": float, "os_version": str}``
    """
    os_version = platform.platform()

    try:
        from .optimizations import detect_hardware  # type: ignore[import]

        hw = detect_hardware()
        return {
            "chip": hw.chip_name,
            "gpu_cores": hw.gpu_cores,
            "memory_gb": hw.total_memory_gb,
            "bandwidth_gbs": hw.memory_bandwidth_gbs,
            "os_version": os_version,
        }
    except Exception:
        pass

    # Fallback: use sysctl for memory, leave chip/cores/bandwidth unknown.
    memory_gb = 0.0
    try:
        import subprocess

        result = subprocess.run(
            ["sysctl", "-n", "hw.memsize"],
            capture_output=True,
            text=True,
            check=True,
        )
        memory_gb = int(result.stdout.strip()) / (1024**3)
    except Exception:
        pass

    return {
        "chip": "",
        "gpu_cores": 0,
        "memory_gb": memory_gb,
        "bandwidth_gbs": 0.0,
        "os_version": os_version,
    }

vllm_mlx.bench_serve.auto_detect_runtime async 

auto_detect_runtime(client: AsyncClient, base_url: str) -> dict

Query the running server and return a runtime descriptor dict.

Hits /health, /v1/models, and /v1/status in sequence. Each call is wrapped in an :exc:httpx.HTTPError guard so a missing endpoint does not abort the whole detection.

Parameters:

  • client (AsyncClient) –

    An open :class:httpx.AsyncClient.

  • base_url (str) –

    Base URL of the server (e.g. "http://localhost:8080").

Returns:

  • dict –

    Dict with keys: model_id, model_type, engine_type,

  • dict –

    mtp_enabled, specprefill, kv_quant, cache_type,

  • dict –

    metal_active_gb, metal_peak_gb, metal_cache_gb.

Source code in vllm_mlx/bench_serve.py 
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
async def auto_detect_runtime(client: httpx.AsyncClient, base_url: str) -> dict:
    """Query the running server and return a runtime descriptor dict.

    Hits ``/health``, ``/v1/models``, and ``/v1/status`` in sequence.
    Each call is wrapped in an :exc:`httpx.HTTPError` guard so a missing
    endpoint does not abort the whole detection.

    Args:
        client: An open :class:`httpx.AsyncClient`.
        base_url: Base URL of the server (e.g. ``"http://localhost:8080"``).

    Returns:
        Dict with keys: ``model_id``, ``model_type``, ``engine_type``,
        ``mtp_enabled``, ``specprefill``, ``kv_quant``, ``cache_type``,
        ``metal_active_gb``, ``metal_peak_gb``, ``metal_cache_gb``.
    """
    result: dict = {
        "model_id": "",
        "model_type": "",
        "engine_type": "",
        "mtp_enabled": False,
        "specprefill": False,
        "kv_quant": "",
        "cache_type": "",
        "metal_active_gb": 0.0,
        "metal_peak_gb": 0.0,
        "metal_cache_gb": 0.0,
    }

    # /health
    try:
        resp = await client.get(f"{base_url}/health")
        resp.raise_for_status()
        health = parse_health_response(resp.json())
        result["model_type"] = health.get("model_type", "")
    except httpx.HTTPError:
        pass

    # /v1/models
    try:
        resp = await client.get(f"{base_url}/v1/models")
        resp.raise_for_status()
        models_data = resp.json()
        models = models_data.get("data") or []
        if models:
            result["model_id"] = models[0].get("id", "")
    except httpx.HTTPError:
        pass

    # /v1/status
    try:
        resp = await client.get(f"{base_url}/v1/status")
        resp.raise_for_status()
        status = parse_status_response(resp.json())
        result["cache_type"] = status.get("cache_type", "")
        result["metal_active_gb"] = status.get("metal_active_gb", 0.0)
        result["metal_peak_gb"] = status.get("metal_peak_gb", 0.0)
        result["metal_cache_gb"] = status.get("metal_cache_gb", 0.0)
        raw = resp.json()
        result["engine_type"] = raw.get("engine_type", "")
        result["mtp_enabled"] = bool(raw.get("mtp_enabled", False))
        result["specprefill"] = bool(raw.get("specprefill", False))
        result["kv_quant"] = raw.get("kv_quant", "") or ""
    except httpx.HTTPError:
        pass

    return result

vllm_mlx.bench_serve.scrape_metrics async 

scrape_metrics(client: AsyncClient, base_url: str) -> dict

Scrape Prometheus metrics from the server.

Parameters:

  • client (AsyncClient) –

    An open :class:httpx.AsyncClient.

  • base_url (str) –

    Base URL of the server.

Returns:

  • dict –

    Parsed metrics dict (see :func:parse_metrics_text), or an empty

  • dict –

    dict if the endpoint is unreachable.

Source code in vllm_mlx/bench_serve.py 
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
async def scrape_metrics(client: httpx.AsyncClient, base_url: str) -> dict:
    """Scrape Prometheus metrics from the server.

    Args:
        client: An open :class:`httpx.AsyncClient`.
        base_url: Base URL of the server.

    Returns:
        Parsed metrics dict (see :func:`parse_metrics_text`), or an empty
        dict if the endpoint is unreachable.
    """
    try:
        resp = await client.get(f"{base_url}/metrics")
        resp.raise_for_status()
        return parse_metrics_text(resp.text)
    except Exception:
        return {}

vllm_mlx.bench_serve.clear_runtime_cache async 

clear_runtime_cache(client: AsyncClient, base_url: str) -> dict

Clear server-side runtime caches and return a JSON-serializable event.

Source code in vllm_mlx/bench_serve.py 
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
async def clear_runtime_cache(client: httpx.AsyncClient, base_url: str) -> dict:
    """Clear server-side runtime caches and return a JSON-serializable event."""
    event: dict[str, Any] = {
        "attempted": True,
        "ok": False,
        "status_code": 0,
        "response": {},
        "error": "",
    }
    try:
        resp = await client.delete(f"{base_url}/v1/cache")
        event["status_code"] = resp.status_code
        try:
            event["response"] = resp.json()
        except ValueError:
            event["response"] = {"text": resp.text}
        resp.raise_for_status()
        event["ok"] = True
    except Exception as exc:
        event["error"] = str(exc)
    return event

vllm_mlx.bench_serve._normalize_cache_policy 

_normalize_cache_policy(value: Optional[str]) -> str

Normalize cache-policy spelling from CLI or workload JSON.

CLI choices are hyphenated, but workload JSON may use underscores when it follows common Python/YAML identifier style.

Source code in vllm_mlx/bench_serve.py 
687
688
689
690
691
692
693
694
695
696
697
698
def _normalize_cache_policy(value: Optional[str]) -> str:
    """Normalize cache-policy spelling from CLI or workload JSON.

    CLI choices are hyphenated, but workload JSON may use underscores when it
    follows common Python/YAML identifier style.
    """
    policy = (value or "preserve").strip().lower().replace("_", "-")
    if policy not in {"preserve", "before-run", "before-case"}:
        raise ValueError(
            "cache policy must be one of: preserve, before-run, before-case"
        )
    return policy

vllm_mlx.bench_serve.parse_sse_line 

parse_sse_line(line: str) -> Optional[dict]

Parse one Server-Sent Events line from a streaming chat completion.

Parameters:

  • line (str) –

    A single raw line from the SSE stream (may or may not include a trailing newline — it is stripped before processing).

Returns:

  • Optional[dict] –

    None for blank lines, comment lines (starting with :) and the

  • Optional[dict] –

    data: [DONE] sentinel. For all other data: lines the JSON is

  • Optional[dict] –

    parsed and a dict is returned::

    {"id": Optional[str], "content": str, "finish_reason": Optional[str], "usage": Optional[dict], "tool_calls_delta": Optional[list]}

  • Optional[dict] –

    Missing keys (choices, delta, content) are handled

  • Optional[dict] –

    gracefully and default to empty string / None.

Source code in vllm_mlx/bench_serve.py 
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
def parse_sse_line(line: str) -> Optional[dict]:
    """Parse one Server-Sent Events line from a streaming chat completion.

    Args:
        line: A single raw line from the SSE stream (may or may not include
            a trailing newline — it is stripped before processing).

    Returns:
        ``None`` for blank lines, comment lines (starting with ``:``) and the
        ``data: [DONE]`` sentinel.  For all other ``data:`` lines the JSON is
        parsed and a dict is returned::

            {"id": Optional[str], "content": str,
             "finish_reason": Optional[str], "usage": Optional[dict],
             "tool_calls_delta": Optional[list]}

        Missing keys (``choices``, ``delta``, ``content``) are handled
        gracefully and default to empty string / ``None``.
    """
    line = line.strip()
    if not line:
        return None
    if line.startswith(":"):
        return None
    if line == "data: [DONE]":
        return None
    if not line.startswith("data: "):
        return None

    payload = line[len("data: ") :]
    try:
        chunk = json.loads(payload)
    except json.JSONDecodeError:
        return None

    choices = chunk.get("choices") or []
    delta = choices[0].get("delta", {}) if choices else {}
    content = delta.get("content", "") or ""
    finish_reason = choices[0].get("finish_reason") if choices else None
    usage = chunk.get("usage")
    tool_calls_delta = delta.get("tool_calls")

    return {
        "id": chunk.get("id"),
        "content": content,
        "finish_reason": finish_reason,
        "usage": usage,
        "tool_calls_delta": tool_calls_delta,
    }

vllm_mlx.bench_serve._cancel_server_request async 

_cancel_server_request(client: AsyncClient, base_url: str, request_id: Optional[str]) -> None

Best-effort server-side cancellation for timed-out workload streams.

Source code in vllm_mlx/bench_serve.py 
757
758
759
760
761
762
763
764
765
766
767
768
769
770
async def _cancel_server_request(
    client: httpx.AsyncClient,
    base_url: str,
    request_id: Optional[str],
) -> None:
    """Best-effort server-side cancellation for timed-out workload streams."""
    if not request_id:
        return
    try:
        await client.post(f"{base_url}/v1/requests/{request_id}/cancel", timeout=5.0)
    except Exception:
        # Older servers may not expose request cancellation. Closing the stream
        # still gives the server disconnect signal; this endpoint is best effort.
        pass

vllm_mlx.bench_serve.accumulate_tool_calls 

accumulate_tool_calls(acc: dict[int, dict], delta_list: list[dict]) -> None

Merge streamed OpenAI tool-call deltas into acc by index.

Source code in vllm_mlx/bench_serve.py 
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
def accumulate_tool_calls(acc: dict[int, dict], delta_list: list[dict]) -> None:
    """Merge streamed OpenAI tool-call deltas into *acc* by index."""
    for tc_delta in delta_list:
        idx = int(tc_delta.get("index", 0))
        if idx not in acc:
            acc[idx] = {
                "id": tc_delta.get("id", ""),
                "type": tc_delta.get("type", "function"),
                "function": {"name": "", "arguments": ""},
            }
        entry = acc[idx]
        if tc_delta.get("id"):
            entry["id"] = tc_delta["id"]
        if tc_delta.get("type"):
            entry["type"] = tc_delta["type"]
        function_delta = tc_delta.get("function") or {}
        if function_delta.get("name"):
            entry["function"]["name"] += function_delta["name"]
        if function_delta.get("arguments"):
            entry["function"]["arguments"] += function_delta["arguments"]

vllm_mlx.bench_serve.finalize_tool_calls 

finalize_tool_calls(acc: dict[int, dict]) -> list[dict]

Return accumulated tool calls in stream index order.

Source code in vllm_mlx/bench_serve.py 
795
796
797
def finalize_tool_calls(acc: dict[int, dict]) -> list[dict]:
    """Return accumulated tool calls in stream index order."""
    return [acc[idx] for idx in sorted(acc)]

vllm_mlx.bench_serve.compute_request_metrics 

compute_request_metrics(t_start: float, t_first_token: float, token_times: list, t_end: float, prompt_tokens: int, completion_tokens: int) -> dict

Compute standard latency and throughput metrics for a single request.

All time arguments are :func:time.perf_counter values (seconds as floats).

Parameters:

  • t_start (float) –

    Timestamp immediately before the request was sent.

  • t_first_token (float) –

    Timestamp when the first content token was received.

  • token_times (list) –

    List of timestamps, one per content token (including the first). When there is only one token tpot_ms is 0.0.

  • t_end (float) –

    Timestamp after the final SSE chunk was consumed.

  • prompt_tokens (int) –

    Number of prompt tokens reported by the server.

  • completion_tokens (int) –

    Number of completion tokens generated.

Returns:

  • dict –

    Dict with keys ttft_ms, tpot_ms, e2e_latency_ms,

  • dict –

    gen_tps, prompt_tps — all floats.

Source code in vllm_mlx/bench_serve.py 
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
def compute_request_metrics(
    t_start: float,
    t_first_token: float,
    token_times: list,
    t_end: float,
    prompt_tokens: int,
    completion_tokens: int,
) -> dict:
    """Compute standard latency and throughput metrics for a single request.

    All time arguments are :func:`time.perf_counter` values (seconds as
    floats).

    Args:
        t_start: Timestamp immediately before the request was sent.
        t_first_token: Timestamp when the first content token was received.
        token_times: List of timestamps, one per content token (including the
            first).  When there is only one token ``tpot_ms`` is ``0.0``.
        t_end: Timestamp after the final SSE chunk was consumed.
        prompt_tokens: Number of prompt tokens reported by the server.
        completion_tokens: Number of completion tokens generated.

    Returns:
        Dict with keys ``ttft_ms``, ``tpot_ms``, ``e2e_latency_ms``,
        ``gen_tps``, ``prompt_tps`` — all floats.
    """
    ttft_ms = (t_first_token - t_start) * 1000.0
    e2e_latency_ms = (t_end - t_start) * 1000.0

    # TPOT: mean inter-token gap across all generated tokens.
    if len(token_times) > 1:
        intervals = [
            token_times[i] - token_times[i - 1] for i in range(1, len(token_times))
        ]
        tpot_ms = statistics.mean(intervals) * 1000.0
    else:
        tpot_ms = 0.0

    # Use last token time (not t_end which includes HTTP teardown)
    t_last_token = token_times[-1] if token_times else t_end
    gen_duration = t_last_token - t_first_token
    gen_tps = completion_tokens / gen_duration if gen_duration > 0 else 0.0

    prompt_duration = t_first_token - t_start
    prompt_tps = prompt_tokens / prompt_duration if prompt_duration > 0 else 0.0

    return {
        "ttft_ms": ttft_ms,
        "tpot_ms": tpot_ms,
        "e2e_latency_ms": e2e_latency_ms,
        "gen_tps": gen_tps,
        "prompt_tps": prompt_tps,
    }

vllm_mlx.bench_serve.count_prompt_tokens async 

count_prompt_tokens(client: AsyncClient, base_url: str, messages: list[dict], model: str) -> int

Count prompt tokens for a message list by sending a 1-token request.

Sends a non-streaming chat completion with max_tokens=1 and reads usage.prompt_tokens from the response.

Parameters:

  • client (AsyncClient) –

    An open :class:httpx.AsyncClient.

  • base_url (str) –

    Base URL of the server.

  • messages (list[dict]) –

    The message list to send.

  • model (str) –

    Model ID to target.

Returns:

  • int –

    Number of prompt tokens, or 0 on error.

Source code in vllm_mlx/bench_serve.py 
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
async def count_prompt_tokens(
    client: httpx.AsyncClient,
    base_url: str,
    messages: list[dict],
    model: str,
) -> int:
    """Count prompt tokens for a message list by sending a 1-token request.

    Sends a non-streaming chat completion with ``max_tokens=1`` and reads
    ``usage.prompt_tokens`` from the response.

    Args:
        client: An open :class:`httpx.AsyncClient`.
        base_url: Base URL of the server.
        messages: The message list to send.
        model: Model ID to target.

    Returns:
        Number of prompt tokens, or ``0`` on error.
    """
    try:
        resp = await client.post(
            f"{base_url}/v1/chat/completions",
            json={
                "model": model,
                "messages": messages,
                "max_tokens": 1,
                "stream": False,
            },
        )
        resp.raise_for_status()
        data = resp.json()
        return int((data.get("usage") or {}).get("prompt_tokens", 0))
    except Exception:
        return 0

vllm_mlx.bench_serve.stream_chat_completion async 

stream_chat_completion(client: AsyncClient, base_url: str, messages: list[dict], model: str, max_tokens: int = 256, enable_thinking: Optional[bool] = None, extra_body: Optional[dict] = None, timeout_s: Optional[float] = None) -> dict

Send a streaming chat completion and collect per-token timing data.

Tracks TTFT, per-token timestamps, accumulated content, finish reason, and usage (via stream_options: {"include_usage": True}).

Parameters:

  • client (AsyncClient) –

    An open :class:httpx.AsyncClient.

  • base_url (str) –

    Base URL of the server.

  • messages (list[dict]) –

    The message list to send.

  • model (str) –

    Model ID to target.

  • max_tokens (int, default: 256 ) –

    Maximum tokens to generate (default 256).

  • enable_thinking (Optional[bool], default: None ) –

    If not None, passed as enable_thinking in the request body.

  • extra_body (Optional[dict], default: None ) –

    Optional extra keys merged into the request body.

  • timeout_s (Optional[float], default: None ) –

    Optional case-level timeout. When set, the stream is closed and best-effort server cancellation is attempted before raising :class:TimeoutError.

Returns:

  • dict –

    Dict with all :func:compute_request_metrics fields plus

  • dict –

    completion_tokens, prompt_tokens, finish_reason,

  • dict –

    content.

Source code in vllm_mlx/bench_serve.py 
 892
 893
 894
 895
 896
 897
 898
 899
 900
 901
 902
 903
 904
 905
 906
 907
 908
 909
 910
 911
 912
 913
 914
 915
 916
 917
 918
 919
 920
 921
 922
 923
 924
 925
 926
 927
 928
 929
 930
 931
 932
 933
 934
 935
 936
 937
 938
 939
 940
 941
 942
 943
 944
 945
 946
 947
 948
 949
 950
 951
 952
 953
 954
 955
 956
 957
 958
 959
 960
 961
 962
 963
 964
 965
 966
 967
 968
 969
 970
 971
 972
 973
 974
 975
 976
 977
 978
 979
 980
 981
 982
 983
 984
 985
 986
 987
 988
 989
 990
 991
 992
 993
 994
 995
 996
 997
 998
 999
1000
1001
1002
1003
1004
1005
1006
1007
1008
1009
1010
1011
1012
async def stream_chat_completion(
    client: httpx.AsyncClient,
    base_url: str,
    messages: list[dict],
    model: str,
    max_tokens: int = 256,
    enable_thinking: Optional[bool] = None,
    extra_body: Optional[dict] = None,
    timeout_s: Optional[float] = None,
) -> dict:
    """Send a streaming chat completion and collect per-token timing data.

    Tracks TTFT, per-token timestamps, accumulated content, finish reason,
    and usage (via ``stream_options: {"include_usage": True}``).

    Args:
        client: An open :class:`httpx.AsyncClient`.
        base_url: Base URL of the server.
        messages: The message list to send.
        model: Model ID to target.
        max_tokens: Maximum tokens to generate (default ``256``).
        enable_thinking: If not ``None``, passed as ``enable_thinking`` in the
            request body.
        extra_body: Optional extra keys merged into the request body.
        timeout_s: Optional case-level timeout. When set, the stream is closed
            and best-effort server cancellation is attempted before raising
            :class:`TimeoutError`.

    Returns:
        Dict with all :func:`compute_request_metrics` fields plus
        ``completion_tokens``, ``prompt_tokens``, ``finish_reason``,
        ``content``.
    """
    body: dict = {
        "model": model,
        "messages": messages,
        "max_tokens": max_tokens,
        "stream": True,
        "stream_options": {"include_usage": True},
    }
    if enable_thinking is not None:
        body["enable_thinking"] = enable_thinking
    if extra_body:
        body.update(extra_body)

    t_start = time.perf_counter()
    t_first_token: Optional[float] = None
    token_times: list[float] = []
    content_parts: list[str] = []
    finish_reason: Optional[str] = None
    usage: Optional[dict] = None
    request_id: Optional[str] = None
    tool_calls_acc: dict[int, dict] = {}

    async def _consume_stream() -> None:
        nonlocal finish_reason, request_id, t_first_token, usage
        async with client.stream(
            "POST", f"{base_url}/v1/chat/completions", json=body
        ) as response:
            response.raise_for_status()
            async for raw_line in response.aiter_lines():
                parsed = parse_sse_line(raw_line)
                if parsed is None:
                    continue
                if parsed.get("id"):
                    request_id = parsed["id"]
                if parsed.get("usage"):
                    usage = parsed["usage"]
                if parsed.get("finish_reason"):
                    finish_reason = parsed["finish_reason"]
                tc_delta = parsed.get("tool_calls_delta")
                if tc_delta:
                    now = time.perf_counter()
                    if t_first_token is None:
                        t_first_token = now
                    token_times.append(now)
                    accumulate_tool_calls(tool_calls_acc, tc_delta)
                chunk_content = parsed.get("content", "")
                if chunk_content:
                    now = time.perf_counter()
                    if t_first_token is None:
                        t_first_token = now
                    token_times.append(now)
                    content_parts.append(chunk_content)

    try:
        if timeout_s and timeout_s > 0:
            async with asyncio.timeout(timeout_s):
                await _consume_stream()
        else:
            await _consume_stream()
    except TimeoutError:
        await _cancel_server_request(client, base_url, request_id)
        raise TimeoutError(
            f"stream_chat_completion timed out after {timeout_s:.3f}s"
        ) from None

    t_end = time.perf_counter()
    if t_first_token is None:
        t_first_token = t_end

    prompt_tokens = int((usage or {}).get("prompt_tokens", 0))
    completion_tokens = int((usage or {}).get("completion_tokens", 0))

    metrics = compute_request_metrics(
        t_start=t_start,
        t_first_token=t_first_token,
        token_times=token_times,
        t_end=t_end,
        prompt_tokens=prompt_tokens,
        completion_tokens=completion_tokens,
    )

    return {
        **metrics,
        "completion_tokens": completion_tokens,
        "prompt_tokens": prompt_tokens,
        "finish_reason": finish_reason,
        "content": "".join(content_parts),
        "tool_calls": finalize_tool_calls(tool_calls_acc),
    }

vllm_mlx.bench_serve.validate_response 

validate_response(finish_reason: Optional[str], content: str, status_code: int, *, tool_calls: Optional[list[dict]] = None) -> tuple[bool, str]

Validate a single streaming response result.

Parameters:

  • finish_reason (Optional[str]) –

    The finish_reason from the final SSE chunk, or None if not received.

  • content (str) –

    The accumulated text content of the response.

  • status_code (int) –

    The HTTP status code of the response (use 200 for successful streaming requests).

Returns:

  • bool –

    (is_valid, message) — is_valid is True when the response

  • str –

    passes all checks; message is an empty string on success or a

  • tuple[bool, str] –

    human-readable description of the first failure.

Source code in vllm_mlx/bench_serve.py 
1020
1021
1022
1023
1024
1025
1026
1027
1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
1039
1040
1041
1042
1043
1044
1045
1046
1047
1048
1049
def validate_response(
    finish_reason: Optional[str],
    content: str,
    status_code: int,
    *,
    tool_calls: Optional[list[dict]] = None,
) -> tuple[bool, str]:
    """Validate a single streaming response result.

    Args:
        finish_reason: The ``finish_reason`` from the final SSE chunk, or
            ``None`` if not received.
        content: The accumulated text content of the response.
        status_code: The HTTP status code of the response (use ``200`` for
            successful streaming requests).

    Returns:
        ``(is_valid, message)`` — ``is_valid`` is ``True`` when the response
        passes all checks; ``message`` is an empty string on success or a
        human-readable description of the first failure.
    """
    if status_code >= 400:
        return (False, f"HTTP error {status_code}")
    if finish_reason is None:
        return (False, "Missing finish_reason")
    if finish_reason == "length":
        return (False, "Truncated (finish_reason=length)")
    if not content and not tool_calls:
        return (False, "Empty response content")
    return (True, "")

vllm_mlx.bench_serve._check_finish_reason 

_check_finish_reason(allowed: Any, finish_reason: Optional[str]) -> list[str]

Verify finish_reason is in the allowed set, if one is configured.

Source code in vllm_mlx/bench_serve.py 
1052
1053
1054
1055
1056
1057
1058
1059
def _check_finish_reason(allowed: Any, finish_reason: Optional[str]) -> list[str]:
    """Verify ``finish_reason`` is in the allowed set, if one is configured."""
    if allowed is None:
        return []
    allowed_list = [allowed] if isinstance(allowed, str) else list(allowed)
    if finish_reason in allowed_list:
        return []
    return [f"finish_reason {finish_reason!r} not in allowed set {allowed_list!r}"]

vllm_mlx.bench_serve._check_length_bounds 

_check_length_bounds(min_chars: Any, max_chars: Any, content: str) -> list[str]

Apply min_chars / max_chars content-length bounds.

Source code in vllm_mlx/bench_serve.py 
1062
1063
1064
1065
1066
1067
1068
1069
def _check_length_bounds(min_chars: Any, max_chars: Any, content: str) -> list[str]:
    """Apply ``min_chars`` / ``max_chars`` content-length bounds."""
    issues: list[str] = []
    if min_chars is not None and len(content) < int(min_chars):
        issues.append(f"content shorter than min_chars={min_chars}")
    if max_chars is not None and len(content) > int(max_chars):
        issues.append(f"content longer than max_chars={max_chars}")
    return issues

vllm_mlx.bench_serve._check_regex_patterns 

_check_regex_patterns(patterns: Any, content: str, *, kind: str, expect_match: bool) -> list[str]

Validate that each pattern either matches or does not, per expect_match.

kind is the diagnostic name ("required_regex" or "forbidden_regex") and is reused across the resulting issue strings so operators can grep for the failing check.

Source code in vllm_mlx/bench_serve.py 
1072
1073
1074
1075
1076
1077
1078
1079
1080
1081
1082
1083
1084
1085
1086
1087
1088
1089
1090
1091
1092
1093
1094
1095
1096
def _check_regex_patterns(
    patterns: Any,
    content: str,
    *,
    kind: str,
    expect_match: bool,
) -> list[str]:
    """Validate that each pattern either matches or does not, per ``expect_match``.

    ``kind`` is the diagnostic name (``"required_regex"`` or
    ``"forbidden_regex"``) and is reused across the resulting issue
    strings so operators can grep for the failing check.
    """
    issues: list[str] = []
    for pattern in patterns or []:
        try:
            matched = bool(re.search(str(pattern), content, re.MULTILINE))
        except re.error as exc:
            issues.append(f"invalid {kind} {pattern!r}: {exc}")
            continue
        if expect_match and not matched:
            issues.append(f"{kind} did not match: {pattern}")
        elif not expect_match and matched:
            issues.append(f"{kind} matched: {pattern}")
    return issues

vllm_mlx.bench_serve._check_json_content 

_check_json_content(should_be_json: Any, content: str) -> list[str]

Verify content parses as JSON when checks['json'] is truthy.

Source code in vllm_mlx/bench_serve.py 
1099
1100
1101
1102
1103
1104
1105
1106
1107
def _check_json_content(should_be_json: Any, content: str) -> list[str]:
    """Verify ``content`` parses as JSON when ``checks['json']`` is truthy."""
    if not should_be_json:
        return []
    try:
        json.loads(content)
    except json.JSONDecodeError as exc:
        return [f"content is not valid JSON: {exc}"]
    return []

vllm_mlx.bench_serve._check_tool_call_count_and_names 

_check_tool_call_count_and_names(checks: dict, tool_calls: list[dict]) -> list[str]

Apply no_tool_calls / tool_call_count / tool_call_names.

Source code in vllm_mlx/bench_serve.py 
1110
1111
1112
1113
1114
1115
1116
1117
1118
1119
1120
1121
1122
1123
1124
1125
1126
1127
1128
1129
1130
1131
1132
def _check_tool_call_count_and_names(checks: dict, tool_calls: list[dict]) -> list[str]:
    """Apply ``no_tool_calls`` / ``tool_call_count`` / ``tool_call_names``."""
    issues: list[str] = []
    if checks.get("no_tool_calls") and tool_calls:
        issues.append(f"no_tool_calls: expected 0 tool calls, got {len(tool_calls)}")

    expected_count = checks.get("tool_call_count")
    if expected_count is not None and len(tool_calls) != int(expected_count):
        issues.append(
            f"tool_call_count: expected {expected_count}, got {len(tool_calls)}"
        )

    expected_names = checks.get("tool_call_names")
    if expected_names is not None:
        actual_names = sorted(
            tc.get("function", {}).get("name", "") for tc in tool_calls
        )
        expected_sorted = sorted(str(name) for name in expected_names)
        if actual_names != expected_sorted:
            issues.append(
                f"tool_call_names: expected {expected_sorted!r}, got {actual_names!r}"
            )
    return issues

vllm_mlx.bench_serve._check_tool_call_args 

_check_tool_call_args(required_args: Any, tool_calls: list[dict]) -> list[str]

Validate parsed JSON arguments include the required keys per function.

For each named function, looks up matching tool calls, parses their arguments as JSON, and reports issues for: missing tool call, invalid JSON, non-object arguments, or missing required keys.

Source code in vllm_mlx/bench_serve.py 
1135
1136
1137
1138
1139
1140
1141
1142
1143
1144
1145
1146
1147
1148
1149
1150
1151
1152
1153
1154
1155
1156
1157
1158
1159
1160
1161
1162
1163
1164
1165
1166
1167
1168
1169
1170
1171
1172
1173
1174
def _check_tool_call_args(required_args: Any, tool_calls: list[dict]) -> list[str]:
    """Validate parsed JSON arguments include the required keys per function.

    For each named function, looks up matching tool calls, parses their
    ``arguments`` as JSON, and reports issues for: missing tool call,
    invalid JSON, non-object arguments, or missing required keys.
    """
    required_args = required_args or {}
    if not required_args:
        return []
    issues: list[str] = []
    by_name: dict[str, list[dict]] = {}
    for tc in tool_calls:
        name = tc.get("function", {}).get("name", "")
        by_name.setdefault(name, []).append(tc)
    for name, required_keys in required_args.items():
        matches = by_name.get(str(name), [])
        if not matches:
            issues.append(f"tool_call_args_required_keys: no tool call named {name!r}")
            continue
        for tc in matches:
            raw_args = tc.get("function", {}).get("arguments", "")
            try:
                parsed_args = json.loads(raw_args or "{}")
            except json.JSONDecodeError as exc:
                issues.append(
                    f"tool_call_args_required_keys: {name} arguments invalid JSON: {exc}"
                )
                continue
            if not isinstance(parsed_args, dict):
                issues.append(
                    f"tool_call_args_required_keys: {name} arguments not an object"
                )
                continue
            missing = [key for key in required_keys if key not in parsed_args]
            if missing:
                issues.append(
                    f"tool_call_args_required_keys: {name} missing keys {missing!r}"
                )
    return issues

vllm_mlx.bench_serve.validate_quality_checks 

validate_quality_checks(finish_reason: Optional[str], content: str, checks: Optional[dict], *, status_code: int = 200, tool_calls: Optional[list[dict]] = None) -> tuple[bool, list[str]]

Validate content against generic workload quality checks.

Supported checks: - finish_reason: string or list of allowed finish reasons - required_regex: list of regex patterns that must match - forbidden_regex: list of regex patterns that must not match - min_chars / max_chars: length bounds - json: when true, content must parse as JSON - tool_call_count: exact number of streamed tool calls - tool_call_names: expected function names, order-independent - tool_call_args_required_keys: required JSON argument keys by function - no_tool_calls: assert that no tool calls were emitted

Source code in vllm_mlx/bench_serve.py 
1177
1178
1179
1180
1181
1182
1183
1184
1185
1186
1187
1188
1189
1190
1191
1192
1193
1194
1195
1196
1197
1198
1199
1200
1201
1202
1203
1204
1205
1206
1207
1208
1209
1210
1211
1212
1213
1214
1215
1216
1217
1218
1219
1220
1221
1222
1223
1224
1225
1226
1227
1228
1229
1230
1231
def validate_quality_checks(
    finish_reason: Optional[str],
    content: str,
    checks: Optional[dict],
    *,
    status_code: int = 200,
    tool_calls: Optional[list[dict]] = None,
) -> tuple[bool, list[str]]:
    """Validate content against generic workload quality checks.

    Supported checks:
    - ``finish_reason``: string or list of allowed finish reasons
    - ``required_regex``: list of regex patterns that must match
    - ``forbidden_regex``: list of regex patterns that must not match
    - ``min_chars`` / ``max_chars``: length bounds
    - ``json``: when true, content must parse as JSON
    - ``tool_call_count``: exact number of streamed tool calls
    - ``tool_call_names``: expected function names, order-independent
    - ``tool_call_args_required_keys``: required JSON argument keys by function
    - ``no_tool_calls``: assert that no tool calls were emitted
    """
    basic_ok, basic_issue = validate_response(
        finish_reason, content, status_code, tool_calls=tool_calls
    )
    issues: list[str] = [] if basic_ok else [basic_issue]
    checks = checks or {}
    tool_calls = tool_calls or []

    issues.extend(_check_finish_reason(checks.get("finish_reason"), finish_reason))
    issues.extend(
        _check_length_bounds(checks.get("min_chars"), checks.get("max_chars"), content)
    )
    issues.extend(
        _check_regex_patterns(
            checks.get("required_regex"),
            content,
            kind="required_regex",
            expect_match=True,
        )
    )
    issues.extend(
        _check_regex_patterns(
            checks.get("forbidden_regex"),
            content,
            kind="forbidden_regex",
            expect_match=False,
        )
    )
    issues.extend(_check_json_content(checks.get("json"), content))
    issues.extend(_check_tool_call_count_and_names(checks, tool_calls))
    issues.extend(
        _check_tool_call_args(checks.get("tool_call_args_required_keys"), tool_calls)
    )

    return (not issues, issues)

vllm_mlx.bench_serve.compute_summary_stats 

compute_summary_stats(values: list[float]) -> dict

Compute summary statistics over a list of floats.

Parameters:

  • values (list[float]) –

    Non-empty list of floats to summarise.

Returns:

  • dict –

    Dict with keys mean, stddev, min, max, p50,

  • dict –

    p95, p99. Percentiles use linear interpolation on sorted

  • dict –

    values.

Raises:

  • ValueError –

    If values is empty.

Source code in vllm_mlx/bench_serve.py 
1234
1235
1236
1237
1238
1239
1240
1241
1242
1243
1244
1245
1246
1247
1248
1249
1250
1251
1252
1253
1254
1255
1256
1257
1258
1259
1260
1261
1262
1263
1264
1265
1266
1267
1268
1269
1270
1271
1272
1273
1274
1275
1276
def compute_summary_stats(values: list[float]) -> dict:
    """Compute summary statistics over a list of floats.

    Args:
        values: Non-empty list of floats to summarise.

    Returns:
        Dict with keys ``mean``, ``stddev``, ``min``, ``max``, ``p50``,
        ``p95``, ``p99``.  Percentiles use linear interpolation on sorted
        values.

    Raises:
        ValueError: If ``values`` is empty.
    """
    if not values:
        raise ValueError("Cannot compute summary stats on empty list")

    n = len(values)
    mean = statistics.mean(values)
    stddev = 0.0 if n == 1 else statistics.stdev(values)
    sorted_vals = sorted(values)

    def _percentile(p: float) -> float:
        if n == 1:
            return sorted_vals[0]
        # Linear interpolation: index = p/100 * (n-1)
        idx = p / 100.0 * (n - 1)
        lo = int(idx)
        hi = lo + 1
        if hi >= n:
            return sorted_vals[-1]
        frac = idx - lo
        return sorted_vals[lo] + frac * (sorted_vals[hi] - sorted_vals[lo])

    return {
        "mean": mean,
        "stddev": stddev,
        "min": sorted_vals[0],
        "max": sorted_vals[-1],
        "p50": _percentile(50),
        "p95": _percentile(95),
        "p99": _percentile(99),
    }

vllm_mlx.bench_serve.run_concurrent_requests async 

run_concurrent_requests(client: AsyncClient, base_url: str, prompts: list[list[dict]], model: str, concurrency: int, max_tokens: int = 256, enable_thinking: Optional[bool] = None, extra_body: Optional[dict] = None, do_validate: bool = True) -> list[dict]

Fire concurrency concurrent streaming requests and collect results.

Prompts are selected round-robin from prompts. All requests are launched simultaneously with :func:asyncio.gather. Exceptions are caught per-task and wrapped in an error dict rather than propagated.

Parameters:

  • client (AsyncClient) –

    An open :class:httpx.AsyncClient.

  • base_url (str) –

    Base URL of the server.

  • prompts (list[list[dict]]) –

    List of message dicts to cycle through.

  • model (str) –

    Model ID to target.

  • concurrency (int) –

    Number of simultaneous requests to fire.

  • max_tokens (int, default: 256 ) –

    Maximum tokens to generate per request (default 256).

  • enable_thinking (Optional[bool], default: None ) –

    Passed through to :func:stream_chat_completion.

  • extra_body (Optional[dict], default: None ) –

    Passed through to :func:stream_chat_completion.

  • do_validate (bool, default: True ) –

    When True, call :func:validate_response on each result and add a "validated" key.

Returns:

  • list[dict] –

    List of result dicts (one per request). Each dict contains at minimum

  • list[dict] –

    a "validated" key when do_validate is True.

Source code in vllm_mlx/bench_serve.py 
1279
1280
1281
1282
1283
1284
1285
1286
1287
1288
1289
1290
1291
1292
1293
1294
1295
1296
1297
1298
1299
1300
1301
1302
1303
1304
1305
1306
1307
1308
1309
1310
1311
1312
1313
1314
1315
1316
1317
1318
1319
1320
1321
1322
1323
1324
1325
1326
1327
1328
1329
1330
1331
1332
1333
1334
1335
1336
1337
1338
1339
1340
1341
1342
async def run_concurrent_requests(
    client: httpx.AsyncClient,
    base_url: str,
    prompts: list[list[dict]],
    model: str,
    concurrency: int,
    max_tokens: int = 256,
    enable_thinking: Optional[bool] = None,
    extra_body: Optional[dict] = None,
    do_validate: bool = True,
) -> list[dict]:
    """Fire ``concurrency`` concurrent streaming requests and collect results.

    Prompts are selected round-robin from ``prompts``.  All requests are
    launched simultaneously with :func:`asyncio.gather`.  Exceptions are
    caught per-task and wrapped in an error dict rather than propagated.

    Args:
        client: An open :class:`httpx.AsyncClient`.
        base_url: Base URL of the server.
        prompts: List of message dicts to cycle through.
        model: Model ID to target.
        concurrency: Number of simultaneous requests to fire.
        max_tokens: Maximum tokens to generate per request (default ``256``).
        enable_thinking: Passed through to :func:`stream_chat_completion`.
        extra_body: Passed through to :func:`stream_chat_completion`.
        do_validate: When ``True``, call :func:`validate_response` on each
            result and add a ``"validated"`` key.

    Returns:
        List of result dicts (one per request).  Each dict contains at minimum
        a ``"validated"`` key when ``do_validate`` is ``True``.
    """
    prompt_cycle = itertools.cycle(prompts)
    selected = [next(prompt_cycle) for _ in range(concurrency)]

    async def _single(messages: list[dict]) -> dict:
        try:
            result = await stream_chat_completion(
                client=client,
                base_url=base_url,
                messages=messages,
                model=model,
                max_tokens=max_tokens,
                enable_thinking=enable_thinking,
                extra_body=extra_body,
            )
            if do_validate:
                is_valid, _ = validate_response(
                    finish_reason=result.get("finish_reason"),
                    content=result.get("content", ""),
                    status_code=200,
                )
                result["validated"] = is_valid
            return result
        except Exception as exc:
            err: dict = {
                "error": str(exc),
                "validated": False,
            }
            return err

    results = await asyncio.gather(*[_single(msg) for msg in selected])
    return list(results)

vllm_mlx.bench_serve._summary_or_empty 

_summary_or_empty(values: list[float]) -> dict
Source code in vllm_mlx/bench_serve.py 
1345
1346
def _summary_or_empty(values: list[float]) -> dict:
    return compute_summary_stats(values) if values else {}

vllm_mlx.bench_serve._resolve_max_tokens 

_resolve_max_tokens(case: WorkloadCase, workload: Workload) -> int

Return the effective max_tokens for a case, falling back to workload defaults and finally to 256.

Source code in vllm_mlx/bench_serve.py 
1349
1350
1351
1352
def _resolve_max_tokens(case: WorkloadCase, workload: Workload) -> int:
    """Return the effective ``max_tokens`` for a case, falling back to
    workload defaults and finally to 256."""
    return int(case.max_tokens or workload.defaults.get("max_tokens", 256))

vllm_mlx.bench_serve._assemble_case_request_kwargs 

_assemble_case_request_kwargs(case: WorkloadCase, workload: Workload, model: str) -> dict

Build the keyword-arguments dict passed to stream_chat_completion for one case, applying max_tokens fallback and converting policy_timeout_ms to seconds.

Source code in vllm_mlx/bench_serve.py 
1355
1356
1357
1358
1359
1360
1361
1362
1363
1364
1365
1366
1367
1368
1369
1370
1371
1372
def _assemble_case_request_kwargs(
    case: WorkloadCase, workload: Workload, model: str
) -> dict:
    """Build the keyword-arguments dict passed to ``stream_chat_completion``
    for one case, applying max_tokens fallback and converting
    ``policy_timeout_ms`` to seconds."""
    return {
        "messages": case.messages,
        "model": model,
        "max_tokens": _resolve_max_tokens(case, workload),
        "enable_thinking": case.enable_thinking,
        "extra_body": case.extra_body,
        "timeout_s": (
            case.policy_timeout_ms / 1000
            if case.policy_timeout_ms is not None
            else None
        ),
    }

vllm_mlx.bench_serve._empty_completion_result 

_empty_completion_result() -> dict

Zero-valued completion result used when stream_chat_completion raises. The structure matches a real successful response so downstream code can read result.get(...) without branching on the failure.

Source code in vllm_mlx/bench_serve.py 
1375
1376
1377
1378
1379
1380
1381
1382
1383
1384
1385
1386
1387
1388
1389
1390
def _empty_completion_result() -> dict:
    """Zero-valued completion result used when ``stream_chat_completion``
    raises. The structure matches a real successful response so downstream
    code can read ``result.get(...)`` without branching on the failure."""
    return {
        "ttft_ms": 0.0,
        "tpot_ms": 0.0,
        "e2e_latency_ms": 0.0,
        "gen_tps": 0.0,
        "prompt_tps": 0.0,
        "prompt_tokens": 0,
        "completion_tokens": 0,
        "finish_reason": None,
        "content": "",
        "tool_calls": [],
    }

vllm_mlx.bench_serve._fetch_post_run_status async 

_fetch_post_run_status(client: AsyncClient, base_url: str) -> dict

GET /v1/status after a case run, swallowing transport errors so a missing or temporarily-unavailable status endpoint does not fail the case record.

Source code in vllm_mlx/bench_serve.py 
1393
1394
1395
1396
1397
1398
1399
1400
1401
1402
async def _fetch_post_run_status(client: httpx.AsyncClient, base_url: str) -> dict:
    """GET ``/v1/status`` after a case run, swallowing transport errors so
    a missing or temporarily-unavailable status endpoint does not fail
    the case record."""
    try:
        resp = await client.get(f"{base_url}/v1/status")
        resp.raise_for_status()
        return resp.json()
    except Exception:
        return {}

vllm_mlx.bench_serve._compute_within_policy_timeout 

_compute_within_policy_timeout(timeout_ms: Optional[int], *, error_present: bool, e2e_latency_ms: float) -> Optional[bool]

Resolve the policy.within_timeout field.

None when the case did not configure a policy timeout, False when the request errored (any latency claim would be misleading), and otherwise the latency comparison result.

Source code in vllm_mlx/bench_serve.py 
1405
1406
1407
1408
1409
1410
1411
1412
1413
1414
1415
1416
1417
1418
def _compute_within_policy_timeout(
    timeout_ms: Optional[int], *, error_present: bool, e2e_latency_ms: float
) -> Optional[bool]:
    """Resolve the ``policy.within_timeout`` field.

    ``None`` when the case did not configure a policy timeout, ``False``
    when the request errored (any latency claim would be misleading),
    and otherwise the latency comparison result.
    """
    if timeout_ms is None:
        return None
    if error_present:
        return False
    return e2e_latency_ms <= timeout_ms

vllm_mlx.bench_serve._build_tool_calls_summary 

_build_tool_calls_summary(tool_calls: Any) -> Optional[dict]

Compact summary of streamed tool calls for the case record.

Returns None when no tool calls were emitted so consumers can distinguish "feature not exercised" from "feature exercised, zero calls" if that ever matters.

Source code in vllm_mlx/bench_serve.py 
1421
1422
1423
1424
1425
1426
1427
1428
1429
1430
1431
1432
1433
1434
def _build_tool_calls_summary(tool_calls: Any) -> Optional[dict]:
    """Compact summary of streamed tool calls for the case record.

    Returns ``None`` when no tool calls were emitted so consumers can
    distinguish "feature not exercised" from "feature exercised, zero
    calls" if that ever matters.
    """
    if not tool_calls:
        return None
    return {
        "count": len(tool_calls),
        "names": sorted(tc.get("function", {}).get("name", "") for tc in tool_calls),
        "raw": tool_calls,
    }

vllm_mlx.bench_serve._build_workload_record 

_build_workload_record(*, case: WorkloadCase, workload: Workload, model: str, runtime: dict, hardware: dict, run_id: str, timestamp: str, started_wall: str, repetition: int, result: dict, error: str, quality_ok: bool, quality_issues: list[str], content: str, cache_hits_delta: int, cache_misses_delta: int, tokens_saved_delta: int, status_after: dict, cache_reset: Optional[dict], include_content: bool) -> dict

Assemble the JSON-serializable workload-case record from the raw inputs and the completion result. Pure function: no I/O, deterministic given its arguments.

Source code in vllm_mlx/bench_serve.py 
1437
1438
1439
1440
1441
1442
1443
1444
1445
1446
1447
1448
1449
1450
1451
1452
1453
1454
1455
1456
1457
1458
1459
1460
1461
1462
1463
1464
1465
1466
1467
1468
1469
1470
1471
1472
1473
1474
1475
1476
1477
1478
1479
1480
1481
1482
1483
1484
1485
1486
1487
1488
1489
1490
1491
1492
1493
1494
1495
1496
1497
1498
1499
1500
1501
1502
1503
1504
1505
1506
1507
1508
1509
1510
1511
1512
1513
1514
1515
def _build_workload_record(
    *,
    case: WorkloadCase,
    workload: Workload,
    model: str,
    runtime: dict,
    hardware: dict,
    run_id: str,
    timestamp: str,
    started_wall: str,
    repetition: int,
    result: dict,
    error: str,
    quality_ok: bool,
    quality_issues: list[str],
    content: str,
    cache_hits_delta: int,
    cache_misses_delta: int,
    tokens_saved_delta: int,
    status_after: dict,
    cache_reset: Optional[dict],
    include_content: bool,
) -> dict:
    """Assemble the JSON-serializable workload-case record from the raw
    inputs and the completion result. Pure function: no I/O, deterministic
    given its arguments."""
    record = {
        "run_id": run_id,
        "timestamp": timestamp,
        "started_at": started_wall,
        "workload": workload.name,
        "case_id": case.case_id,
        "repetition": repetition,
        "tags": list(case.tags),
        "model_id": model,
        "runtime": runtime,
        "hardware": hardware,
        "request": {
            "max_tokens": _resolve_max_tokens(case, workload),
            "request_path": case.request_path,
            "enable_thinking": case.enable_thinking,
            "extra_body": case.extra_body or {},
            "message_count": len(case.messages),
        },
        "policy": {
            "timeout_ms": case.policy_timeout_ms,
            "within_timeout": _compute_within_policy_timeout(
                case.policy_timeout_ms,
                error_present=bool(error),
                e2e_latency_ms=result["e2e_latency_ms"],
            ),
        },
        "cache_reset": cache_reset or {"attempted": False},
        "metrics": {
            "ttft_ms": result["ttft_ms"],
            "tpot_ms": result["tpot_ms"],
            "e2e_latency_ms": result["e2e_latency_ms"],
            "gen_tps": result["gen_tps"],
            "prompt_tps": result["prompt_tps"],
            "prompt_tokens": result["prompt_tokens"],
            "completion_tokens": result["completion_tokens"],
            "cache_hits": cache_hits_delta,
            "cache_misses": cache_misses_delta,
            "tokens_saved": tokens_saved_delta,
            "metal": parse_status_response(status_after),
        },
        "quality": {
            "ok": quality_ok,
            "issues": quality_issues,
            "finish_reason": result.get("finish_reason"),
            "content_chars": len(content),
            "content_preview": content[:240],
        },
        "tool_calls": _build_tool_calls_summary(result.get("tool_calls") or []),
        "ok": quality_ok,
    }
    if include_content:
        record["quality"]["content"] = content
    return record

vllm_mlx.bench_serve.run_workload_case async 

run_workload_case(client: AsyncClient, base_url: str, *, workload: Workload, case: WorkloadCase, model: str, runtime: dict, hardware: dict, run_id: str, timestamp: str, repetition: int = 0, scrape: bool = True, include_content: bool = False, cache_reset: Optional[dict] = None) -> dict

Run one workload case and return a JSON-serializable result.

Source code in vllm_mlx/bench_serve.py 
1518
1519
1520
1521
1522
1523
1524
1525
1526
1527
1528
1529
1530
1531
1532
1533
1534
1535
1536
1537
1538
1539
1540
1541
1542
1543
1544
1545
1546
1547
1548
1549
1550
1551
1552
1553
1554
1555
1556
1557
1558
1559
1560
1561
1562
1563
1564
1565
1566
1567
1568
1569
1570
1571
1572
1573
1574
1575
1576
1577
1578
1579
1580
1581
1582
1583
1584
1585
1586
1587
1588
1589
1590
1591
1592
1593
async def run_workload_case(
    client: httpx.AsyncClient,
    base_url: str,
    *,
    workload: Workload,
    case: WorkloadCase,
    model: str,
    runtime: dict,
    hardware: dict,
    run_id: str,
    timestamp: str,
    repetition: int = 0,
    scrape: bool = True,
    include_content: bool = False,
    cache_reset: Optional[dict] = None,
) -> dict:
    """Run one workload case and return a JSON-serializable result."""
    metrics_before = await scrape_metrics(client, base_url) if scrape else {}
    started_wall = datetime.now(timezone.utc).isoformat()

    request_kwargs = _assemble_case_request_kwargs(case, workload, model)
    try:
        result = await stream_chat_completion(
            client=client, base_url=base_url, **request_kwargs
        )
        error = ""
    except Exception as exc:
        result = _empty_completion_result()
        error = str(exc)

    metrics_after = await scrape_metrics(client, base_url) if scrape else {}
    status_after = await _fetch_post_run_status(client, base_url)

    cache_hits_delta = metrics_after.get("cache_hits", 0) - metrics_before.get(
        "cache_hits", 0
    )
    cache_misses_delta = metrics_after.get("cache_misses", 0) - metrics_before.get(
        "cache_misses", 0
    )
    tokens_saved_delta = metrics_after.get("tokens_saved", 0) - metrics_before.get(
        "tokens_saved", 0
    )

    content = str(result.get("content") or "")
    quality_ok, quality_issues = validate_quality_checks(
        result.get("finish_reason"),
        content,
        case.checks,
        status_code=500 if error else 200,
        tool_calls=result.get("tool_calls") or [],
    )
    if error:
        quality_issues.append(f"request error: {error}")

    return _build_workload_record(
        case=case,
        workload=workload,
        model=model,
        runtime=runtime,
        hardware=hardware,
        run_id=run_id,
        timestamp=timestamp,
        started_wall=started_wall,
        repetition=repetition,
        result=result,
        error=error,
        quality_ok=quality_ok,
        quality_issues=quality_issues,
        content=content,
        cache_hits_delta=cache_hits_delta,
        cache_misses_delta=cache_misses_delta,
        tokens_saved_delta=tokens_saved_delta,
        status_after=status_after,
        cache_reset=cache_reset,
        include_content=include_content,
    )

vllm_mlx.bench_serve._group_results_by_case_id 

_group_results_by_case_id(results: list[dict]) -> dict[str, list[dict]]

Bucket workload case records by their case_id field, defaulting a missing case_id to the empty string so the grouping is stable.

Source code in vllm_mlx/bench_serve.py 
1596
1597
1598
1599
1600
1601
1602
def _group_results_by_case_id(results: list[dict]) -> dict[str, list[dict]]:
    """Bucket workload case records by their ``case_id`` field, defaulting
    a missing ``case_id`` to the empty string so the grouping is stable."""
    cases: dict[str, list[dict]] = {}
    for result in results:
        cases.setdefault(str(result.get("case_id", "")), []).append(result)
    return cases

vllm_mlx.bench_serve._summarize_case 

_summarize_case(case_results: list[dict]) -> dict

Build the per-case summary block.

Mirrors the shape used at the run-level (sample counts, pass/fail rates, policy-timeout outcome, latency / ttft / gen_tps summaries) and adds two case-only fields: sample_count and repetitions (the sorted set of repetition indices the case was run under), plus content_chars since content length is more useful per-case than per-run.

Source code in vllm_mlx/bench_serve.py 
1605
1606
1607
1608
1609
1610
1611
1612
1613
1614
1615
1616
1617
1618
1619
1620
1621
1622
1623
1624
1625
1626
1627
1628
1629
1630
1631
1632
1633
1634
1635
1636
1637
1638
1639
1640
1641
1642
1643
1644
1645
1646
1647
1648
def _summarize_case(case_results: list[dict]) -> dict:
    """Build the per-case summary block.

    Mirrors the shape used at the run-level (sample counts, pass/fail
    rates, policy-timeout outcome, latency / ttft / gen_tps summaries)
    and adds two case-only fields: ``sample_count`` and ``repetitions``
    (the sorted set of repetition indices the case was run under), plus
    ``content_chars`` since content length is more useful per-case than
    per-run.
    """
    quality_failures = [r for r in case_results if not r["quality"].get("ok")]
    policy_trials = [
        r for r in case_results if r["policy"].get("within_timeout") is not None
    ]
    policy_failures = [
        r for r in policy_trials if r["policy"].get("within_timeout") is False
    ]
    return {
        "sample_count": len(case_results),
        "repetitions": sorted(
            {
                int(r.get("repetition", 0))
                for r in case_results
                if r.get("repetition") is not None
            }
        ),
        "passed": not quality_failures,
        "failure_count": len(quality_failures),
        "failure_rate": (
            round(len(quality_failures) / len(case_results), 4) if case_results else 0.0
        ),
        "policy_timeout_passed": (not policy_failures if policy_trials else None),
        "policy_timeout_failure_count": (
            len(policy_failures) if policy_trials else None
        ),
        "latency_ms": _summary_or_empty(
            [r["metrics"]["e2e_latency_ms"] for r in case_results]
        ),
        "ttft_ms": _summary_or_empty([r["metrics"]["ttft_ms"] for r in case_results]),
        "gen_tps": _summary_or_empty([r["metrics"]["gen_tps"] for r in case_results]),
        "content_chars": _summary_or_empty(
            [r["quality"].get("content_chars", 0) for r in case_results]
        ),
    }

vllm_mlx.bench_serve.summarize_workload_results 

summarize_workload_results(results: list[dict]) -> dict

Aggregate workload case records into stable qualification summary stats.

Source code in vllm_mlx/bench_serve.py 
1651
1652
1653
1654
1655
1656
1657
1658
1659
1660
1661
1662
1663
1664
1665
1666
1667
1668
1669
1670
1671
1672
1673
1674
1675
1676
1677
1678
1679
1680
1681
1682
1683
1684
1685
1686
1687
1688
1689
def summarize_workload_results(results: list[dict]) -> dict:
    """Aggregate workload case records into stable qualification summary stats."""
    failures = [r for r in results if not r["quality"]["ok"]]
    policy_trials = [
        r for r in results if r["policy"].get("within_timeout") is not None
    ]
    policy_failures = [
        r for r in policy_trials if r["policy"].get("within_timeout") is False
    ]

    cases = _group_results_by_case_id(results)
    case_summaries = {
        case_id: _summarize_case(case_results)
        for case_id, case_results in sorted(cases.items())
    }

    return {
        "case_count": len(results),
        "unique_case_count": len(cases),
        "repetition_count": max(
            (len(summary["repetitions"]) for summary in case_summaries.values()),
            default=0,
        ),
        "passed": not failures,
        "failure_count": len(failures),
        "failure_rate": round(len(failures) / len(results), 4) if results else 0.0,
        "quality_passed": not failures,
        "quality_failure_count": len(failures),
        "policy_timeout_passed": not policy_failures if policy_trials else None,
        "policy_timeout_failure_count": (
            len(policy_failures) if policy_trials else None
        ),
        "latency_ms": _summary_or_empty(
            [r["metrics"]["e2e_latency_ms"] for r in results]
        ),
        "ttft_ms": _summary_or_empty([r["metrics"]["ttft_ms"] for r in results]),
        "gen_tps": _summary_or_empty([r["metrics"]["gen_tps"] for r in results]),
        "case_summaries": case_summaries,
    }

vllm_mlx.bench_serve.run_bench_serve_workload async 

run_bench_serve_workload(*, url: str, workload_path: str, model: Optional[str] = None, output_path: Optional[str] = None, output_format: str = 'json', scrape: bool = True, include_content: bool = False, request_timeout_s: Optional[float] = 300.0, repetitions: int = 1, cache_policy: Optional[str] = None) -> dict

Run a declarative workload against a running server.

This is the contract-style counterpart to prompt sweeps: it keeps product policy knobs in the manifest, records them as evidence, and measures what the server actually does before anyone promotes a model or feature stack.

Source code in vllm_mlx/bench_serve.py 
1692
1693
1694
1695
1696
1697
1698
1699
1700
1701
1702
1703
1704
1705
1706
1707
1708
1709
1710
1711
1712
1713
1714
1715
1716
1717
1718
1719
1720
1721
1722
1723
1724
1725
1726
1727
1728
1729
1730
1731
1732
1733
1734
1735
1736
1737
1738
1739
1740
1741
1742
1743
1744
1745
1746
1747
1748
1749
1750
1751
1752
1753
1754
1755
1756
1757
1758
1759
1760
1761
1762
1763
1764
1765
1766
1767
1768
1769
1770
1771
1772
1773
1774
1775
1776
1777
1778
1779
1780
1781
1782
1783
1784
1785
1786
1787
1788
1789
1790
1791
1792
1793
1794
1795
1796
1797
1798
1799
1800
1801
1802
1803
1804
1805
1806
1807
1808
1809
1810
1811
1812
1813
1814
1815
1816
1817
1818
async def run_bench_serve_workload(
    *,
    url: str,
    workload_path: str,
    model: Optional[str] = None,
    output_path: Optional[str] = None,
    output_format: str = "json",
    scrape: bool = True,
    include_content: bool = False,
    request_timeout_s: Optional[float] = 300.0,
    repetitions: int = 1,
    cache_policy: Optional[str] = None,
) -> dict:
    """Run a declarative workload against a running server.

    This is the contract-style counterpart to prompt sweeps: it keeps product
    policy knobs in the manifest, records them as evidence, and measures what
    the server actually does before anyone promotes a model or feature stack.
    """
    if repetitions < 1:
        raise ValueError("repetitions must be at least 1")

    workload = load_workload(workload_path)
    resolved_cache_policy = _normalize_cache_policy(
        cache_policy or workload.defaults.get("cache_policy")
    )
    run_id = str(uuid.uuid4())[:8]
    timestamp = datetime.now(timezone.utc).isoformat()
    timeout = httpx.Timeout(request_timeout_s) if request_timeout_s else None

    async with httpx.AsyncClient(timeout=timeout) as client:
        runtime = await auto_detect_runtime(client, url)
        hardware = detect_hardware_fingerprint()
        model_id = model or runtime.get("model_id", "")
        if not model_id:
            raise ValueError("could not determine model ID; pass --model")

        records = []
        cache_events = []
        if resolved_cache_policy == "before-run":
            cache_events.append(
                {
                    "scope": "before-run",
                    "event": await clear_runtime_cache(client, url),
                }
            )
        total_cases = len(workload.cases) * repetitions
        completed = 0
        for repetition in range(repetitions):
            for case in workload.cases:
                print(
                    f"  [{repetition + 1}/{repetitions}] {case.case_id} ...",
                    end="",
                    flush=True,
                )
                cache_reset = None
                if resolved_cache_policy == "before-case":
                    cache_reset = await clear_runtime_cache(client, url)
                    cache_events.append(
                        {
                            "scope": "before-case",
                            "case_id": case.case_id,
                            "repetition": repetition,
                            "event": cache_reset,
                        }
                    )
                record = await run_workload_case(
                    client,
                    url,
                    workload=workload,
                    case=case,
                    model=model_id,
                    runtime=runtime,
                    hardware=hardware,
                    run_id=run_id,
                    timestamp=timestamp,
                    repetition=repetition,
                    scrape=scrape,
                    include_content=include_content,
                    cache_reset=cache_reset,
                )
                records.append(record)
                completed += 1
                latency = record.get("metrics", {}).get("e2e_latency_ms", 0)
                ok = record.get("ok", False)
                status = "ok" if ok else "FAIL"
                print(f" {latency:.0f}ms {status} ({completed}/{total_cases})")

    payload = {
        "run_id": run_id,
        "timestamp": timestamp,
        "workload": {
            "name": workload.name,
            "description": workload.description,
            "path": str(Path(workload_path).expanduser()),
            "defaults": workload.defaults,
            "repetitions": repetitions,
        },
        "transport": {
            "request_timeout_s": request_timeout_s,
            "note": "transport safety only; product policy timeouts live in workload cases",
        },
        "policy": {
            "note": "comparison-only unless your product contract explicitly requires it",
        },
        "cache_policy": {
            "mode": resolved_cache_policy,
            "events": cache_events,
        },
        "summary": summarize_workload_results(records),
        "results": records,
    }

    if output_format == "sqlite":
        if not output_path:
            raise ValueError("--output is required when --format sqlite")
        write_workload_sqlite(payload, output_path)
        print(f"Workload SQLite results written to {output_path}")
        return payload

    rendered = format_workload_payload(payload, output_format)
    if output_path:
        Path(output_path).expanduser().write_text(rendered)
        print(f"Workload results written to {output_path}")
    else:
        print(rendered)
    return payload

vllm_mlx.bench_serve._result_to_dict 

_result_to_dict(r: BenchServeResult) -> dict

Convert a :class:BenchServeResult to an ordered dict.

Returns an OrderedDict-style plain dict whose keys follow the dataclass field declaration order (as listed in :data:RESULT_COLUMNS).

Source code in vllm_mlx/bench_serve.py 
1840
1841
1842
1843
1844
1845
1846
def _result_to_dict(r: BenchServeResult) -> dict:
    """Convert a :class:`BenchServeResult` to an ordered dict.

    Returns an ``OrderedDict``-style plain ``dict`` whose keys follow the
    dataclass field declaration order (as listed in :data:`RESULT_COLUMNS`).
    """
    return {f.name: getattr(r, f.name) for f in _dataclasses.fields(r)}

vllm_mlx.bench_serve.format_table 

format_table(results: list[BenchServeResult]) -> str

Render a human-readable terminal table of benchmark results.

Only the columns in :data:_TABLE_COLUMNS are shown. Float values are rounded to one decimal place.

Parameters:

  • results (list[BenchServeResult]) –

    List of :class:BenchServeResult instances.

Returns:

  • str –

    Formatted string using tabulate with tablefmt="simple".

Source code in vllm_mlx/bench_serve.py 
1849
1850
1851
1852
1853
1854
1855
1856
1857
1858
1859
1860
1861
1862
1863
1864
1865
1866
1867
1868
1869
1870
1871
def format_table(results: list[BenchServeResult]) -> str:
    """Render a human-readable terminal table of benchmark results.

    Only the columns in :data:`_TABLE_COLUMNS` are shown.  Float values are
    rounded to one decimal place.

    Args:
        results: List of :class:`BenchServeResult` instances.

    Returns:
        Formatted string using ``tabulate`` with ``tablefmt="simple"``.
    """
    rows = []
    for r in results:
        d = _result_to_dict(r)
        row = []
        for col in _TABLE_COLUMNS:
            val = d.get(col)
            if isinstance(val, float):
                val = round(val, 1)
            row.append(val)
        rows.append(row)
    return _tabulate(rows, headers=_TABLE_COLUMNS, tablefmt="simple")

vllm_mlx.bench_serve.format_json 

format_json(results: list[BenchServeResult]) -> str

Serialize benchmark results as a JSON array.

All fields from :data:RESULT_COLUMNS are included.

Parameters:

  • results (list[BenchServeResult]) –

    List of :class:BenchServeResult instances.

Returns:

  • str –

    JSON string with indent=2.

Source code in vllm_mlx/bench_serve.py 
1874
1875
1876
1877
1878
1879
1880
1881
1882
1883
1884
1885
def format_json(results: list[BenchServeResult]) -> str:
    """Serialize benchmark results as a JSON array.

    All fields from :data:`RESULT_COLUMNS` are included.

    Args:
        results: List of :class:`BenchServeResult` instances.

    Returns:
        JSON string with ``indent=2``.
    """
    return json.dumps([_result_to_dict(r) for r in results], indent=2)

vllm_mlx.bench_serve.format_csv 

format_csv(results: list[BenchServeResult]) -> str

Serialize benchmark results as CSV with a header row.

All columns are included.

Parameters:

  • results (list[BenchServeResult]) –

    List of :class:BenchServeResult instances.

Returns:

  • str –

    CSV string (header + one row per result).

Source code in vllm_mlx/bench_serve.py 
1888
1889
1890
1891
1892
1893
1894
1895
1896
1897
1898
1899
1900
1901
1902
1903
1904
def format_csv(results: list[BenchServeResult]) -> str:
    """Serialize benchmark results as CSV with a header row.

    All columns are included.

    Args:
        results: List of :class:`BenchServeResult` instances.

    Returns:
        CSV string (header + one row per result).
    """
    buf = io.StringIO()
    writer = csv_mod.DictWriter(buf, fieldnames=RESULT_COLUMNS)
    writer.writeheader()
    for r in results:
        writer.writerow(_result_to_dict(r))
    return buf.getvalue()

vllm_mlx.bench_serve._sql_escape 

_sql_escape(value) -> str

Escape a Python value for use as a SQL literal.

  • None -> "NULL"
  • bool -> "1" or "0"
  • int / float -> string representation
  • str -> single-quoted with internal single-quotes doubled
Source code in vllm_mlx/bench_serve.py 
1907
1908
1909
1910
1911
1912
1913
1914
1915
1916
1917
1918
1919
1920
1921
1922
1923
1924
1925
1926
1927
def _sql_escape(value) -> str:
    """Escape a Python value for use as a SQL literal.

    - ``None`` -> ``"NULL"``
    - ``bool`` -> ``"1"`` or ``"0"``
    - ``int`` / ``float`` -> string representation
    - ``str`` -> single-quoted with internal single-quotes doubled
    """
    if value is None:
        return "NULL"
    if isinstance(value, bool):
        return "1" if value else "0"
    if isinstance(value, float):
        if math.isnan(value) or math.isinf(value):
            return "NULL"
        return str(value)
    if isinstance(value, int):
        return str(value)
    # str
    escaped = str(value).replace("'", "''")
    return f"'{escaped}'"

vllm_mlx.bench_serve.format_sql 

format_sql(results: list[BenchServeResult]) -> str

Emit a SQL CREATE TABLE IF NOT EXISTS statement and INSERT rows.

The schema follows the exact column order defined in the bench-serve spec.

Parameters:

  • results (list[BenchServeResult]) –

    List of :class:BenchServeResult instances.

Returns:

  • str –

    SQL string containing the CREATE TABLE statement followed by one

  • str –

    INSERT statement per result.

Source code in vllm_mlx/bench_serve.py 
1945
1946
1947
1948
1949
1950
1951
1952
1953
1954
1955
1956
1957
1958
1959
1960
1961
1962
1963
1964
def format_sql(results: list[BenchServeResult]) -> str:
    """Emit a SQL ``CREATE TABLE IF NOT EXISTS`` statement and INSERT rows.

    The schema follows the exact column order defined in the bench-serve spec.

    Args:
        results: List of :class:`BenchServeResult` instances.

    Returns:
        SQL string containing the CREATE TABLE statement followed by one
        INSERT statement per result.
    """
    lines = [
        f"CREATE TABLE IF NOT EXISTS bench_serve ({_SQL_SCHEMA});",
    ]
    for r in results:
        d = _result_to_dict(r)
        values = ", ".join(_sql_escape(d[col]) for col in RESULT_COLUMNS)
        lines.append(f"INSERT INTO bench_serve VALUES ({values});")
    return "\n".join(lines)

vllm_mlx.bench_serve._write_sqlite_rows 

_write_sqlite_rows(output_path: str, *, table: str, schema: str, columns: list[str], rows: list[dict]) -> None

Append benchmark rows to a SQLite database.

Source code in vllm_mlx/bench_serve.py 
1967
1968
1969
1970
1971
1972
1973
1974
1975
1976
1977
1978
1979
1980
1981
1982
1983
1984
1985
1986
1987
1988
1989
1990
def _write_sqlite_rows(
    output_path: str,
    *,
    table: str,
    schema: str,
    columns: list[str],
    rows: list[dict],
) -> None:
    """Append benchmark rows to a SQLite database."""
    db_path = Path(output_path).expanduser()
    _validate_sql_identifier(table, kind="table")
    for column in columns:
        _validate_sql_identifier(column, kind="column")
    placeholders = ", ".join("?" for _ in columns)
    column_list = ", ".join(columns)
    values = [[row.get(col) for col in columns] for row in rows]
    with sqlite3.connect(db_path) as conn:
        conn.execute(f"CREATE TABLE IF NOT EXISTS {table} ({schema})")
        if values:
            conn.executemany(
                f"INSERT INTO {table} ({column_list}) VALUES ({placeholders})",
                values,
            )
        conn.commit()

vllm_mlx.bench_serve._validate_sql_identifier 

_validate_sql_identifier(identifier: str, *, kind: str) -> None

Reject unsafe SQL identifiers before string interpolation.

Source code in vllm_mlx/bench_serve.py 
1993
1994
1995
1996
def _validate_sql_identifier(identifier: str, *, kind: str) -> None:
    """Reject unsafe SQL identifiers before string interpolation."""
    if not _SQL_IDENTIFIER_RE.fullmatch(identifier):
        raise ValueError(f"invalid SQLite {kind} identifier: {identifier!r}")

vllm_mlx.bench_serve.write_sqlite 

write_sqlite(results: list[BenchServeResult], output_path: str) -> None

Append prompt-sweep benchmark results to a SQLite database.

Source code in vllm_mlx/bench_serve.py 
1999
2000
2001
2002
2003
2004
2005
2006
2007
2008
2009
def write_sqlite(results: list[BenchServeResult], output_path: str) -> None:
    """Append prompt-sweep benchmark results to a SQLite database."""

    rows = [_result_to_dict(r) for r in results]
    _write_sqlite_rows(
        output_path,
        table="bench_serve",
        schema=_SQL_SCHEMA,
        columns=RESULT_COLUMNS,
        rows=rows,
    )

vllm_mlx.bench_serve._workload_record_to_row 

_workload_record_to_row(record: dict) -> dict
Source code in vllm_mlx/bench_serve.py 
2069
2070
2071
2072
2073
2074
2075
2076
2077
2078
2079
2080
2081
2082
2083
2084
2085
2086
2087
2088
2089
2090
2091
2092
2093
2094
2095
2096
2097
2098
2099
2100
2101
2102
2103
2104
2105
2106
2107
2108
2109
2110
2111
2112
2113
2114
2115
2116
2117
2118
2119
def _workload_record_to_row(record: dict) -> dict:
    runtime = record.get("runtime") or {}
    hardware = record.get("hardware") or {}
    request = record.get("request") or {}
    policy = record.get("policy") or {}
    metrics = record.get("metrics") or {}
    metal = metrics.get("metal") or {}
    quality = record.get("quality") or {}
    return {
        "run_id": record.get("run_id", ""),
        "timestamp": record.get("timestamp", ""),
        "workload": record.get("workload", ""),
        "case_id": record.get("case_id", ""),
        "repetition": record.get("repetition", 0),
        "tags": ",".join(record.get("tags") or []),
        "model_id": record.get("model_id", ""),
        "chip": hardware.get("chip", ""),
        "memory_gb": hardware.get("memory_gb", 0.0),
        "os_version": hardware.get("os_version", ""),
        "engine_type": runtime.get("engine_type", ""),
        "model_type": runtime.get("model_type", ""),
        "mtp_enabled": runtime.get("mtp_enabled", False),
        "specprefill": runtime.get("specprefill", False),
        "kv_quant": runtime.get("kv_quant", ""),
        "cache_type": runtime.get("cache_type", ""),
        "request_max_tokens": request.get("max_tokens"),
        "request_enable_thinking": request.get("enable_thinking"),
        "request_extra_body": json.dumps(
            request.get("extra_body") or {}, sort_keys=True
        ),
        "policy_timeout_ms": policy.get("timeout_ms"),
        "within_policy_timeout": policy.get("within_timeout"),
        "ttft_ms": metrics.get("ttft_ms", 0.0),
        "tpot_ms": metrics.get("tpot_ms", 0.0),
        "e2e_latency_ms": metrics.get("e2e_latency_ms", 0.0),
        "gen_tps": metrics.get("gen_tps", 0.0),
        "prompt_tps": metrics.get("prompt_tps", 0.0),
        "prompt_tokens": metrics.get("prompt_tokens", 0),
        "completion_tokens": metrics.get("completion_tokens", 0),
        "cache_hits": metrics.get("cache_hits", 0),
        "cache_misses": metrics.get("cache_misses", 0),
        "tokens_saved": metrics.get("tokens_saved", 0),
        "metal_active_gb": metal.get("metal_active_gb", 0.0),
        "metal_peak_gb": metal.get("metal_peak_gb", 0.0),
        "metal_cache_gb": metal.get("metal_cache_gb", 0.0),
        "quality_ok": quality.get("ok", False),
        "quality_issues": json.dumps(quality.get("issues") or []),
        "finish_reason": quality.get("finish_reason"),
        "content_chars": quality.get("content_chars", 0),
        "content_preview": quality.get("content_preview", ""),
    }

vllm_mlx.bench_serve.format_workload_table 

format_workload_table(payload: dict) -> str

Format workload result records as a compact human-readable table.

Source code in vllm_mlx/bench_serve.py 
2122
2123
2124
2125
2126
2127
2128
2129
2130
2131
2132
2133
2134
def format_workload_table(payload: dict) -> str:
    """Format workload result records as a compact human-readable table."""

    rows = []
    for record in payload.get("results") or []:
        row = _workload_record_to_row(record)
        rows.append(
            [
                round(value, 1) if isinstance(value, float) else value
                for value in (row[col] for col in _WORKLOAD_TABLE_COLUMNS)
            ]
        )
    return _tabulate(rows, headers=_WORKLOAD_TABLE_COLUMNS, tablefmt="simple")

vllm_mlx.bench_serve.format_workload_json 

format_workload_json(payload: dict) -> str

Serialize a workload result payload as indented JSON.

Source code in vllm_mlx/bench_serve.py 
2137
2138
2139
2140
def format_workload_json(payload: dict) -> str:
    """Serialize a workload result payload as indented JSON."""

    return json.dumps(payload, indent=2)

vllm_mlx.bench_serve.format_workload_csv 

format_workload_csv(payload: dict) -> str

Serialize workload result records with the stable CSV column contract.

Source code in vllm_mlx/bench_serve.py 
2143
2144
2145
2146
2147
2148
2149
2150
2151
def format_workload_csv(payload: dict) -> str:
    """Serialize workload result records with the stable CSV column contract."""

    buf = io.StringIO()
    writer = csv_mod.DictWriter(buf, fieldnames=WORKLOAD_RESULT_COLUMNS)
    writer.writeheader()
    for record in payload.get("results") or []:
        writer.writerow(_workload_record_to_row(record))
    return buf.getvalue()

vllm_mlx.bench_serve.format_workload_sql 

format_workload_sql(payload: dict) -> str

Render SQL statements that create and populate the workload table.

Source code in vllm_mlx/bench_serve.py 
2170
2171
2172
2173
2174
2175
2176
2177
2178
2179
2180
def format_workload_sql(payload: dict) -> str:
    """Render SQL statements that create and populate the workload table."""

    lines = [
        f"CREATE TABLE IF NOT EXISTS bench_serve_workload ({_WORKLOAD_SQL_SCHEMA});",
    ]
    for record in payload.get("results") or []:
        row = _workload_record_to_row(record)
        values = ", ".join(_sql_escape(row[col]) for col in WORKLOAD_RESULT_COLUMNS)
        lines.append(f"INSERT INTO bench_serve_workload VALUES ({values});")
    return "\n".join(lines)

vllm_mlx.bench_serve.write_workload_sqlite 

write_workload_sqlite(payload: dict, output_path: str) -> None

Append workload result records to a SQLite database.

Source code in vllm_mlx/bench_serve.py 
2183
2184
2185
2186
2187
2188
2189
2190
2191
2192
2193
def write_workload_sqlite(payload: dict, output_path: str) -> None:
    """Append workload result records to a SQLite database."""

    rows = [_workload_record_to_row(record) for record in payload.get("results") or []]
    _write_sqlite_rows(
        output_path,
        table="bench_serve_workload",
        schema=_WORKLOAD_SQL_SCHEMA,
        columns=WORKLOAD_RESULT_COLUMNS,
        rows=rows,
    )

vllm_mlx.bench_serve.format_workload_payload 

format_workload_payload(payload: dict, fmt: str = 'json') -> str

Serialize a workload payload in the requested text output format.

Raises:

  • ValueError –

    If fmt is not json, csv, sql, or table.

Source code in vllm_mlx/bench_serve.py 
2196
2197
2198
2199
2200
2201
2202
2203
2204
2205
2206
2207
2208
2209
2210
2211
def format_workload_payload(payload: dict, fmt: str = "json") -> str:
    """Serialize a workload payload in the requested text output format.

    Raises:
        ValueError: If ``fmt`` is not ``json``, ``csv``, ``sql``, or ``table``.
    """

    if fmt == "json":
        return format_workload_json(payload)
    if fmt == "csv":
        return format_workload_csv(payload)
    if fmt == "sql":
        return format_workload_sql(payload)
    if fmt == "table":
        return format_workload_table(payload)
    raise ValueError(f"Unsupported workload output format: {fmt}")

vllm_mlx.bench_serve.run_bench_serve async 

run_bench_serve(url: str = 'http://127.0.0.1:8080', model: Optional[str] = None, prompt_sets: list[str] = None, prompt_file: Optional[str] = None, concurrencies: list[int] = None, max_tokens: int = 256, repetitions: int = 3, warmup: int = 1, thinking_values: list[Optional[bool]] = None, extra_bodies: list[str] = None, output_path: Optional[str] = None, fmt: str = 'table', do_validate: bool = True, scrape: bool = True, tag: Optional[str] = None, override_fields: Optional[dict] = None, system_prompt_file: Optional[str] = None, skip_preflight_token_count: bool = False) -> list[BenchServeResult]

Run the full bench-serve sweep against a running vllm-mlx server.

Parameters:

  • url (str, default: 'http://127.0.0.1:8080' ) –

    Base URL of the server.

  • model (Optional[str], default: None ) –

    Model ID to use. If None, auto-detected from the server.

  • prompt_sets (list[str], default: None ) –

    List of prompt set names or paths. Defaults to ["short", "medium", "long"].

  • prompt_file (Optional[str], default: None ) –

    Optional path to an extra prompt file to include.

  • concurrencies (list[int], default: None ) –

    Concurrency levels to sweep. Defaults to [1, 4].

  • max_tokens (int, default: 256 ) –

    Maximum tokens to generate per request.

  • repetitions (int, default: 3 ) –

    Number of repetitions per sweep config.

  • warmup (int, default: 1 ) –

    Number of warmup rounds before the first measured repetition.

  • thinking_values (list[Optional[bool]], default: None ) –

    Values for enable_thinking. Defaults to [None].

  • extra_bodies (list[str], default: None ) –

    JSON strings for extra body parameters. Defaults to [""] (no extra body).

  • output_path (Optional[str], default: None ) –

    File path to write results to. If None, prints to stdout.

  • fmt (str, default: 'table' ) –

    Output format — one of "table", "json", "csv", "sql", or "sqlite".

  • do_validate (bool, default: True ) –

    Whether to validate each response.

  • scrape (bool, default: True ) –

    Whether to scrape /metrics before and after each run.

  • tag (Optional[str], default: None ) –

    Optional tag string stored in every result row.

  • override_fields (Optional[dict], default: None ) –

    Dict of field names to override on every result.

Returns:

Source code in vllm_mlx/bench_serve.py 
2221
2222
2223
2224
2225
2226
2227
2228
2229
2230
2231
2232
2233
2234
2235
2236
2237
2238
2239
2240
2241
2242
2243
2244
2245
2246
2247
2248
2249
2250
2251
2252
2253
2254
2255
2256
2257
2258
2259
2260
2261
2262
2263
2264
2265
2266
2267
2268
2269
2270
2271
2272
2273
2274
2275
2276
2277
2278
2279
2280
2281
2282
2283
2284
2285
2286
2287
2288
2289
2290
2291
2292
2293
2294
2295
2296
2297
2298
2299
2300
2301
2302
2303
2304
2305
2306
2307
2308
2309
2310
2311
2312
2313
2314
2315
2316
2317
2318
2319
2320
2321
2322
2323
2324
2325
2326
2327
2328
2329
2330
2331
2332
2333
2334
2335
2336
2337
2338
2339
2340
2341
2342
2343
2344
2345
2346
2347
2348
2349
2350
2351
2352
2353
2354
2355
2356
2357
2358
2359
2360
2361
2362
2363
2364
2365
2366
2367
2368
2369
2370
2371
2372
2373
2374
2375
2376
2377
2378
2379
2380
2381
2382
2383
2384
2385
2386
2387
2388
2389
2390
2391
2392
2393
2394
2395
2396
2397
2398
2399
2400
2401
2402
2403
2404
2405
2406
2407
2408
2409
2410
2411
2412
2413
2414
2415
2416
2417
2418
2419
2420
2421
2422
2423
2424
2425
2426
2427
2428
2429
2430
2431
2432
2433
2434
2435
2436
2437
2438
2439
2440
2441
2442
2443
2444
2445
2446
2447
2448
2449
2450
2451
2452
2453
2454
2455
2456
2457
2458
2459
2460
2461
2462
2463
2464
2465
2466
2467
2468
2469
2470
2471
2472
2473
2474
2475
2476
2477
2478
2479
2480
2481
2482
2483
2484
2485
2486
2487
2488
2489
2490
2491
2492
2493
2494
2495
2496
2497
2498
2499
2500
2501
2502
2503
2504
2505
2506
2507
2508
2509
2510
2511
2512
2513
2514
2515
2516
2517
2518
2519
2520
2521
2522
2523
2524
2525
2526
2527
2528
2529
2530
2531
2532
2533
2534
2535
2536
2537
2538
2539
2540
2541
2542
2543
2544
2545
2546
2547
2548
2549
2550
2551
2552
2553
2554
2555
2556
2557
2558
2559
2560
2561
2562
2563
2564
2565
2566
2567
2568
2569
2570
2571
2572
2573
2574
2575
2576
2577
2578
2579
2580
2581
2582
2583
2584
2585
2586
2587
2588
2589
2590
2591
2592
2593
2594
2595
2596
2597
2598
2599
2600
2601
2602
2603
2604
2605
2606
2607
2608
2609
2610
2611
2612
2613
2614
2615
2616
2617
2618
2619
2620
2621
2622
2623
2624
2625
2626
2627
2628
2629
2630
2631
2632
2633
2634
2635
2636
2637
2638
async def run_bench_serve(
    url: str = "http://127.0.0.1:8080",
    model: Optional[str] = None,
    prompt_sets: list[str] = None,
    prompt_file: Optional[str] = None,
    concurrencies: list[int] = None,
    max_tokens: int = 256,
    repetitions: int = 3,
    warmup: int = 1,
    thinking_values: list[Optional[bool]] = None,
    extra_bodies: list[str] = None,
    output_path: Optional[str] = None,
    fmt: str = "table",
    do_validate: bool = True,
    scrape: bool = True,
    tag: Optional[str] = None,
    override_fields: Optional[dict] = None,
    system_prompt_file: Optional[str] = None,
    skip_preflight_token_count: bool = False,
) -> list[BenchServeResult]:
    """Run the full bench-serve sweep against a running vllm-mlx server.

    Args:
        url: Base URL of the server.
        model: Model ID to use. If ``None``, auto-detected from the server.
        prompt_sets: List of prompt set names or paths. Defaults to
            ``["short", "medium", "long"]``.
        prompt_file: Optional path to an extra prompt file to include.
        concurrencies: Concurrency levels to sweep. Defaults to ``[1, 4]``.
        max_tokens: Maximum tokens to generate per request.
        repetitions: Number of repetitions per sweep config.
        warmup: Number of warmup rounds before the first measured repetition.
        thinking_values: Values for ``enable_thinking``. Defaults to
            ``[None]``.
        extra_bodies: JSON strings for extra body parameters. Defaults to
            ``[""]`` (no extra body).
        output_path: File path to write results to. If ``None``, prints to
            stdout.
        fmt: Output format — one of ``"table"``, ``"json"``, ``"csv"``,
            ``"sql"``, or ``"sqlite"``.
        do_validate: Whether to validate each response.
        scrape: Whether to scrape ``/metrics`` before and after each run.
        tag: Optional tag string stored in every result row.
        override_fields: Dict of field names to override on every result.

    Returns:
        List of :class:`BenchServeResult` instances.
    """
    # 1. Set defaults
    if prompt_sets is None:
        prompt_sets = ["short", "medium", "long"]
    if concurrencies is None:
        concurrencies = [1, 4]
    if thinking_values is None:
        thinking_values = [None]
    if extra_bodies is None:
        extra_bodies = [""]
    if override_fields is None:
        override_fields = {}

    # 2. Generate run_id and timestamp
    run_id = str(uuid.uuid4())[:8]
    timestamp = datetime.now(timezone.utc).isoformat()

    # 3. Open HTTP client
    async with httpx.AsyncClient(timeout=httpx.Timeout(300.0)) as client:
        # 4. Auto-detect runtime and hardware
        print(f"Connecting to {url}...")
        runtime = await auto_detect_runtime(client, url)
        hw = detect_hardware_fingerprint()

        # 5. Resolve model_id
        model_id = model or runtime.get("model_id", "")
        if not model_id:
            print(
                "Error: could not determine model ID. Use --model to specify.",
                file=sys.stderr,
            )
            return []

        # 6. Print hardware and runtime info
        print(
            f"Hardware: {hw.get('chip', 'unknown')} / {hw.get('memory_gb', 0):.0f}GB / {hw.get('os_version', '')}"
        )
        print(
            f"Runtime:  model={model_id}  engine={runtime.get('engine_type', '')}  cache={runtime.get('cache_type', '')}"
        )

        # 7. Load prompts
        all_prompts: dict[str, list[list[dict]]] = {}
        for ps in prompt_sets:
            try:
                all_prompts[ps] = load_prompt_set(ps)
            except FileNotFoundError as exc:
                print(f"Warning: skipping prompt set '{ps}': {exc}", file=sys.stderr)
        if prompt_file:
            try:
                all_prompts[prompt_file] = load_prompt_set(prompt_file)
            except FileNotFoundError as exc:
                print(
                    f"Warning: skipping prompt file '{prompt_file}': {exc}",
                    file=sys.stderr,
                )

        if not all_prompts:
            print("Error: no prompt sets could be loaded.", file=sys.stderr)
            return []

        # 7b. If --system-prompt-file given, prepend that system message to
        # every prompt across every set. This is the warm-prompts path: the
        # server was started with the same system in its warm-up file, so
        # every request here hits the prefix cache.
        if system_prompt_file:
            sys_path = Path(system_prompt_file).expanduser()
            if not sys_path.exists():
                print(
                    f"Error: --system-prompt-file not found: {sys_path}",
                    file=sys.stderr,
                )
                return []
            sys_content = sys_path.read_text()
            system_msg = {"role": "system", "content": sys_content}
            for ps, prompts in all_prompts.items():
                patched: list[list[dict]] = []
                for msgs in prompts:
                    # Do not double-prepend if the prompt already has a system
                    # as its first message.
                    if msgs and msgs[0].get("role") == "system":
                        patched.append(msgs)
                    else:
                        patched.append([system_msg] + msgs)
                all_prompts[ps] = patched
            print(f"System prompt prepended from {sys_path} ({len(sys_content)} chars)")

        # 8. Token-count first prompt from each set.
        # This sends a non-streaming max_tokens=1 request per prompt set, which
        # populates the server's prefix cache with the full prompt. That is
        # harmless for ordinary benchmarking but DEFEATS cold-vs-warm
        # comparisons (both paths end up warm after the pre-flight). Skip it
        # when --skip-preflight-token-count is set; prompt_tokens will be
        # populated from the first measured request's usage instead.
        prompt_token_counts: dict[str, int] = {}
        if skip_preflight_token_count:
            for ps in all_prompts:
                prompt_token_counts[ps] = 0
        else:
            for ps, prompts in all_prompts.items():
                try:
                    count = await count_prompt_tokens(client, url, prompts[0], model_id)
                    prompt_token_counts[ps] = count
                except Exception:
                    prompt_token_counts[ps] = 0

        # 9. Expand sweep
        sweep = expand_sweep(
            list(all_prompts.keys()),
            concurrencies,
            thinking_values,
            extra_bodies,
            repetitions,
        )

        # Account for warmup rounds: insert warmup configs at rep==0 boundaries.
        # We handle warmup inline during the sweep by tracking which
        # (ps, conc, think, eb) combos have been warmed up.
        total_runs = len(sweep)
        warmup_note = f" (+ {warmup} warmup per config)" if warmup > 0 else ""
        print(f"Total runs: {total_runs}{warmup_note}")

        results: list[BenchServeResult] = []
        warmed_up: set[tuple] = set()

        # 11. Iterate over sweep
        for ps, conc, think, eb, rep in sweep:
            prompts = all_prompts[ps]

            # Parse extra_body
            extra_body_dict: Optional[dict] = None
            if eb:
                try:
                    extra_body_dict = json.loads(eb)
                except json.JSONDecodeError:
                    extra_body_dict = None

            # Format progress label
            think_label = f"think={think}" if think is not None else ""
            eb_label = f"eb={eb[:20]}" if eb else ""
            label_parts = [ps, f"conc={conc}", f"rep={rep}"]
            if think_label:
                label_parts.append(think_label)
            if eb_label:
                label_parts.append(eb_label)
            label = " ".join(label_parts)

            # d. Warmup on first rep of each unique config
            warmup_key = (ps, conc, think, eb)
            if rep == 0 and warmup_key not in warmed_up and warmup > 0:
                warmed_up.add(warmup_key)
                for _ in range(warmup):
                    try:
                        await run_concurrent_requests(
                            client=client,
                            base_url=url,
                            prompts=prompts,
                            model=model_id,
                            concurrency=conc,
                            max_tokens=max_tokens,
                            enable_thinking=think,
                            extra_body=extra_body_dict,
                            do_validate=False,
                        )
                    except Exception:
                        pass

            # e. Scrape /metrics before
            metrics_before: dict = {}
            if scrape:
                metrics_before = await scrape_metrics(client, url)

            # f. Run concurrent requests
            req_results = await run_concurrent_requests(
                client=client,
                base_url=url,
                prompts=prompts,
                model=model_id,
                concurrency=conc,
                max_tokens=max_tokens,
                enable_thinking=think,
                extra_body=extra_body_dict,
                do_validate=do_validate,
            )

            # g. Scrape /metrics after, compute cache delta
            metrics_after: dict = {}
            if scrape:
                metrics_after = await scrape_metrics(client, url)

            cache_hits_delta = metrics_after.get("cache_hits", 0) - metrics_before.get(
                "cache_hits", 0
            )
            cache_misses_delta = metrics_after.get(
                "cache_misses", 0
            ) - metrics_before.get("cache_misses", 0)
            tokens_saved_delta = metrics_after.get(
                "tokens_saved", 0
            ) - metrics_before.get("tokens_saved", 0)
            total_events = cache_hits_delta + cache_misses_delta
            cache_hit_rate = (
                cache_hits_delta / total_events if total_events > 0 else 0.0
            )

            # h. Get metal memory from /v1/status
            metal_active_gb = runtime.get("metal_active_gb", 0.0)
            metal_peak_gb = runtime.get("metal_peak_gb", 0.0)
            metal_cache_gb = runtime.get("metal_cache_gb", 0.0)
            try:
                resp = await client.get(f"{url}/v1/status")
                resp.raise_for_status()
                status_data = parse_status_response(resp.json())
                metal_active_gb = status_data.get("metal_active_gb", metal_active_gb)
                metal_peak_gb = status_data.get("metal_peak_gb", metal_peak_gb)
                metal_cache_gb = status_data.get("metal_cache_gb", metal_cache_gb)
            except Exception:
                pass

            # i. Aggregate per-request metrics
            valid_results = [r for r in req_results if "error" not in r]
            if not valid_results:
                # All requests errored — build a failed result
                result_obj = BenchServeResult(
                    run_id=run_id,
                    timestamp=timestamp,
                    tag=tag or "",
                    # Hardware
                    chip=hw.get("chip", ""),
                    gpu_cores=hw.get("gpu_cores", 0),
                    memory_gb=hw.get("memory_gb", 0.0),
                    bandwidth_gbs=hw.get("bandwidth_gbs", 0.0),
                    os_version=hw.get("os_version", ""),
                    # Runtime
                    model_id=model_id,
                    model_type=runtime.get("model_type", ""),
                    engine_type=runtime.get("engine_type", ""),
                    mtp_enabled=runtime.get("mtp_enabled", False),
                    specprefill=runtime.get("specprefill", False),
                    kv_quant=runtime.get("kv_quant", ""),
                    cache_type=runtime.get("cache_type", ""),
                    # Config
                    prompt_set=ps,
                    concurrency=conc,
                    max_tokens=max_tokens,
                    enable_thinking=think,
                    extra_body=eb,
                    repetition=rep,
                    prompt_tokens=prompt_token_counts.get(ps, 0),
                    # Latency / throughput all zero
                    validated=False,
                )
                print(f"  {label}: FAIL (all requests errored)")
            else:

                def _mean(key: str) -> float:
                    vals = [
                        r[key] for r in valid_results if key in r and r[key] is not None
                    ]
                    return statistics.mean(vals) if vals else 0.0

                mean_ttft = _mean("ttft_ms")
                mean_tpot = _mean("tpot_ms")
                mean_gen_tps = _mean("gen_tps")
                mean_prompt_tps = _mean("prompt_tps")
                mean_e2e = _mean("e2e_latency_ms")

                total_completion_tokens = sum(
                    r.get("completion_tokens", 0) for r in valid_results
                )
                max_e2e_seconds = (
                    max(
                        (r.get("e2e_latency_ms", 0.0) for r in valid_results),
                        default=0.0,
                    )
                    / 1000.0
                )
                throughput_tps = (
                    total_completion_tokens / max_e2e_seconds
                    if max_e2e_seconds > 0
                    else 0.0
                )
                requests_per_s = conc / max_e2e_seconds if max_e2e_seconds > 0 else 0.0

                all_validated = all(r.get("validated", True) for r in valid_results)

                result_obj = BenchServeResult(
                    run_id=run_id,
                    timestamp=timestamp,
                    tag=tag or "",
                    # Hardware
                    chip=hw.get("chip", ""),
                    gpu_cores=hw.get("gpu_cores", 0),
                    memory_gb=hw.get("memory_gb", 0.0),
                    bandwidth_gbs=hw.get("bandwidth_gbs", 0.0),
                    os_version=hw.get("os_version", ""),
                    # Runtime
                    model_id=model_id,
                    model_type=runtime.get("model_type", ""),
                    engine_type=runtime.get("engine_type", ""),
                    mtp_enabled=runtime.get("mtp_enabled", False),
                    specprefill=runtime.get("specprefill", False),
                    kv_quant=runtime.get("kv_quant", ""),
                    cache_type=runtime.get("cache_type", ""),
                    # Config
                    prompt_set=ps,
                    concurrency=conc,
                    max_tokens=max_tokens,
                    enable_thinking=think,
                    extra_body=eb,
                    repetition=rep,
                    prompt_tokens=prompt_token_counts.get(ps, 0),
                    # Latency
                    ttft_ms=mean_ttft,
                    tpot_ms=mean_tpot,
                    e2e_latency_ms=mean_e2e,
                    # Throughput
                    gen_tps=mean_gen_tps,
                    prompt_tps=mean_prompt_tps,
                    throughput_tps=throughput_tps,
                    requests_per_s=requests_per_s,
                    # Memory
                    metal_active_gb=metal_active_gb,
                    metal_peak_gb=metal_peak_gb,
                    metal_cache_gb=metal_cache_gb,
                    # Cache
                    cache_hits=cache_hits_delta,
                    cache_misses=cache_misses_delta,
                    cache_hit_rate=cache_hit_rate,
                    tokens_saved=tokens_saved_delta,
                    # Validation
                    validated=all_validated,
                )

                status = "PASS" if all_validated else "FAIL"
                print(
                    f"  {label}: TTFT={mean_ttft:.0f}ms  TPS={mean_gen_tps:.1f}  {status}"
                )

            # j. Apply override_fields
            for field_name, field_val in override_fields.items():
                if hasattr(result_obj, field_name):
                    setattr(result_obj, field_name, field_val)

            results.append(result_obj)

        # 12. Format output
        if fmt == "sqlite":
            if not output_path:
                raise ValueError("--output is required when --format sqlite")
            write_sqlite(results, output_path)
            print(f"\nSQLite results written to {output_path}")
            return results

        formatters = {
            "table": format_table,
            "json": format_json,
            "csv": format_csv,
            "sql": format_sql,
        }
        formatter = formatters.get(fmt, format_table)
        output = formatter(results)

        # 13. Write to file or stdout
        if output_path:
            Path(output_path).write_text(output)
            print(f"\nResults written to {output_path}")
        else:
            print()
            print(output)

        return results

Complete contract reference

Expand any definition for its exact inputs, annotations, defaults, return contract, directly raised exceptions, source-grounded behavior, and immutable line link. This section includes private and nested definitions that ordinary API generators omit.

vllm_mlx.bench_serve.WorkloadCase · class 
vllm_mlx.bench_serve.WorkloadCase(case_id: str, messages: list[dict], request_path: Optional[str] = None, max_tokens: Optional[int] = None, enable_thinking: Optional[bool] = None, extra_body: Optional[dict] = None, policy_timeout_ms: Optional[int] = None, checks: Optional[dict] = None, tags: tuple[str, ...] = ())

One declarative benchmark case for contract-style serving tests.

Parameters

Name Type Required Default Description
case_id str yes none Required constructor field.
messages list[dict] yes none Required constructor field.
request_path Optional[str] no None Optional constructor field; defaults to None.
max_tokens Optional[int] no None Optional constructor field; defaults to None.
enable_thinking Optional[bool] no None Optional constructor field; defaults to None.
extra_body Optional[dict] no None Optional constructor field; defaults to None.
policy_timeout_ms Optional[int] no None Optional constructor field; defaults to None.
checks Optional[dict] no None Optional constructor field; defaults to None.
tags tuple[str, ...] no () Optional constructor field; defaults to ().

Returns

  • Constructs: vllm_mlx.bench_serve.WorkloadCase

Exceptions and behavior

Class WorkloadCase declares 0 direct member(s). No direct raise statement appears in this definition.

View source #L51-L62.

vllm_mlx.bench_serve.Workload · class 
vllm_mlx.bench_serve.Workload(name: str, description: str, defaults: dict, cases: list[WorkloadCase])

Normalized bench-serve workload manifest.

Parameters

Name Type Required Default Description
name str yes none Required constructor field.
description str yes none Required constructor field.
defaults dict yes none Required constructor field.
cases list[WorkloadCase] yes none Required constructor field.

Returns

  • Constructs: vllm_mlx.bench_serve.Workload

Exceptions and behavior

Class Workload declares 0 direct member(s). No direct raise statement appears in this definition.

View source #L66-L72.

vllm_mlx.bench_serve.load_prompt_set · function 
vllm_mlx.bench_serve.load_prompt_set(name_or_path: str) -> list[list[dict]]

Load a prompt set by builtin name or file path.

Parameters

Name Type Required Default Description
name_or_path str yes none Required positional or keyword input.

Returns

  • Type: list[list[dict]]
  • Direct return expressions: [[msg] for msg in raw]; raw

Exceptions and behavior

Function load_prompt_set calls target.exists, FileNotFoundError, target.open, json.load; can raise FileNotFoundError, ValueError; has 2 explicit return paths. Directly raised exceptions: FileNotFoundError, ValueError.

View source #L75-L137.

vllm_mlx.bench_serve._require_message_list · function 
vllm_mlx.bench_serve._require_message_list(value: Any, *, label: str) -> list[dict]

Function _require_message_list calls isinstance, ValueError, enumerate; can raise ValueError; returns value.

Parameters

Name Type Required Default Description
value Any yes none Required positional or keyword input.
label str yes none Required keyword-only input.

Returns

  • Type: list[dict]
  • Direct return expressions: value

Exceptions and behavior

Function _require_message_list calls isinstance, ValueError, enumerate; can raise ValueError; returns value. Directly raised exceptions: ValueError.

View source #L140-L148.

vllm_mlx.bench_serve._load_case_request · function 
vllm_mlx.bench_serve._load_case_request(path: str, *, workload_path: Path, case_id: str) -> dict

Function _load_case_request calls Path(path).expanduser, Path, request_path.is_absolute, request_path.open; can raise ValueError; returns request.

Parameters

Name Type Required Default Description
path str yes none Required positional or keyword input.
workload_path Path yes none Required keyword-only input.
case_id str yes none Required keyword-only input.

Returns

  • Type: dict
  • Direct return expressions: request

Exceptions and behavior

Function _load_case_request calls Path(path).expanduser, Path, request_path.is_absolute, request_path.open; can raise ValueError; returns request. Directly raised exceptions: ValueError.

View source #L151-L159.

vllm_mlx.bench_serve._request_extra_body · function 
vllm_mlx.bench_serve._request_extra_body(request: dict) -> dict

Function _request_extra_body calls request.items; returns {key: value for key, value in request.items() if key not in reserved}.

Parameters

Name Type Required Default Description
request dict yes none Required positional or keyword input.

Returns

  • Type: dict
  • Direct return expressions: {key: value for key, value in request.items() if key not in reserved}

Exceptions and behavior

Function _request_extra_body calls request.items; returns {key: value for key, value in request.items() if key not in reserved}. No direct raise statement appears in this definition.

View source #L162-L171.

vllm_mlx.bench_serve._first_not_none · function 
vllm_mlx.bench_serve._first_not_none(*values: Any) -> Any

Function _first_not_none has 2 explicit return paths.

Parameters

Name Type Required Default Description
*values Any no none Additional variadic positional inputs accepted by this callable.

Returns

  • Type: Any
  • Direct return expressions: value; None

Exceptions and behavior

Function _first_not_none has 2 explicit return paths. No direct raise statement appears in this definition.

View source #L174-L178.

vllm_mlx.bench_serve._normalize_tags · function 
vllm_mlx.bench_serve._normalize_tags(tags: Any, *, case_id: str) -> tuple[str, ...]

Coerce a workload case's tags field to a tuple of strings.

Parameters

Name Type Required Default Description
tags Any yes none Required positional or keyword input.
case_id str yes none Required keyword-only input.

Returns

  • Type: tuple[str, ...]
  • Direct return expressions: tuple((str(tag) for tag in tags))

Exceptions and behavior

Function _normalize_tags calls isinstance, ValueError, tuple, str; can raise ValueError; returns tuple((str(tag) for tag in tags)). Directly raised exceptions: ValueError.

View source #L181-L191.

vllm_mlx.bench_serve._merge_case_checks · function 
vllm_mlx.bench_serve._merge_case_checks(default_checks: Any, case_checks: Any, *, case_id: str) -> Optional[dict]

Merge a case's checks over the workload defaults.

Parameters

Name Type Required Default Description
default_checks Any yes none Required positional or keyword input.
case_checks Any yes none Required positional or keyword input.
case_id str yes none Required keyword-only input.

Returns

  • Type: Optional[dict]
  • Direct return expressions: merged or None

Exceptions and behavior

Function _merge_case_checks calls dict, isinstance, ValueError, case_checks.items; can raise ValueError; returns merged or None. Directly raised exceptions: ValueError.

View source #L194-L227.

vllm_mlx.bench_serve._build_workload_case · function 
vllm_mlx.bench_serve._build_workload_case(item: Any, idx: int, *, defaults: dict, workload_path: Path) -> WorkloadCase

Construct one WorkloadCase from a raw workload entry.

Parameters

Name Type Required Default Description
item Any yes none Required positional or keyword input.
idx int yes none Required positional or keyword input.
defaults dict yes none Required keyword-only input.
workload_path Path yes none Required keyword-only input.

Returns

  • Type: WorkloadCase
  • Direct return expressions: WorkloadCase(case_id=case_id, messages=messages, request_path=str(request_path) if request_path is not None else None, …

Exceptions and behavior

Function _build_workload_case calls isinstance, ValueError, str, item.get; can raise ValueError; returns WorkloadCase(case_id=case_id, messages=messages, request_path=str(request_path) if request_path is not None else None, …. Directly raised exceptions: ValueError.

View source #L230-L300.

vllm_mlx.bench_serve.load_workload · function 
vllm_mlx.bench_serve.load_workload(path: str | Path) -> Workload

Load a declarative serving benchmark workload.

Parameters

Name Type Required Default Description
path str \| Path yes none Required positional or keyword input.

Returns

  • Type: Workload
  • Direct return expressions: Workload(name=str(raw.get('name') or workload_path.stem), description=str(raw.get('description') or ''), defaults=defau…

Exceptions and behavior

Function load_workload calls Path(path).expanduser, Path, workload_path.open, json.load; can raise ValueError; returns Workload(name=str(raw.get('name') or workload_path.stem), description=str(raw.get('description') or ''), defaults=defau…. Directly raised exceptions: ValueError.

View source #L303-L335.

vllm_mlx.bench_serve.BenchServeResult · class 
vllm_mlx.bench_serve.BenchServeResult(run_id: str = '', timestamp: str = '', tag: str = '', chip: str = '', gpu_cores: int = 0, memory_gb: float = 0.0, bandwidth_gbs: float = 0.0, os_version: str = '', model_id: str = '', model_type: str = '', engine_type: str = '', mtp_enabled: bool = False, specprefill: bool = False, kv_quant: str = '', cache_type: str = '', prompt_set: str = '', concurrency: int = 1, max_tokens: int = 256, enable_thinking: Optional[bool] = None, extra_body: str = '', repetition: int = 0, prompt_tokens: int = 0, ttft_ms: float = 0.0, tpot_ms: float = 0.0, e2e_latency_ms: float = 0.0, gen_tps: float = 0.0, prompt_tps: float = 0.0, throughput_tps: float = 0.0, requests_per_s: float = 0.0, metal_active_gb: float = 0.0, metal_peak_gb: float = 0.0, metal_cache_gb: float = 0.0, cache_hits: int = 0, cache_misses: int = 0, cache_hit_rate: float = 0.0, tokens_saved: int = 0, validated: bool = True)

Aggregated results from a single bench-serve run configuration.

Parameters

Name Type Required Default Description
run_id str no '' Optional constructor field; defaults to ''.
timestamp str no '' Optional constructor field; defaults to ''.
tag str no '' Optional constructor field; defaults to ''.
chip str no '' Optional constructor field; defaults to ''.
gpu_cores int no 0 Optional constructor field; defaults to 0.
memory_gb float no 0.0 Optional constructor field; defaults to 0.0.
bandwidth_gbs float no 0.0 Optional constructor field; defaults to 0.0.
os_version str no '' Optional constructor field; defaults to ''.
model_id str no '' Optional constructor field; defaults to ''.
model_type str no '' Optional constructor field; defaults to ''.
engine_type str no '' Optional constructor field; defaults to ''.
mtp_enabled bool no False Optional constructor field; defaults to False.
specprefill bool no False Optional constructor field; defaults to False.
kv_quant str no '' Optional constructor field; defaults to ''.
cache_type str no '' Optional constructor field; defaults to ''.
prompt_set str no '' Optional constructor field; defaults to ''.
concurrency int no 1 Optional constructor field; defaults to 1.
max_tokens int no 256 Optional constructor field; defaults to 256.
enable_thinking Optional[bool] no None Optional constructor field; defaults to None.
extra_body str no '' Optional constructor field; defaults to ''.
repetition int no 0 Optional constructor field; defaults to 0.
prompt_tokens int no 0 Optional constructor field; defaults to 0.
ttft_ms float no 0.0 Optional constructor field; defaults to 0.0.
tpot_ms float no 0.0 Optional constructor field; defaults to 0.0.
e2e_latency_ms float no 0.0 Optional constructor field; defaults to 0.0.
gen_tps float no 0.0 Optional constructor field; defaults to 0.0.
prompt_tps float no 0.0 Optional constructor field; defaults to 0.0.
throughput_tps float no 0.0 Optional constructor field; defaults to 0.0.
requests_per_s float no 0.0 Optional constructor field; defaults to 0.0.
metal_active_gb float no 0.0 Optional constructor field; defaults to 0.0.
metal_peak_gb float no 0.0 Optional constructor field; defaults to 0.0.
metal_cache_gb float no 0.0 Optional constructor field; defaults to 0.0.
cache_hits int no 0 Optional constructor field; defaults to 0.
cache_misses int no 0 Optional constructor field; defaults to 0.
cache_hit_rate float no 0.0 Optional constructor field; defaults to 0.0.
tokens_saved int no 0 Optional constructor field; defaults to 0.
validated bool no True Optional constructor field; defaults to True.

Returns

  • Constructs: vllm_mlx.bench_serve.BenchServeResult

Exceptions and behavior

Class BenchServeResult declares 0 direct member(s). No direct raise statement appears in this definition.

View source #L344-L400.

vllm_mlx.bench_serve.expand_sweep · function 
vllm_mlx.bench_serve.expand_sweep(prompt_sets: list[str], concurrencies: list[int], thinking_values: list[Optional[bool]], extra_bodies: list[str], repetitions: int) -> list[SweepConfig]

Expand sweep parameters into a flat list of configurations.

Parameters

Name Type Required Default Description
prompt_sets list[str] yes none Names or paths of prompt sets to include.
concurrencies list[int] yes none Concurrency levels to test (e.g. [1, 4, 16]).
thinking_values list[Optional[bool]] yes none Values for enable_thinking (e.g. [None, True, False]).
extra_bodies list[str] yes none JSON strings (or empty string) to pass as extra body parameters on each request.
repetitions int yes none Number of times to repeat each unique combination. Each repeat gets a distinct 0-based repetition index.

Returns

  • Type: list[SweepConfig]
  • Direct return expressions: configs

Exceptions and behavior

Function expand_sweep calls itertools.product, range, configs.append; returns configs. No direct raise statement appears in this definition.

View source #L411-L444.

vllm_mlx.bench_serve.parse_health_response · function 
vllm_mlx.bench_serve.parse_health_response(data: dict) -> dict

Extract model identity fields from a GET /health response.

Parameters

Name Type Required Default Description
data dict yes none Parsed JSON body from the /health endpoint. Expected shape::

Returns

  • Type: dict
  • Direct return expressions: {'model_name': data.get('model_name', ''), 'model_type': data.get('model_type', '')}

Exceptions and behavior

Function parse_health_response calls data.get; returns {'model_name': data.get('model_name', ''), 'model_type': data.get('model_type', '')}. No direct raise statement appears in this definition.

View source #L452-L467.

vllm_mlx.bench_serve.parse_status_response · function 
vllm_mlx.bench_serve.parse_status_response(data: dict) -> dict

Extract metal and cache info from a GET /v1/status response.

Parameters

Name Type Required Default Description
data dict yes none Parsed JSON body from the /v1/status endpoint. Metal info is expected under data["metal"] and cache info under data["cache"]. Missing keys are handled gracefully.

Returns

  • Type: dict
  • Direct return expressions: {'model': data.get('model', ''), 'metal_active_gb': float(metal.get('active_memory_gb') or metal.get('active_gb') or 0.…

Exceptions and behavior

Function parse_status_response calls data.get, float, metal.get, cache.get; returns {'model': data.get('model', ''), 'metal_active_gb': float(metal.get('active_memory_gb') or metal.get('active_gb') or 0.…. No direct raise statement appears in this definition.

View source #L470-L496.

vllm_mlx.bench_serve.parse_metrics_text · function 
vllm_mlx.bench_serve.parse_metrics_text(text: str) -> dict

Parse Prometheus text exposition format from GET /metrics.

Parameters

Name Type Required Default Description
text str yes none Raw response body from the /metrics endpoint.

Returns

  • Type: dict
  • Direct return expressions: {'cache_hits': _extract('vllm_prefix_cache_hits_total'), 'cache_misses': _extract('vllm_prefix_cache_misses_total'), 't…

Exceptions and behavior

Function parse_metrics_text calls _extract; returns {'cache_hits': _extract('vllm_prefix_cache_hits_total'), 'cache_misses': _extract('vllm_prefix_cache_misses_total'), 't…. No direct raise statement appears in this definition.

View source #L499-L521.

vllm_mlx.bench_serve.parse_metrics_text._extract · nested function 
vllm_mlx.bench_serve.parse_metrics_text._extract(metric_name: str) -> int

Nested Function parse_metrics_text._extract calls re.escape, re.search, int, m.group; returns int(m.group(1)) if m else 0.

Parameters

Name Type Required Default Description
metric_name str yes none Required positional or keyword input.

Returns

  • Type: int
  • Direct return expressions: int(m.group(1)) if m else 0

Exceptions and behavior

Nested Function parse_metrics_text._extract calls re.escape, re.search, int, m.group; returns int(m.group(1)) if m else 0. No direct raise statement appears in this definition.

View source #L512-L515.

vllm_mlx.bench_serve.detect_hardware_fingerprint · function 
vllm_mlx.bench_serve.detect_hardware_fingerprint() -> dict

Return a hardware fingerprint dict for the current machine.

Parameters

This callable has no explicit inputs.

Returns

  • Type: dict
  • Direct return expressions: {'chip': hw.chip_name, 'gpu_cores': hw.gpu_cores, 'memory_gb': hw.total_memory_gb, 'bandwidth_gbs': hw.memory_bandwidth…; {'chip': '', 'gpu_cores': 0, 'memory_gb': memory_gb, 'bandwidth_gbs': 0.0, 'os_version': os_version}

Exceptions and behavior

Function detect_hardware_fingerprint calls platform.platform, detect_hardware, subprocess.run, int; has 2 explicit return paths. No direct raise statement appears in this definition.

View source #L524-L573.

vllm_mlx.bench_serve.auto_detect_runtime · function 
async vllm_mlx.bench_serve.auto_detect_runtime(client: httpx.AsyncClient, base_url: str) -> dict

Query the running server and return a runtime descriptor dict.

Parameters

Name Type Required Default Description
client httpx.AsyncClient yes none An open :class:httpx.AsyncClient.
base_url str yes none Base URL of the server (e.g. "http://localhost:8080").

Returns

  • Type: dict
  • Direct return expressions: result

Exceptions and behavior

Function auto_detect_runtime calls client.get, resp.raise_for_status, parse_health_response, resp.json; awaits asynchronous work; returns result. No direct raise statement appears in this definition.

View source #L576-L642.

vllm_mlx.bench_serve.scrape_metrics · function 
async vllm_mlx.bench_serve.scrape_metrics(client: httpx.AsyncClient, base_url: str) -> dict

Scrape Prometheus metrics from the server.

Parameters

Name Type Required Default Description
client httpx.AsyncClient yes none An open :class:httpx.AsyncClient.
base_url str yes none Base URL of the server.

Returns

  • Type: dict
  • Direct return expressions: parse_metrics_text(resp.text); {}

Exceptions and behavior

Function scrape_metrics calls client.get, resp.raise_for_status, parse_metrics_text; awaits asynchronous work; has 2 explicit return paths. No direct raise statement appears in this definition.

View source #L645-L661.

vllm_mlx.bench_serve.clear_runtime_cache · function 
async vllm_mlx.bench_serve.clear_runtime_cache(client: httpx.AsyncClient, base_url: str) -> dict

Clear server-side runtime caches and return a JSON-serializable event.

Parameters

Name Type Required Default Description
client httpx.AsyncClient yes none Required positional or keyword input.
base_url str yes none Required positional or keyword input.

Returns

  • Type: dict
  • Direct return expressions: event

Exceptions and behavior

Function clear_runtime_cache calls client.delete, resp.json, resp.raise_for_status, str; awaits asynchronous work; returns event. No direct raise statement appears in this definition.

View source #L664-L684.

vllm_mlx.bench_serve._normalize_cache_policy · function 
vllm_mlx.bench_serve._normalize_cache_policy(value: Optional[str]) -> str

Normalize cache-policy spelling from CLI or workload JSON.

Parameters

Name Type Required Default Description
value Optional[str] yes none Required positional or keyword input.

Returns

  • Type: str
  • Direct return expressions: policy

Exceptions and behavior

Function _normalize_cache_policy calls (value or 'preserve').strip().lower().replace, (value or 'preserve').strip().lower, (value or 'preserve').strip, ValueError; can raise ValueError; returns policy. Directly raised exceptions: ValueError.

View source #L687-L698.

vllm_mlx.bench_serve.parse_sse_line · function 
vllm_mlx.bench_serve.parse_sse_line(line: str) -> Optional[dict]

Parse one Server-Sent Events line from a streaming chat completion.

Parameters

Name Type Required Default Description
line str yes none A single raw line from the SSE stream (may or may not include a trailing newline — it is stripped before processing).

Returns

  • Type: Optional[dict]
  • Direct return expressions: None; {'id': chunk.get('id'), 'content': content, 'finish_reason': finish_reason, 'usage': usage, 'tool_calls_delta': tool_ca…

Exceptions and behavior

Function parse_sse_line calls line.strip, line.startswith, len, json.loads; has 2 explicit return paths. No direct raise statement appears in this definition.

View source #L706-L754.

vllm_mlx.bench_serve._cancel_server_request · function 
async vllm_mlx.bench_serve._cancel_server_request(client: httpx.AsyncClient, base_url: str, request_id: Optional[str]) -> None

Best-effort server-side cancellation for timed-out workload streams.

Parameters

Name Type Required Default Description
client httpx.AsyncClient yes none Required positional or keyword input.
base_url str yes none Required positional or keyword input.
request_id Optional[str] yes none Required positional or keyword input.

Returns

  • Type: None
  • Direct return expressions: None

Exceptions and behavior

Function _cancel_server_request calls client.post; awaits asynchronous work; returns None. No direct raise statement appears in this definition.

View source #L757-L770.

vllm_mlx.bench_serve.accumulate_tool_calls · function 
vllm_mlx.bench_serve.accumulate_tool_calls(acc: dict[int, dict], delta_list: list[dict]) -> None

Merge streamed OpenAI tool-call deltas into acc by index.

Parameters

Name Type Required Default Description
acc dict[int, dict] yes none Required positional or keyword input.
delta_list list[dict] yes none Required positional or keyword input.

Returns

  • Type: None

Exceptions and behavior

Function accumulate_tool_calls calls int, tc_delta.get, function_delta.get. No direct raise statement appears in this definition.

View source #L773-L792.

vllm_mlx.bench_serve.finalize_tool_calls · function 
vllm_mlx.bench_serve.finalize_tool_calls(acc: dict[int, dict]) -> list[dict]

Return accumulated tool calls in stream index order.

Parameters

Name Type Required Default Description
acc dict[int, dict] yes none Required positional or keyword input.

Returns

  • Type: list[dict]
  • Direct return expressions: [acc[idx] for idx in sorted(acc)]

Exceptions and behavior

Function finalize_tool_calls calls sorted; returns [acc[idx] for idx in sorted(acc)]. No direct raise statement appears in this definition.

View source #L795-L797.

vllm_mlx.bench_serve.compute_request_metrics · function 
vllm_mlx.bench_serve.compute_request_metrics(t_start: float, t_first_token: float, token_times: list, t_end: float, prompt_tokens: int, completion_tokens: int) -> dict

Compute standard latency and throughput metrics for a single request.

Parameters

Name Type Required Default Description
t_start float yes none Timestamp immediately before the request was sent.
t_first_token float yes none Timestamp when the first content token was received.
token_times list yes none List of timestamps, one per content token (including the first). When there is only one token tpot_ms is 0.0.
t_end float yes none Timestamp after the final SSE chunk was consumed.
prompt_tokens int yes none Number of prompt tokens reported by the server.
completion_tokens int yes none Number of completion tokens generated.

Returns

  • Type: dict
  • Direct return expressions: {'ttft_ms': ttft_ms, 'tpot_ms': tpot_ms, 'e2e_latency_ms': e2e_latency_ms, 'gen_tps': gen_tps, 'prompt_tps': prompt_tps}

Exceptions and behavior

Function compute_request_metrics calls len, range, statistics.mean; returns {'ttft_ms': ttft_ms, 'tpot_ms': tpot_ms, 'e2e_latency_ms': e2e_latency_ms, 'gen_tps': gen_tps, 'prompt_tps': prompt_tps}. No direct raise statement appears in this definition.

View source #L800-L852.

vllm_mlx.bench_serve.count_prompt_tokens · function 
async vllm_mlx.bench_serve.count_prompt_tokens(client: httpx.AsyncClient, base_url: str, messages: list[dict], model: str) -> int

Count prompt tokens for a message list by sending a 1-token request.

Parameters

Name Type Required Default Description
client httpx.AsyncClient yes none An open :class:httpx.AsyncClient.
base_url str yes none Base URL of the server.
messages list[dict] yes none The message list to send.
model str yes none Model ID to target.

Returns

  • Type: int
  • Direct return expressions: int((data.get('usage') or {}).get('prompt_tokens', 0)); 0

Exceptions and behavior

Function count_prompt_tokens calls client.post, resp.raise_for_status, resp.json, int; awaits asynchronous work; has 2 explicit return paths. No direct raise statement appears in this definition.

View source #L855-L889.

vllm_mlx.bench_serve.stream_chat_completion · function 
async vllm_mlx.bench_serve.stream_chat_completion(client: httpx.AsyncClient, base_url: str, messages: list[dict], model: str, max_tokens: int = 256, enable_thinking: Optional[bool] = None, extra_body: Optional[dict] = None, timeout_s: Optional[float] = None) -> dict

Send a streaming chat completion and collect per-token timing data.

Parameters

Name Type Required Default Description
client httpx.AsyncClient yes none An open :class:httpx.AsyncClient.
base_url str yes none Base URL of the server.
messages list[dict] yes none The message list to send.
model str yes none Model ID to target.
max_tokens int no 256 Maximum tokens to generate (default 256).
enable_thinking Optional[bool] no None If not None, passed as enable_thinking in the request body.
extra_body Optional[dict] no None Optional extra keys merged into the request body.
timeout_s Optional[float] no None Optional case-level timeout. When set, the stream is closed and best-effort server cancellation is attempted before raising :class:TimeoutError.

Returns

  • Type: dict
  • Direct return expressions: {**metrics, 'completion_tokens': completion_tokens, 'prompt_tokens': prompt_tokens, 'finish_reason': finish_reason, 'co…

Exceptions and behavior

Function stream_chat_completion calls body.update, time.perf_counter, asyncio.timeout, _consume_stream; awaits asynchronous work; can raise TimeoutError; returns {**metrics, 'completion_tokens': completion_tokens, 'prompt_tokens': prompt_tokens, 'finish_reason': finish_reason, 'co…. Directly raised exceptions: TimeoutError.

View source #L892-L1012.

vllm_mlx.bench_serve.stream_chat_completion._consume_stream · nested function 
async vllm_mlx.bench_serve.stream_chat_completion._consume_stream() -> None

Nested Function stream_chat_completion._consume_stream calls client.stream, response.raise_for_status, response.aiter_lines, parse_sse_line.

Parameters

This callable has no explicit inputs.

Returns

  • Type: None

Exceptions and behavior

Nested Function stream_chat_completion._consume_stream calls client.stream, response.raise_for_status, response.aiter_lines, parse_sse_line. No direct raise statement appears in this definition.

View source #L946-L975.

vllm_mlx.bench_serve.validate_response · function 
vllm_mlx.bench_serve.validate_response(finish_reason: Optional[str], content: str, status_code: int, *, tool_calls: Optional[list[dict]] = None) -> tuple[bool, str]

Validate a single streaming response result.

Parameters

Name Type Required Default Description
finish_reason Optional[str] yes none The finish_reason from the final SSE chunk, or None if not received.
content str yes none The accumulated text content of the response.
status_code int yes none The HTTP status code of the response (use 200 for successful streaming requests).
tool_calls Optional[list[dict]] no None Optional keyword-only input; defaults to None.

Returns

  • Type: tuple[bool, str]
  • Direct return expressions: (False, f'HTTP error {status_code}'); (False, 'Missing finish_reason'); (False, 'Truncated (finish_reason=length)'); (False, 'Empty response content'); (True, '')

Exceptions and behavior

Function validate_response has 5 explicit return paths. No direct raise statement appears in this definition.

View source #L1020-L1049.

vllm_mlx.bench_serve._check_finish_reason · function 
vllm_mlx.bench_serve._check_finish_reason(allowed: Any, finish_reason: Optional[str]) -> list[str]

Verify finish_reason is in the allowed set, if one is configured.

Parameters

Name Type Required Default Description
allowed Any yes none Required positional or keyword input.
finish_reason Optional[str] yes none Required positional or keyword input.

Returns

  • Type: list[str]
  • Direct return expressions: []; [f'finish_reason {finish_reason!r} not in allowed set {allowed_list!r}']

Exceptions and behavior

Function _check_finish_reason calls isinstance, list; has 2 explicit return paths. No direct raise statement appears in this definition.

View source #L1052-L1059.

vllm_mlx.bench_serve._check_length_bounds · function 
vllm_mlx.bench_serve._check_length_bounds(min_chars: Any, max_chars: Any, content: str) -> list[str]

Apply min_chars / max_chars content-length bounds.

Parameters

Name Type Required Default Description
min_chars Any yes none Required positional or keyword input.
max_chars Any yes none Required positional or keyword input.
content str yes none Required positional or keyword input.

Returns

  • Type: list[str]
  • Direct return expressions: issues

Exceptions and behavior

Function _check_length_bounds calls len, int, issues.append; returns issues. No direct raise statement appears in this definition.

View source #L1062-L1069.

vllm_mlx.bench_serve._check_regex_patterns · function 
vllm_mlx.bench_serve._check_regex_patterns(patterns: Any, content: str, *, kind: str, expect_match: bool) -> list[str]

Validate that each pattern either matches or does not, per expect_match.

Parameters

Name Type Required Default Description
patterns Any yes none Required positional or keyword input.
content str yes none Required positional or keyword input.
kind str yes none Required keyword-only input.
expect_match bool yes none Required keyword-only input.

Returns

  • Type: list[str]
  • Direct return expressions: issues

Exceptions and behavior

Function _check_regex_patterns calls bool, re.search, str, issues.append; returns issues. No direct raise statement appears in this definition.

View source #L1072-L1096.

vllm_mlx.bench_serve._check_json_content · function 
vllm_mlx.bench_serve._check_json_content(should_be_json: Any, content: str) -> list[str]

Verify content parses as JSON when checks['json'] is truthy.

Parameters

Name Type Required Default Description
should_be_json Any yes none Required positional or keyword input.
content str yes none Required positional or keyword input.

Returns

  • Type: list[str]
  • Direct return expressions: []; [f'content is not valid JSON: {exc}']

Exceptions and behavior

Function _check_json_content calls json.loads; has 2 explicit return paths. No direct raise statement appears in this definition.

View source #L1099-L1107.

vllm_mlx.bench_serve._check_tool_call_count_and_names · function 
vllm_mlx.bench_serve._check_tool_call_count_and_names(checks: dict, tool_calls: list[dict]) -> list[str]

Apply no_tool_calls / tool_call_count / tool_call_names.

Parameters

Name Type Required Default Description
checks dict yes none Required positional or keyword input.
tool_calls list[dict] yes none Required positional or keyword input.

Returns

  • Type: list[str]
  • Direct return expressions: issues

Exceptions and behavior

Function _check_tool_call_count_and_names calls checks.get, issues.append, len, int; returns issues. No direct raise statement appears in this definition.

View source #L1110-L1132.

vllm_mlx.bench_serve._check_tool_call_args · function 
vllm_mlx.bench_serve._check_tool_call_args(required_args: Any, tool_calls: list[dict]) -> list[str]

Validate parsed JSON arguments include the required keys per function.

Parameters

Name Type Required Default Description
required_args Any yes none Required positional or keyword input.
tool_calls list[dict] yes none Required positional or keyword input.

Returns

  • Type: list[str]
  • Direct return expressions: []; issues

Exceptions and behavior

Function _check_tool_call_args calls tc.get('function', {}).get, tc.get, by_name.setdefault(name, []).append, by_name.setdefault; has 2 explicit return paths. No direct raise statement appears in this definition.

View source #L1135-L1174.

vllm_mlx.bench_serve.validate_quality_checks · function 
vllm_mlx.bench_serve.validate_quality_checks(finish_reason: Optional[str], content: str, checks: Optional[dict], *, status_code: int = 200, tool_calls: Optional[list[dict]] = None) -> tuple[bool, list[str]]

Validate content against generic workload quality checks.

Parameters

Name Type Required Default Description
finish_reason Optional[str] yes none Required positional or keyword input.
content str yes none Required positional or keyword input.
checks Optional[dict] yes none Required positional or keyword input.
status_code int no 200 Optional keyword-only input; defaults to 200.
tool_calls Optional[list[dict]] no None Optional keyword-only input; defaults to None.

Returns

  • Type: tuple[bool, list[str]]
  • Direct return expressions: (not issues, issues)

Exceptions and behavior

Function validate_quality_checks calls validate_response, issues.extend, _check_finish_reason, checks.get; returns (not issues, issues). No direct raise statement appears in this definition.

View source #L1177-L1231.

vllm_mlx.bench_serve.compute_summary_stats · function 
vllm_mlx.bench_serve.compute_summary_stats(values: list[float]) -> dict

Compute summary statistics over a list of floats.

Parameters

Name Type Required Default Description
values list[float] yes none Non-empty list of floats to summarise.

Returns

  • Type: dict
  • Direct return expressions: {'mean': mean, 'stddev': stddev, 'min': sorted_vals[0], 'max': sorted_vals[-1], 'p50': _percentile(50), 'p95': _percent…

Exceptions and behavior

Function compute_summary_stats calls ValueError, len, statistics.mean, statistics.stdev; can raise ValueError; returns {'mean': mean, 'stddev': stddev, 'min': sorted_vals[0], 'max': sorted_vals[-1], 'p50': _percentile(50), 'p95': _percent…. Directly raised exceptions: ValueError.

View source #L1234-L1276.

vllm_mlx.bench_serve.compute_summary_stats._percentile · nested function 
vllm_mlx.bench_serve.compute_summary_stats._percentile(p: float) -> float

Nested Function compute_summary_stats._percentile calls int; has 3 explicit return paths.

Parameters

Name Type Required Default Description
p float yes none Required positional or keyword input.

Returns

  • Type: float
  • Direct return expressions: sorted_vals[0]; sorted_vals[-1]; sorted_vals[lo] + frac * (sorted_vals[hi] - sorted_vals[lo])

Exceptions and behavior

Nested Function compute_summary_stats._percentile calls int; has 3 explicit return paths. No direct raise statement appears in this definition.

View source #L1256-L1266.

vllm_mlx.bench_serve.run_concurrent_requests · function 
async vllm_mlx.bench_serve.run_concurrent_requests(client: httpx.AsyncClient, base_url: str, prompts: list[list[dict]], model: str, concurrency: int, max_tokens: int = 256, enable_thinking: Optional[bool] = None, extra_body: Optional[dict] = None, do_validate: bool = True) -> list[dict]

Fire concurrency concurrent streaming requests and collect results.

Parameters

Name Type Required Default Description
client httpx.AsyncClient yes none An open :class:httpx.AsyncClient.
base_url str yes none Base URL of the server.
prompts list[list[dict]] yes none List of message dicts to cycle through.
model str yes none Model ID to target.
concurrency int yes none Number of simultaneous requests to fire.
max_tokens int no 256 Maximum tokens to generate per request (default 256).
enable_thinking Optional[bool] no None Passed through to :func:stream_chat_completion.
extra_body Optional[dict] no None Passed through to :func:stream_chat_completion.
do_validate bool no True When True, call :func:validate_response on each result and add a "validated" key.

Returns

  • Type: list[dict]
  • Direct return expressions: list(results)

Exceptions and behavior

Function run_concurrent_requests calls itertools.cycle, next, range, asyncio.gather; awaits asynchronous work; returns list(results). No direct raise statement appears in this definition.

View source #L1279-L1342.

vllm_mlx.bench_serve.run_concurrent_requests._single · nested function 
async vllm_mlx.bench_serve.run_concurrent_requests._single(messages: list[dict]) -> dict

Nested Function run_concurrent_requests._single calls stream_chat_completion, validate_response, result.get, str; awaits asynchronous work; has 2 explicit return paths.

Parameters

Name Type Required Default Description
messages list[dict] yes none Required positional or keyword input.

Returns

  • Type: dict
  • Direct return expressions: result; err

Exceptions and behavior

Nested Function run_concurrent_requests._single calls stream_chat_completion, validate_response, result.get, str; awaits asynchronous work; has 2 explicit return paths. No direct raise statement appears in this definition.

View source #L1315-L1339.

vllm_mlx.bench_serve._summary_or_empty · function 
vllm_mlx.bench_serve._summary_or_empty(values: list[float]) -> dict

Function _summary_or_empty calls compute_summary_stats; returns compute_summary_stats(values) if values else {}.

Parameters

Name Type Required Default Description
values list[float] yes none Required positional or keyword input.

Returns

  • Type: dict
  • Direct return expressions: compute_summary_stats(values) if values else {}

Exceptions and behavior

Function _summary_or_empty calls compute_summary_stats; returns compute_summary_stats(values) if values else {}. No direct raise statement appears in this definition.

View source #L1345-L1346.

vllm_mlx.bench_serve._resolve_max_tokens · function 
vllm_mlx.bench_serve._resolve_max_tokens(case: WorkloadCase, workload: Workload) -> int

Return the effective max_tokens for a case, falling back to workload defaults and finally to 256.

Parameters

Name Type Required Default Description
case WorkloadCase yes none Required positional or keyword input.
workload Workload yes none Required positional or keyword input.

Returns

  • Type: int
  • Direct return expressions: int(case.max_tokens or workload.defaults.get('max_tokens', 256))

Exceptions and behavior

Function _resolve_max_tokens calls int, workload.defaults.get; returns int(case.max_tokens or workload.defaults.get('max_tokens', 256)). No direct raise statement appears in this definition.

View source #L1349-L1352.

vllm_mlx.bench_serve._assemble_case_request_kwargs · function 
vllm_mlx.bench_serve._assemble_case_request_kwargs(case: WorkloadCase, workload: Workload, model: str) -> dict

Build the keyword-arguments dict passed to stream_chat_completion for one case, applying max_tokens fallback and converting policy_timeout_ms to seconds.

Parameters

Name Type Required Default Description
case WorkloadCase yes none Required positional or keyword input.
workload Workload yes none Required positional or keyword input.
model str yes none Required positional or keyword input.

Returns

  • Type: dict
  • Direct return expressions: {'messages': case.messages, 'model': model, 'max_tokens': _resolve_max_tokens(case, workload), 'enable_thinking': case.…

Exceptions and behavior

Function _assemble_case_request_kwargs calls _resolve_max_tokens; returns {'messages': case.messages, 'model': model, 'max_tokens': _resolve_max_tokens(case, workload), 'enable_thinking': case.…. No direct raise statement appears in this definition.

View source #L1355-L1372.

vllm_mlx.bench_serve._empty_completion_result · function 
vllm_mlx.bench_serve._empty_completion_result() -> dict

Zero-valued completion result used when stream_chat_completion raises.

Parameters

This callable has no explicit inputs.

Returns

  • Type: dict
  • Direct return expressions: {'ttft_ms': 0.0, 'tpot_ms': 0.0, 'e2e_latency_ms': 0.0, 'gen_tps': 0.0, 'prompt_tps': 0.0, 'prompt_tokens': 0, 'complet…

Exceptions and behavior

Function _empty_completion_result returns {'ttft_ms': 0.0, 'tpot_ms': 0.0, 'e2e_latency_ms': 0.0, 'gen_tps': 0.0, 'prompt_tps': 0.0, 'prompt_tokens': 0, 'complet…. No direct raise statement appears in this definition.

View source #L1375-L1390.

vllm_mlx.bench_serve._fetch_post_run_status · function 
async vllm_mlx.bench_serve._fetch_post_run_status(client: httpx.AsyncClient, base_url: str) -> dict

GET /v1/status after a case run, swallowing transport errors so a missing or temporarily-unavailable status endpoint does not fail the case record.

Parameters

Name Type Required Default Description
client httpx.AsyncClient yes none Required positional or keyword input.
base_url str yes none Required positional or keyword input.

Returns

  • Type: dict
  • Direct return expressions: resp.json(); {}

Exceptions and behavior

Function _fetch_post_run_status calls client.get, resp.raise_for_status, resp.json; awaits asynchronous work; has 2 explicit return paths. No direct raise statement appears in this definition.

View source #L1393-L1402.

vllm_mlx.bench_serve._compute_within_policy_timeout · function 
vllm_mlx.bench_serve._compute_within_policy_timeout(timeout_ms: Optional[int], *, error_present: bool, e2e_latency_ms: float) -> Optional[bool]

Resolve the policy.within_timeout field.

Parameters

Name Type Required Default Description
timeout_ms Optional[int] yes none Required positional or keyword input.
error_present bool yes none Required keyword-only input.
e2e_latency_ms float yes none Required keyword-only input.

Returns

  • Type: Optional[bool]
  • Direct return expressions: None; False; e2e_latency_ms <= timeout_ms

Exceptions and behavior

Function _compute_within_policy_timeout has 3 explicit return paths. No direct raise statement appears in this definition.

View source #L1405-L1418.

vllm_mlx.bench_serve._build_tool_calls_summary · function 
vllm_mlx.bench_serve._build_tool_calls_summary(tool_calls: Any) -> Optional[dict]

Compact summary of streamed tool calls for the case record.

Parameters

Name Type Required Default Description
tool_calls Any yes none Required positional or keyword input.

Returns

  • Type: Optional[dict]
  • Direct return expressions: None; {'count': len(tool_calls), 'names': sorted((tc.get('function', {}).get('name', '') for tc in tool_calls)), 'raw': tool_…

Exceptions and behavior

Function _build_tool_calls_summary calls len, sorted, tc.get('function', {}).get, tc.get; has 2 explicit return paths. No direct raise statement appears in this definition.

View source #L1421-L1434.

vllm_mlx.bench_serve._build_workload_record · function 
vllm_mlx.bench_serve._build_workload_record(*, case: WorkloadCase, workload: Workload, model: str, runtime: dict, hardware: dict, run_id: str, timestamp: str, started_wall: str, repetition: int, result: dict, error: str, quality_ok: bool, quality_issues: list[str], content: str, cache_hits_delta: int, cache_misses_delta: int, tokens_saved_delta: int, status_after: dict, cache_reset: Optional[dict], include_content: bool) -> dict

Assemble the JSON-serializable workload-case record from the raw inputs and the completion result.

Parameters

Name Type Required Default Description
case WorkloadCase yes none Required keyword-only input.
workload Workload yes none Required keyword-only input.
model str yes none Required keyword-only input.
runtime dict yes none Required keyword-only input.
hardware dict yes none Required keyword-only input.
run_id str yes none Required keyword-only input.
timestamp str yes none Required keyword-only input.
started_wall str yes none Required keyword-only input.
repetition int yes none Required keyword-only input.
result dict yes none Required keyword-only input.
error str yes none Required keyword-only input.
quality_ok bool yes none Required keyword-only input.
quality_issues list[str] yes none Required keyword-only input.
content str yes none Required keyword-only input.
cache_hits_delta int yes none Required keyword-only input.
cache_misses_delta int yes none Required keyword-only input.
tokens_saved_delta int yes none Required keyword-only input.
status_after dict yes none Required keyword-only input.
cache_reset Optional[dict] yes none Required keyword-only input.
include_content bool yes none Required keyword-only input.

Returns

  • Type: dict
  • Direct return expressions: record

Exceptions and behavior

Function _build_workload_record calls list, _resolve_max_tokens, len, _compute_within_policy_timeout; returns record. No direct raise statement appears in this definition.

View source #L1437-L1515.

vllm_mlx.bench_serve.run_workload_case · function 
async vllm_mlx.bench_serve.run_workload_case(client: httpx.AsyncClient, base_url: str, *, workload: Workload, case: WorkloadCase, model: str, runtime: dict, hardware: dict, run_id: str, timestamp: str, repetition: int = 0, scrape: bool = True, include_content: bool = False, cache_reset: Optional[dict] = None) -> dict

Run one workload case and return a JSON-serializable result.

Parameters

Name Type Required Default Description
client httpx.AsyncClient yes none Required positional or keyword input.
base_url str yes none Required positional or keyword input.
workload Workload yes none Required keyword-only input.
case WorkloadCase yes none Required keyword-only input.
model str yes none Required keyword-only input.
runtime dict yes none Required keyword-only input.
hardware dict yes none Required keyword-only input.
run_id str yes none Required keyword-only input.
timestamp str yes none Required keyword-only input.
repetition int no 0 Optional keyword-only input; defaults to 0.
scrape bool no True Optional keyword-only input; defaults to True.
include_content bool no False Optional keyword-only input; defaults to False.
cache_reset Optional[dict] no None Optional keyword-only input; defaults to None.

Returns

  • Type: dict
  • Direct return expressions: _build_workload_record(case=case, workload=workload, model=model, runtime=runtime, hardware=hardware, run_id=run_id, ti…

Exceptions and behavior

Function run_workload_case calls scrape_metrics, datetime.now(timezone.utc).isoformat, datetime.now, _assemble_case_request_kwargs; awaits asynchronous work; returns _build_workload_record(case=case, workload=workload, model=model, runtime=runtime, hardware=hardware, run_id=run_id, ti…. No direct raise statement appears in this definition.

View source #L1518-L1593.

vllm_mlx.bench_serve._group_results_by_case_id · function 
vllm_mlx.bench_serve._group_results_by_case_id(results: list[dict]) -> dict[str, list[dict]]

Bucket workload case records by their case_id field, defaulting a missing case_id to the empty string so the grouping is stable.

Parameters

Name Type Required Default Description
results list[dict] yes none Required positional or keyword input.

Returns

  • Type: dict[str, list[dict]]
  • Direct return expressions: cases

Exceptions and behavior

Function _group_results_by_case_id calls cases.setdefault(str(result.get('case_id', '')), []).append, cases.setdefault, str, result.get; returns cases. No direct raise statement appears in this definition.

View source #L1596-L1602.

vllm_mlx.bench_serve._summarize_case · function 
vllm_mlx.bench_serve._summarize_case(case_results: list[dict]) -> dict

Build the per-case summary block.

Parameters

Name Type Required Default Description
case_results list[dict] yes none Required positional or keyword input.

Returns

  • Type: dict
  • Direct return expressions: {'sample_count': len(case_results), 'repetitions': sorted({int(r.get('repetition', 0)) for r in case_results if r.get('…

Exceptions and behavior

Function _summarize_case calls r['quality'].get, r['policy'].get, len, sorted; returns {'sample_count': len(case_results), 'repetitions': sorted({int(r.get('repetition', 0)) for r in case_results if r.get('…. No direct raise statement appears in this definition.

View source #L1605-L1648.

vllm_mlx.bench_serve.summarize_workload_results · function 
vllm_mlx.bench_serve.summarize_workload_results(results: list[dict]) -> dict

Aggregate workload case records into stable qualification summary stats.

Parameters

Name Type Required Default Description
results list[dict] yes none Required positional or keyword input.

Returns

  • Type: dict
  • Direct return expressions: {'case_count': len(results), 'unique_case_count': len(cases), 'repetition_count': max((len(summary['repetitions']) for …

Exceptions and behavior

Function summarize_workload_results calls r['policy'].get, _group_results_by_case_id, _summarize_case, sorted; returns {'case_count': len(results), 'unique_case_count': len(cases), 'repetition_count': max((len(summary['repetitions']) for …. No direct raise statement appears in this definition.

View source #L1651-L1689.

vllm_mlx.bench_serve.run_bench_serve_workload · function 
async vllm_mlx.bench_serve.run_bench_serve_workload(*, url: str, workload_path: str, model: Optional[str] = None, output_path: Optional[str] = None, output_format: str = 'json', scrape: bool = True, include_content: bool = False, request_timeout_s: Optional[float] = 300.0, repetitions: int = 1, cache_policy: Optional[str] = None) -> dict

Run a declarative workload against a running server.

Parameters

Name Type Required Default Description
url str yes none Required keyword-only input.
workload_path str yes none Required keyword-only input.
model Optional[str] no None Optional keyword-only input; defaults to None.
output_path Optional[str] no None Optional keyword-only input; defaults to None.
output_format str no 'json' Optional keyword-only input; defaults to 'json'.
scrape bool no True Optional keyword-only input; defaults to True.
include_content bool no False Optional keyword-only input; defaults to False.
request_timeout_s Optional[float] no 300.0 Optional keyword-only input; defaults to 300.0.
repetitions int no 1 Optional keyword-only input; defaults to 1.
cache_policy Optional[str] no None Optional keyword-only input; defaults to None.

Returns

  • Type: dict
  • Direct return expressions: payload

Exceptions and behavior

Function run_bench_serve_workload calls ValueError, load_workload, _normalize_cache_policy, workload.defaults.get; awaits asynchronous work; can raise ValueError; returns payload. Directly raised exceptions: ValueError.

View source #L1692-L1818.

vllm_mlx.bench_serve._result_to_dict · function 
vllm_mlx.bench_serve._result_to_dict(r: BenchServeResult) -> dict

Convert a :class:BenchServeResult to an ordered dict.

Parameters

Name Type Required Default Description
r BenchServeResult yes none Required positional or keyword input.

Returns

  • Type: dict
  • Direct return expressions: {f.name: getattr(r, f.name) for f in _dataclasses.fields(r)}

Exceptions and behavior

Function _result_to_dict calls getattr, _dataclasses.fields; returns {f.name: getattr(r, f.name) for f in _dataclasses.fields(r)}. No direct raise statement appears in this definition.

View source #L1840-L1846.

vllm_mlx.bench_serve.format_table · function 
vllm_mlx.bench_serve.format_table(results: list[BenchServeResult]) -> str

Render a human-readable terminal table of benchmark results.

Parameters

Name Type Required Default Description
results list[BenchServeResult] yes none List of :class:BenchServeResult instances.

Returns

  • Type: str
  • Direct return expressions: _tabulate(rows, headers=_TABLE_COLUMNS, tablefmt='simple')

Exceptions and behavior

Function format_table calls _result_to_dict, d.get, isinstance, round; returns _tabulate(rows, headers=_TABLE_COLUMNS, tablefmt='simple'). No direct raise statement appears in this definition.

View source #L1849-L1871.

vllm_mlx.bench_serve.format_json · function 
vllm_mlx.bench_serve.format_json(results: list[BenchServeResult]) -> str

Serialize benchmark results as a JSON array.

Parameters

Name Type Required Default Description
results list[BenchServeResult] yes none List of :class:BenchServeResult instances.

Returns

  • Type: str
  • Direct return expressions: json.dumps([_result_to_dict(r) for r in results], indent=2)

Exceptions and behavior

Function format_json calls json.dumps, _result_to_dict; returns json.dumps([_result_to_dict(r) for r in results], indent=2). No direct raise statement appears in this definition.

View source #L1874-L1885.

vllm_mlx.bench_serve.format_csv · function 
vllm_mlx.bench_serve.format_csv(results: list[BenchServeResult]) -> str

Serialize benchmark results as CSV with a header row.

Parameters

Name Type Required Default Description
results list[BenchServeResult] yes none List of :class:BenchServeResult instances.

Returns

  • Type: str
  • Direct return expressions: buf.getvalue()

Exceptions and behavior

Function format_csv calls io.StringIO, csv_mod.DictWriter, writer.writeheader, writer.writerow; returns buf.getvalue(). No direct raise statement appears in this definition.

View source #L1888-L1904.

vllm_mlx.bench_serve._sql_escape · function 
vllm_mlx.bench_serve._sql_escape(value) -> str

Escape a Python value for use as a SQL literal.

Parameters

Name Type Required Default Description
value not annotated yes none Required positional or keyword input.

Returns

  • Type: str
  • Direct return expressions: 'NULL'; '1' if value else '0'; str(value); f"'{escaped}'"

Exceptions and behavior

Function _sql_escape calls isinstance, math.isnan, math.isinf, str; has 4 explicit return paths. No direct raise statement appears in this definition.

View source #L1907-L1927.

vllm_mlx.bench_serve.format_sql · function 
vllm_mlx.bench_serve.format_sql(results: list[BenchServeResult]) -> str

Emit a SQL CREATE TABLE IF NOT EXISTS statement and INSERT rows.

Parameters

Name Type Required Default Description
results list[BenchServeResult] yes none List of :class:BenchServeResult instances.

Returns

  • Type: str
  • Direct return expressions: '\n'.join(lines)

Exceptions and behavior

Function format_sql calls _result_to_dict, ', '.join, _sql_escape, lines.append; returns '\n'.join(lines). No direct raise statement appears in this definition.

View source #L1945-L1964.

vllm_mlx.bench_serve._write_sqlite_rows · function 
vllm_mlx.bench_serve._write_sqlite_rows(output_path: str, *, table: str, schema: str, columns: list[str], rows: list[dict]) -> None

Append benchmark rows to a SQLite database.

Parameters

Name Type Required Default Description
output_path str yes none Required positional or keyword input.
table str yes none Required keyword-only input.
schema str yes none Required keyword-only input.
columns list[str] yes none Required keyword-only input.
rows list[dict] yes none Required keyword-only input.

Returns

  • Type: None

Exceptions and behavior

Function _write_sqlite_rows calls Path(output_path).expanduser, Path, _validate_sql_identifier, ', '.join. No direct raise statement appears in this definition.

View source #L1967-L1990.

vllm_mlx.bench_serve._validate_sql_identifier · function 
vllm_mlx.bench_serve._validate_sql_identifier(identifier: str, *, kind: str) -> None

Reject unsafe SQL identifiers before string interpolation.

Parameters

Name Type Required Default Description
identifier str yes none Required positional or keyword input.
kind str yes none Required keyword-only input.

Returns

  • Type: None

Exceptions and behavior

Function _validate_sql_identifier calls _SQL_IDENTIFIER_RE.fullmatch, ValueError; can raise ValueError. Directly raised exceptions: ValueError.

View source #L1993-L1996.

vllm_mlx.bench_serve.write_sqlite · function 
vllm_mlx.bench_serve.write_sqlite(results: list[BenchServeResult], output_path: str) -> None

Append prompt-sweep benchmark results to a SQLite database.

Parameters

Name Type Required Default Description
results list[BenchServeResult] yes none Required positional or keyword input.
output_path str yes none Required positional or keyword input.

Returns

  • Type: None

Exceptions and behavior

Function write_sqlite calls _result_to_dict, _write_sqlite_rows. No direct raise statement appears in this definition.

View source #L1999-L2009.

vllm_mlx.bench_serve._workload_record_to_row · function 
vllm_mlx.bench_serve._workload_record_to_row(record: dict) -> dict

Function _workload_record_to_row calls record.get, metrics.get, ','.join, hardware.get; returns {'run_id': record.get('run_id', ''), 'timestamp': record.get('timestamp', ''), 'workload': record.get('workload', ''), ….

Parameters

Name Type Required Default Description
record dict yes none Required positional or keyword input.

Returns

  • Type: dict
  • Direct return expressions: {'run_id': record.get('run_id', ''), 'timestamp': record.get('timestamp', ''), 'workload': record.get('workload', ''), …

Exceptions and behavior

Function _workload_record_to_row calls record.get, metrics.get, ','.join, hardware.get; returns {'run_id': record.get('run_id', ''), 'timestamp': record.get('timestamp', ''), 'workload': record.get('workload', ''), …. No direct raise statement appears in this definition.

View source #L2069-L2119.

vllm_mlx.bench_serve.format_workload_table · function 
vllm_mlx.bench_serve.format_workload_table(payload: dict) -> str

Format workload result records as a compact human-readable table.

Parameters

Name Type Required Default Description
payload dict yes none Required positional or keyword input.

Returns

  • Type: str
  • Direct return expressions: _tabulate(rows, headers=_WORKLOAD_TABLE_COLUMNS, tablefmt='simple')

Exceptions and behavior

Function format_workload_table calls payload.get, _workload_record_to_row, rows.append, isinstance; returns _tabulate(rows, headers=_WORKLOAD_TABLE_COLUMNS, tablefmt='simple'). No direct raise statement appears in this definition.

View source #L2122-L2134.

vllm_mlx.bench_serve.format_workload_json · function 
vllm_mlx.bench_serve.format_workload_json(payload: dict) -> str

Serialize a workload result payload as indented JSON.

Parameters

Name Type Required Default Description
payload dict yes none Required positional or keyword input.

Returns

  • Type: str
  • Direct return expressions: json.dumps(payload, indent=2)

Exceptions and behavior

Function format_workload_json calls json.dumps; returns json.dumps(payload, indent=2). No direct raise statement appears in this definition.

View source #L2137-L2140.

vllm_mlx.bench_serve.format_workload_csv · function 
vllm_mlx.bench_serve.format_workload_csv(payload: dict) -> str

Serialize workload result records with the stable CSV column contract.

Parameters

Name Type Required Default Description
payload dict yes none Required positional or keyword input.

Returns

  • Type: str
  • Direct return expressions: buf.getvalue()

Exceptions and behavior

Function format_workload_csv calls io.StringIO, csv_mod.DictWriter, writer.writeheader, payload.get; returns buf.getvalue(). No direct raise statement appears in this definition.

View source #L2143-L2151.

vllm_mlx.bench_serve.format_workload_sql · function 
vllm_mlx.bench_serve.format_workload_sql(payload: dict) -> str

Render SQL statements that create and populate the workload table.

Parameters

Name Type Required Default Description
payload dict yes none Required positional or keyword input.

Returns

  • Type: str
  • Direct return expressions: '\n'.join(lines)

Exceptions and behavior

Function format_workload_sql calls payload.get, _workload_record_to_row, ', '.join, _sql_escape; returns '\n'.join(lines). No direct raise statement appears in this definition.

View source #L2170-L2180.

vllm_mlx.bench_serve.write_workload_sqlite · function 
vllm_mlx.bench_serve.write_workload_sqlite(payload: dict, output_path: str) -> None

Append workload result records to a SQLite database.

Parameters

Name Type Required Default Description
payload dict yes none Required positional or keyword input.
output_path str yes none Required positional or keyword input.

Returns

  • Type: None

Exceptions and behavior

Function write_workload_sqlite calls _workload_record_to_row, payload.get, _write_sqlite_rows. No direct raise statement appears in this definition.

View source #L2183-L2193.

vllm_mlx.bench_serve.format_workload_payload · function 
vllm_mlx.bench_serve.format_workload_payload(payload: dict, fmt: str = 'json') -> str

Serialize a workload payload in the requested text output format.

Parameters

Name Type Required Default Description
payload dict yes none Required positional or keyword input.
fmt str no 'json' Optional positional or keyword input; defaults to 'json'.

Returns

  • Type: str
  • Direct return expressions: format_workload_json(payload); format_workload_csv(payload); format_workload_sql(payload); format_workload_table(payload)

Exceptions and behavior

Function format_workload_payload calls format_workload_json, format_workload_csv, format_workload_sql, format_workload_table; can raise ValueError; has 4 explicit return paths. Directly raised exceptions: ValueError.

View source #L2196-L2211.

vllm_mlx.bench_serve.run_bench_serve · function 
async vllm_mlx.bench_serve.run_bench_serve(url: str = 'http://127.0.0.1:8080', model: Optional[str] = None, prompt_sets: list[str] = None, prompt_file: Optional[str] = None, concurrencies: list[int] = None, max_tokens: int = 256, repetitions: int = 3, warmup: int = 1, thinking_values: list[Optional[bool]] = None, extra_bodies: list[str] = None, output_path: Optional[str] = None, fmt: str = 'table', do_validate: bool = True, scrape: bool = True, tag: Optional[str] = None, override_fields: Optional[dict] = None, system_prompt_file: Optional[str] = None, skip_preflight_token_count: bool = False) -> list[BenchServeResult]

Run the full bench-serve sweep against a running vllm-mlx server.

Parameters

Name Type Required Default Description
url str no 'http://127.0.0.1:8080' Base URL of the server.
model Optional[str] no None Model ID to use. If None, auto-detected from the server.
prompt_sets list[str] no None List of prompt set names or paths. Defaults to ["short", "medium", "long"].
prompt_file Optional[str] no None Optional path to an extra prompt file to include.
concurrencies list[int] no None Concurrency levels to sweep. Defaults to [1, 4].
max_tokens int no 256 Maximum tokens to generate per request.
repetitions int no 3 Number of repetitions per sweep config.
warmup int no 1 Number of warmup rounds before the first measured repetition.
thinking_values list[Optional[bool]] no None Values for enable_thinking. Defaults to [None].
extra_bodies list[str] no None JSON strings for extra body parameters. Defaults to [""] (no extra body).
output_path Optional[str] no None File path to write results to. If None, prints to stdout.
fmt str no 'table' Output format — one of "table", "json", "csv", "sql", or "sqlite".
do_validate bool no True Whether to validate each response.
scrape bool no True Whether to scrape /metrics before and after each run.
tag Optional[str] no None Optional tag string stored in every result row.
override_fields Optional[dict] no None Dict of field names to override on every result.
system_prompt_file Optional[str] no None Optional positional or keyword input; defaults to None.
skip_preflight_token_count bool no False Optional positional or keyword input; defaults to False.

Returns

  • Type: list[BenchServeResult]
  • Direct return expressions: []; results

Exceptions and behavior

Function run_bench_serve calls str, uuid.uuid4, datetime.now(timezone.utc).isoformat, datetime.now; awaits asynchronous work; can raise ValueError; has 2 explicit return paths. Directly raised exceptions: ValueError.

View source #L2221-L2638.

vllm_mlx.bench_serve.run_bench_serve._mean · nested function 
vllm_mlx.bench_serve.run_bench_serve._mean(key: str) -> float

Nested Function run_bench_serve._mean calls statistics.mean; returns statistics.mean(vals) if vals else 0.0.

Parameters

Name Type Required Default Description
key str yes none Required positional or keyword input.

Returns

  • Type: float
  • Direct return expressions: statistics.mean(vals) if vals else 0.0

Exceptions and behavior

Nested Function run_bench_serve._mean calls statistics.mean; returns statistics.mean(vals) if vals else 0.0. No direct raise statement appears in this definition.

View source #L2522-L2526.

Complete symbol map

This map also includes private definitions and nested helpers. The signature column exposes every explicit input even when an internal helper has no dedicated parameter prose.

Symbol Kind Signature and inputs What it does Source
WorkloadCase class WorkloadCase(case_id: str, messages: list[dict], request_path: Optional[str] = None, max_tokens: Optional[int] = None, enable_thinking: Optional[bool] = None, extra_body: Optional[dict] = None, policy_timeout_ms: Optional[int] = None, checks: Optional[dict] = None, tags: tuple[str, ...] = ()) One declarative benchmark case for contract-style serving tests. #L51-L62
Workload class Workload(name: str, description: str, defaults: dict, cases: list[WorkloadCase]) Normalized bench-serve workload manifest. #L66-L72
load_prompt_set function load_prompt_set(name_or_path: str) -> list[list[dict]] Load a prompt set by builtin name or file path. #L75-L137
_require_message_list function _require_message_list(value: Any, *, label: str) -> list[dict] Function _require_message_list calls isinstance, ValueError, enumerate; can raise ValueError; returns value. #L140-L148
_load_case_request function _load_case_request(path: str, *, workload_path: Path, case_id: str) -> dict Function _load_case_request calls Path(path).expanduser, Path, request_path.is_absolute, request_path.open; can raise ValueError; returns request. #L151-L159
_request_extra_body function _request_extra_body(request: dict) -> dict Function _request_extra_body calls request.items; returns {key: value for key, value in request.items() if key not in reserved}. #L162-L171
_first_not_none function _first_not_none(*values: Any) -> Any Function _first_not_none has 2 explicit return paths. #L174-L178
_normalize_tags function _normalize_tags(tags: Any, *, case_id: str) -> tuple[str, ...] Coerce a workload case's tags field to a tuple of strings. #L181-L191
_merge_case_checks function _merge_case_checks(default_checks: Any, case_checks: Any, *, case_id: str) -> Optional[dict] Merge a case's checks over the workload defaults. #L194-L227
_build_workload_case function _build_workload_case(item: Any, idx: int, *, defaults: dict, workload_path: Path) -> WorkloadCase Construct one WorkloadCase from a raw workload entry. #L230-L300
load_workload function load_workload(path: str \| Path) -> Workload Load a declarative serving benchmark workload. #L303-L335
BenchServeResult class BenchServeResult(run_id: str = '', timestamp: str = '', tag: str = '', chip: str = '', gpu_cores: int = 0, memory_gb: float = 0.0, bandwidth_gbs: float = 0.0, os_version: str = '', model_id: str = '', model_type: str = '', engine_type: str = '', mtp_enabled: bool = False, specprefill: bool = False, kv_quant: str = '', cache_type: str = '', prompt_set: str = '', concurrency: int = 1, max_tokens: int = 256, enable_thinking: Optional[bool] = None, extra_body: str = '', repetition: int = 0, prompt_tokens: int = 0, ttft_ms: float = 0.0, tpot_ms: float = 0.0, e2e_latency_ms: float = 0.0, gen_tps: float = 0.0, prompt_tps: float = 0.0, throughput_tps: float = 0.0, requests_per_s: float = 0.0, metal_active_gb: float = 0.0, metal_peak_gb: float = 0.0, metal_cache_gb: float = 0.0, cache_hits: int = 0, cache_misses: int = 0, cache_hit_rate: float = 0.0, tokens_saved: int = 0, validated: bool = True) Aggregated results from a single bench-serve run configuration. #L344-L400
expand_sweep function expand_sweep(prompt_sets: list[str], concurrencies: list[int], thinking_values: list[Optional[bool]], extra_bodies: list[str], repetitions: int) -> list[SweepConfig] Expand sweep parameters into a flat list of configurations. #L411-L444
parse_health_response function parse_health_response(data: dict) -> dict Extract model identity fields from a GET /health response. #L452-L467
parse_status_response function parse_status_response(data: dict) -> dict Extract metal and cache info from a GET /v1/status response. #L470-L496
parse_metrics_text function parse_metrics_text(text: str) -> dict Parse Prometheus text exposition format from GET /metrics. #L499-L521
parse_metrics_text._extract nested function parse_metrics_text._extract(metric_name: str) -> int Nested Function parse_metrics_text._extract calls re.escape, re.search, int, m.group; returns int(m.group(1)) if m else 0. #L512-L515
detect_hardware_fingerprint function detect_hardware_fingerprint() -> dict Return a hardware fingerprint dict for the current machine. #L524-L573
auto_detect_runtime function async auto_detect_runtime(client: httpx.AsyncClient, base_url: str) -> dict Query the running server and return a runtime descriptor dict. #L576-L642
scrape_metrics function async scrape_metrics(client: httpx.AsyncClient, base_url: str) -> dict Scrape Prometheus metrics from the server. #L645-L661
clear_runtime_cache function async clear_runtime_cache(client: httpx.AsyncClient, base_url: str) -> dict Clear server-side runtime caches and return a JSON-serializable event. #L664-L684
_normalize_cache_policy function _normalize_cache_policy(value: Optional[str]) -> str Normalize cache-policy spelling from CLI or workload JSON. #L687-L698
parse_sse_line function parse_sse_line(line: str) -> Optional[dict] Parse one Server-Sent Events line from a streaming chat completion. #L706-L754
_cancel_server_request function async _cancel_server_request(client: httpx.AsyncClient, base_url: str, request_id: Optional[str]) -> None Best-effort server-side cancellation for timed-out workload streams. #L757-L770
accumulate_tool_calls function accumulate_tool_calls(acc: dict[int, dict], delta_list: list[dict]) -> None Merge streamed OpenAI tool-call deltas into acc by index. #L773-L792
finalize_tool_calls function finalize_tool_calls(acc: dict[int, dict]) -> list[dict] Return accumulated tool calls in stream index order. #L795-L797
compute_request_metrics function compute_request_metrics(t_start: float, t_first_token: float, token_times: list, t_end: float, prompt_tokens: int, completion_tokens: int) -> dict Compute standard latency and throughput metrics for a single request. #L800-L852
count_prompt_tokens function async count_prompt_tokens(client: httpx.AsyncClient, base_url: str, messages: list[dict], model: str) -> int Count prompt tokens for a message list by sending a 1-token request. #L855-L889
stream_chat_completion function async stream_chat_completion(client: httpx.AsyncClient, base_url: str, messages: list[dict], model: str, max_tokens: int = 256, enable_thinking: Optional[bool] = None, extra_body: Optional[dict] = None, timeout_s: Optional[float] = None) -> dict Send a streaming chat completion and collect per-token timing data. #L892-L1012
stream_chat_completion._consume_stream nested function async stream_chat_completion._consume_stream() -> None Nested Function stream_chat_completion._consume_stream calls client.stream, response.raise_for_status, response.aiter_lines, parse_sse_line. #L946-L975
validate_response function validate_response(finish_reason: Optional[str], content: str, status_code: int, *, tool_calls: Optional[list[dict]] = None) -> tuple[bool, str] Validate a single streaming response result. #L1020-L1049
_check_finish_reason function _check_finish_reason(allowed: Any, finish_reason: Optional[str]) -> list[str] Verify finish_reason is in the allowed set, if one is configured. #L1052-L1059
_check_length_bounds function _check_length_bounds(min_chars: Any, max_chars: Any, content: str) -> list[str] Apply min_chars / max_chars content-length bounds. #L1062-L1069
_check_regex_patterns function _check_regex_patterns(patterns: Any, content: str, *, kind: str, expect_match: bool) -> list[str] Validate that each pattern either matches or does not, per expect_match. #L1072-L1096
_check_json_content function _check_json_content(should_be_json: Any, content: str) -> list[str] Verify content parses as JSON when checks['json'] is truthy. #L1099-L1107
_check_tool_call_count_and_names function _check_tool_call_count_and_names(checks: dict, tool_calls: list[dict]) -> list[str] Apply no_tool_calls / tool_call_count / tool_call_names. #L1110-L1132
_check_tool_call_args function _check_tool_call_args(required_args: Any, tool_calls: list[dict]) -> list[str] Validate parsed JSON arguments include the required keys per function. #L1135-L1174
validate_quality_checks function validate_quality_checks(finish_reason: Optional[str], content: str, checks: Optional[dict], *, status_code: int = 200, tool_calls: Optional[list[dict]] = None) -> tuple[bool, list[str]] Validate content against generic workload quality checks. #L1177-L1231
compute_summary_stats function compute_summary_stats(values: list[float]) -> dict Compute summary statistics over a list of floats. #L1234-L1276
compute_summary_stats._percentile nested function compute_summary_stats._percentile(p: float) -> float Nested Function compute_summary_stats._percentile calls int; has 3 explicit return paths. #L1256-L1266
run_concurrent_requests function async run_concurrent_requests(client: httpx.AsyncClient, base_url: str, prompts: list[list[dict]], model: str, concurrency: int, max_tokens: int = 256, enable_thinking: Optional[bool] = None, extra_body: Optional[dict] = None, do_validate: bool = True) -> list[dict] Fire concurrency concurrent streaming requests and collect results. #L1279-L1342
run_concurrent_requests._single nested function async run_concurrent_requests._single(messages: list[dict]) -> dict Nested Function run_concurrent_requests._single calls stream_chat_completion, validate_response, result.get, str; awaits asynchronous work; has 2 explicit return paths. #L1315-L1339
_summary_or_empty function _summary_or_empty(values: list[float]) -> dict Function _summary_or_empty calls compute_summary_stats; returns compute_summary_stats(values) if values else {}. #L1345-L1346
_resolve_max_tokens function _resolve_max_tokens(case: WorkloadCase, workload: Workload) -> int Return the effective max_tokens for a case, falling back to workload defaults and finally to 256. #L1349-L1352
_assemble_case_request_kwargs function _assemble_case_request_kwargs(case: WorkloadCase, workload: Workload, model: str) -> dict Build the keyword-arguments dict passed to stream_chat_completion for one case, applying max_tokens fallback and converting policy_timeout_ms to seconds. #L1355-L1372
_empty_completion_result function _empty_completion_result() -> dict Zero-valued completion result used when stream_chat_completion raises. #L1375-L1390
_fetch_post_run_status function async _fetch_post_run_status(client: httpx.AsyncClient, base_url: str) -> dict GET /v1/status after a case run, swallowing transport errors so a missing or temporarily-unavailable status endpoint does not fail the case record. #L1393-L1402
_compute_within_policy_timeout function _compute_within_policy_timeout(timeout_ms: Optional[int], *, error_present: bool, e2e_latency_ms: float) -> Optional[bool] Resolve the policy.within_timeout field. #L1405-L1418
_build_tool_calls_summary function _build_tool_calls_summary(tool_calls: Any) -> Optional[dict] Compact summary of streamed tool calls for the case record. #L1421-L1434
_build_workload_record function _build_workload_record(*, case: WorkloadCase, workload: Workload, model: str, runtime: dict, hardware: dict, run_id: str, timestamp: str, started_wall: str, repetition: int, result: dict, error: str, quality_ok: bool, quality_issues: list[str], content: str, cache_hits_delta: int, cache_misses_delta: int, tokens_saved_delta: int, status_after: dict, cache_reset: Optional[dict], include_content: bool) -> dict Assemble the JSON-serializable workload-case record from the raw inputs and the completion result. #L1437-L1515
run_workload_case function async run_workload_case(client: httpx.AsyncClient, base_url: str, *, workload: Workload, case: WorkloadCase, model: str, runtime: dict, hardware: dict, run_id: str, timestamp: str, repetition: int = 0, scrape: bool = True, include_content: bool = False, cache_reset: Optional[dict] = None) -> dict Run one workload case and return a JSON-serializable result. #L1518-L1593
_group_results_by_case_id function _group_results_by_case_id(results: list[dict]) -> dict[str, list[dict]] Bucket workload case records by their case_id field, defaulting a missing case_id to the empty string so the grouping is stable. #L1596-L1602
_summarize_case function _summarize_case(case_results: list[dict]) -> dict Build the per-case summary block. #L1605-L1648
summarize_workload_results function summarize_workload_results(results: list[dict]) -> dict Aggregate workload case records into stable qualification summary stats. #L1651-L1689
run_bench_serve_workload function async run_bench_serve_workload(*, url: str, workload_path: str, model: Optional[str] = None, output_path: Optional[str] = None, output_format: str = 'json', scrape: bool = True, include_content: bool = False, request_timeout_s: Optional[float] = 300.0, repetitions: int = 1, cache_policy: Optional[str] = None) -> dict Run a declarative workload against a running server. #L1692-L1818
_result_to_dict function _result_to_dict(r: BenchServeResult) -> dict Convert a :class:BenchServeResult to an ordered dict. #L1840-L1846
format_table function format_table(results: list[BenchServeResult]) -> str Render a human-readable terminal table of benchmark results. #L1849-L1871
format_json function format_json(results: list[BenchServeResult]) -> str Serialize benchmark results as a JSON array. #L1874-L1885
format_csv function format_csv(results: list[BenchServeResult]) -> str Serialize benchmark results as CSV with a header row. #L1888-L1904
_sql_escape function _sql_escape(value) -> str Escape a Python value for use as a SQL literal. #L1907-L1927
format_sql function format_sql(results: list[BenchServeResult]) -> str Emit a SQL CREATE TABLE IF NOT EXISTS statement and INSERT rows. #L1945-L1964
_write_sqlite_rows function _write_sqlite_rows(output_path: str, *, table: str, schema: str, columns: list[str], rows: list[dict]) -> None Append benchmark rows to a SQLite database. #L1967-L1990
_validate_sql_identifier function _validate_sql_identifier(identifier: str, *, kind: str) -> None Reject unsafe SQL identifiers before string interpolation. #L1993-L1996
write_sqlite function write_sqlite(results: list[BenchServeResult], output_path: str) -> None Append prompt-sweep benchmark results to a SQLite database. #L1999-L2009
_workload_record_to_row function _workload_record_to_row(record: dict) -> dict Function _workload_record_to_row calls record.get, metrics.get, ','.join, hardware.get; returns {'run_id': record.get('run_id', ''), 'timestamp': record.get('timestamp', ''), 'workload': record.get('workload', ''), …. #L2069-L2119
format_workload_table function format_workload_table(payload: dict) -> str Format workload result records as a compact human-readable table. #L2122-L2134
format_workload_json function format_workload_json(payload: dict) -> str Serialize a workload result payload as indented JSON. #L2137-L2140
format_workload_csv function format_workload_csv(payload: dict) -> str Serialize workload result records with the stable CSV column contract. #L2143-L2151
format_workload_sql function format_workload_sql(payload: dict) -> str Render SQL statements that create and populate the workload table. #L2170-L2180
write_workload_sqlite function write_workload_sqlite(payload: dict, output_path: str) -> None Append workload result records to a SQLite database. #L2183-L2193
format_workload_payload function format_workload_payload(payload: dict, fmt: str = 'json') -> str Serialize a workload payload in the requested text output format. #L2196-L2211
run_bench_serve function async run_bench_serve(url: str = 'http://127.0.0.1:8080', model: Optional[str] = None, prompt_sets: list[str] = None, prompt_file: Optional[str] = None, concurrencies: list[int] = None, max_tokens: int = 256, repetitions: int = 3, warmup: int = 1, thinking_values: list[Optional[bool]] = None, extra_bodies: list[str] = None, output_path: Optional[str] = None, fmt: str = 'table', do_validate: bool = True, scrape: bool = True, tag: Optional[str] = None, override_fields: Optional[dict] = None, system_prompt_file: Optional[str] = None, skip_preflight_token_count: bool = False) -> list[BenchServeResult] Run the full bench-serve sweep against a running vllm-mlx server. #L2221-L2638
run_bench_serve._mean nested function run_bench_serve._mean(key: str) -> float Nested Function run_bench_serve._mean calls statistics.mean; returns statistics.mean(vals) if vals else 0.0. #L2522-L2526