| from __future__ import annotations |
| from dataclasses import asdict |
| import uuid |
| from .config import RuntimeConfig |
| from .models import build_adapter, ModelResult |
| from .observability import AuditLedger, configure_logging, capture_error |
|
|
| class RuntimeFailure(RuntimeError): |
| pass |
|
|
| class NATO1000Runtime: |
| def __init__(self, config: RuntimeConfig | None = None, ledger: AuditLedger | None = None): |
| self.config = config or RuntimeConfig() |
| self.config.validate() |
| self.logger = configure_logging(self.config.log_level) |
| self.ledger = ledger or AuditLedger() |
| self.adapter = build_adapter(self.config.model_id) |
| self.ready = True |
|
|
| def health(self) -> dict: |
| return {"ready": self.ready, "model_id": self.config.model_id, "backend": self.config.backend, |
| "audit_chain_valid": self.ledger.verify()} |
|
|
| def run(self, prompt: str) -> ModelResult: |
| run_id = str(uuid.uuid4()) |
| self.ledger.record("run_started", run_id, self.config.model_id, "started", "inference requested") |
| try: |
| result = self.adapter.generate(prompt, temperature=self.config.temperature, |
| top_p=self.config.top_p, max_tokens=self.config.max_tokens, |
| policy_mode=self.config.policy_mode) |
| self.ledger.record("run_finished", run_id, self.config.model_id, "success", "inference completed") |
| return result |
| except Exception as exc: |
| capture_error(self.logger, self.ledger, run_id, self.config.model_id, exc) |
| if self.config.fail_closed: |
| raise RuntimeFailure(f"inference failed for {self.config.model_id}") from exc |
| raise |
|
|
| __all__ = ["NATO1000Runtime", "RuntimeFailure"] |
|
|
