Source code for stormlog.infer.analysis

"""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 "-"