"""ScenarioMiningAgent — similar-failure retrieval + clustering. Embedding + attribute retrieval of similar historical failures, following the megaeval embedding recipe (structural features + one-hot attributes + seeded random projection, cosine retrieval — see MegaStore._build_embeddings in sensorflow/megaeval/runs.py, which operates on run artifacts and is therefore mirrored here for failure records rather than re-invented). Clustering method is SELECTED FROM THE DATA SHAPE and the choice is justified in the output (it is not always DBSCAN): * n < 15 -> single-linkage agglomerative on cosine distance (density parameters are unidentifiable at tiny n) * mostly categorical -> attribute-signature grouping (exact modes) * otherwise -> density-based (DBSCAN-style) via pairwise radius graph connected components """ from __future__ import annotations from typing import Any, Dict, List, Tuple import numpy as np from sensorflow.agentic.agents.base import BaseAgent, compact, no_escalation from sensorflow.agentic.models import AgentEscalation, FailureEvent GT_CLASSES = ["pedestrian", "cyclist", "motorcycle", "vehicle", "truck"] PRED_CLASSES = ["pedestrian", "cyclist", "motorcycle", "vehicle", "truck", "construction_cone", "background"] TIMES = ["day", "night"] WEATHERS = ["clear", "rain"] EMBED_DIM = 16 EMBED_SEED = 1234 # same seeded-projection convention as megaeval def historical_corpus(seed: int = 77, n: int = 48) -> List[Dict]: """Deterministic synthetic archive of previously-triaged failures. Deliberately contains NO pedestrian->construction_cone precedent, so the motivating failure is genuinely novel against the archive. """ rng = np.random.default_rng(seed) pairs = [("pedestrian", "cyclist"), ("cyclist", "motorcycle"), ("vehicle", "truck"), ("truck", "vehicle"), ("cyclist", "vehicle"), ("pedestrian", "background"), ("motorcycle", "cyclist"), ("vehicle", "background")] out = [] for i in range(n): gt, pred = pairs[int(rng.integers(0, len(pairs)))] out.append({ "historical_id": f"hist-{i:04d}", "gt_class": gt, "predicted_class": pred, "construction_zone": bool(rng.random() < 0.15), "time_of_day": TIMES[int(rng.random() < 0.4)], "weather": WEATHERS[int(rng.random() < 0.3)], "distance_m": round(float(rng.uniform(8, 70)), 1), "resolved": bool(rng.random() < 0.8), "resolution": "retrained" if rng.random() < 0.5 else "data_fix", }) return out def _featurize(rec: Dict) -> np.ndarray: f: List[float] = [] f.extend(1.0 if rec["gt_class"] == c else 0.0 for c in GT_CLASSES) f.extend(1.0 if rec["predicted_class"] == c else 0.0 for c in PRED_CLASSES) f.append(1.0 if rec.get("construction_zone") else 0.0) f.append(1.0 if rec.get("time_of_day") == "night" else 0.0) f.append(1.0 if rec.get("weather") == "rain" else 0.0) f.append(min(float(rec.get("distance_m", 30.0)) / 70.0, 1.0)) return np.asarray(f, dtype=np.float64) def embed(records: List[Dict]) -> np.ndarray: X = np.vstack([_featurize(r) for r in records]) rng = np.random.default_rng(EMBED_SEED) proj = rng.normal(size=(X.shape[1], EMBED_DIM)) / np.sqrt(X.shape[1]) E = X @ proj return E / np.clip(np.linalg.norm(E, axis=1, keepdims=True), 1e-9, None) def _select_clustering(n: int, categorical_share: float) -> Tuple[str, str]: if n < 15: return ("single_linkage_agglomerative", f"only n={n} member instances: density-based methods " "(DBSCAN) need an eps/minPts neighborhood that is " "unidentifiable at this sample size; single-linkage on " "cosine distance degrades gracefully") if categorical_share > 0.8: return ("attribute_signature_grouping", f"{categorical_share:.0%} of features are categorical " "one-hots: exact signature grouping is better-defined than " "euclidean density in a nearly-discrete space") return ("density_radius_components", f"n={n} with mixed features: density-based clustering " "(DBSCAN-style radius graph components) is appropriate") def _cluster(records: List[Dict], emb: np.ndarray, method: str) -> List[List[int]]: n = len(records) if method == "attribute_signature_grouping": groups: Dict[Tuple, List[int]] = {} for i, r in enumerate(records): key = (r.get("construction_zone"), r.get("time_of_day"), r.get("weather")) groups.setdefault(key, []).append(i) return list(groups.values()) # graph-based (single-linkage threshold == radius components on cosine) sims = emb @ emb.T thr = 0.92 if method == "single_linkage_agglomerative" else 0.95 parent = list(range(n)) def find(a): while parent[a] != a: parent[a] = parent[parent[a]] a = parent[a] return a for i in range(n): for j in range(i + 1, n): if sims[i, j] >= thr: parent[find(i)] = find(j) groups2: Dict[int, List[int]] = {} for i in range(n): groups2.setdefault(find(i), []).append(i) return list(groups2.values()) class ScenarioMiningAgent(BaseAgent): name = "scenario_mining" version = "1.0.0" epistemic_status = "DERIVED" def _analyze(self, failure_id: str, failure: FailureEvent = None, **inputs) -> Tuple[Dict[str, Any], float, str, AgentEscalation]: assert failure is not None corpus = historical_corpus() # ---- retrieval: embedding + attribute filters -------------------- query = { "gt_class": failure.gt_class or "vehicle", "predicted_class": failure.predicted_class or "background", "construction_zone": any(i.construction_zone for i in failure.instances), "time_of_day": (failure.instances[0].time_of_day if failure.instances else "day"), "weather": (failure.instances[0].weather if failure.instances else "clear"), "distance_m": (float(np.mean([i.distance_m for i in failure.instances])) if failure.instances else 30.0), } all_emb = embed(corpus + [query]) emb, q = all_emb[:-1], all_emb[-1] sims = emb @ q order = np.argsort(-sims)[:8] similar = [{**corpus[int(i)], "similarity": round(float(sims[int(i)]), 4)} for i in order] exact_precedent = [r for r in corpus if r["gt_class"] == query["gt_class"] and r["predicted_class"] == query["predicted_class"]] novelty = "novel" if not exact_precedent else ( "known_regression" if any(not r["resolved"] for r in exact_precedent) else "known_stable") # ---- clustering of the failure's own instances ------------------- inst_records = [{ "gt_class": i.gt_class, "predicted_class": i.predicted_class, "construction_zone": i.construction_zone, "time_of_day": i.time_of_day, "weather": i.weather, "distance_m": i.distance_m, } for i in failure.instances] cluster_reports: List[Dict] = [] method = justification = None if inst_records: feat_dim = _featurize(inst_records[0]).size categorical_share = (feat_dim - 1) / feat_dim # all but distance method, justification = _select_clustering(len(inst_records), categorical_share) inst_emb = embed(inst_records) clusters = _cluster(inst_records, inst_emb, method) total = len(inst_records) for idx, members in enumerate(sorted(clusters, key=len, reverse=True)): recs = [inst_records[m] for m in members] dominant = { "construction_zone": (sum(r["construction_zone"] for r in recs) / len(recs)), "night_share": (sum(r["time_of_day"] == "night" for r in recs) / len(recs)), "rain_share": (sum(r["weather"] == "rain" for r in recs) / len(recs)), "mean_distance_m": round(float(np.mean( [r["distance_m"] for r in recs])), 1), } cluster_reports.append({ "cluster_id": f"{failure.failure_id}-c{idx}", "size": len(members), "population_share": round(len(members) / total, 3), "dominant_conditions": dominant, "representative_examples": [ failure.instances[m].instance_id for m in members[:3]], }) output = { "similar_historical": similar, "exact_precedent_count": len(exact_precedent), "novelty": novelty, "novelty_basis": ("no archived failure shares the " f"{query['gt_class']}->{query['predicted_class']} " "signature" if novelty == "novel" else f"{len(exact_precedent)} archived precedents"), "clustering_method": method, "clustering_justification": justification, "clusters": cluster_reports, "retrieval": ("cosine over seeded random-projection embeddings of " "structural+categorical features (megaeval recipe) " "+ attribute retrieval"), } confidence = 0.75 if inst_records else 0.4 return (output, confidence, "retrieval and clustering are deterministic; confidence " "reflects instance coverage", no_escalation()) def _llm_prompt(self, failure_id, output, **inputs): return ("Summarize the cluster structure and novelty finding of this " "failure mining result: " + compact(output))