From dda2a2f078ae57040c3adbb65509dce4429cc879 Mon Sep 17 00:00:00 2001 From: BANADDA MUBARAKA <83466862+BANADDA@users.noreply.github.com> Date: Thu, 15 Oct 2026 21:59:48 +0300 Subject: [PATCH] engine: task names and aliases come from the live catalogue --- README.md | 5 +- microtensor/__init__.py | 17 +++- microtensor/engine.py | 192 +++++++++++++++++++++++++++++++--------- tests/test_engine.py | 51 ++++++++--- 4 files changed, 209 insertions(+), 56 deletions(-) diff --git a/README.md b/README.md index a50df8a..69d355a 100644 --- a/README.md +++ b/README.md @@ -9,6 +9,7 @@ The hosted API needs no SDK; any OpenAI client works. The Engine is the option f from microtensor import Engine Engine.frontier_models() # live list of frontier models our inference miners serve + Engine.served_tasks() # live list of tasks, their aliases and whether each is live engine = Engine.load( tasks=["text-to-sql", "classification", "extraction"], @@ -25,7 +26,9 @@ The hosted API needs no SDK; any OpenAI client works. The Engine is the option f ## Behaviour -- `tasks` maps each task to the current certified specialist in its arena. Users never see arenas, and get upgrades automatically when a better specialist is certified. A task the network does not serve yet raises `TaskUnavailable` at load time, naming what is served. +- `tasks` are resolved against the live catalogue when the engine loads; the SDK holds no list of its own. A name can be a task id (`sql_generation`), its hyphenated form, a model id (`mt/sql_gen_s1`) or any alias the arena declares (`text-to-sql`, `sql`). New tasks and renamed aliases reach every installed SDK the moment the catalogue changes. +- `Engine.served_tasks()` returns what the network serves right now: each task, its current specialist, its aliases and whether an operator is live. A task the network does not serve raises `TaskUnavailable` at load time, naming what is served. +- Each task maps to the current certified specialist in its arena. Users never see arenas and get upgrades automatically when a better specialist is certified. - The certified router identifies the task for each request; `result.task` says which specialist answered. Pass `task=` to `run` to skip classification. - Escalation stays inside the Microtensor network, served by frontier models hosted by inference miners. `escalation_model` takes `"auto"`, a model id from `frontier_models()`, or a dict per task such as `{"text-to-sql": "deepseek-v4-pro", "classification": False}`. - With escalation off, with no frontier model served, or once the daily budget is used, answers below the floor come back with `below_floor=True` and the application decides. diff --git a/microtensor/__init__.py b/microtensor/__init__.py index b52f628..18fb766 100644 --- a/microtensor/__init__.py +++ b/microtensor/__init__.py @@ -1,11 +1,26 @@ from microtensor.engine import ( Budget, + Catalogue, Engine, EngineError, Result, + Served, TaskUnavailable, + catalogue, frontier_models, + served_tasks, ) -__all__ = ["Budget", "Engine", "EngineError", "Result", "TaskUnavailable", "frontier_models"] +__all__ = [ + "Budget", + "Catalogue", + "Engine", + "EngineError", + "Result", + "Served", + "TaskUnavailable", + "catalogue", + "frontier_models", + "served_tasks", +] __version__ = "0.1.0" diff --git a/microtensor/engine.py b/microtensor/engine.py index f1e4265..96dcac1 100644 --- a/microtensor/engine.py +++ b/microtensor/engine.py @@ -12,13 +12,6 @@ AUTO: Final[str] = "mt/auto" DEFAULT_FLOOR: Final[float] = 0.9 TIMEOUT_SECONDS: Final[float] = 180.0 -TASK_NAMES: Final[dict[str, str]] = { - "text-to-sql": "sql_generation", - "sql": "sql_generation", - "sql_generation": "sql_generation", - "classification": "classification", - "extraction": "extraction", -} NO_KEY: Final[str] = "set MICROTENSOR_API_KEY or pass api_key" LOCAL_PREVIEW: Final[str] = ( "the local backend ships with the Enterprise preview; install microtensor-engine[local] " @@ -34,6 +27,88 @@ class TaskUnavailable(EngineError): pass +def _normal(name: str) -> str: + return name.strip().lower().replace(" ", "-") + + +@dataclass(frozen=True, slots=True) +class Served: + task: str + model: str + title: str + aliases: tuple[str, ...] + live: bool + + @property + def names(self) -> tuple[str, ...]: + raw = [ + self.task, + self.task.replace("_", "-"), + self.model, + self.model.removeprefix("mt/"), + *self.aliases, + ] + return tuple(dict.fromkeys(_normal(n) for n in raw if n)) + + @property + def label(self) -> str: + return self.aliases[0] if self.aliases else self.task + + +@dataclass(frozen=True, slots=True) +class Catalogue: + served: tuple[Served, ...] + frontier: tuple[str, ...] + + @classmethod + def from_models(cls, models: Sequence[Mapping[str, Any]]) -> Catalogue: + entries = [dict(m) for m in models] + picked: dict[str, str] = {} + live: dict[str, bool] = {} + for entry in entries: + if entry.get("id") != AUTO: + continue + for item in entry.get("tasks") or []: + task = str(item.get("task", "")) + if task: + picked[task] = str(item.get("model", "")) + live[task] = bool(item.get("live", False)) + for entry in entries: + task = str(entry.get("task", "")) + if task and entry.get("id") != AUTO and task not in picked: + picked[task] = str(entry.get("id", "")) + live[task] = bool((entry.get("capacity") or {}).get("live", False)) + by_id = {str(e.get("id", "")): e for e in entries} + served = tuple( + Served( + task=task, + model=model, + title=str(by_id.get(model, {}).get("title", "")), + aliases=tuple(str(a) for a in by_id.get(model, {}).get("aliases") or []), + live=live.get(task, False), + ) + for task, model in sorted(picked.items()) + ) + frontier = tuple( + str(e["id"]) for e in entries if e.get("kind") == "frontier" and e.get("id") + ) + return cls(served=served, frontier=frontier) + + def resolve(self, name: str) -> Served | None: + wanted = _normal(name) + for entry in self.served: + if wanted in entry.names: + return entry + return None + + def describe(self) -> str: + parts = [] + for entry in self.served: + extra = f" ({', '.join(entry.aliases)})" if entry.aliases else "" + parts.append(f"{entry.task}{extra}") + return ", ".join(parts) or "none" + + @dataclass(frozen=True, slots=True) class Result: answer: str @@ -69,11 +144,6 @@ def _roll(self) -> None: self.spent = 0.0 -def task_id(name: str) -> str: - key = name.strip().lower().replace(" ", "-") - return TASK_NAMES.get(key, key.replace("-", "_")) - - def _messages(question: str, context: str) -> list[dict[str, str]]: found: list[dict[str, str]] = [] if context.strip(): @@ -117,6 +187,9 @@ def models(self) -> list[dict[str, Any]]: body = self._body(self._http.get("/models")) return [dict(m) for m in body.get("data", [])] + def catalogue(self) -> Catalogue: + return Catalogue.from_models(self.models()) + def chat( self, *, model: str, messages: Sequence[Mapping[str, str]], quality_floor: float ) -> dict[str, Any]: @@ -129,42 +202,62 @@ def chat( return self._body(self._http.post("/chat/completions", json=payload)) -def frontier_models( +def _hosted(api_key: str | None, base_url: str, transport: httpx.BaseTransport | None) -> Hosted: + return Hosted( + api_key or os.environ.get("MICROTENSOR_API_KEY", ""), base_url, transport=transport + ) + + +def catalogue( *, api_key: str | None = None, base_url: str = DEFAULT_BASE_URL, transport: httpx.BaseTransport | None = None, -) -> list[str]: - hosted = Hosted( - api_key or os.environ.get("MICROTENSOR_API_KEY", ""), base_url, transport=transport - ) +) -> Catalogue: + hosted = _hosted(api_key, base_url, transport) try: - return [ - str(m["id"]) for m in hosted.models() if m.get("kind") == "frontier" and m.get("id") - ] + return hosted.catalogue() finally: hosted.close() +def served_tasks( + *, + api_key: str | None = None, + base_url: str = DEFAULT_BASE_URL, + transport: httpx.BaseTransport | None = None, +) -> list[Served]: + return list(catalogue(api_key=api_key, base_url=base_url, transport=transport).served) + + +def frontier_models( + *, + api_key: str | None = None, + base_url: str = DEFAULT_BASE_URL, + transport: httpx.BaseTransport | None = None, +) -> list[str]: + return list(catalogue(api_key=api_key, base_url=base_url, transport=transport).frontier) + + class Engine: def __init__( self, *, hosted: Hosted, + catalogue: Catalogue, tasks: dict[str, str], quality_floor: float, escalation: bool, escalation_model: str | Mapping[str, Any], budget: Budget, - frontier: list[str], ) -> None: self._hosted = hosted + self._catalogue = catalogue self._tasks = tasks self.quality_floor = quality_floor self.escalation = escalation self._escalation_model = escalation_model self.budget = budget - self._frontier = frontier @staticmethod def frontier_models( @@ -172,6 +265,12 @@ def frontier_models( ) -> list[str]: return frontier_models(api_key=api_key, base_url=base_url) + @staticmethod + def served_tasks( + *, api_key: str | None = None, base_url: str = DEFAULT_BASE_URL + ) -> list[Served]: + return served_tasks(api_key=api_key, base_url=base_url) + @classmethod def load( cls, @@ -189,40 +288,38 @@ def load( ) -> Engine: if backend != "hosted": raise EngineError(LOCAL_PREVIEW) - hosted = Hosted( - api_key or os.environ.get("MICROTENSOR_API_KEY", ""), base_url, transport=transport - ) - served = { - str(m.get("task", "")): str(m.get("id", "")) - for m in hosted.models() - if m.get("task") and m.get("id") != AUTO - } + hosted = _hosted(api_key, base_url, transport) + found = hosted.catalogue() wanted: dict[str, str] = {} for name in tasks: - ident = task_id(name) - if ident not in served: + entry = found.resolve(name) + if entry is None: hosted.close() - known = ", ".join(sorted(served)) or "none" - raise TaskUnavailable(f"{name!r} is not served yet; served tasks: {known}") - wanted[ident] = served[ident] - frontier = [str(m["id"]) for m in hosted.models() if m.get("kind") == "frontier"] + raise TaskUnavailable( + f"{name!r} is not served yet; served tasks: {found.describe()}" + ) + wanted[entry.task] = entry.model budget = Budget( usd_per_day=float((escalation_budget or {}).get("usd_per_day", 0) or 0) or None ) return cls( hosted=hosted, + catalogue=found, tasks=wanted, quality_floor=quality_floor, escalation=escalation, escalation_model=escalation_model, budget=budget, - frontier=frontier, ) @property def tasks(self) -> dict[str, str]: return dict(self._tasks) + @property + def catalogue(self) -> Catalogue: + return self._catalogue + def close(self) -> None: self._hosted.close() @@ -235,20 +332,26 @@ def __exit__(self, *_: object) -> None: def _escalation_for(self, task: str) -> str | None: chosen: Any = self._escalation_model if isinstance(chosen, Mapping): - chosen = chosen.get(task, chosen.get(task.replace("_", "-"), "auto")) + entry = self._catalogue.resolve(task) + names = entry.names if entry else (_normal(task),) + chosen = next((chosen[k] for k in chosen if _normal(str(k)) in names), "auto") if chosen is False or chosen is None: return None + frontier = self._catalogue.frontier if chosen == "auto": - return self._frontier[0] if self._frontier else None + return frontier[0] if frontier else None name = str(chosen) - if name not in self._frontier: + if name not in frontier: raise EngineError(f"{name!r} is not a frontier model the network serves") return name def run(self, question: str, *, context: str = "", task: str | None = None) -> Result: - model = AUTO if task is None else self._tasks.get(task_id(task), "") - if task is not None and not model: - raise TaskUnavailable(f"{task!r} was not loaded") + model = AUTO + if task is not None: + entry = self._catalogue.resolve(task) + model = self._tasks.get(entry.task, "") if entry else "" + if not model: + raise TaskUnavailable(f"{task!r} was not loaded; loaded: {self._loaded()}") messages = _messages(question, context) found = self._hosted.chat(model=model, messages=messages, quality_floor=self.quality_floor) result = self._result(found) @@ -273,6 +376,9 @@ def run(self, question: str, *, context: str = "", task: str | None = None) -> R raw=again, ) + def _loaded(self) -> str: + return ", ".join(sorted(self._tasks)) or "none" + def _result(self, found: dict[str, Any]) -> Result: choices = found.get("choices") or [{}] message = choices[0].get("message") or {} diff --git a/tests/test_engine.py b/tests/test_engine.py index f5c9cd5..c00998c 100644 --- a/tests/test_engine.py +++ b/tests/test_engine.py @@ -5,7 +5,7 @@ import httpx import pytest -from microtensor import Engine, EngineError, TaskUnavailable, frontier_models +from microtensor import Engine, EngineError, TaskUnavailable, frontier_models, served_tasks MODELS = { "data": [ @@ -13,7 +13,20 @@ "id": "mt/auto", "tasks": [{"task": "sql_generation", "model": "mt/sql_gen_s1", "live": True}], }, - {"id": "mt/sql_gen_s1", "task": "sql_generation", "quality": 0.73}, + { + "id": "mt/sql_gen_s1", + "task": "sql_generation", + "title": "SQL generation, season 1", + "aliases": ["text-to-sql", "sql"], + "quality": 0.73, + "capacity": {"operators": 2, "live": True}, + }, + { + "id": "mt/classify_s1", + "task": "classification", + "title": "Classification, season 1", + "capacity": {"operators": 0, "live": False}, + }, ] } @@ -27,7 +40,7 @@ def gateway(request: httpx.Request) -> httpx.Response: 200, json={ "id": "cmpl_1", - "model": "mt/sql_gen_s1", + "model": body["model"], "choices": [{"message": {"content": "SELECT count(*) FROM orders"}}], "routing": { "task": "sql_generation", @@ -41,26 +54,42 @@ def gateway(request: httpx.Request) -> httpx.Response: ) -def test_hosted_engine_maps_tasks_and_reports_the_floor() -> None: +def test_task_names_come_from_the_live_catalogue() -> None: + transport = httpx.MockTransport(gateway) + served = served_tasks(api_key="mt_sk_x", base_url="http://g/v1", transport=transport) + assert [s.task for s in served] == ["classification", "sql_generation"] + assert served[1].aliases == ("text-to-sql", "sql") and served[1].live + assert not served[0].live and served[0].label == "classification" + + for name in ("text-to-sql", "SQL", "sql_generation", "sql-generation", "sql_gen_s1"): + engine = Engine.load([name], api_key="mt_sk_x", base_url="http://g/v1", transport=transport) + assert engine.tasks == {"sql_generation": "mt/sql_gen_s1"} + engine.close() + + with pytest.raises(TaskUnavailable, match="text-to-sql"): + Engine.load( + ["summarisation"], api_key="mt_sk_x", base_url="http://g/v1", transport=transport + ) + assert frontier_models(api_key="mt_sk_x", base_url="http://g/v1", transport=transport) == [] + + +def test_hosted_engine_reports_the_floor_and_pins_a_loaded_task() -> None: transport = httpx.MockTransport(gateway) engine = Engine.load( ["text-to-sql"], api_key="mt_sk_x", base_url="http://g/v1", transport=transport ) - assert engine.tasks == {"sql_generation": "mt/sql_gen_s1"} - sure = engine.run("How many orders in 2025?", context="CREATE TABLE orders(x)") assert sure.answer.startswith("SELECT") and sure.task == "sql_generation" assert sure.confidence == 0.96 and not sure.below_floor and not sure.escalated unsure = engine.run("How many orders?", context="CREATE TABLE orders(x)") assert unsure.below_floor and not unsure.escalated and unsure.escalated_to is None - engine.close() + pinned = engine.run("How many orders in 2025?", context="x", task="text-to-sql") + assert pinned.raw["model"] == "mt/sql_gen_s1" with pytest.raises(TaskUnavailable): - Engine.load( - ["classification"], api_key="mt_sk_x", base_url="http://g/v1", transport=transport - ) - assert frontier_models(api_key="mt_sk_x", base_url="http://g/v1", transport=transport) == [] + engine.run("Is this spam?", task="classification") + engine.close() def test_the_local_backend_is_named_as_a_preview_and_a_key_is_required() -> None: