Spaces:
Paused
Paused
File size: 8,032 Bytes
5d10875 | 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 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 | """Failure Evidence Graph.
Structured, typed nodes (Object / Environment / Sensors / Prediction /
GroundTruth / Tracking / Planner / HistoricalSimilarity / Frequency /
SafetyConsequence) each carrying an explicit evidence status:
OBSERVED directly measured from the campaign data / replay
DERIVED computed deterministically from observed data
HYPOTHESIS produced by an advisory agent; requires confirmation
UNAVAILABLE the evidence does not exist yet; the graph says so instead
of letting anything downstream pretend it does
The graph starts from deterministic scene data and is enriched as analysis
stages complete. Nothing here consults an LLM.
"""
from __future__ import annotations
from typing import Optional
from sensorflow.agentic import data as data_mod
from sensorflow.agentic.models import (EvidenceEdge, EvidenceGraph, EvidenceNode,
FailureEvent, new_id)
def _node(node_type: str, status: str, summary: str, fields: dict,
source: str, caveats: Optional[list] = None) -> EvidenceNode:
return EvidenceNode(node_id=new_id("ev"), node_type=node_type, status=status,
summary=summary, fields=fields, source=source,
caveats=caveats or [])
def build_graph(failure: FailureEvent, campaign: data_mod.Campaign) -> EvidenceGraph:
"""Initial graph from deterministic campaign evidence for the failure's
representative instance (first instance)."""
graph = EvidenceGraph(failure_id=failure.failure_id)
inst = failure.instances[0] if failure.instances else None
if inst is None:
graph.nodes.append(_node(
"Object", "UNAVAILABLE",
"No concrete failure instance is attached to this pattern",
{}, "pipeline"))
return graph
history = data_mod.observations_for_instance(
campaign, inst.sequence_id, inst.object_instance_id)
obs = next((o for o in history if o.frame_id == inst.frame_id), None)
graph.nodes.append(_node(
"Object", "OBSERVED",
f"{inst.gt_class} instance {inst.object_instance_id} at "
f"{inst.distance_m:.1f} m",
{"instance_id": inst.object_instance_id, "gt_class": inst.gt_class,
"distance_m": inst.distance_m,
"bbox_3d": obs.gt_bbox_3d if obs else None},
"bevfusion synthetic scene ground truth"))
graph.nodes.append(_node(
"Environment", "OBSERVED",
f"{inst.time_of_day}/{inst.weather}, construction_zone="
f"{inst.construction_zone}, geo={inst.geo_bucket}",
{"time_of_day": inst.time_of_day, "weather": inst.weather,
"construction_zone": inst.construction_zone,
"geo_bucket": inst.geo_bucket, "occluded": inst.occluded},
"sequence context (campaign metadata)"))
graph.nodes.append(_node(
"Sensors", "DERIVED",
"Camera/LiDAR coherence not yet verified — pending fusion agent",
{"modalities": ["camera", "lidar"], "verified": False},
"sensorflow.bevfusion simulated sensors",
caveats=["fusion verification runs in FAILURE_ANALYSIS"]))
if obs is not None:
graph.nodes.append(_node(
"Prediction", "OBSERVED",
f"candidate predicted {obs.candidate.predicted_class} "
f"(conf {obs.candidate.confidence}); baseline predicted "
f"{obs.baseline.predicted_class} (conf {obs.baseline.confidence})",
{"candidate": obs.candidate.model_dump(),
"baseline": obs.baseline.model_dump(),
"candidate_model": failure.candidate_model,
"baseline_model": failure.baseline_model},
"campaign prediction log"))
graph.nodes.append(_node(
"GroundTruth", "OBSERVED",
f"GT class {obs.gt_class} (synthetic vendor-style GT)",
{"gt_class": obs.gt_class, "gt_bbox_3d": obs.gt_bbox_3d,
"gt_source": "synthetic_vendor_gt"},
"bevfusion synthetic scene ground truth"))
graph.nodes.append(_node(
"Tracking", "OBSERVED",
f"candidate track {obs.candidate.track_id}; "
f"{len(history)} frames observed for this instance",
{"candidate_track_id": obs.candidate.track_id,
"baseline_track_id": obs.baseline.track_id,
"frames_observed": len(history)},
"campaign prediction log"))
else:
graph.nodes.append(_node("Prediction", "UNAVAILABLE",
"prediction record missing", {}, "campaign"))
graph.nodes.append(_node("GroundTruth", "UNAVAILABLE",
"ground truth record missing", {}, "campaign"))
graph.nodes.append(_node("Tracking", "UNAVAILABLE",
"tracking record missing", {}, "campaign"))
if inst.has_planner_trace:
graph.nodes.append(_node(
"Planner", "OBSERVED",
"Replay planner-response trace exists for this instance",
{"replay_available": True, "analyzed": False},
"synthetic planner replay trace",
caveats=["safety-impact agent quantifies behavior in FAILURE_ANALYSIS"]))
else:
graph.nodes.append(_node(
"Planner", "UNAVAILABLE",
"No behavioral replay evidence for this instance — downstream "
"consequence cannot be observed, only hypothesized",
{"replay_available": False},
"synthetic planner replay trace"))
graph.nodes.append(_node(
"HistoricalSimilarity", "DERIVED",
"Similar-failure retrieval pending (scenario mining agent)",
{"retrieved": False}, "scenario mining (embedding + attribute retrieval)"))
basis = failure.detection_basis
graph.nodes.append(_node(
"Frequency", "DERIVED",
f"candidate {basis.candidate_events}/{basis.denominator} "
f"({basis.candidate_rate:.6f}) vs baseline {basis.baseline_events}"
f"/{basis.denominator} ({basis.baseline_rate:.6f}) — CI/significance "
"pending statistical agent",
{"candidate_events": basis.candidate_events,
"baseline_events": basis.baseline_events,
"denominator": basis.denominator,
"candidate_rate": basis.candidate_rate,
"baseline_rate": basis.baseline_rate},
"deterministic detection scan"))
graph.nodes.append(_node(
"SafetyConsequence", "UNAVAILABLE",
"Not yet assessed — requires safety-impact analysis over replay "
"evidence; hypotheses are never recorded as consequences",
{}, "safety impact agent (FAILURE_ANALYSIS)"))
ids = {n.node_type: n.node_id for n in graph.nodes}
for src, dst, rel in [
("Object", "Prediction", "was_predicted_as"),
("Object", "GroundTruth", "labeled_as"),
("Object", "Environment", "situated_in"),
("Prediction", "Tracking", "feeds"),
("Tracking", "Planner", "feeds"),
("Prediction", "Frequency", "aggregated_into"),
("Planner", "SafetyConsequence", "determines"),
("Object", "HistoricalSimilarity", "compared_against"),
("Sensors", "Prediction", "produced"),
]:
if src in ids and dst in ids:
graph.edges.append(EvidenceEdge(src=ids[src], dst=ids[dst], relation=rel))
return graph
def set_node(graph: EvidenceGraph, node_type: str, status: str, summary: str,
fields: dict, source: str, caveats: Optional[list] = None) -> None:
"""Replace (or add) a node after an analysis stage produced its evidence."""
for i, n in enumerate(graph.nodes):
if n.node_type == node_type:
graph.nodes[i] = EvidenceNode(
node_id=n.node_id, node_type=node_type, status=status,
summary=summary, fields=fields, source=source,
caveats=caveats or [])
return
graph.nodes.append(_node(node_type, status, summary, fields, source, caveats))
|