"""Provider-independent voice state and synchronous finite-decision inference. Hosts own async execution, cancellation, and device admission. Importing this module does not import Studio, VoiceMem, Torch, Transformers, or ONNX Runtime. """ from dataclasses import dataclass import json import math from pathlib import Path import threading import numpy as np LABELS = ('ordinary', 'deep') FIELDS = (('eot_probability', 1.), ('silence_s', 2.), ('span_s', 30.)) def observation_contract(): return {'version':1,'audio_fields':[name for name,_ in FIELDS], 'scales':[scale for _,scale in FIELDS], 'layout':['eot','silence','span','eot_present','silence_present','span_present','history_present'], 'text_preprocessing':'voicemem-decision-text', 'missing_value':0,'present_zero_has_mask':True} def prepare(text, history=None, observations=None): """Normalize shared state; missing observations retain explicit masks.""" current = ' '.join(str(text).split()) if len(current) > 256: current = current[:128] + ' … ' + current[-128:] recent = [] for row in list(history or [])[-4:]: role = row.get('role') if role not in ('user', 'assistant'): continue content = ' '.join(str(row.get('content') or '').split())[:80] if content: recent.append(f'{role}: {content}') state = {} if observations is None else observations if not isinstance(state, dict) or set(state) - {name for name, _ in FIELDS}: raise ValueError('Only shared physical observations and EOT probability are accepted') values, masks = [], [] for name, scale in FIELDS: value = state.get(name) if value is None: values.append(0.); masks.append(0.); continue if (isinstance(value, bool) or not isinstance(value, (int, float)) or not math.isfinite(value) or value < 0 or (name == 'eot_probability' and value > 1)): raise ValueError('Invalid shared observation') values.append(min(value / scale, 1.)); masks.append(1.) history_text = '\n'.join(recent) return current, history_text, np.asarray(values + masks + [bool(history_text)], np.float32) def collate(tokenizer, rows, *, tensors='pt'): state = [prepare(row['text'], row.get('history'), row.get('audio_state')) for row in rows] encoded = tokenizer([text for a,b,_ in state for text in (a,b)], padding=True, truncation=True, max_length=384, return_tensors=tensors) observed = np.stack([r[2] for r in state]) if tensors == 'pt': import torch observed = torch.from_numpy(observed) return {'input_ids':encoded['input_ids'], 'attention_mask':encoded['attention_mask'], 'observed':observed} def build_model(config, pretrained=None): import torch from torch import nn from transformers import AutoModel class StateDecisionModel(nn.Module): def __init__(self): super().__init__() if config.model_type not in ('bert','modernbert'): raise ValueError('Unsupported portable encoder') options = {'trust_remote_code':False, 'attn_implementation':'sdpa' if config.model_type == 'modernbert' else 'eager'} if config.model_type == 'modernbert': config.reference_compile = False else: options['add_pooling_layer'] = False self.encoder = (AutoModel.from_pretrained(str(pretrained),config=config,local_files_only=True, weights_only=True,**options) if pretrained else AutoModel.from_config(config,**options)) self.hidden_size = config.hidden_size width = 3*self.hidden_size + 7 self.classifier = nn.Sequential(nn.LayerNorm(width),nn.Linear(width,256),nn.GELU(), nn.Dropout(.15),nn.Linear(256,2)) def forward(self,input_ids,attention_mask,observed): hidden = self.encoder(input_ids=input_ids,attention_mask=attention_mask).last_hidden_state[:,0] pair = hidden.reshape(observed.shape[0],2,self.hidden_size) current = pair[:,0] recent = pair[:,1] * observed[:,-1:] return self.classifier(torch.cat((current,recent,current*recent,observed),dim=-1)) return StateDecisionModel() def load_model(path, device='cpu', *, pretrained=False): from transformers import AutoConfig, AutoTokenizer from safetensors.torch import load_file path = Path(path) config_path = path if pretrained else path/'encoder' config = AutoConfig.from_pretrained(config_path,local_files_only=True,trust_remote_code=False) model = build_model(config,path if pretrained else None) if not pretrained: model.load_state_dict(load_file(str(path/'model.safetensors')),strict=True) tokenizer = AutoTokenizer.from_pretrained(path if pretrained else path/'tokenizer', local_files_only=True,trust_remote_code=False,fix_mistral_regex=False) return model.to(device).eval(),tokenizer @dataclass(frozen=True) class Decision: label: str p_ordinary: float p_deep: float calibration_scope: str = 'machine-supervised, not independently human-validated' class Router: """Run a local model independently of ASR, TTS, LLM, and host application.""" def __init__(self,path,*,backend='onnx',device='cpu',admission=None): if backend not in ('onnx','eager') or (backend=='onnx' and device!='cpu'): raise ValueError('Unsupported runtime') self.path = Path(path) self.meta = json.loads((self.path/'depth_encoder.json').read_text()) if (self.meta.get('version')!=1 or self.meta.get('features_version')!=1 or self.meta.get('observation_contract')!=observation_contract() or self.meta.get('task')!='voice-depth-encoder' or self.meta.get('labels')!=list(LABELS) or self.meta.get('max_length')!=384 or self.meta.get('threshold')!=.5 or self.meta.get('history_policy')!='first80'): raise ValueError('Incompatible portable decision contract') self.temperature = float(self.meta['temperature']) if not math.isfinite(self.temperature) or self.temperature<=0: raise ValueError('Invalid calibration') self.backend,self.device,self.admission = backend,device,admission self.model = self.tokenizer = None self.lock = threading.Lock() def _load(self): if self.backend=='eager': self.model,self.tokenizer = load_model(self.path,self.device) else: import onnxruntime as ort from transformers import AutoTokenizer self.tokenizer = AutoTokenizer.from_pretrained(self.path/'tokenizer',local_files_only=True, trust_remote_code=False,fix_mistral_regex=False) options = ort.SessionOptions() options.intra_op_num_threads,options.inter_op_num_threads = 2,1 self.model = ort.InferenceSession(str(self.path/'model.onnx'),sess_options=options, providers=['CPUExecutionProvider']) def predict(self,text,*,history=None,observations=None): from contextlib import nullcontext current,recent,observed = prepare(text,history,observations) if not current: return Decision('ordinary',1.,0.,'empty-input compatibility rule') with self.lock, (self.admission() if self.admission else nullcontext()): if self.model is None: self._load() encoded = self.tokenizer([current,recent],padding=True,truncation=True, max_length=384,return_tensors='np') inputs = {'input_ids':encoded['input_ids'],'attention_mask':encoded['attention_mask'], 'observed':observed[None]} if self.backend=='onnx': logits = self.model.run(['logits'],inputs)[0][0] else: import torch with torch.inference_mode(): logits = self.model(**{k:torch.from_numpy(v).to(self.device) for k,v in inputs.items()})[0].float().cpu().numpy() scaled = np.asarray(logits,np.float64)/self.temperature if scaled.shape!=(2,) or not np.isfinite(scaled).all(): raise ValueError('Invalid decision logits') probability = np.exp(scaled-scaled.max()); probability /= probability.sum() deep = float(probability[1]) return Decision('deep' if deep>=.5 else 'ordinary',1.-deep,deep)