File size: 4,009 Bytes
1d2de8a | 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 | """Preserve five-task trained-head confidence/routing with optional evidence."""
import hashlib,json
from pathlib import Path
import numpy as np
from solomon.engine import SolomonEngine,sha
from solomon.engine_reasoning import ReasoningEngine
from solomon.service_heads import HeadsService
from solomon.service_answers import Service as ConfidenceService
from solomon.service_evidence import EvidenceService,CONTRACT
class ServiceEngine(SolomonEngine,ReasoningEngine):
def __init__(self,*args,correctness_report_sha256=None,**kwargs):
super().__init__(*args,**kwargs)
self.identity.update(trained_heads_report_sha256=correctness_report_sha256,serving_sha256=sha(__file__))
self.identity.pop('fingerprint',None)
self.identity['fingerprint']=hashlib.sha256(json.dumps(self.identity,sort_keys=True).encode()).hexdigest()
def estimate(self,state,block,n,stage='fast',max_new_tokens=512):
if not any('image' in p for p in state['parts']):
return ReasoningEngine.estimate(self,state,block,n,stage=stage,max_new_tokens=max_new_tokens)
if stage!='fast':raise ValueError('image reasoning is unqualified')
with self._lock,self.torch.inference_mode():
ids,_,_,_=self._encode(state['parts'],self._render(state['parts'],block),state['features'])
p=state['prefix_tokens']
if not self.torch.equal(ids[:,:p],state['prefix_ids']):raise ValueError('image estimate prefix mismatch')
return {'input_tokens':int(ids.shape[1])-p,'generated_tokens':0,'branches':1,
'full_prompt_tokens':int(ids.shape[1]),'reused_prefix_tokens':p}
class ConfidenceHeadsService(HeadsService):
def expected_identity(self,stage):
return {**super().expected_identity(stage),'contract':CONTRACT,'serving_sha256':sha(__file__)}
def _validate_artifact(self,stage,artifact):
ConfidenceService._validate_artifact(self,stage,artifact)
if artifact.get('schema')!='solomon-heads-envelope-v1':raise ValueError('the evidence layer confidence envelope required')
selected=artifact.get('trained_head_identity',{});runtime=self.runtime_identity
if (selected.get('weights_sha256')!=runtime.get('trained_heads_sha256') or
selected.get('report_sha256')!=runtime.get('trained_heads_report_sha256') or
selected.get('feature_identity',{}).get('runtime_fingerprint')!=runtime.get('base_fingerprint') or
selected.get('feature_identity',{}).get('adapter_sha256')!=runtime.get('adapter_sha256')):
raise ValueError('the evidence layer correctness/head/backbone identity mismatch')
class FreshDecider:
def __init__(self,engine,design,store):self.engine=engine;self.design=design;self.store=store
def __call__(self,parts,task,request):
from solomon.service_states import Service as Readout
engine=self.engine
class Capture:
def __init__(self):self.records=[]
def ask(self,*args,**kwargs):
result=engine.ask(*args,**kwargs);self.records.append(result);return result
capture=Capture();state=engine.prefill(parts or [{'text':''}])
class Fixed(Readout):
def _warm(self,key):return state
readout=Fixed(self.store,capture,self.design)
with engine.task_context(task):answer=readout.ask('fresh-evidence',task,**request)
return {'answer':answer,'cost':{'input_tokens':state['prefix_tokens']+sum(r['branch_tokens'] for r in capture.records),
'branches':len(capture.records),'prefill_tokens':state['prefix_tokens']}}
def service(store,engine,design,*,artifact=None,views_artifact=None,views_tasks=(),selector=None,page_selector=None):
answers=ConfidenceHeadsService(store,engine,design,artifact=artifact,views_artifact=views_artifact,views_tasks=views_tasks)
return EvidenceService(answers,FreshDecider(engine,design,Path(store)/'.evidence-readout'),selector,page_selector)
|