"""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()