File size: 8,450 Bytes
78ac017 | 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 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 | #!/usr/bin/env python3
"""Materialize a router-free third-view agent on event-view disagreements.
The frozen first stage decides whether to acquire the event view. A third
32-frame view is acquired only when that event proposal disagrees with the
recent proposal, and its answer is committed directly. Runtime decisions use
predictions and acquisition state only; task names, benchmark identity, and
gold labels are never inputs to the policy.
"""
from __future__ import annotations
import argparse
import copy
import json
from pathlib import Path
from statistics import fmean
from typing import Any
IDENTITY_FIELDS = (
"benchmark",
"model_manifest_sha256",
"annotation_sha256",
"evaluation_protocol_sha256",
)
def _index(value: dict[str, Any]) -> dict[str, dict[str, Any]]:
rows = value.get("rows") or []
indexed = {str(row.get("uid") or ""): row for row in rows}
if "" in indexed or len(indexed) != len(rows):
raise ValueError("result rows contain empty or duplicate UIDs")
return indexed
def _frame_indices(row: dict[str, Any]) -> set[int]:
return {int(value) for value in (row.get("sampling") or {}).get("frame_indices") or []}
def materialize(
recent: dict[str, Any],
event: dict[str, Any],
base_agent: dict[str, Any],
third: dict[str, Any],
) -> dict[str, Any]:
for field in IDENTITY_FIELDS:
values = {value.get(field) for value in (recent, event, base_agent, third)}
if len(values) != 1:
raise ValueError(f"view identity mismatch: {field}")
recent_rows, event_rows, base_rows, third_rows = map(
_index, (recent, event, base_agent, third)
)
if set(base_rows) != set(recent_rows) or not set(event_rows) <= set(recent_rows):
raise ValueError("base/recent/event UID coverage diverged")
disagreements = {
uid
for uid, row in event_rows.items()
if str(row.get("prediction") or "")
!= str(recent_rows[uid].get("prediction") or "")
}
if set(third_rows) != disagreements:
raise ValueError("third-view coverage must equal the event disagreement cohort")
output_rows = []
for recent_row in recent.get("rows") or []:
uid = str(recent_row["uid"])
base = copy.deepcopy(base_rows[uid])
if uid not in third_rows:
base["third_view_policy"] = {
"triggered": False,
"reason": "no_acquired_event_disagreement",
"question_task_group_or_benchmark_router": False,
}
output_rows.append(base)
continue
proposal = event_rows[uid]
adjudicator = third_rows[uid]
if proposal.get("error") or adjudicator.get("error"):
raise ValueError(f"{uid}: acquired view contains an error")
if not bool((base.get("agent_policy") or {}).get("escalated")):
raise ValueError(f"{uid}: third view was triggered outside the acquired cohort")
timestamps = (adjudicator.get("sampling") or {}).get("frame_timestamps") or []
if timestamps and max(map(float, timestamps)) > float(adjudicator["query_time"]) + 1e-3:
raise ValueError(f"{uid}: third view crosses the causal frontier")
recent_prediction = str(recent_row.get("prediction") or "")
event_prediction = str(proposal.get("prediction") or "")
third_prediction = str(adjudicator.get("prediction") or "")
if not recent_prediction or not event_prediction or not third_prediction:
raise ValueError(f"{uid}: an agent proposal is empty")
if recent_prediction == event_prediction:
raise ValueError(f"{uid}: third view lacks a disagreement trigger")
unique_indices = _frame_indices(proposal) | _frame_indices(adjudicator)
base.update(
{
"prediction": third_prediction,
"raw_response": adjudicator.get("raw_response"),
"third_view_prediction": third_prediction,
"third_view_policy": {
"triggered": True,
"trigger": "acquired_event_prediction_differs_from_recent_prediction",
"commit": "direct_third_view_prediction",
"recent_prediction": recent_prediction,
"event_prediction": event_prediction,
"question_task_group_or_benchmark_router": False,
"gold_available_to_policy": False,
},
"media_calls": int(base.get("media_calls") or 2) + 1,
"num_turns": int(base.get("num_turns") or 2) + 1,
"model_processed_frames": int(base.get("model_processed_frames") or 24)
+ int(adjudicator.get("frames_observed") or 32),
"unique_frames": len(unique_indices) if unique_indices else 32,
"latency_seconds": float(base.get("latency_seconds") or 0.0)
+ float(adjudicator.get("latency_seconds") or 0.0),
"visual_tokens": int(recent_row.get("visual_tokens") or 0)
+ int(proposal.get("visual_tokens") or 0)
+ int(adjudicator.get("visual_tokens") or 0),
}
)
output_rows.append(base)
if any(not str(row.get("prediction") or "") for row in output_rows):
raise ValueError("materialized agent contains an empty prediction")
triggered = len(third_rows)
return {
**{
key: base_agent.get(key)
for key in (
"annotation",
"annotation_sha256",
"benchmark",
"evaluation_protocol",
"evaluation_protocol_sha256",
"model",
"model_manifest_sha256",
"processor",
)
},
"format": "streameva.third_view_disagreement_agent.v0",
"status": "benchmark_selected_router_free_development_agent",
"classification": "benchmark_selected_development_not_confirmatory",
"comparable_under_locked_recent4_protocol": False,
"policy": {
"first_view": "recent4",
"second_view": "frozen_uncertainty_triggered_causalbudget20",
"third_view": "causalbudget32r8_on_acquired_answer_disagreement",
"third_view_commit": "direct_prediction",
"question_task_group_or_benchmark_router": False,
"benchmark_identity_available_to_policy": False,
"gold_available_to_policy": False,
},
"num_samples": len(output_rows),
"num_errors": 0,
"third_view_triggered": triggered,
"third_view_trigger_rate": triggered / len(output_rows),
"mean_model_calls": fmean(int(row.get("media_calls") or 1) for row in output_rows),
"mean_unique_frames": fmean(int(row.get("unique_frames") or 4) for row in output_rows),
"mean_model_processed_frames": fmean(
int(row.get("model_processed_frames") or 4) for row in output_rows
),
"rows": output_rows,
}
def main() -> None:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--recent", type=Path, required=True)
parser.add_argument("--event", type=Path, required=True)
parser.add_argument("--base-agent", type=Path, required=True)
parser.add_argument("--third-view", type=Path, required=True)
parser.add_argument("--output", type=Path, required=True)
args = parser.parse_args()
if args.output.exists():
raise FileExistsError(args.output)
values = [
json.loads(path.read_text(encoding="utf-8"))
for path in (args.recent, args.event, args.base_agent, args.third_view)
]
result = materialize(*values)
result["artifacts"] = {
"recent": str(args.recent.resolve()),
"event": str(args.event.resolve()),
"base_agent": str(args.base_agent.resolve()),
"third_view": str(args.third_view.resolve()),
}
args.output.parent.mkdir(parents=True, exist_ok=True)
args.output.write_text(json.dumps(result, indent=2, sort_keys=True) + "\n")
print(
json.dumps(
{
"output": str(args.output),
"num_samples": result["num_samples"],
"third_view_triggered": result["third_view_triggered"],
},
sort_keys=True,
)
)
if __name__ == "__main__":
main()
|