From af8f8cc25db78212d10d9c4603a2e59bf157e4d2 Mon Sep 17 00:00:00 2001 From: kantan-kanto Date: Sat, 26 Sep 2026 17:06:14 +0900 Subject: [PATCH] Add MultiOutputScript (Jev) node with jev namespace for scripts --- README.md | 36 +++++++++++ __init__.py | 159 +++++++++++++++++++++++++++++++++++++++---------- jev_backend.py | 49 +++++++++++++++ 3 files changed, 211 insertions(+), 33 deletions(-) create mode 100644 jev_backend.py diff --git a/README.md b/README.md index 9382a82..7d53929 100644 --- a/README.md +++ b/README.md @@ -74,6 +74,42 @@ w, h = iv1, iv2 ov1, ov2 = (512, 384) if w >= h else (384, 512) ``` +### `MultiOutputScript (Jev)` +**Category:** `utils` + +Same inputs, outputs, and script rules as `MultiOutputScript`, plus a `jev` namespace that asks [Jev](https://typesafe.ai) (TypeSafe AI's decision model) typed questions about text. + +Setup: +```bash +pip install typesafe-sdk +export TYPESAFE_API_KEY="your-api-key" +``` + +Extra input: +- `max_jev_calls` (INT, default 8): maximum Jev API requests per run. Identical questions within a run are asked only once. + +Functions (`state` is a string, dict, or list; Jev accepts text only): +- `jev.yes(state, question[, threshold])` → `bool` (yes-probability ≥ `threshold`, default `0.5`) +- `jev.noul(state, question)` → yes-probability (`0.0`–`1.0`). Compare it with a threshold; using it directly as a condition raises an error. +- `jev.choice(state, question, options)` → the most likely option (`str`). `options` is a list of labels, or a dict of labels to descriptions. +- `jev.score(state, question, levels)` → expected level (`float`, `0` to `len(levels) - 1`) + +```python +# it1: a prompt text +if jev.yes(it1, "Does this prompt describe a night scene?", 0.7): + ov1 = 30 +else: + ov1 = 20 + +kind = jev.choice(it1, "Which orientation suits this prompt?", ["portrait", "landscape"]) +ov2, ov3 = (384, 512) if kind == "portrait" else (512, 384) +``` + +Notes: +- This node sends `state` and questions to the TypeSafe API over the internet. Do not use it with text you cannot share externally. +- Results are probabilistic; test thresholds on your own inputs. +- `MultiOutputScript` itself never makes network requests; `jev` is not available there. + ### `centi` **Category:** `utils` diff --git a/__init__.py b/__init__.py index 8d0f36b..9a46fc6 100644 --- a/__init__.py +++ b/__init__.py @@ -16,6 +16,8 @@ import math import random from typing import Any, Dict +from .jev_backend import JevBackend, TypeSafeClient + _DEFAULT_MAX_STEPS = 10000 _DEFAULT_MAX_CALL_DEPTH = 32 @@ -148,6 +150,10 @@ class _SafeNamespace: self.name = name +class _JevProbability(float): + pass + + class _SafeFunction: def __init__(self, name: str, args: ast.arguments, body: list[ast.stmt]): self.name = name @@ -182,8 +188,10 @@ class _SafeScriptInterpreter: variables: Dict[str, Any], max_steps: int = _DEFAULT_MAX_STEPS, max_call_depth: int = _DEFAULT_MAX_CALL_DEPTH, + jev: JevBackend | None = None, ): self.frames = [variables] + self.jev = jev self.max_steps = max_steps self.max_call_depth = max_call_depth self.steps = 0 @@ -491,8 +499,30 @@ class _SafeScriptInterpreter: return datetime.date.today() if namespace == "math" and name in _SAFE_MATH_FUNCTIONS: return self._call_math(name, args) + if namespace == "jev": + return self._call_jev(name, args) raise ValueError(f"Unsupported namespace call: {namespace}.{name}") + def _call_jev(self, name: str, args: list[Any]) -> Any: + if name == "yes": + if len(args) not in (2, 3): + raise ValueError("jev.yes() expects 2 or 3 arguments") + threshold = args[2] if len(args) == 3 else 0.5 + return self.jev.noul(args[0], args[1]) >= threshold + if name == "noul": + if len(args) != 2: + raise ValueError("jev.noul() expects 2 arguments") + return _JevProbability(self.jev.noul(*args)) + if name == "choice": + if len(args) != 3: + raise ValueError("jev.choice() expects 3 arguments") + return self.jev.choice(*args) + if name == "score": + if len(args) != 3: + raise ValueError("jev.score() expects 3 arguments") + return self.jev.score(*args) + raise ValueError(f"Unsupported namespace call: jev.{name}") + def _call_math(self, name: str, args: list[Any]) -> Any: if name == "ceil": return math.ceil(*args) @@ -680,6 +710,8 @@ class _SafeScriptInterpreter: return frame[name] if name in ("random", "datetime", "math"): return _SafeNamespace(name) + if name == "jev" and self.jev is not None: + return _SafeNamespace(name) return None def _get_name(self, name: str) -> Any: @@ -695,6 +727,8 @@ class _SafeScriptInterpreter: @staticmethod def _truthy(value: Any) -> bool: + if isinstance(value, _JevProbability): + raise ValueError("Compare jev.noul() with a threshold, or use jev.yes()") return bool(value) @@ -821,42 +855,99 @@ class MultiOutputScript: in_value_2: Any = None, in_value_3: Any = None, ): - it1 = _select_text_value("", in_text_1, "in_text_1") - it2 = _select_text_value("", in_text_2, "in_text_2") - it3 = _select_text_value("", in_text_3, "in_text_3") - iv1 = _select_numeric_value(0.0, in_value_1, "in_value_1") - iv2 = _select_numeric_value(0.0, in_value_2, "in_value_2") - iv3 = _select_numeric_value(0.0, in_value_3, "in_value_3") - - locals_dict: Dict[str, Any] = { - "it1": it1, - "it2": it2, - "it3": it3, - "iv1": iv1, - "iv2": iv2, - "iv3": iv3, - "ot1": None, - "ot2": None, - "ot3": None, - "ov1": None, - "ov2": None, - "ov3": None, - } - - locals_dict = _SafeScriptInterpreter(locals_dict).run(code) - - _validate_outputs(locals_dict) - - return ( - locals_dict.get("ot1"), - locals_dict.get("ot2"), - locals_dict.get("ot3"), - None if locals_dict.get("ov1") is None else int(locals_dict.get("ov1")), - None if locals_dict.get("ov2") is None else int(locals_dict.get("ov2")), - None if locals_dict.get("ov3") is None else int(locals_dict.get("ov3")), + return _run_script( + code, None, in_text_1, in_text_2, in_text_3, in_value_1, in_value_2, in_value_3 ) +class MultiOutputScriptJev(MultiOutputScript): + @classmethod + def INPUT_TYPES(cls): + types = super().INPUT_TYPES() + types["required"]["code"][1]["default"] = ( + "# Jev: jev.yes(state, question[, threshold]) -> bool\n" + "# jev.noul(state, question) -> probability\n" + "# jev.choice(state, question, options) -> str\n" + "# jev.score(state, question, levels) -> float\n" + "# Example:\n" + "kind = jev.choice(it1, \"What orientation suits this prompt?\", [\"portrait\", \"landscape\"])\n" + "ov1, ov2 = (384, 512) if kind == \"portrait\" else (512, 384)\n" + ) + types["required"]["max_jev_calls"] = ("INT", {"default": 8, "min": 1, "max": 64}) + return types + + def run( + self, + code: str, + max_jev_calls: int, + in_text_1: Any = None, + in_text_2: Any = None, + in_text_3: Any = None, + in_value_1: Any = None, + in_value_2: Any = None, + in_value_3: Any = None, + ): + if TypeSafeClient is None: + raise RuntimeError("typesafe-sdk is not installed. Run: pip install typesafe-sdk") + with TypeSafeClient() as client: + return _run_script( + code, + JevBackend(client, max_jev_calls), + in_text_1, + in_text_2, + in_text_3, + in_value_1, + in_value_2, + in_value_3, + ) + + +def _run_script( + code: str, + jev: JevBackend | None, + in_text_1: Any, + in_text_2: Any, + in_text_3: Any, + in_value_1: Any, + in_value_2: Any, + in_value_3: Any, +): + it1 = _select_text_value("", in_text_1, "in_text_1") + it2 = _select_text_value("", in_text_2, "in_text_2") + it3 = _select_text_value("", in_text_3, "in_text_3") + iv1 = _select_numeric_value(0.0, in_value_1, "in_value_1") + iv2 = _select_numeric_value(0.0, in_value_2, "in_value_2") + iv3 = _select_numeric_value(0.0, in_value_3, "in_value_3") + + locals_dict: Dict[str, Any] = { + "it1": it1, + "it2": it2, + "it3": it3, + "iv1": iv1, + "iv2": iv2, + "iv3": iv3, + "ot1": None, + "ot2": None, + "ot3": None, + "ov1": None, + "ov2": None, + "ov3": None, + } + + locals_dict = _SafeScriptInterpreter(locals_dict, jev=jev).run(code) + + _validate_outputs(locals_dict) + + return ( + locals_dict.get("ot1"), + locals_dict.get("ot2"), + locals_dict.get("ot3"), + None if locals_dict.get("ov1") is None else int(locals_dict.get("ov1")), + None if locals_dict.get("ov2") is None else int(locals_dict.get("ov2")), + None if locals_dict.get("ov3") is None else int(locals_dict.get("ov3")), + ) + + class Centi: @classmethod def INPUT_TYPES(cls): @@ -889,10 +980,12 @@ class Centi: NODE_CLASS_MAPPINGS = { "MultiOutputScript": MultiOutputScript, + "MultiOutputScriptJev": MultiOutputScriptJev, "centi": Centi, } NODE_DISPLAY_NAME_MAPPINGS = { "MultiOutputScript": "MultiOutputScript", + "MultiOutputScriptJev": "MultiOutputScript (Jev)", "centi": "centi", } diff --git a/jev_backend.py b/jev_backend.py new file mode 100644 index 0000000..acf0658 --- /dev/null +++ b/jev_backend.py @@ -0,0 +1,49 @@ +# ComfyUI-ScriptFlow +# Copyright (C) 2026 kantan-kanto (https://github.com/kantan-kanto) +# +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. + +from __future__ import annotations + +import json +from typing import Any + +try: + from typesafe_sdk import Choice, Noul, Score, TypeSafeClient +except ImportError: + TypeSafeClient = None + + +class JevBackend: + """Asks Jev one question per call. Scoped to a single script run.""" + + def __init__(self, client: Any, max_calls: int): + self.client = client + self.max_calls = max_calls + self.calls = 0 + self.cache: dict[str, Any] = {} + + def noul(self, state: Any, question: str) -> float: + return self._ask(state, Noul(instructions=question)).noul + + def choice(self, state: Any, question: str, options: list | dict) -> str: + if isinstance(options, list): + options = {str(option): None for option in options} + return self._ask(state, Choice(instructions=question, criteria=options)).choice + + def score(self, state: Any, question: str, levels: list) -> float: + return self._ask(state, Score(instructions=question, criteria=levels)).score + + def _ask(self, state: Any, question: Any) -> Any: + key = json.dumps([state, question.model_dump()], sort_keys=True, ensure_ascii=False) + if key in self.cache: + return self.cache[key] + if self.calls >= self.max_calls: + raise RuntimeError(f"Script exceeded max_jev_calls ({self.max_calls})") + self.calls += 1 + answer = self.client.system_one(state=state, questions={"q": question}).answers["q"] + self.cache[key] = answer + return answer