| """Reusable text-choice and compact arithmetic decisions over one resident model.""" |
| import json,time |
| from .numeric import applicable,compact_unit,compact_prompt,decode,INSTRUCTION |
| from .readout import prompt_for,readout |
|
|
|
|
| def validate(request): |
| if not isinstance(request,dict) or set(request)!={'state','question','options'}:raise ValueError('Provide exactly state, question, and options') |
| if not all(isinstance(request[k],str) and request[k].strip() for k in ['state','question']):raise ValueError('State and question must be nonempty strings') |
| options=request['options'] |
| if not isinstance(options,list) or not 2<=len(options)<=26:raise ValueError('Provide 2 to 26 options') |
| if not all(isinstance(o,(list,tuple)) and len(o)==2 and all(isinstance(v,str) and v for v in o) for o in options):raise ValueError('Options must be pairs of nonempty strings') |
| if len({o[0] for o in options})!=len(options):raise ValueError('Option IDs must be unique') |
| if len(json.dumps(request))>65536:raise ValueError('Request exceeds the 64 KiB input limit') |
|
|
|
|
| class DecisionEngine: |
| def __init__(self,client): |
| client.verify_profile() |
| self.client=client |
|
|
| def decide(self,request): |
| validate(request) |
| |
| with self.client.lock:return self._decide(request) |
|
|
| def _decide(self,request): |
| self.client.calls.clear() |
| tick=time.perf_counter();start=0;trace={};attempted=applicable(request) |
| if attempted: |
| unit=compact_unit(request) |
| prompt=compact_prompt(request) if unit is not None else INSTRUCTION+'\n'+prompt_for(request['state'],request['question'],request['options']).replace('Answer with the letter of the best option only.','Return only the JSON calculation, or a null expression.') |
| response=self.client.post('/v1/chat/completions',dict(model='qwen',messages=[dict(role='user',content=prompt)],max_tokens=64 if unit is not None else 96,temperature=0)) |
| choice=response['choices'][0];content=choice['message'].get('content') or '' |
| trace=dict(tool_expression=content,tool_tokens=(response.get('usage') or {}).get('completion_tokens'),tool_finish_reason=choice['finish_reason'],tool_error=None) |
| if unit is not None:trace['output_unit']=unit |
| try: |
| if choice['finish_reason']=='length':raise ValueError('truncated') |
| answer,_=decode(json.dumps(dict(expression=content,unit=unit)) if unit is not None else content,request['options']) |
| return dict(answer=answer,method='numeric-tool',tool_attempted=True,seconds=time.perf_counter()-tick,score=None,trace=trace,calls=self.client.calls[start:]) |
| except (ValueError,TypeError,AssertionError,SyntaxError,ZeroDivisionError,OverflowError,KeyError,IndexError) as exc:trace['tool_error']=str(exc) |
| result=readout(self.client,request['state'],request['question'],request['options'],.85,dict(cache_prompt=False,top_logprobs=1024)) |
| if result['missing_letters']:raise RuntimeError('Backend did not return all option scores') |
| return dict(answer=result['argmax'],method='fast',tool_attempted=attempted,seconds=time.perf_counter()-tick, |
| score=max(p['p'] for p in result['probs']),score_kind='option-relative softmax; not a correctness guarantee', |
| probs=result['probs'],trace=trace,calls=self.client.calls[start:]) |
|
|
