283 lines
9.7 KiB
Python
283 lines
9.7 KiB
Python
"""
|
|
NeMo/Nemotron Replay Finalizer
|
|
==============================
|
|
|
|
Single-command final gate for externally produced NeMo/Nemotron replay results.
|
|
This module does not call NIM, NVIDIA APIs, tools, production systems, or LLMs.
|
|
It only imports already-produced external JSONL and runs AWOOOI's local gates.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from src.services.agent_nemotron_replay_adapter import (
|
|
NEMOTRON_CANDIDATE_ID,
|
|
import_nemotron_external_results_with_report,
|
|
)
|
|
from src.services.agent_replacement_evaluator import (
|
|
BASELINE_CANDIDATE_ID,
|
|
MIN_INCIDENTS_FOR_CANARY,
|
|
AgentReplayRecord,
|
|
score_replay_records,
|
|
)
|
|
from src.services.agent_replay_contract import validate_candidate_replay_contract
|
|
from src.services.agent_replay_label_grader import grade_replay_records_with_fixtures
|
|
from src.services.agent_replay_normalizer import (
|
|
CandidateReplayResult,
|
|
normalize_candidate_result,
|
|
)
|
|
from src.services.agent_replay_promotion_gate import (
|
|
evaluate_agent_replay_promotion_gate,
|
|
)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class NemotronReplayFinalizerOutputs:
|
|
"""Output path bundle for one finalized NeMo replay batch."""
|
|
|
|
candidate_raw: Path
|
|
import_report: Path
|
|
contract_report: Path
|
|
normalized_output: Path
|
|
graded_output: Path
|
|
grading_report: Path
|
|
scorecard: Path
|
|
pipeline_report: Path
|
|
promotion_gate: Path
|
|
summary: Path
|
|
|
|
@classmethod
|
|
def from_prefix(cls, prefix: Path) -> NemotronReplayFinalizerOutputs:
|
|
text = str(prefix)
|
|
return cls(
|
|
candidate_raw=Path(f"{text}-candidate-raw.jsonl"),
|
|
import_report=Path(f"{text}-import-report.json"),
|
|
contract_report=Path(f"{text}-contract-report.json"),
|
|
normalized_output=Path(f"{text}-candidate-normalized.jsonl"),
|
|
graded_output=Path(f"{text}-candidate-graded.jsonl"),
|
|
grading_report=Path(f"{text}-grading-report.json"),
|
|
scorecard=Path(f"{text}-scorecard.json"),
|
|
pipeline_report=Path(f"{text}-pipeline-report.json"),
|
|
promotion_gate=Path(f"{text}-promotion-gate.json"),
|
|
summary=Path(f"{text}-finalizer-summary.json"),
|
|
)
|
|
|
|
def to_dict(self) -> dict[str, str]:
|
|
return {
|
|
"candidate_raw": str(self.candidate_raw),
|
|
"import_report": str(self.import_report),
|
|
"contract_report": str(self.contract_report),
|
|
"normalized_output": str(self.normalized_output),
|
|
"graded_output": str(self.graded_output),
|
|
"grading_report": str(self.grading_report),
|
|
"scorecard": str(self.scorecard),
|
|
"pipeline_report": str(self.pipeline_report),
|
|
"promotion_gate": str(self.promotion_gate),
|
|
"summary": str(self.summary),
|
|
}
|
|
|
|
|
|
def finalize_nemotron_replay(
|
|
*,
|
|
requests: list[dict[str, Any]],
|
|
external_results: list[dict[str, Any]],
|
|
candidate_inputs: list[dict[str, Any]],
|
|
fixtures: list[dict[str, Any]],
|
|
baseline_records: list[AgentReplayRecord | dict[str, Any]],
|
|
target_stage: str = "shadow",
|
|
baseline_candidate_id: str = BASELINE_CANDIDATE_ID,
|
|
min_incidents_for_canary: int = MIN_INCIDENTS_FOR_CANARY,
|
|
) -> tuple[dict[str, Any], dict[str, list[Any]]]:
|
|
"""Run import -> contract -> normalize -> grade -> score -> promotion gate."""
|
|
artifacts: dict[str, list[Any]] = {
|
|
"candidate_raw": [],
|
|
"normalized": [],
|
|
"graded": [],
|
|
}
|
|
failures: list[str] = []
|
|
|
|
candidate_raw, import_report = import_nemotron_external_results_with_report(
|
|
external_results,
|
|
requests=requests,
|
|
)
|
|
import_report_payload = import_report.to_dict()
|
|
if not import_report.valid:
|
|
failures.append("import_report_invalid")
|
|
summary = _summary(
|
|
import_report=import_report_payload,
|
|
contract_report=None,
|
|
pipeline_report=None,
|
|
promotion_gate=None,
|
|
failures=failures,
|
|
stage="import",
|
|
)
|
|
return summary, artifacts
|
|
|
|
artifacts["candidate_raw"] = candidate_raw
|
|
contract_report = validate_candidate_replay_contract(
|
|
candidate_inputs=candidate_inputs,
|
|
candidate_results=candidate_raw,
|
|
expected_candidate_id=NEMOTRON_CANDIDATE_ID,
|
|
).to_dict()
|
|
if not contract_report["valid"]:
|
|
failures.append("contract_invalid")
|
|
summary = _summary(
|
|
import_report=import_report_payload,
|
|
contract_report=contract_report,
|
|
pipeline_report=_pipeline_report(
|
|
contract_report=contract_report,
|
|
normalized_records=0,
|
|
graded_records=0,
|
|
scorecard_written=False,
|
|
label_grading_applied=False,
|
|
),
|
|
promotion_gate=None,
|
|
failures=failures,
|
|
stage="contract",
|
|
)
|
|
return summary, artifacts
|
|
|
|
normalized_records = [
|
|
normalize_candidate_result(CandidateReplayResult.from_dict(payload))
|
|
for payload in candidate_raw
|
|
]
|
|
artifacts["normalized"] = normalized_records
|
|
graded_records, grading_report = grade_replay_records_with_fixtures(
|
|
fixtures=fixtures,
|
|
replay_records=normalized_records,
|
|
)
|
|
artifacts["graded"] = graded_records
|
|
baseline_only = _baseline_records_only(
|
|
baseline_records,
|
|
baseline_candidate_id=baseline_candidate_id,
|
|
)
|
|
if not baseline_only:
|
|
failures.append("baseline_records_missing")
|
|
pipeline_report = _pipeline_report(
|
|
contract_report=contract_report,
|
|
normalized_records=len(normalized_records),
|
|
graded_records=len(graded_records),
|
|
scorecard_written=False,
|
|
label_grading_applied=True,
|
|
baseline_records=0,
|
|
ignored_nonbaseline_records=0,
|
|
)
|
|
summary = _summary(
|
|
import_report=import_report_payload,
|
|
contract_report=contract_report,
|
|
pipeline_report=pipeline_report,
|
|
promotion_gate=None,
|
|
failures=failures,
|
|
stage="baseline",
|
|
grading_report=grading_report.to_dict(),
|
|
)
|
|
return summary, artifacts
|
|
|
|
scorecard = score_replay_records(
|
|
baseline_only + graded_records,
|
|
baseline_candidate_id=baseline_candidate_id,
|
|
min_incidents_for_canary=min_incidents_for_canary,
|
|
).to_dict()
|
|
promotion_gate = evaluate_agent_replay_promotion_gate(
|
|
candidate_id=NEMOTRON_CANDIDATE_ID,
|
|
scorecard_report=scorecard,
|
|
contract_report=contract_report,
|
|
raw_results=candidate_raw,
|
|
import_report=import_report_payload,
|
|
target_stage=target_stage,
|
|
).to_dict()
|
|
if promotion_gate["approved"] is not True:
|
|
failures.extend(str(item) for item in promotion_gate.get("failures") or [])
|
|
|
|
pipeline_report = _pipeline_report(
|
|
contract_report=contract_report,
|
|
normalized_records=len(normalized_records),
|
|
graded_records=len(graded_records),
|
|
scorecard_written=True,
|
|
label_grading_applied=True,
|
|
baseline_records=len(baseline_only),
|
|
ignored_nonbaseline_records=len(baseline_records) - len(baseline_only),
|
|
)
|
|
summary = _summary(
|
|
import_report=import_report_payload,
|
|
contract_report=contract_report,
|
|
pipeline_report=pipeline_report,
|
|
promotion_gate=promotion_gate,
|
|
failures=failures,
|
|
stage="promotion_gate",
|
|
scorecard=scorecard,
|
|
grading_report=grading_report.to_dict(),
|
|
)
|
|
return summary, artifacts
|
|
|
|
|
|
def _summary(
|
|
*,
|
|
import_report: dict[str, Any],
|
|
contract_report: dict[str, Any] | None,
|
|
pipeline_report: dict[str, Any] | None,
|
|
promotion_gate: dict[str, Any] | None,
|
|
failures: list[str],
|
|
stage: str,
|
|
scorecard: dict[str, Any] | None = None,
|
|
grading_report: dict[str, Any] | None = None,
|
|
) -> dict[str, Any]:
|
|
return {
|
|
"schema_version": "agent_nemotron_replay_finalizer_report_v1",
|
|
"candidate_id": NEMOTRON_CANDIDATE_ID,
|
|
"stage": stage,
|
|
"approved": bool((promotion_gate or {}).get("approved")),
|
|
"decision": "approved" if bool((promotion_gate or {}).get("approved")) else "blocked",
|
|
"failures": list(failures),
|
|
"import_report": import_report,
|
|
"contract_report": contract_report,
|
|
"pipeline_report": pipeline_report,
|
|
"grading_report": grading_report,
|
|
"scorecard": scorecard,
|
|
"promotion_gate": promotion_gate,
|
|
}
|
|
|
|
|
|
def _pipeline_report(
|
|
*,
|
|
contract_report: dict[str, Any],
|
|
normalized_records: int,
|
|
graded_records: int,
|
|
scorecard_written: bool,
|
|
label_grading_applied: bool,
|
|
baseline_records: int = 0,
|
|
ignored_nonbaseline_records: int = 0,
|
|
) -> dict[str, Any]:
|
|
return {
|
|
"schema_version": "agent_replay_pipeline_report_v1",
|
|
"candidate_id": NEMOTRON_CANDIDATE_ID,
|
|
"contract_valid": bool(contract_report.get("valid")),
|
|
"input_records": int(contract_report.get("inputs", 0)),
|
|
"result_records": int(contract_report.get("results", 0)),
|
|
"normalized_records": normalized_records,
|
|
"graded_records": graded_records,
|
|
"baseline_records": baseline_records,
|
|
"ignored_nonbaseline_records": ignored_nonbaseline_records,
|
|
"label_grading_applied": label_grading_applied,
|
|
"scorecard_written": scorecard_written,
|
|
}
|
|
|
|
|
|
def _baseline_records_only(
|
|
records: list[AgentReplayRecord | dict[str, Any]],
|
|
*,
|
|
baseline_candidate_id: str,
|
|
) -> list[AgentReplayRecord]:
|
|
parsed = [
|
|
record if isinstance(record, AgentReplayRecord) else AgentReplayRecord.from_dict(record)
|
|
for record in records
|
|
]
|
|
return [
|
|
record
|
|
for record in parsed
|
|
if record.candidate_id == baseline_candidate_id
|
|
]
|