Prompt
Given a list of judge result dictionaries, return the top issue groups
sorted by weighted impact. Weight each case by severity and user flag
strength. Keep only aggregate counts and example IDs.
SEVERITY_WEIGHT = {"low": 1, "medium": 3, "high": 8}
FLAG_WEIGHT = {
"sampled": 1,
"low_confidence": 2,
"user_flagged": 5,
"user_unsatisfied": 8,
}
def top_audio_issues(results, limit=5):
if limit < 1:
return []
groups = {}
seen_event_ids = set()
for row in results:
if not isinstance(row, dict):
continue
diagnostics = row.get("diagnostics", {})
metadata = row.get("metadata", {})
if not isinstance(diagnostics, dict) or not isinstance(metadata, dict):
continue
event_id = row.get("id")
if event_id in seen_event_ids:
continue
issue = diagnostics.get("primary_issue")
severity = diagnostics.get("severity", "medium")
flag = metadata.get("flag_reason", "sampled")
if (
not isinstance(event_id, str)
or not isinstance(issue, str)
or severity not in SEVERITY_WEIGHT
or flag not in FLAG_WEIGHT
):
continue
seen_event_ids.add(event_id)
weight = SEVERITY_WEIGHT[severity] * FLAG_WEIGHT[flag]
bucket = groups.setdefault(issue, {
"issue": issue,
"count": 0,
"impact": 0,
"examples": [],
"models": {},
"versions": {},
})
bucket["count"] += 1
bucket["impact"] += weight
bucket["examples"] = (bucket["examples"] + [event_id])[:3]
model = metadata.get("model", "unknown")
if not isinstance(model, str):
model = "unknown"
bucket["models"][model] = bucket["models"].get(model, 0) + 1
version = metadata.get("app_version", "unknown")
if not isinstance(version, str):
version = "unknown"
bucket["versions"][version] = bucket["versions"].get(version, 0) + 1
return sorted(groups.values(), key=lambda item: (-item["impact"], -item["count"]))[:limit]
Hidden answer: what to test
Test that user-unsatisfied high-severity cases outrank many sampled
low-severity cases, malformed judge outputs are excluded before
aggregation, duplicate event IDs do not inflate impact, example IDs
are capped, and model/version breakdowns stay aggregate-only.