Spaces:
Paused
Paused
| """MITL Copilot: advisory analysis of evaluation failures. | |
| Routes to local Ollama/Gemma when reachable; otherwise produces a deterministic | |
| offline analysis derived from the actual evaluation evidence. The Copilot is | |
| advisory only: it never changes metrics, approves/rejects labels, or overrides | |
| gates (spec §35). | |
| """ | |
| from __future__ import annotations | |
| import json | |
| from typing import Dict, List, Optional | |
| import httpx | |
| from sensorflow.evaluation import reporting | |
| from sensorflow.evaluation.records import EvalStore | |
| OLLAMA_ENDPOINTS = [ | |
| {"url": "http://localhost:11434/api/chat", "model": "gemma3:latest"}, | |
| {"url": "http://dgx-spark.tail16d8d9.ts.net:11434/api/chat", "model": "gemma4:26b"}, | |
| ] | |
| REASON_HINTS = { | |
| "LOW_IOU": ("Geometric mismatch vs reference", "box regression head under-fitting for this class/range"), | |
| "POSITION_ERROR": ("Center offset beyond tolerance", "LiDAR-camera extrinsic drift or late fusion latency"), | |
| "ORIENTATION_ERROR": ("Heading estimate off", "symmetric object ambiguity / yaw flipping in the detector"), | |
| "INSUFFICIENT_POINT_SUPPORT": ("Too few LiDAR returns inside box", "occlusion, range sparsity, or hallucinated detection"), | |
| "ANOMALY": ("Statistical outlier vs population", "feature combination rarely seen in training distribution"), | |
| "GRADER_DISAGREEMENT": ("Independent graders disagree", "genuinely ambiguous object class or degraded viewpoint"), | |
| "TRACK_FRAGMENTATION": ("Track split into fragments", "missed detections breaking association continuity"), | |
| "ID_SWITCH": ("Identity switched mid-track", "crossing objects with similar appearance/kinematics"), | |
| "MODEL_REGRESSION": ("Model version regressed", "training data shift or a bad checkpoint promotion"), | |
| "LOW_CONFIDENCE": ("Detector confidence very low", "borderline detection near the score threshold"), | |
| "SENSOR_DISAGREEMENT": ("Camera and LiDAR disagree", "single-sensor failure mode (glare, rain scatter, dropout)"), | |
| } | |
| def _gather_context(store: EvalStore, req: Dict) -> Dict: | |
| ctx: Dict = {"context_type": req.get("context_type", "general")} | |
| aid = req.get("annotation_id") | |
| if aid: | |
| rec = reporting.evaluation_record(store, aid) | |
| if rec: | |
| ctx["evaluation_record"] = rec | |
| eid = req.get("event_id") | |
| if eid: | |
| ev = store.get("rare_events", eid) | |
| if ev: | |
| ctx["rare_event"] = ev.model_dump() | |
| if req.get("context_type") == "regression" or req.get("model_version"): | |
| regs = sorted(store.all("regressions"), key=lambda r: r.date) | |
| if req.get("model_version"): | |
| regs = [r for r in regs if r.model_version == req["model_version"]] or regs | |
| if regs: | |
| ctx["regression"] = regs[-1].model_dump() | |
| return ctx | |
| def _offline_analysis(ctx: Dict) -> Dict: | |
| """Deterministic rule-based analysis assembled from real evidence.""" | |
| ctype = ctx.get("context_type", "general") | |
| evidence: List[str] = [] | |
| factors: List[str] = [] | |
| investigation: List[str] = [] | |
| classification = ctype.replace("_", " ").title() | |
| likely_cause = "Insufficient evidence gathered for a specific cause." | |
| confidence = 0.5 | |
| rec = ctx.get("evaluation_record") | |
| if rec: | |
| geo = rec["geometry"] | |
| an = rec["anomaly"] | |
| gr = rec["grading"] | |
| dec = rec.get("decision") or {} | |
| reasons = dec.get("failure_reasons", []) | |
| if reasons: | |
| classification = reasons[0].replace("_", " ").title() | |
| if geo.get("iou_3d") is not None: | |
| evidence.append(f"3D IoU vs reference = {geo['iou_3d']:.3f}") | |
| if geo.get("position_error") is not None: | |
| evidence.append(f"Position error = {geo['position_error']:.2f} m") | |
| if geo.get("orientation_error_deg") is not None: | |
| evidence.append(f"Orientation error = {geo['orientation_error_deg']:.1f}°") | |
| evidence.append(f"Anomaly ensemble score = {an['score']:.3f} " | |
| f"(threshold {an['decision_threshold']:.2f}, strategy {an['ensemble_strategy']})") | |
| if an.get("detector_scores"): | |
| top = sorted(an["normalized_scores"].items(), key=lambda kv: -kv[1])[:3] | |
| evidence.append("Highest detector votes: " + ", ".join(f"{k}={v:.2f}" for k, v in top)) | |
| if gr.get("consensus") is not None: | |
| evidence.append(f"Grader consensus = {gr['consensus']:.2%} across {gr['grader_count']} graders") | |
| if rec["tracking"]["id_switch"]: | |
| evidence.append("Tracking evidence shows an identity switch on this object's track") | |
| if rec["tracking"]["fragmentation"]: | |
| evidence.append("Tracking evidence shows track fragmentation") | |
| primary = dec.get("primary_failure_reason") | |
| if primary and primary in REASON_HINTS: | |
| headline, cause = REASON_HINTS[primary] | |
| likely_cause = f"{headline}: most consistent with {cause}." | |
| confidence = 0.72 | |
| for r in reasons[1:4]: | |
| if r in REASON_HINTS: | |
| factors.append(f"{r}: {REASON_HINTS[r][1]}") | |
| investigation.extend([ | |
| "Open the frame in HITL review and compare camera vs LiDAR evidence", | |
| "Check neighboring frames of the same track for consistent behavior", | |
| "Verify the reference ground-truth type and coverage before trusting the comparison", | |
| ]) | |
| ev = ctx.get("rare_event") | |
| if ev: | |
| classification = f"Rare Event: {ev['scenario_type'].replace('_', ' ').title()}" | |
| evidence.append(f"Scenario {ev['scenario_type']} severity={ev['severity']} " | |
| f"rarity={ev['rarity_score']:.2f} anomaly={ev['anomaly_score']:.2f}") | |
| for sensor, detail in ev.get("sensor_evidence", {}).items(): | |
| evidence.append(f"{sensor}: {detail}") | |
| likely_cause = "Genuine rare scenario rather than a labeling defect; value lies in adding it to training." | |
| investigation.append("Verify the event, then include its frames in the next training dataset") | |
| confidence = 0.68 | |
| reg = ctx.get("regression") | |
| if reg and ctype == "regression": | |
| classification = "Model Performance Regression" | |
| for d in reg.get("deltas", [])[:6]: | |
| if d["regressed"]: | |
| evidence.append(f"{d['metric']}: {d['baseline']:.3f} → {d['current']:.3f} " | |
| f"(Δ {d['delta']:+.3f}, tolerance {d['tolerance']})") | |
| if reg.get("affected_classes"): | |
| factors.append("Affected classes: " + ", ".join(reg["affected_classes"])) | |
| likely_cause = ("Training-data distribution shift or label-quality drop between model versions; " | |
| "the regressed metrics cluster in " + ", ".join(reg.get("kinds", ["performance"])) + ".") | |
| investigation.extend([ | |
| "Diff the training dataset lineage between the two model versions", | |
| "Re-run the benchmark on the affected classes only", | |
| "Inspect flagged labels from the newer model for systematic geometry errors", | |
| ]) | |
| confidence = 0.7 | |
| if not evidence: | |
| evidence.append("No structured evidence was attached to this request.") | |
| investigation.append("Select a specific annotation, rare event, or regression entry and retry") | |
| hypothesis = (f"HYPOTHESIS (not a conclusion): {likely_cause} " | |
| "This is generated from deterministic evidence rules and requires human confirmation.") | |
| lines = [f"# {classification}", "", "## Observed evidence"] | |
| lines += [f"- {e}" for e in evidence] | |
| if factors: | |
| lines += ["", "## Contributing factors"] + [f"- {f}" for f in factors] | |
| lines += ["", "## Likely cause", likely_cause, "", "## Hypothesis", hypothesis, | |
| "", "## Recommended investigation"] + [f"- {i}" for i in investigation] | |
| lines += ["", f"_Advisory only — confidence {confidence:.0%}. The Copilot never changes metrics or overrides gates._"] | |
| return { | |
| "analysis": "\n".join(lines), | |
| "structured": { | |
| "failure_classification": classification, | |
| "observed_evidence": evidence, | |
| "likely_cause": likely_cause, | |
| "contributing_factors": factors, | |
| "hypothesis": hypothesis, | |
| "recommended_investigation": investigation, | |
| "confidence": confidence, | |
| }, | |
| } | |
| def explain(store: EvalStore, req: Dict) -> Dict: | |
| """Try Ollama first; fall back to the deterministic offline analysis.""" | |
| ctx = _gather_context(store, req) | |
| offline = _offline_analysis(ctx) | |
| prompt = ( | |
| "You are an advisory MITL Copilot for an autonomous-driving label evaluation platform. " | |
| "You never change metrics or approve/reject labels; you explain evidence.\n\n" | |
| f"CONTEXT TYPE: {ctx.get('context_type')}\n" | |
| f"EVIDENCE (JSON):\n{json.dumps({k: v for k, v in ctx.items() if k != 'context_type'}, indent=2)[:6000]}\n\n" | |
| "Produce: failure classification, observed evidence, likely cause, contributing factors, " | |
| "a clearly-labeled hypothesis, recommended investigation steps, and a confidence estimate. " | |
| "Be concise and concrete." | |
| ) | |
| for ep in OLLAMA_ENDPOINTS: | |
| try: | |
| res = httpx.post(ep["url"], json={ | |
| "model": ep["model"], | |
| "messages": [{"role": "user", "content": prompt}], | |
| "stream": False, | |
| }, timeout=20.0) | |
| if res.status_code == 200: | |
| text = res.json().get("message", {}).get("content", "") | |
| if text: | |
| return { | |
| "status": "ok", | |
| "provider": ep["url"], | |
| "analysis": text, | |
| "structured": offline["structured"], # deterministic evidence summary retained | |
| } | |
| except Exception: | |
| continue | |
| return { | |
| "status": "ok", | |
| "provider": "offline_deterministic", | |
| "analysis": offline["analysis"], | |
| "structured": offline["structured"], | |
| } | |