File size: 2,616 Bytes
8d20d7d | 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 | """Typed adapter from the frozen learned predictor to core suggestion artifacts."""
from dataclasses import dataclass
from pathlib import Path
from yaml_edit.engine import inspect, preview
from yaml_edit.models import EditProposal
from yaml_edit.suggestion import SuggestionDisposition, SuggestionResult
from .checkpoint import ArtifactConfig, load_checkpoint
from .model import CandidateScorer
from .predictor import LearnedPredictor
@dataclass(slots=True)
class LoadedLearnedRuntime:
"""One explicitly loaded checkpoint reused for repeated warm inference calls."""
config: ArtifactConfig
model: CandidateScorer
predictor: LearnedPredictor
@property
def device(self) -> str:
"""Return the predictor's actual explicitly selected device."""
return self.predictor.device
@classmethod
def load(cls, config_path: Path, weights_path: Path, device: str) -> "LoadedLearnedRuntime":
"""Load through the hardened checkpoint API with no device fallback."""
config, model = load_checkpoint(config_path, weights_path)
return cls(config, model, LearnedPredictor(config, model, device))
def __call__(self, source: str, instruction: str) -> SuggestionResult:
"""Provide the existing evaluator's two-argument predictor interface."""
return self.suggest(source, instruction)
def suggest(self, source: str, instruction: str) -> SuggestionResult:
"""Adapt one prediction into an immutable source-bound typed result."""
prediction = self.predictor(source, instruction)
disposition = prediction.get("disposition")
proposal_data = prediction.get("proposal")
if disposition == SuggestionDisposition.SUPPORTED:
proposal = EditProposal.from_dict(proposal_data)
preview(source, proposal)
return SuggestionResult(
SuggestionDisposition.SUPPORTED,
"Experimental learned scorer selected one core-validated existing field.",
proposal,
)
if disposition == SuggestionDisposition.NEEDS_CLARIFICATION:
candidates = inspect(source).candidates
if candidates:
return SuggestionResult(
SuggestionDisposition.NEEDS_CLARIFICATION,
"Experimental learned scorer requires a specific existing field.",
candidates=candidates,
)
return SuggestionResult(
SuggestionDisposition.UNSUPPORTED,
"Experimental learned scorer abstained from proposing an edit.",
)
|