File size: 3,038 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 | """Real callback-driven escalation. Internal budget/escalation layer: the answer
contract above it always answers and refuses to emit any deferral field."""
from dataclasses import dataclass
import math
from numbers import Integral, Real
@dataclass(frozen=True)
class Budget:
branches: int = 8
input_tokens: int = 8192
generated_tokens: int = 3072
def route(stages, budget=Budget()):
"""stages are (name, reserved_cost, callback). Callback returns answer/confidence/cost.
Reservations are hard maxima enforced by the backend, not optimistic estimates.
Callbacks must honor the reservation passed to them. Overspending fails closed.
"""
used = {'branches':0,'input_tokens':0,'generated_tokens':0}
valid_cost=lambda value:isinstance(value,Integral) and not isinstance(value,bool) and value>=0
valid_probability=lambda value:isinstance(value,Real) and not isinstance(value,bool) and math.isfinite(value) and 0<=value<=1
if any(not valid_cost(getattr(budget,k)) for k in used):
raise ValueError('invalid compute budget')
traces, last = [], None
reasons = {'fast':'full_depth','views':'escalated_views','reasoning':'escalated_reasoning','exit':'early_exit'}
for name,reserve,call in stages:
if name not in reasons or set(reserve)!=set(used) or any(not valid_cost(v) for v in reserve.values()):
raise ValueError('invalid stage reservation')
if any(used[k]+reserve.get(k,0) > getattr(budget,k) for k in used):
return {'answer':None,'candidate':last,'abstained':True,'compute':{'stop_reason':'budget_exhausted','cost':used,'stages':traces}}
result = call(dict(reserve))
cost = result['cost']
if set(cost)!=set(used) or any(not valid_cost(cost[k]) or cost[k] > reserve[k] for k in used):
raise ValueError('backend exceeded reserved compute budget')
for k in used:
used[k] += cost.get(k,0)
last = result.get('answer')
confidence = result.get('confidence') or {}
if not isinstance(confidence,dict):
confidence={}
traces.append({'stage':name,'cost':cost,'confidence':confidence})
certified = confidence.get('calibration_support') == 'in_validated_scope' and confidence.get('policy_status') == 'confirmed'
accepted = (certified and last is not None and valid_probability(confidence.get('reliability'))
and valid_probability(confidence.get('threshold')) and confidence['reliability'] >= confidence['threshold'])
if accepted:
return {'answer':last,'abstained':False,'confidence':confidence,'compute':{'stop_reason':reasons[name],'cost':used,'stages':traces}}
return {'answer':None,'candidate':last,'abstained':True,'compute':{'stop_reason':'abstain','cost':used,'stages':traces}}
def forced_answer_suffix(trace):
"""Backend appends this to a capped trace and performs a letter-logit forward."""
return ('' if '</think>' in trace else '\n</think>') + '\nAnswer (one letter):'
|