| """Decision Index engine for Dinah-1 (https://github.com/apolinario/decision-index). |
| |
| python -m decision_index run --engine decision_index_engine:Dinah01Engine \ |
| --option repo=<path or repo> --edition 0.3 --suite-dir suite-0.3 --out runs/dinah-1 |
| |
| Put this file next to `experts.py`, `dinah.py` and `tools.py` (all in the model repository) and on PYTHONPATH. |
| Kit rules: nothing is truncated, no option is dropped and the rendering is the same for every benchmark. The gate |
| reads only the request content. A request that does not fit in 8,192 tokens, or a question type the model does not |
| support, raises `Unsupported` (unless an expert settles all of its questions without the encoder). |
| """ |
| from __future__ import annotations |
|
|
| import sys |
| from pathlib import Path |
|
|
| from decision_index.engines import Engine, Unsupported |
|
|
| sys.path.insert(0, str(Path(__file__).resolve().parent)) |
| from dinah import RequestTooLong |
| from experts import Dinah01 |
|
|
|
|
| class Dinah01Engine(Engine): |
| name = "dinah-1" |
| latency = ("In-process request wall time: code and calculator experts (when they fire), then one encoder pass over the " |
| "request (all questions batched together); excludes model loading.") |
|
|
| def __init__(self, repo: str = ".", revision: str | None = None, device: str | None = None, dtype: str | None = None, |
| batch_size: int = 16, tools: str = "code,calculator", backend: str = "torch", calibrate: str | bool = True, **options): |
| super().__init__(**options) |
| self.model = Dinah01.from_pretrained(repo, device=device, dtype=dtype, revision=revision, tools=tools, backend=backend) |
| self.batch_size = int(batch_size) |
| enc = self.model.encoder |
| self.calibrated = str(calibrate).lower() not in ("0", "false", "no", "off") |
| if not self.calibrated: |
| enc.temperature, enc.platt = None, None |
| self.provenance = {"repo": repo, "revision": revision or "main", "backend": backend, |
| "device": getattr(enc, "device", "cpu"), "dtype": getattr(enc, "dtype", "int8"), "calibrated": self.calibrated, |
| "max_length": self.model.max_length, "experts": self.model.tools, |
| "truncation": "none: requests above 8,192 tokens raise Unsupported"} |
|
|
| def __call__(self, state, questions): |
| for q in questions.values(): |
| if q["type"] not in ("choice", "noul"): |
| raise Unsupported(f"question type {q['type']}") |
| try: |
| answers = self.model.predict(state, questions, batch_size=self.batch_size) |
| except RequestTooLong as e: |
| raise Unsupported(f"context window: {e}") |
| return {"model": "dinah-1", "answers": answers}, None |
|
|
| def synchronize(self): |
| enc = self.model.encoder |
| if getattr(enc, "device", "cpu").startswith("cuda"): |
| enc.torch.cuda.synchronize() |
|
|
