"""Analysis helpers for inference profiling artifacts."""
from __future__ import annotations
import json
from pathlib import Path
from typing import Any, Iterable, TypeGuard
[docs]
def analyze_inference_events(path: str | Path) -> dict[str, Any]:
"""Analyze an inference profiling JSONL artifact."""
records = _load_jsonl(path)
requests = [
record
for record in records
if record.get("event_type") == "infer.request"
and record.get("phase") == "measured"
]
samples = [
record
for record in records
if record.get("event_type") == "infer.system_sample"
]
ok_requests = [record for record in requests if record.get("status") == "ok"]
grouped: dict[str, list[dict[str, Any]]] = {}
for record in ok_requests:
case_id = str(record.get("case_id", "unknown"))
grouped.setdefault(case_id, []).append(record)
cases = {}
for case_id, case_requests in sorted(grouped.items()):
cases[case_id] = _summarize_requests(
case_requests,
samples=_samples_for_request_window(samples, case_requests),
)
failed = [record for record in requests if record.get("status") != "ok"]
return {
"summary": {
"total_requests": len(requests),
"successful_requests": len(ok_requests),
"failed_requests": len(failed),
"failure_rate": (len(failed) / len(requests)) if requests else 0.0,
"case_count": len(cases),
},
"cases": cases,
}
[docs]
def format_analysis_text(report: dict[str, Any]) -> str:
"""Render an inference analysis report as text."""
summary = report.get("summary", {})
lines = [
"Inference Profile Analysis",
"-" * 28,
f"Total requests: {summary.get('total_requests', 0)}",
f"Successful requests: {summary.get('successful_requests', 0)}",
f"Failed requests: {summary.get('failed_requests', 0)}",
f"Failure rate: {float(summary.get('failure_rate', 0.0)):.2%}",
]
cases = report.get("cases", {})
if isinstance(cases, dict) and cases:
lines.append("")
lines.append("Cases:")
for case_id, case in cases.items():
latency = case.get("latency_ms", {}) if isinstance(case, dict) else {}
throughput = case.get("throughput", {}) if isinstance(case, dict) else {}
lines.append(
f"- {case_id}: "
f"p50 E2E={_fmt(latency.get('e2e_p50'))} ms, "
f"p95 E2E={_fmt(latency.get('e2e_p95'))} ms, "
f"p50 TTFT={_fmt(latency.get('ttft_p50'))} ms, "
f"output={_fmt(throughput.get('output_tokens_per_second'))} tok/s, "
f"requests={_fmt(throughput.get('requests_per_second'))} req/s"
)
return "\n".join(lines)
def _load_jsonl(path: str | Path) -> list[dict[str, Any]]:
records: list[dict[str, Any]] = []
with Path(path).open("r", encoding="utf-8") as handle:
for line_number, raw_line in enumerate(handle, start=1):
line = raw_line.strip()
if not line:
continue
payload = json.loads(line)
if not isinstance(payload, dict):
raise ValueError(f"Line {line_number} is not a JSON object")
records.append(payload)
return records
def _summarize_requests(
requests: list[dict[str, Any]],
*,
samples: list[dict[str, Any]],
) -> dict[str, Any]:
e2e = _number_values(requests, "e2e_latency_ms")
ttft = _number_values(requests, "ttft_ms")
first_chunk = _number_values(requests, "first_chunk_latency_ms")
output_tokens = sum(_int_value(record.get("output_tokens")) for record in requests)
total_tokens = sum(_int_value(record.get("total_tokens")) for record in requests)
request_window = _request_time_window(requests)
duration_seconds = (
max(request_window[1] - request_window[0], 0) / 1_000_000_000
if request_window is not None
else 0.0
)
request_count = len(requests)
output_tps = output_tokens / duration_seconds if duration_seconds > 0 else 0.0
total_tps = total_tokens / duration_seconds if duration_seconds > 0 else 0.0
request_rate = request_count / duration_seconds if duration_seconds > 0 else 0.0
peak_device_used = _peak_sample_value(samples, "device_used_bytes")
peak_process_rss = _peak_sample_value(samples, "process_rss_bytes")
return {
"request_count": request_count,
"latency_ms": {
"e2e_p50": _percentile(e2e, 50),
"e2e_p95": _percentile(e2e, 95),
"e2e_p99": _percentile(e2e, 99),
"ttft_p50": _percentile(ttft, 50),
"ttft_p95": _percentile(ttft, 95),
"ttft_p99": _percentile(ttft, 99),
"first_chunk_p50": _percentile(first_chunk, 50),
"first_chunk_p95": _percentile(first_chunk, 95),
},
"throughput": {
"duration_seconds": duration_seconds,
"requests_per_second": request_rate,
"output_tokens_per_second": output_tps,
"total_tokens_per_second": total_tps,
},
"tokens": {
"output_tokens": output_tokens,
"total_tokens": total_tokens,
"output_token_sources": sorted(
{
str(record.get("output_token_source", "unknown"))
for record in requests
}
),
},
"memory": {
"peak_device_used_bytes": peak_device_used,
"peak_process_rss_bytes": peak_process_rss,
},
}
def _samples_for_request_window(
samples: list[dict[str, Any]],
requests: list[dict[str, Any]],
) -> list[dict[str, Any]]:
request_window = _request_time_window(requests)
if request_window is None:
return []
start_ns, end_ns = request_window
return [
sample
for sample in samples
if _is_number(sample.get("timestamp_ns"))
and start_ns <= _int_value(sample.get("timestamp_ns")) <= end_ns
]
def _request_time_window(requests: list[dict[str, Any]]) -> tuple[int, int] | None:
bounds: list[tuple[int, int]] = []
for record in requests:
started_at = record.get("started_at_ns")
ended_at = record.get("ended_at_ns")
if not _is_number(started_at) or not _is_number(ended_at):
continue
bounds.append((_int_value(started_at), _int_value(ended_at)))
if not bounds:
return None
return min(start for start, _end in bounds), max(end for _start, end in bounds)
def _peak_sample_value(samples: list[dict[str, Any]], field: str) -> int | None:
return max(
(
_int_value(sample.get(field))
for sample in samples
if _is_number(sample.get(field))
),
default=None,
)
def _number_values(records: Iterable[dict[str, Any]], field: str) -> list[float]:
values: list[float] = []
for record in records:
value = record.get(field)
if _is_number(value):
values.append(float(value))
return values
def _percentile(values: list[float], percentile: int) -> float | None:
if not values:
return None
sorted_values = sorted(values)
if len(sorted_values) == 1:
return sorted_values[0]
rank = (percentile / 100.0) * (len(sorted_values) - 1)
lower = int(rank)
upper = min(lower + 1, len(sorted_values) - 1)
weight = rank - lower
return sorted_values[lower] * (1.0 - weight) + sorted_values[upper] * weight
def _int_value(value: Any) -> int:
if isinstance(value, int) and not isinstance(value, bool):
return value
if isinstance(value, float):
return int(value)
return 0
def _is_number(value: Any) -> TypeGuard[int | float]:
return isinstance(value, (int, float)) and not isinstance(value, bool)
def _fmt(value: Any) -> str:
if isinstance(value, (int, float)):
return f"{float(value):.2f}"
return "-"