| |
| |
| """Apertus Decide: shared input contract, scoring and direct CPU serving.""" |
| import json |
| import math |
| from pathlib import Path |
| import numpy as np |
|
|
| SCHEMA_VERSION = 1 |
|
|
| def text(value): |
| return value if isinstance(value, str) else json.dumps(value, ensure_ascii=False, sort_keys=True) |
|
|
| def compile_question(state, question): |
| """Identical text construction during training, evaluation and deployment.""" |
| kind = question.get('type') |
| instruction = question.get('instructions') |
| if kind not in ('choice', 'score', 'noul') or instruction is None: |
| raise ValueError('Each question requires type and instructions.') |
| criteria = question.get('criteria') |
| query = 'State:\n' + text(state) + '\nQuestion:\n' + text(instruction) |
| if kind == 'choice': |
| if not isinstance(criteria, dict) or not 2 <= len(criteria) <= 64: |
| raise ValueError('Choice requires 2–64 options.') |
| keys = list(criteria) |
| candidates = [k if v is None else k + ': ' + text(v) for k, v in criteria.items()] |
| elif kind == 'score': |
| if not isinstance(criteria, list) or not 2 <= len(criteria) <= 10: |
| raise ValueError('Score requires 2–10 ordered descriptions.') |
| keys = [str(i) for i in range(len(criteria))] |
| candidates = [text(v) for v in criteria] |
| else: |
| if criteria is not None and (not isinstance(criteria, dict) or set(criteria) != {'false', 'true'}): |
| raise ValueError('Noul criteria must provide both false and true.') |
| keys = ['false', 'true'] |
| candidates = [text(criteria[k]) for k in keys] if criteria else ['The answer is no.', 'The answer is yes.'] |
| |
| return ['query: ' + query] + ['passage: ' + c for c in candidates], keys |
|
|
| def target_distribution(question, answer, keys): |
| n = len(keys) |
| if 'probabilities' in answer: |
| if set(answer['probabilities']) != set(keys): |
| raise ValueError('Target distribution keys do not match options.') |
| y = np.array([answer['probabilities'][k] for k in keys], dtype=np.float32) |
| elif question['type'] == 'noul': |
| p = float(answer['noul']); y = np.array([1-p, p], dtype=np.float32) |
| elif 'choice' in answer: |
| y = np.eye(n, dtype=np.float32)[keys.index(answer['choice'])] |
| elif 'level' in answer: |
| level = int(answer['level']) |
| if level != answer['level'] or not 0 <= level < n: raise ValueError('Invalid score level.') |
| y = np.eye(n, dtype=np.float32)[level] |
| else: |
| score = float(answer['score']) |
| if not 0 <= score <= n-1: raise ValueError('Invalid expected score.') |
| lo, hi = math.floor(score), math.ceil(score) |
| y = np.zeros(n, dtype=np.float32) |
| if lo == hi: y[lo] = 1 |
| else: y[lo], y[hi] = hi-score, score-lo |
| if not np.isfinite(y).all() or (y < 0).any() or not np.isclose(y.sum(), 1, atol=1e-5): |
| raise ValueError('Invalid target probabilities.') |
| return y |
|
|
| def probabilities(logits, temperature=1.): |
| z = np.asarray(logits, dtype=np.float64) / temperature |
| z -= z.max(); p = np.exp(z); return p / p.sum() |
|
|
| def render_answer(question, keys, p): |
| kind = question['type'] |
| if kind == 'noul': return {'type': kind, 'noul': float(p[1])} |
| confidence = float(np.clip(1 + np.sum(p * np.log(np.maximum(p, 1e-15))) / np.log(len(p)), 0, 1)) |
| out = {'type': kind, 'probabilities': dict(zip(keys, map(float, p))), 'confidence': confidence} |
| if kind == 'choice': out['choice'] = keys[int(np.argmax(p))] |
| else: |
| out['score'] = float(np.dot(np.arange(len(p)), p)) |
| out['legend'] = dict(zip(keys, question['criteria'])) |
| return out |
|
|
| class DecisionModel: |
| def __init__(self, directory, batch_size=8): |
| import torch |
| from sentence_transformers import SentenceTransformer |
| self.directory = Path(directory) |
| self.config = json.loads((self.directory / 'decision_config.json').read_text()) |
| self.batch_size = int(batch_size) |
| if self.batch_size < 1: |
| raise ValueError('batch_size must be positive.') |
| if self.config.get('schema_version', 1) != SCHEMA_VERSION: |
| raise ValueError('Unsupported decision schema version.') |
| scale = float(self.config['logit_scale']) |
| if not math.isfinite(scale) or scale <= 0: |
| raise ValueError('Invalid logit_scale.') |
| for kind in ('choice', 'score', 'noul'): |
| temperature = float(self.config['temperatures'][kind]) |
| if not math.isfinite(temperature) or temperature <= 0: |
| raise ValueError('Invalid calibration temperature.') |
| self.model = SentenceTransformer(str(self.directory), device='cpu', |
| trust_remote_code=False, local_files_only=True, |
| model_kwargs={'attn_implementation': 'sdpa'}) |
| self.model.float().eval() |
| self.model.max_seq_length = int(self.config['max_length']) |
| if self.model.max_seq_length < 1: |
| raise ValueError('Invalid model context limit.') |
|
|
| def encode(self, inputs): |
| sizes = [len(x) for x in self.model.tokenizer(inputs, truncation=False)['input_ids']] |
| if max(sizes) > self.config['max_length']: |
| raise ValueError('Input exceeds the supported token limit; shorten state or criteria.') |
| result = self.model.encode(inputs, prompt='', batch_size=self.batch_size, |
| normalize_embeddings=True, convert_to_numpy=True, show_progress_bar=False) |
| if not np.isfinite(result).all(): |
| raise RuntimeError('Encoder returned non-finite embeddings.') |
| return result |
|
|
| def evaluate(self, request): |
| if not isinstance(request, dict) or 'state' not in request: |
| raise ValueError('Provide a state and named questions.') |
| if not isinstance(request['state'], (str, dict, list)): |
| raise ValueError('state must be text, an object, or an array.') |
| questions = request.get('questions') |
| if not isinstance(questions, dict) or not 1 <= len(questions) <= 64: |
| raise ValueError('Provide 1-64 named questions.') |
| compiled = [] |
| unique = {} |
| for name, question in questions.items(): |
| if not isinstance(name, str) or not name or not isinstance(question, dict): |
| raise ValueError('Each named question must be an object.') |
| inputs, keys = compile_question(request['state'], question) |
| indices = [] |
| for value in inputs: |
| if value not in unique: |
| unique[value] = len(unique) |
| indices.append(unique[value]) |
| compiled.append((name, question, keys, indices)) |
| embeddings = self.encode(list(unique)) |
| answers = {} |
| for name, question, keys, indices in compiled: |
| e = embeddings[indices] |
| logits = (e[1:] @ e[0]) * self.config['logit_scale'] |
| p = probabilities(logits, self.config['temperatures'][question['type']]) |
| answers[name] = render_answer(question, keys, p) |
| return {'model': request.get('model', self.config['model_id']), 'answers': answers} |
|
|
|
|
| def app(directory, batch_size=8): |
| import os |
| import hmac |
| import threading |
| from fastapi import FastAPI, HTTPException, Request |
| from starlette.concurrency import run_in_threadpool |
| model = DecisionModel(directory, batch_size=batch_size) |
| service = FastAPI(title='Apertus Decide System One API') |
| lock = threading.Lock() |
| def evaluate(body): |
| with lock: |
| return model.evaluate(body) |
| @service.get('/health') |
| def health(): |
| return {'status': 'ok', 'backend': 'sentence-transformers', 'device': 'cpu'} |
| @service.post('/v1/systemone') |
| async def systemone(request: Request): |
| key = os.environ.get('DECIDE_API_KEY') |
| if key and not hmac.compare_digest(request.headers.get('x-api-key') or request.headers.get('authorization', '').removeprefix('Bearer '), key): |
| raise HTTPException(401, 'Invalid API key.') |
| raw = bytearray() |
| async for chunk in request.stream(): |
| raw.extend(chunk) |
| if len(raw) > 65536: |
| raise HTTPException(413, 'Request exceeds 64 KiB.') |
| try: |
| def reject_constant(value): |
| raise ValueError('Non-finite JSON numbers are not supported.') |
| body = json.loads(raw, parse_constant=reject_constant) |
| if isinstance(body, dict) and body.get('model') not in (None, 'apertus-decide', model.config['model_id']): |
| raise HTTPException(404, 'The requested model is not served here.') |
| return await run_in_threadpool(evaluate, body) |
| except (ValueError, KeyError, TypeError, AttributeError) as exc: |
| raise HTTPException(422, str(exc)) from exc |
| return service |
|
|
|
|
| if __name__ == '__main__': |
| import argparse |
| import uvicorn |
| parser = argparse.ArgumentParser() |
| parser.add_argument('--model-dir', required=True) |
| parser.add_argument('--batch-size', type=int, default=8) |
| parser.add_argument('--host', default='127.0.0.1') |
| parser.add_argument('--port', type=int, default=8000) |
| args = parser.parse_args() |
| uvicorn.run(app(args.model_dir, args.batch_size), host=args.host, port=args.port) |
|
|
