| """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.", |
| ) |
|
|
