Spaces:

---

multimodalart

/

jev-decision-index

Like 773

Running

"""Decision Index kit Engine for Jev-Style-0.8B-Decision-v3 GGUF (llama.cpp via the repo's own runtime).

Uses the Hub repo chaoliangUNSW/Jev-Style-0.8B-Decision-v3-GGUF as shipped: jev_style_decision_gguf.py +
jev-score (build: sh build_jev_score.sh /path/to/llama.cpp, add LLAMA_CMAKE_FLAGS="-DGGML_CUDA=ON" on CUDA).

Fixed protocol for every benchmark: the runtime's one rendering (macjev-render-v1), ONE global temperature
(category=None), no prompt tuning, no option filtering, options in request order, nothing truncated. Object
instructions / option descriptions are passed as compact JSON (the kit's `text`). A choice question whose options
exceed the 2,048-token head budget is scored by the runtime in option chunks (one softmax over all options).
Requests the runtime rejects (InputBudgetError / QuestionError) raise the kit's Unsupported (scored as wrong).

Options: model_dir=<repo snapshot> (default: download the repo), quant=F16|Q8_0|Q4_K_M (default F16),
binary=<jev-score path> (default: $JEV_SCORE_BIN or <model_dir>/build/jev-score), runtime=<path to an alternative
jev_style_decision_gguf.py>, split_options=true|false.
"""
import hashlib
import importlib.util
import sys
from pathlib import Path

from decision_index.engines.base import Engine, Unsupported, text

REPO = "chaoliangUNSW/Jev-Style-0.8B-Decision-v3-GGUF"


def load_runtime(path):
    sys.dont_write_bytecode = True
    spec = importlib.util.spec_from_file_location("jev_style_decision_gguf", path)
    mod = importlib.util.module_from_spec(spec)
    sys.modules["jev_style_decision_gguf"] = mod
    spec.loader.exec_module(mod)
    return mod


def to_runtime_question(q):
    ins = q.get("instructions", "")
    ins = ins if isinstance(ins, str) else text(ins)
    if q["type"] == "choice":
        crit = {str(k): (v if (v is None or isinstance(v, str)) else text(v)) for k, v in q["criteria"].items()}
        return {"t": "choice", "ins": ins, "crit": crit}
    if q["type"] == "noul":
        c = q.get("criteria") or {}
        crit = {k: (c[k] if isinstance(c[k], str) else text(c[k])) for k in ("false", "true") if k in c and c[k] not in (None, "")}
        return {"t": "noul", "ins": ins, "crit": crit or None}
    raise Unsupported("unsupported question type " + str(q["type"]))


class JevStyleGGUFEngine(Engine):
    name = "jev-style-0.8b-decision-v3-gguf"
    latency = "In-process request wall time: rendering + jev-score (llama.cpp) scoring; excludes model loading."

    def __init__(self, model_dir=None, quant="F16", binary=None, runtime=None, split_options=True, **options):
        super().__init__(**options)
        if model_dir is None:
            from huggingface_hub import snapshot_download
            model_dir = snapshot_download(REPO)
        model_dir = Path(model_dir)
        src = Path(runtime) if runtime else model_dir / "jev_style_decision_gguf.py"
        self.rt = load_runtime(src)
        extra = {} if split_options else {"split_options": False}      # older runtimes have no such argument
        self.m = self.rt.JevStyleDecisionGGUF(model_dir, quant=quant, binary=binary, verify=True, **extra)
        self.provenance = {"model": self.rt.MODEL_NAME, "hf_repo": REPO, "quant": quant, "model_dir": str(model_dir),
                           "runtime_sha256": hashlib.sha256(src.read_bytes()).hexdigest(),
                           "gguf": self.m.gguf.name, "manifest_verified": True, "many_mode": self.m.many_mode,
                           "split_options": bool(split_options), "template": self.rt.TEMPLATE_VERSION,
                           "context_limit": self.rt.CONTEXT_LIMIT, "head_max": self.rt.HARD_HEAD_MAX,
                           "temperature": "global (category=None) = %s" % self.m.temperatures.get("global"),
                           "no_truncation": True, "option_filtering": False, "per_benchmark_prompts": False}

    def __call__(self, state, questions):
        keys = list(questions)
        qs = [to_runtime_question(questions[k]) for k in keys]
        try:
            out = self.m.decide_many(state, qs, category=None)
        except self.rt.InputBudgetError as e:
            raise Unsupported(f"InputBudgetError: {e}") from None
        except self.rt.QuestionError as e:
            raise Unsupported(f"QuestionError: {e}") from None
        answers, raw, tokens = {}, {}, 0
        for k, q, r in zip(keys, questions.values(), out):
            if q["type"] == "choice":
                probs = {str(n): float(r["probabilities"][str(n)]) for n in q["criteria"]}
                answers[k] = {"type": "choice", "choice": r["answer"], "probabilities": probs}
            else:
                answers[k] = {"type": "noul", "noul": float(r["probabilities"]["true"])}
            raw[k] = {"scores": r["scores"], "temperature": r["temperature"], "input_tokens": r["input_tokens"],
                      "head_tokens": r["head_tokens"], **({"option_chunks": r["option_chunks"]} if "option_chunks" in r else {})}
            tokens += r["input_tokens"]
        return {"model": self.rt.MODEL_NAME, "answers": answers, "usage": {"input_tokens": tokens}}, raw

    def runtime(self):
        return {"backend": "llama.cpp", "jev_score": str(self.m.binary), "jev_score_info": self.m.info}

    def close(self):
        self.m.close()