File size: 3,752 Bytes
35d483e | 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 | #!/usr/bin/env python3
"""Replay model probabilities through the production turn controller."""
from __future__ import annotations
import argparse
import json
from pathlib import Path
from turn_detection.runtime.controller import ControllerConfig
from turn_detection.runtime.predictor import ModelMetadata
from turn_detection.runtime.replay import replay_jsonl
ROOT = Path(__file__).resolve().parents[1]
DEFAULT_METADATA_CANDIDATES = (
ROOT / "artifacts" / "model_metadata.json",
ROOT / "model_metadata.json",
)
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--input", required=True, help="JSONL of PauseCheckpoint fields")
parser.add_argument("--output", required=True, help="Destination decision JSONL")
parser.add_argument(
"--metadata",
help="Exported model_metadata.json; packaged layouts are auto-detected",
)
parser.add_argument("--threshold", type=float)
parser.add_argument("--long-pause-threshold", type=float)
parser.add_argument("--min-silence-ms", type=float)
parser.add_argument("--relax-after-ms", type=float)
parser.add_argument("--max-silence-ms", type=float)
parser.add_argument("--confirmations", type=int)
parser.add_argument(
"--evidence-scope",
choices=("synthetic_integration", "unqualified_sequence"),
default="unqualified_sequence",
help="Label replay provenance; this does not promote a replay to test evidence",
)
return parser.parse_args()
def _metadata_path(value: str | None) -> Path | None:
if value:
path = Path(value)
if not path.is_absolute():
path = ROOT / path
if not path.is_file():
raise SystemExit(f"model metadata does not exist: {path}")
return path
return next((path for path in DEFAULT_METADATA_CANDIDATES if path.is_file()), None)
def _controller_config(args: argparse.Namespace) -> ControllerConfig:
metadata_path = _metadata_path(args.metadata)
if metadata_path is not None:
base = ModelMetadata.from_path(metadata_path).controller
elif args.threshold is not None:
base = ControllerConfig(
endpoint_threshold=args.threshold,
long_pause_threshold=max(0.0, args.threshold - 0.18),
)
else:
raise SystemExit("provide --metadata or --threshold; refusing an unbound replay policy")
endpoint_threshold = base.endpoint_threshold if args.threshold is None else args.threshold
relaxation_delta = base.endpoint_threshold - base.long_pause_threshold
long_pause_threshold = (
max(0.0, endpoint_threshold - relaxation_delta)
if args.long_pause_threshold is None
else args.long_pause_threshold
)
return ControllerConfig(
endpoint_threshold=endpoint_threshold,
long_pause_threshold=long_pause_threshold,
min_silence_ms=(
base.min_silence_ms if args.min_silence_ms is None else args.min_silence_ms
),
relax_after_ms=(
base.relax_after_ms if args.relax_after_ms is None else args.relax_after_ms
),
max_silence_ms=(
base.max_silence_ms if args.max_silence_ms is None else args.max_silence_ms
),
required_confirmations=(
base.required_confirmations if args.confirmations is None else args.confirmations
),
)
def main() -> None:
args = parse_args()
config = _controller_config(args)
summary = replay_jsonl(
args.input,
args.output,
config,
evidence_scope=args.evidence_scope,
)
print(json.dumps(summary, indent=2, sort_keys=True))
if __name__ == "__main__":
main()
|