diff --git a/docs/easycom_annotated_floor_evaluation.md b/docs/easycom_annotated_floor_evaluation.md index 4e5bf82..4e1ed11 100644 --- a/docs/easycom_annotated_floor_evaluation.md +++ b/docs/easycom_annotated_floor_evaluation.md @@ -29,6 +29,44 @@ acoustic boundary can occur anywhere inside a 200 ms target bin, do not reinterpret it as an exact continuous-time lead without separately reporting the quantization convention. +## Provisional LLM-only pipeline smoke + +Before human adjudication is complete, unanimous two-pass LLM suggestions can +be materialized into the review-file schema for a pipeline smoke test: + +```bash +python3 materialize_provisional_llm_reviews.py \ + --events reports/easycom_entry_int_session{1,2,3,4}_v14/easycom_eot_int_deepseek_official*.jsonl \ + --output reports/easycom_entry_int_sessions1_4_v14/easycom_eot_int_provisional_llm_reviews.jsonl \ + --acknowledge-provisional-llm-auto-accept +``` + +The converter skips pass disagreements, pending API failures, and candidates +without an LLM suggestion. It writes JSONL, CSV, and a summary JSON. Every +materialized row carries `review_source=PROVISIONAL_LLM_AUTO_ACCEPT` and +`human_reviewed=false`; the nested `human_status=REVIEWED` value exists only +for review-schema compatibility and must never be described as human review. +The discouraged `--include-nonunanimous` option is available only for debugging +the fallback path. + +The evaluator rejects these rows by default. A provisional run requires an +explicit opt-in and marks its output protocol and semantic scope accordingly: + +```bash +python3 evaluate_annotated_floor_events.py \ + --reviews reports/easycom_entry_int_sessions1_4_v14/easycom_eot_int_provisional_llm_reviews.jsonl \ + --run egospeak=results/egospeak_miniroada_floor_eval \ + --sessions Session_4 \ + --allow-provisional-llm-reviews \ + --print-event-examples 5 \ + --output results/easycom_egospeak_event_smoke/session4_summary.json +``` + +This output is suitable for alignment debugging and preliminary analysis, not +for a manuscript claim of human-adjudicated ground truth. Official test +Sessions 1--3 still require disjoint training-side calibration unless the +separate exploratory in-sample override is used. + ## Development smoke run Use only training-side sessions while the annotation and region rules are diff --git a/evaluate_annotated_floor_events.py b/evaluate_annotated_floor_events.py index eaab58c..5b57c88 100644 --- a/evaluate_annotated_floor_events.py +++ b/evaluate_annotated_floor_events.py @@ -1,5 +1,5 @@ #!/usr/bin/env python3 -"""Evaluate human-adjudicated wearer entry and INT-yield opportunities. +"""Evaluate reviewed wearer entry and INT-yield opportunities. The forecasting model emits dense EgoSpeak-compatible probabilities rather than event detections. At each nominal horizon, a confirmed positive event is @@ -42,6 +42,7 @@ from evaluate_floor_state_changes import ( SOURCE_FPS = 20.0 Family = Literal["entry", "int"] FAMILIES: tuple[Family, ...] = ("entry", "int") +PROVISIONAL_LLM_SOURCE = "PROVISIONAL_LLM_AUTO_ACCEPT" @dataclass(frozen=True) @@ -65,7 +66,7 @@ def parse_args() -> argparse.Namespace: type=Path, action="append", required=True, - help="Human review JSONL. Repeat to combine independently stored files.", + help="Review JSONL. Repeat to combine independently stored files.", ) parser.add_argument( "--run", @@ -84,7 +85,7 @@ def parse_args() -> argparse.Namespace: type=Path, action="append", default=None, - help="Disjoint training-side human review JSONL used for thresholds.", + help="Disjoint training-side review JSONL used for thresholds.", ) parser.add_argument( "--calibration-run", @@ -132,6 +133,14 @@ def parse_args() -> argparse.Namespace: metavar="N", help="Print up to N scored examples per family for the first run.", ) + parser.add_argument( + "--allow-provisional-llm-reviews", + action="store_true", + help=( + "Explicitly allow auto-accepted LLM labels. Outputs remain marked " + "provisional and must not be reported as human-adjudicated ground truth." + ), + ) return parser.parse_args() @@ -189,7 +198,9 @@ def _seconds_to_end_row_exclusive(seconds: float) -> int: return int(math.ceil(seconds / STEP_SECONDS - 1e-7)) -def _review_payload(row: dict[str, Any], uid: str) -> tuple[dict[str, Any], dict[str, Any]]: +def _review_payload( + row: dict[str, Any], uid: str, *, allow_provisional_llm: bool +) -> tuple[dict[str, Any], dict[str, Any]]: review = row.get("review") candidate = row.get("candidate") if not isinstance(review, dict) or not isinstance(candidate, dict): @@ -198,17 +209,26 @@ def _review_payload(row: dict[str, Any], uid: str) -> tuple[dict[str, Any], dict ) if review.get("human_status") != "REVIEWED": raise ValueError(f"{uid}: human_status must be REVIEWED") + review_source = row.get("review_source", review.get("review_source")) + if review_source == PROVISIONAL_LLM_SOURCE and not allow_provisional_llm: + raise ValueError( + f"{uid}: provisional LLM auto-accept requires " + "--allow-provisional-llm-reviews" + ) return review, candidate def load_annotated_opportunities( paths: Iterable[Path], sessions: Iterable[str], + *, + allow_provisional_llm: bool = False, ) -> tuple[list[AnnotatedOpportunity], dict[str, int]]: selected_sessions = set(sessions) opportunities: list[AnnotatedOpportunity] = [] seen: set[str] = set() excluded = 0 + provisional_llm = 0 for row_number, row in enumerate(_read_jsonl(paths), 1): uid = str(row.get("candidate_id") or row.get("candidate_uid") or f"row-{row_number}") @@ -218,7 +238,12 @@ def load_annotated_opportunities( session = str(row.get("session", "")) if session not in selected_sessions: continue - review, candidate = _review_payload(row, uid) + review, candidate = _review_payload( + row, uid, allow_provisional_llm=allow_provisional_llm + ) + review_source = row.get("review_source", review.get("review_source")) + if review_source == PROVISIONAL_LLM_SOURCE: + provisional_llm += 1 verdict = review.get("human_verdict") if verdict == "EXCLUDE": excluded += 1 @@ -310,6 +335,7 @@ def load_annotated_opportunities( "num_review_rows": len(seen), "num_scoring_eligible": len(opportunities), "num_excluded": excluded, + "num_provisional_llm": provisional_llm, } return opportunities, diagnostics @@ -558,7 +584,12 @@ def main() -> None: runs = parse_runs(args.run) calibration_runs, has_calibration = _validate_protocol_args(args, runs) - opportunities, review_diagnostics = load_annotated_opportunities(args.reviews, args.sessions) + allow_provisional = bool(getattr(args, "allow_provisional_llm_reviews", False)) + opportunities, review_diagnostics = load_annotated_opportunities( + args.reviews, + args.sessions, + allow_provisional_llm=allow_provisional, + ) if not opportunities: raise ValueError("No scoring-eligible reviewed opportunities for evaluated sessions") @@ -568,6 +599,7 @@ def main() -> None: calibration_opportunities, calibration_diagnostics = load_annotated_opportunities( args.calibration_reviews, args.calibration_sessions, + allow_provisional_llm=allow_provisional, ) if not calibration_opportunities: raise ValueError("No scoring-eligible calibration opportunities") @@ -632,13 +664,25 @@ def main() -> None: } ) + provisional_count = review_diagnostics["num_provisional_llm"] + ( + 0 if calibration_diagnostics is None + else calibration_diagnostics["num_provisional_llm"] + ) payload = { "schema_version": 1, - "protocol": "human-adjudicated-fixed-horizon-floor-events-v1", - "semantic_scope": ( - "Human-confirmed wearer entry, INT-yield, and hard-negative opportunities; " - "positive events remain single 5-Hz target bins." + "protocol": ( + "provisional-llm-fixed-horizon-floor-events-v1" + if provisional_count + else "human-adjudicated-fixed-horizon-floor-events-v1" ), + "semantic_scope": ( + "Provisional LLM auto-accepted wearer entry, INT-yield, and hard-negative " + "opportunities; no human review is implied." + if provisional_count + else "Human-confirmed wearer entry, INT-yield, and hard-negative " + "opportunities; positive events remain single 5-Hz target bins." + ), + "provisional_llm": bool(provisional_count), "prediction_layout": "[target_time, class, future_horizon]", "class_names": list(CLASS_NAMES), "sessions": list(args.sessions), diff --git a/materialize_provisional_llm_reviews.py b/materialize_provisional_llm_reviews.py new file mode 100644 index 0000000..28c39be --- /dev/null +++ b/materialize_provisional_llm_reviews.py @@ -0,0 +1,188 @@ +#!/usr/bin/env python3 +"""Materialize unanimous LLM pre-labels as explicitly provisional review rows. + +The output is schema-compatible with ``review_floor_events.py`` and +``evaluate_annotated_floor_events.py``, but it is not evidence of human review. +Every row carries machine-readable provenance stating that no human reviewed +the label. This command is intended for pipeline smoke tests and preliminary +analysis only. +""" + +from __future__ import annotations + +import argparse +from collections import Counter +from datetime import datetime, timezone +import json +from pathlib import Path +from typing import Any, Iterable + +from review_floor_events import ( + REVIEW_SCHEMA_VERSION, + ReviewStore, + expand_candidate_paths, + load_candidates, +) + + +LABEL_SOURCE = "PROVISIONAL_LLM_AUTO_ACCEPT" + + +def _confidence(value: Any) -> str: + if isinstance(value, bool) or not isinstance(value, (int, float)): + return "low" + if value >= 0.8: + return "high" + if value >= 0.5: + return "medium" + return "low" + + +def materialize( + event_paths: Iterable[Path], + output_path: Path, + *, + include_nonunanimous: bool = False, +) -> tuple[list[dict[str, Any]], dict[str, Any]]: + candidate_paths = expand_candidate_paths(event_paths, output_path=output_path) + candidates = load_candidates(candidate_paths) + store = ReviewStore(output_path, candidates) + records: list[dict[str, Any]] = [] + skipped: Counter[str] = Counter() + counts: Counter[str] = Counter() + timestamp = datetime.now(timezone.utc).isoformat() + + for candidate in candidates: + suggestion = candidate.payload.get("llm_suggestion") + if not isinstance(suggestion, dict) or not suggestion.get("verdict"): + if candidate.payload.get("llm_adjudication_failure"): + skipped["pending_llm_failure"] += 1 + else: + skipped["missing_llm_suggestion"] += 1 + continue + if suggestion.get("unanimous") is not True and not include_nonunanimous: + skipped["nonunanimous"] += 1 + continue + + review = store._validated_review( + candidate.candidate_id, + { + "human_action": "accept", + "confidence": _confidence(suggestion.get("confidence")), + "boundary_correction_seconds": 0, + "reviewer": "", + "notes": ( + "Automatically accepted from an LLM pre-label for a provisional " + "pipeline smoke test; no human reviewed this item." + ), + }, + ) + review["review_source"] = LABEL_SOURCE + review["human_reviewed"] = False + record = { + "schema_version": REVIEW_SCHEMA_VERSION, + "candidate_id": candidate.candidate_id, + "candidate_uid": candidate.uid, + "source_file": candidate.source_path.name, + "source_row": candidate.source_row, + "session": candidate.session, + "split": candidate.split, + "candidate_family": candidate.family, + "task": candidate.task, + "automatic_label": candidate.automatic_label, + "automatic_protocol_label": candidate.payload.get("automatic_protocol_label"), + "automatic_hard_negative_type": candidate.payload.get( + "automatic_hard_negative_type" + ), + "ai_verdict": candidate.payload.get("ai_verdict"), + "ai_event_type": candidate.payload.get("ai_event_type"), + "ai_subtype": candidate.payload.get("ai_subtype"), + "ai_exclude_reason": candidate.payload.get("ai_exclude_reason"), + "ai_exclude_reason_group": candidate.payload.get("ai_exclude_reason_group"), + "review_source": LABEL_SOURCE, + "human_reviewed": False, + "review": review, + "updated_at": timestamp, + "candidate": candidate.payload, + } + records.append(record) + counts[f"session/{candidate.session}"] += 1 + counts[f"family/{candidate.family}"] += 1 + counts[f"verdict/{review['human_verdict']}"] += 1 + if review.get("human_subtype"): + counts[f"subtype/{review['human_subtype']}"] += 1 + if review.get("exclude_reason_group"): + counts[f"exclude_group/{review['exclude_reason_group']}"] += 1 + + store.reviews = {str(row["candidate_id"]): row for row in records} + store._write_locked() + summary = { + "schema_version": 1, + "label_source": LABEL_SOURCE, + "human_reviewed": False, + "warning": "Provisional LLM auto-accepts are not human-adjudicated ground truth.", + "input_files": [str(path) for path in candidate_paths], + "output": str(output_path), + "num_input_candidates": len(candidates), + "num_materialized": len(records), + "num_skipped": sum(skipped.values()), + "skipped": dict(sorted(skipped.items())), + "counts": dict(sorted(counts.items())), + "include_nonunanimous": include_nonunanimous, + "created_at": timestamp, + } + summary_path = output_path.with_suffix(".summary.json") + summary_path.write_text( + json.dumps(summary, ensure_ascii=False, indent=2) + "\n", encoding="utf-8" + ) + return records, summary + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument( + "--events", + type=Path, + nargs="+", + required=True, + help="LLM-enriched candidate JSONL files or candidate directories.", + ) + parser.add_argument("--output", type=Path, required=True) + parser.add_argument( + "--include-nonunanimous", + action="store_true", + help="Also accept pass-disagreement fallbacks; discouraged even for smoke tests.", + ) + parser.add_argument( + "--acknowledge-provisional-llm-auto-accept", + action="store_true", + help="Required acknowledgement that the output is not human-reviewed ground truth.", + ) + parser.add_argument("--overwrite", action="store_true") + return parser.parse_args() + + +def main() -> None: + args = parse_args() + if not args.acknowledge_provisional_llm_auto_accept: + raise PermissionError( + "Pass --acknowledge-provisional-llm-auto-accept; generated rows are " + "for provisional smoke tests, not human-reviewed ground truth." + ) + output = args.output.expanduser().resolve() + if output.exists() and not args.overwrite: + raise FileExistsError(f"Refusing to overwrite {output}; pass --overwrite") + records, summary = materialize( + args.events, + output, + include_nonunanimous=args.include_nonunanimous, + ) + print(f"Saved {len(records)} provisional review rows to {output}") + print(f"CSV: {output.with_suffix('.csv')}") + print(f"Summary: {output.with_suffix('.summary.json')}") + print("Skipped:", summary["skipped"]) + print("WARNING: no row in this output was human reviewed.") + + +if __name__ == "__main__": + main() diff --git a/test_materialize_provisional_llm_reviews.py b/test_materialize_provisional_llm_reviews.py new file mode 100644 index 0000000..b7334bf --- /dev/null +++ b/test_materialize_provisional_llm_reviews.py @@ -0,0 +1,125 @@ +from __future__ import annotations + +import argparse +import json +from pathlib import Path +import tempfile +import unittest +from unittest import mock + +import numpy as np + +import evaluate_annotated_floor_events as evaluator +from evaluate_annotated_floor_events import load_annotated_opportunities +from evaluate_floor_state_changes import NUM_CLASSES, NUM_HORIZONS +from materialize_provisional_llm_reviews import LABEL_SOURCE, materialize + + +def llm_candidate(uid: str, *, unanimous: bool) -> dict[str, object]: + suggestion = { + "verdict": "POSITIVE", + "event_type": "WEARER_ENTRY", + "subtype": None, + "exclude_reason": None, + "selected_opportunity_uid": None, + "confidence": 0.9, + "reason": "The wearer takes the floor.", + "needs_human_review": True, + "unanimous": unanimous, + } + return { + "uid": uid, + "session": "Session_4", + "split": "train", + "task": "wearer_entry", + "candidate_source": "WEARER_ONSET", + "anchor_dense_frame": 40, + "anchor_dense_seconds": 2.0, + "temporal_evidence": { + "behavior_anchor_frame": 40, + "behavior_anchor_seconds": 2.0, + "opportunity_options": [], + }, + "ai_verdict": "POSITIVE", + "ai_event_type": "WEARER_ENTRY", + "ai_subtype": None, + "ai_exclude_reason": None, + "ai_confidence": 0.9, + "llm_suggestion": suggestion, + } + + +class ProvisionalLLMReviewTests(unittest.TestCase): + def test_only_unanimous_rows_are_materialized_with_nonhuman_provenance(self) -> None: + with tempfile.TemporaryDirectory() as temporary: + root = Path(temporary) + events = root / "events.jsonl" + events.write_text( + "".join( + json.dumps(row) + "\n" + for row in ( + llm_candidate("unanimous", unanimous=True), + llm_candidate("disagreement", unanimous=False), + ) + ), + encoding="utf-8", + ) + output = root / "provisional_reviews.jsonl" + records, summary = materialize([events], output) + + self.assertEqual(len(records), 1) + self.assertEqual(summary["skipped"], {"nonunanimous": 1}) + self.assertEqual(records[0]["review_source"], LABEL_SOURCE) + self.assertFalse(records[0]["human_reviewed"]) + self.assertFalse(records[0]["review"]["human_reviewed"]) + self.assertEqual(records[0]["review"]["human_status"], "REVIEWED") + self.assertTrue(output.with_suffix(".csv").is_file()) + self.assertTrue(output.with_suffix(".summary.json").is_file()) + + with self.assertRaisesRegex(ValueError, "allow-provisional-llm-reviews"): + load_annotated_opportunities([output], ["Session_4"]) + opportunities, diagnostics = load_annotated_opportunities( + [output], ["Session_4"], allow_provisional_llm=True + ) + self.assertEqual(len(opportunities), 1) + self.assertEqual(diagnostics["num_provisional_llm"], 1) + + run_dir = root / "run" + (run_dir / "pred").mkdir(parents=True) + (run_dir / "valid").mkdir() + np.save( + run_dir / "pred" / "Session_4.npy", + np.full((20, NUM_CLASSES, NUM_HORIZONS), 0.5, dtype=np.float32), + ) + np.save( + run_dir / "valid" / "Session_4.npy", + np.ones((20, NUM_HORIZONS), dtype=bool), + ) + evaluation_output = root / "evaluation.json" + args = argparse.Namespace( + reviews=[output], + run=[f"model={run_dir}"], + sessions=["Session_4"], + calibration_reviews=None, + calibration_run=None, + calibration_sessions=None, + matched_recalls=[0.8], + allow_in_sample_thresholds=False, + allow_provisional_llm_reviews=True, + output=evaluation_output, + csv_output=None, + event_csv_output=None, + print_event_examples=0, + ) + with mock.patch.object(evaluator, "parse_args", return_value=args): + evaluator.main() + evaluation = json.loads(evaluation_output.read_text(encoding="utf-8")) + self.assertTrue(evaluation["provisional_llm"]) + self.assertEqual( + evaluation["protocol"], "provisional-llm-fixed-horizon-floor-events-v1" + ) + self.assertIn("no human review", evaluation["semantic_scope"]) + + +if __name__ == "__main__": + unittest.main()