nodes/kantan-kanto-ComfyUI-ScriptFlow
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Add MultiOutputScript (Jev) node with jev namespace for scripts

Browse Source
This commit is contained in:
kantan-kanto
2026-09-26 17:06:14 +09:00
parent a3a9fc5b63
commit af8f8cc25d
3 changed files with 211 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+36
View File
@@ -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`
__init__.py
+126 -33
View File
@@ -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",
}
jev_backend.py
+49
View File
@@ -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