# Copyright 2026 Andreas Martin # SPDX-License-Identifier: Apache-2.0 """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.'] # Explicit prefixes replace inherited retrieval prompts. 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)