vh
ff552abf1f
Straight passthrough to the checkpoint's own DecisionEngine.predict — never the semif mapping, whose different prompt would change the answers. Reuses the one inference thread, bearer auth, MAX_QUEUE, VRAM cap and error envelope; no new concurrency. 1..16 questions in ONE call (never chunked: Jev questions share a prompt); images 422; over MAX_TOKENS 422 before the forward. /health advertises the surface. Response 'model' is a string name@revision (JevBench's runner hashes it; a dict broke its manifest step). Acceptance on the live service (see acceptance/systemone-2026-09-30/): JevBench v1.2.16 typesafe adapter over the 231 public items scores all 202/231, hard 83/111, with 0 changed answers across all 924 rows of the bench's own r1..r4; controls 401/422x3 (token boundary proven at 7168 pass / 7169 refuse); GPU 1 per-process peak 9,866 MiB under the largest accepted request (budget 9,876); /decide/shared positive control unchanged. 120 tests green.
77 lines
3.3 KiB
Python
77 lines
3.3 KiB
Python
"""A torch-free stand-in for the real engine: answers Jev requests in the shape
|
|
DecisionEngine.predict() returns, and records every call it was given."""
|
|
from __future__ import annotations
|
|
|
|
import copy
|
|
import hashlib
|
|
import json
|
|
import math
|
|
|
|
CALIBRATION = {"method": "temperature-scaling", "temperature": 1.99241824}
|
|
|
|
|
|
def softmax(xs: list[float]) -> list[float]:
|
|
top = max(xs)
|
|
e = [math.exp(x - top) for x in xs]
|
|
return [v / sum(e) for v in e]
|
|
|
|
|
|
def score_by_description(field: str, question: dict) -> list[float]:
|
|
"""Default scorer: the option whose description is longest wins; a small first-position bias."""
|
|
descs = list(question["criteria"].values())
|
|
return [len(d) + (0.5 if i == 0 else 0.0) for i, d in enumerate(descs)]
|
|
|
|
|
|
def answer_for(field: str, question: dict, scorer) -> dict:
|
|
"""One Jev answer in the shape DecisionEngine.predict returns, for any of the three types."""
|
|
kind = question["type"]
|
|
if kind == "noul":
|
|
probs = {"yes": 0.75, "no": 0.25}
|
|
return {"type": "noul", "probabilities": probs, "confidence": 0.75, "noul": 0.75,
|
|
"source": "local", "decision": "yes"}
|
|
criteria = question["criteria"]
|
|
if isinstance(criteria, dict):
|
|
keys = list(criteria)
|
|
descriptions = list(criteria.values())
|
|
else: # score: a list of levels
|
|
keys = [str(i) for i in range(len(criteria))]
|
|
descriptions = [str(level) for level in criteria]
|
|
probs = dict(zip(keys, softmax(scorer(field, {**question, "criteria": dict(zip(keys, descriptions))}))))
|
|
best = min(keys, key=lambda k: (-probs[k], k))
|
|
answer = {"type": kind, "probabilities": probs, "confidence": probs[best], "source": "local", "decision": best}
|
|
if kind == "choice":
|
|
answer["choice"] = best
|
|
else:
|
|
answer["score"] = sum(float(k) * p for k, p in probs.items())
|
|
answer["legend"] = {k: (criteria[i] if isinstance(criteria, list) else v)
|
|
for i, (k, v) in enumerate(zip(keys, criteria.values() if isinstance(criteria, dict) else criteria))}
|
|
return answer
|
|
|
|
|
|
class FakeEngine:
|
|
def __init__(self, scorer=score_by_description, tokens_per_question: int = 100):
|
|
self.scorer = scorer
|
|
self.tokens_per_question = tokens_per_question
|
|
self.calls: list[dict] = []
|
|
|
|
metadata = {"name": "Intern-Decision-4B", "revision": "0" * 40}
|
|
|
|
def health(self) -> dict:
|
|
return {**self.metadata, "reserved_gib": 9.1}
|
|
|
|
def predict(self, request: dict) -> tuple[dict, str]:
|
|
self.calls.append(copy.deepcopy(request))
|
|
answers = {field: answer_for(field, question, self.scorer)
|
|
for field, question in request["questions"].items()}
|
|
response = {"answers": answers,
|
|
"usage": {"input_tokens": self.tokens_per_question * len(answers),
|
|
"output_tokens": len(answers), "decision_count": len(answers)},
|
|
"timing": {"inference_ms": 12.5}, "calibration": dict(CALIBRATION),
|
|
"model": "Intern-Decision-4B", "backend": "hf"}
|
|
return response, request_sha(request)
|
|
|
|
|
|
def request_sha(request: dict) -> str:
|
|
"""Stands in for the real prompt hash: the same call always hashes the same."""
|
|
return hashlib.sha256(json.dumps(request, sort_keys=True).encode()).hexdigest()
|