File size: 9,307 Bytes
1209c2f f843101 5adf069 1209c2f 5adf069 1209c2f 5adf069 1209c2f 5adf069 1209c2f f843101 | 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 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 | """Offline Fev loader and the unchanged benchmarked routing composition.
The imported fev primitives are byte-identical to the frozen benchmark runtime.
No customer route definitions or default customer rules are embedded here.
"""
import copy
import json
import os
import sys
import threading
import time
from pathlib import Path
from fev.intent_membership import request_for
from fev.ranked_requests import prepare
from fev.route_applicability import build_request
from fev.ranked_membership import resolve
from fev.research_controls import compare_responses
from .integrity import verify_package
from .decisions import decide_with_predict, validate_request
from .version import VERSION
def route_with_predict(request, *, unknown_key, predict):
"""Compose the frozen primary and independent applicability calls.
predict must accept a complete native request and return a native response.
Clarification is a decision signal; this function does not generate text.
"""
request_for(request, unknown_key) # Strict validation before any inference.
primary = predict(request)
compare_responses(primary, primary)
if set(primary['answers']) != {'route'}:
raise ValueError('Unexpected primary answer scope')
choice = primary['answers']['route']['probabilities']
config = prepare(request, choice, unknown_key=unknown_key)
secondary_request = None
secondary = None
evidence = dict.fromkeys(k for k in choice if k != unknown_key)
if config['request'] is not None:
secondary_request = build_request(request, unknown_key=unknown_key, keys=config['queriedKeys'])
secondary = predict(secondary_request)
compare_responses(secondary, secondary)
if set(secondary['answers']) != set(config['queriedKeys']):
raise ValueError('Unexpected applicability answer scope')
evidence.update({key: value['noul'] for key, value in secondary['answers'].items()})
decision = resolve(evidence, choice, config['precedence'], unknown_key=unknown_key, yes=.8, no=.2)
responses = [primary] + ([secondary] if secondary is not None else [])
return {'model': 'Fev-4B', 'version': VERSION, 'decision': decision,
'route': decision.get('key'), 'probabilities': choice, 'applicability': evidence,
'queriedKeys': config['queriedKeys'], 'precedence': config['precedence'],
'usage': {key: sum(part['usage'][key] for part in responses)
for key in ['input_tokens', 'output_tokens', 'sequences']},
'native': {'primary': primary, 'secondaryRequest': secondary_request, 'secondary': secondary}}
class Fev4B:
"""Load one complete local package; CUDA BF16 reference execution only.
One instance keeps weights resident. Calls are serialized because concurrency
and concurrent GPU-memory use have not been qualified for this release.
"""
def __init__(self, directory, *, device='cuda:0', execution='reference', adaptation_dir=None):
if execution not in ('reference', 'compiled'):
raise ValueError('execution must be reference or compiled')
self.root = Path(directory).resolve()
self.integrity = verify_package(self.root)
# All files are bundled. Runtime inference must not resolve Hub resources.
os.environ['HF_HUB_OFFLINE'] = '1'
os.environ['TRANSFORMERS_OFFLINE'] = '1'
import torch
from peft import LoraConfig, inject_adapter_in_model, set_peft_model_state_dict, get_peft_model_state_dict
from safetensors.torch import load_file
if torch.device(device).type != 'cuda' or not torch.cuda.is_available():
raise RuntimeError('This release requires a BF16-capable CUDA GPU; CPU/MPS inference is not validated.')
with torch.cuda.device(device):
if not torch.cuda.is_bf16_supported():
raise RuntimeError('CUDA BF16 support is required')
torch.set_num_threads(4)
self.config = json.loads((self.root / 'fev_config.json').read_text())
expected = {'maxLength': 16384, 'batchTokens': 16384, 'seed': 20261030,
'nativeTemperature': 1.5539092221462472, 'thresholds': {'yes': .8, 'no': .2}}
if self.config['runtime'] != expected:
raise ValueError('Runtime configuration differs from the benchmarked release')
torch.manual_seed(expected['seed'])
sdk = self.root / 'checkpoint' / 'source' / 'src'
sys.path.insert(0, str(sdk))
from pplx_decider import Decider
module_path = Path(sys.modules['pplx_decider'].__file__).resolve()
if not module_path.is_relative_to(sdk):
raise RuntimeError('A conflicting native SDK is already imported; use an isolated Python process')
self.model = Decider.from_pretrained(self.root / 'checkpoint', device=device, fast=False,
max_length=expected['maxLength'], batch_tokens=expected['batchTokens'], cpu_threads=4)
if self.model.temperature != expected['nativeTemperature']:
raise ValueError('Native temperature changed')
adapter = self.root / 'adapter'
inject_adapter_in_model(LoraConfig.from_pretrained(adapter), self.model.backbone.language_model)
for name, param in self.model.backbone.named_parameters():
if 'lora_' in name: param.data = param.data.float()
param.requires_grad_(False)
for param in self.model.readout.parameters(): param.requires_grad_(False)
state = load_file(str(adapter / 'adapter_model.safetensors'))
if set(state) != set(get_peft_model_state_dict(self.model.backbone.language_model)):
raise ValueError('Adapter key set changed')
restored = set_peft_model_state_dict(self.model.backbone.language_model, state)
if restored.unexpected_keys or any('lora_' in k for k in restored.missing_keys):
raise ValueError('Adapter was not restored completely')
loras = [p for n,p in self.model.backbone.named_parameters() if 'lora_' in n]
if len(loras) != 240 or sum(p.numel() for p in loras) != 10387456 or not all(p.dtype == torch.float32 for p in loras):
raise ValueError('Adapter shape or dtype differs from the benchmark')
self.model.backbone.eval()
self.adaptation = None
if adaptation_dir is not None:
from .adaptation import load_adaptation
self.adaptation = load_adaptation(self, adaptation_dir)
# Construct after restoring the shared adapter. Graphs, FP8, pruning and
# prefetching were not qualified and are intentionally disabled.
if execution == 'compiled':
from pplx_decider.fast import FastPath
from pplx_decider.model import FastOptions
self.model.fast = FastPath(self.model, FastOptions(
compile='layer', cuda_graphs=False, prune_last=False,
fp8=False, prefetch=False, token_budget=16384))
self.execution = execution
self.parameters = list(self.model.backbone.parameters()) + list(self.model.readout.parameters())
if any(p.requires_grad for p in self.parameters):
raise ValueError('Inference parameters must be frozen')
self._versions = [(p,p._version) for p in self.parameters]
self._lock = threading.Lock()
self.device = device
def _predict(self, request):
import torch
chunks = self.model.encoder.plan(request, 16384)
if len(chunks) != 1 or chunks[0].length > 16384:
raise ValueError('Request exceeds the 16,384-token single-sequence limit; rules are never truncated')
with torch.inference_mode(): response = self.model.decide(request)
if response['usage'] != {'input_tokens': chunks[0].length, 'output_tokens': 0, 'sequences': 1}:
raise ValueError('Unexpected native token usage')
if set(response['answers']) != set(request['questions']):
raise ValueError('Incomplete native answer')
return response
def route(self, request, *, unknown_key):
import torch
request = copy.deepcopy(request)
with self._lock:
torch.cuda.synchronize(self.device)
started = time.perf_counter()
result = route_with_predict(request, unknown_key=unknown_key, predict=self._predict)
torch.cuda.synchronize(self.device)
if any(p._version != version or p.requires_grad for p,version in self._versions):
raise RuntimeError('Model parameters changed during inference')
result['latencyMs'] = 1000 * (time.perf_counter() - started)
return result
def decide(self, request):
"""Answer Choice/Noul/Score questions in one joint forward pass.
This exposes the native decision head, without the two-stage routing
policy. Question quality and calibration need workload-specific checks.
"""
request = validate_request(request)
with self._lock:
result = decide_with_predict(request, predict=self._predict)
if any(p._version != version or p.requires_grad for p, version in self._versions):
raise RuntimeError('Model parameters changed during inference')
return result
|