120 lines
4.4 KiB
Python
120 lines
4.4 KiB
Python
from __future__ import annotations
|
|
|
|
import argparse
|
|
import csv
|
|
import json
|
|
from collections import defaultdict
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
DEFAULT_REPORT_DIR = ROOT / "reports" / "model-chat"
|
|
|
|
|
|
def iter_records(report_dir: Path) -> list[dict[str, Any]]:
|
|
records: list[dict[str, Any]] = []
|
|
if not report_dir.exists():
|
|
return records
|
|
for path in sorted(report_dir.glob("*.jsonl")):
|
|
with path.open("r", encoding="utf-8") as handle:
|
|
for line_number, line in enumerate(handle, start=1):
|
|
line = line.strip()
|
|
if not line:
|
|
continue
|
|
try:
|
|
record = json.loads(line)
|
|
except json.JSONDecodeError as exc:
|
|
raise ValueError(f"{path}:{line_number}: invalid JSONL: {exc}") from exc
|
|
if isinstance(record, dict):
|
|
records.append(record)
|
|
return records
|
|
|
|
|
|
def collect_rows(records: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
|
rows = []
|
|
for record in records:
|
|
record_type = record.get("type")
|
|
if record_type == "chat":
|
|
rows.append(
|
|
{
|
|
"type": "chat",
|
|
"plugin": record.get("plugin"),
|
|
"registry_model_id": record.get("registry_model_id"),
|
|
"served_model_name": record.get("served_model_name"),
|
|
"status": record.get("status"),
|
|
"latency_ms": record.get("latency_ms"),
|
|
"error": record.get("error"),
|
|
}
|
|
)
|
|
elif record_type == "compare":
|
|
for result in record.get("results") or []:
|
|
if not isinstance(result, dict):
|
|
continue
|
|
rows.append(
|
|
{
|
|
"type": "compare",
|
|
"plugin": record.get("plugin"),
|
|
"registry_model_id": result.get("registry_model_id"),
|
|
"served_model_name": result.get("served_model_name"),
|
|
"status": result.get("status"),
|
|
"latency_ms": result.get("latency_ms"),
|
|
"error": result.get("error"),
|
|
}
|
|
)
|
|
return rows
|
|
|
|
|
|
def summarize(rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
|
groups: dict[tuple[str, str], dict[str, Any]] = defaultdict(
|
|
lambda: {"requests": 0, "answered": 0, "errors": 0, "latencies": []}
|
|
)
|
|
for row in rows:
|
|
key = (str(row.get("plugin") or ""), str(row.get("served_model_name") or ""))
|
|
group = groups[key]
|
|
group["plugin"], group["served_model_name"] = key
|
|
group["registry_model_id"] = row.get("registry_model_id")
|
|
group["requests"] += 1
|
|
if row.get("status") == "answered":
|
|
group["answered"] += 1
|
|
if row.get("status") == "error":
|
|
group["errors"] += 1
|
|
latency = row.get("latency_ms")
|
|
if isinstance(latency, int | float):
|
|
group["latencies"].append(float(latency))
|
|
|
|
summary = []
|
|
for group in groups.values():
|
|
latencies = group.pop("latencies")
|
|
group["avg_latency_ms"] = round(sum(latencies) / len(latencies)) if latencies else None
|
|
summary.append(group)
|
|
return sorted(summary, key=lambda item: (item["plugin"], item["served_model_name"]))
|
|
|
|
|
|
def write_csv(path: Path, rows: list[dict[str, Any]]) -> None:
|
|
fields = ["plugin", "served_model_name", "registry_model_id", "requests", "answered", "errors", "avg_latency_ms"]
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
with path.open("w", encoding="utf-8", newline="") as handle:
|
|
writer = csv.DictWriter(handle, fieldnames=fields)
|
|
writer.writeheader()
|
|
writer.writerows(rows)
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description="Summarize model-chat JSONL reports.")
|
|
parser.add_argument("--report-dir", type=Path, default=DEFAULT_REPORT_DIR)
|
|
parser.add_argument("--csv", type=Path, help="Optional CSV output path.")
|
|
args = parser.parse_args()
|
|
|
|
records = iter_records(args.report_dir)
|
|
rows = collect_rows(records)
|
|
summary = summarize(rows)
|
|
print(json.dumps({"records": len(records), "rows": len(rows), "models": summary}, ensure_ascii=False, indent=2))
|
|
if args.csv:
|
|
write_csv(args.csv, summary)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|