Spaces:
Paused
Paused
File size: 5,097 Bytes
729628a 6342806 729628a 6342806 729628a | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 | """MITL copilot: LLM failure-mode critique and edge-case routing."""
from __future__ import annotations
import json
import os
from pathlib import Path
from typing import Any, Dict, List, Optional
import httpx
def _llm_endpoints() -> List[Dict[str, str]]:
"""LLM endpoints in fallback order, configurable via environment.
SENSORFLOW_LLM_URL / SENSORFLOW_LLM_MODEL prepend a primary endpoint
(e.g. a remote GPU box); local Ollama is always the last fallback.
Deployment-specific hostnames must never be hard-coded in source.
"""
endpoints: List[Dict[str, str]] = []
url = os.environ.get("SENSORFLOW_LLM_URL")
if url:
endpoints.append({
"url": url,
"model": os.environ.get("SENSORFLOW_LLM_MODEL", "gemma4:latest"),
})
endpoints.append({"url": "http://localhost:11434/api/chat", "model": "gemma4:latest"})
return endpoints
OLLAMA_ENDPOINTS = _llm_endpoints()
MITL_QUEUE_PATH = Path("runs/pipeline/mitl_queue.json")
class MitlCopilot:
"""Route edge cases to human review with LLM-generated critiques."""
def __init__(self, queue_path: Optional[Path] = None):
self.queue_path = queue_path or MITL_QUEUE_PATH
def route_edge_cases(
self,
sequence_id: str,
metric_card: Dict[str, float],
pred_tracks: List[Dict],
thresholds: Optional[Dict[str, float]] = None,
) -> List[Dict]:
thresholds = thresholds or {
"id_swap_rate": 0.02,
"position_error_m": 2.0,
"track_fragmentation_rate": 0.05,
}
queue = self._load_queue()
edge_cases = []
if metric_card.get("id_swap_rate", 0) > thresholds["id_swap_rate"]:
edge_cases.append({
"type": "id_swap",
"severity": "high",
"metric_value": metric_card["id_swap_rate"],
"tracks": pred_tracks[:3],
})
if metric_card.get("position_error_m", 0) > thresholds["position_error_m"]:
edge_cases.append({
"type": "position_error",
"severity": "high",
"metric_value": metric_card["position_error_m"],
"tracks": pred_tracks[:3],
})
if metric_card.get("track_fragmentation_rate", 0) > thresholds["track_fragmentation_rate"]:
edge_cases.append({
"type": "fragmentation",
"severity": "medium",
"metric_value": metric_card["track_fragmentation_rate"],
"tracks": pred_tracks[:3],
})
for case in edge_cases:
case["sequence_id"] = sequence_id
case["critique"] = self.generate_critique(case, metric_card)
queue.append(case)
self._save_queue(queue)
return edge_cases
def generate_critique(
self,
edge_case: Dict[str, Any],
metric_card: Dict[str, float],
) -> str:
prompt = f"""You are a Lead QA Auditor for an autonomous driving 3D perception pipeline.
Analyze this failure mode and provide a concise structural critique.
FAILURE TYPE: {edge_case['type']}
SEVERITY: {edge_case['severity']}
METRIC VALUE: {edge_case['metric_value']}
FULL METRIC CARD:
{json.dumps(metric_card, indent=2)}
AFFECTED TRACKS (sample):
{json.dumps(edge_case.get('tracks', [])[:2], indent=2)}
Provide a professional markdown critique covering:
1. Root cause hypothesis
2. Impact on downstream safety
3. Recommended remediation steps
Be concise and actionable."""
for ep in OLLAMA_ENDPOINTS:
try:
payload = {
"model": ep["model"],
"messages": [{"role": "user", "content": prompt}],
"stream": False,
}
res = httpx.post(ep["url"], json=payload, timeout=25.0)
if res.status_code == 200:
text = res.json().get("message", {}).get("content", "")
if text:
return text
except Exception:
continue
return self._mock_critique(edge_case)
def _mock_critique(self, edge_case: Dict) -> str:
return (
f"### Edge Case: {edge_case['type']}\n\n"
f"Metric value **{edge_case['metric_value']:.4f}** exceeds threshold. "
f"Review affected tracks manually. LLM critique unavailable (Ollama offline)."
)
def _load_queue(self) -> List[Dict]:
if self.queue_path.exists():
with open(self.queue_path) as f:
return json.load(f)
return []
def _save_queue(self, queue: List[Dict]):
self.queue_path.parent.mkdir(parents=True, exist_ok=True)
with open(self.queue_path, "w") as f:
json.dump(queue, f, indent=2)
def get_queue(self, sequence_id: Optional[str] = None) -> List[Dict]:
queue = self._load_queue()
if sequence_id:
return [q for q in queue if q.get("sequence_id") == sequence_id]
return queue
|