37 Commits
Author SHA1 Message Date
draconicdragon 9e91b2dac6 refactor: cleanup/remove code for testing video stuff 2026-01-04 18:46:41 +01:00
draconicdragon 2f1e76a03b refactor: improve i2i enhance prompt placeholder 2026-01-04 18:39:01 +01:00
draconicdragon 6ee38ed168 feat: img2img enhance/upscale node converted to v3 schema 2026-01-04 18:36:42 +01:00
draconicdragon bc05784401 feat: add seed and debug output and log 2026-01-04 18:20:15 +01:00
draconicdragon 36f8025ba0 refactor: set default of vp include venice sysprompt false 2026-01-04 17:43:58 +01:00
draconicdragon 0d2aa51773 feat: make gen text node complete (probably)
add basically all inputs to one node, venice parameters etc, use dynamic combo for hiding unsupported capabilities
fix encode tensor for vision
add text model spec get to utils/catalog
2026-01-04 17:40:03 +01:00
draconicdragon b58aa96687 feat: add text gen/llm node converted to v3 schema 2026-01-04 14:55:19 +01:00
draconicdragon 766b5a1933 chore: add todos 2026-01-04 13:46:26 +01:00
draconicdragon 3284c0ca62 refactor: cleanup 2026-01-03 00:47:56 +01:00
draconicdragon 942c29212e feat: TTS gen node with v3 schema
update backend of gen speech node to try torchaudio with sox and soundfile first and then imageio-ffmpeg as fallback
2026-01-03 00:44:19 +01:00
draconicdragon 1369879ea5 refactor: update tooltips 2026-01-02 20:49:20 +01:00
draconicdragon 3b9edb93b2 refactor: set default seed to 42 2026-01-02 20:49:04 +01:00
draconicdragon 58d6b0e286 fix: forgot batch size input in dynamiccombo 2026-01-02 20:43:09 +01:00
draconicdragon ed2232da5b refactor: reorder inputs and move some inputs out of dynamiccombo 2026-01-02 20:39:35 +01:00
draconicdragon 28985868ef feat: dynamic combo for image gen node for dynamic input constraints 2026-01-02 20:18:19 +01:00
draconicdragon 85e5dd28c5 chore: change comment 2025-12-31 21:19:52 +01:00
draconicdragon ddbbf0fa4f refactor: exclude nano banana pro from gen image node for now
it does work technically but 1024x1024 only, it requires different params, with dynamic combo widgets can be changed accordingly but thats work for future
adjust prompt field tooltip
2025-12-31 21:19:29 +01:00
draconicdragon 6149497c09 feat: per image model constraints for prompt length and multiple of
added image model by id like video node uses to backend for future image node change
2025-12-31 21:02:37 +01:00
draconicdragon 0abaab027a chore: remove duplicate file 2025-12-31 19:55:32 +01:00
draconicdragon e793549006 chore: cleanup and add more tooltips
also change default style preset to none_available instead of just none
2025-12-31 19:50:14 +01:00
draconicdragon e1570a2dec feat: more optional widgets for video node
add more optional widgets for dynamiccombo, durations, aspect ratios, resolutions, audio, and change first or default func to constraint_values
also move some code
2025-12-31 19:26:21 +01:00
draconicdragon 8c5ccaafa1 feat: gen image node in v3 schema
incomplete because some image models require dynamiccombo, eg nano banana
2025-12-31 19:25:44 +01:00
draconicdragon 5ecfcba9fb style: add * to ensure prompt length function as param divider 2025-12-14 10:47:05 +01:00
draconicdragon fa3a05b093 feat: improve text2video node with DynamicCombo widget 2025-12-14 10:25:55 +01:00
draconicdragon dc55bd8964 refactor: convert text to video node to v3 schema 2025-12-14 05:41:26 +01:00
draconicdragon c2121d2515 temp: move/delete most node files to make v3 migration less painful
only gen video from text node remains
2025-12-14 03:56:30 +01:00
draconicdragon 2f934d3d9e chore: remove gen video from image node 2025-12-14 03:55:14 +01:00
draconicdragon 7e1391535c refactor: more work on video node
not finished at all but will generate video now, and outputs newer "VIDEO" datatype output
ProgressBar should show more correct based on polling response
adjusted polling interval to be a bit slower
add (temporary) ability to bypass api api call for video and load one from testing_video folder instead for testing video output usage downstream, if any
the debug env var will trigger any api generated video to be saved to the same testing_video folder instaed of just being saved as temp file
2025-12-13 06:30:28 +01:00
draconicdragon c54bdbbb97 feat: more (debug) logging for venice_client.py
and move setting some env vars to globals
2025-12-13 06:26:42 +01:00
draconicdragon e4caea1af6 fix: .gitignore entry 2025-12-13 06:23:48 +01:00
draconicdragon 19c9a3f212 chore: update .gitignore 2025-12-13 06:23:12 +01:00
draconicdragon 2ef0200c4c chore: increase default cache time to live to 15 minutes 2025-12-13 06:23:04 +01:00
draconicdragon b9b43e7ef5 chore: add README 2025-12-13 06:19:40 +01:00
draconicdragon b5bc49753e chore: add TODO.md 2025-12-13 06:18:00 +01:00
draconicdragon 67d4747e9c feat: add basic primitive video node, untested but might work 2025-12-12 08:15:17 +01:00
draconicdragon 583b7afa78 refactor: comment out VENICE_CLIENT_DRY_RUN 2025-12-12 05:50:13 +01:00
draconicdragon e6266cc049 refactor with new API client and config management
my shitty description:
refactor a bunch of stuff
better model loading, removed the stupid js workaround to populating the COMBO fields
removed model loading from js and model fetching from routes/pyserver
unified model/character/styles lists internally as "catalog"

changed textbox in settings to by type: password

replaced primitive requests with client session for better/cleaner/modular request managing, added some better error logging and responses
added dry run env var VENICE_CLIENT_DRY_RUN can be set to 'true' or '1' (it just returns dummy response and doesnt hit the veniceai api)

make __init__.py module/node loading better by loading any .py file with NODE_*_MAPPING in nodes folder instead of hardcoding names

remove old redundant files

AI summarized description with some editing:
- Added `venice_client.py` for handling API requests to VeniceAI.
- Introduced `venice_config.py` for managing API key configuration.
- Created `venice_catalog.py` to manage and cache models, styles, and characters.
- Removed outdated key management logic from `pyserver/get_key_from_jssetting.py`.
- Consolidated character and model update logic into the new catalog management system.
- Move utility functions to `nodes/utils.py` containing validation stuff mostly.
- Removed legacy update scripts for characters, models, and styles.
- Added error handling and logging for API interactions.
- remove obsolete workflow JSON file.
2025-12-12 04:02:43 +01:00
38 changed files with 2398 additions and 2319 deletions
+2
View File
@@ -3,6 +3,8 @@ test.py
js/test.js
veniceai_config.json
data/*
testing_video/*
nodes/test/*
# Ignore Mac system files
.DS_Store
+3 -109
View File
@@ -1,110 +1,4 @@
# ComfyUI Venice.AI API Custom Nodes
### Hi, hello
An unofficial custom node implementation for ComfyUI that integrates with venice.ai's Generative AI services such as Image, Text and TTS Generation models (as well as Upscale and Enhance). This project is adapted from [ComfyUI-FLUX-TOGETHER-API](https://github.com/BZcreativ/ComfyUI-FLUX-TOGETHER-API) to work with the venice.ai API.
**Note**: TTS is in Beta per Venice (as of 15th June 2025). The node for it "Generate Speech (Venice)" is set to BETA/Experimental too. To be able to find it with node search you have to enable "Show experimental nodes in search" setting in ComfyUI.
Disclaimer: I originally made this on a whim because someone wanted something similar to Together.AI custom nodes but have them use venice instead. I'm also not affiliated with Venice.AI. Idk is this good enough for a disclaimer or something
## Nodes: (Text gen node is missing but will be updated soon)
![all nodes showcased](./gh_assets/nodes_showcase.png)
Image and Text models (as of 15th June 2025 | for Speech gen only tts-kokoro is available)
For updated text generation experience please use `Generate Text Advanced BETA (Venice)` node and additionally `Textgen Parameters (Venice) for extra venice.ai specific parameters to pass onto the generation process.
![Image gen models](./gh_assets/image_gen_models_15june.png) ![Text gen models](./gh_assets/text_gen_models15june.png)
~~# todo: inpainting~~ Deprecated by venice, new thing is coming for it at some point
todo: actual log, maybe separate logging file for less clutter from comfyui stuff, maybe, maybe... eeeee
todo: add settings to set default model by user
todo: less convoluted approach to downloading and loading model/character/styles lists
~~todo: LLM characters list~~ done, but its in beta so subject to big changes, use with `Textgen Parameters (Venice)` node
todo: use variants api (currently in beta) | idk what happened to this, probably gone
Todo: chat history/memory/context for LLM | got a vague idea but thats pretty much it
todo
- Some Validation on queue for API limits like prompt max length or width/height
- these are different for different models so this isnt planned to be implemented unless the api exposes those limits somehow
- i'd have an error thrown before it gets sent to api about some value being too high or too low (though api should send error back for now either way)
- maybe the api already exposes those limits, the steps limit is afaik at least, need to look into it
### Below ReadMe text is only slightly altered from original Flux Together API readme, it was not really reworked or anything so its likely not correct or up to date
### Installation - these instructions are a mess
0. Before proceeding, check if you can find these nodes through ComfyUI-Manager interface rather than following the instructions below.
<details><summary>Expand me to see harder instructions</summary>
1. Clone this repository into your ComfyUI custom_nodes directory:
```bash
cd ComfyUI/custom_nodes
git clone https://github.com/DraconicDragon/ComfyUI-Venice-API.git
```
2. Install the required dependencies: (this might be done by comfyui automatically on restart already?)
```bash
pip install -r requirements.txt
```
OR From the Comfyui Folder (this one is usually preferred if you have portable edition)
```bash
./python_embeded\python.exe -m pip install -r ComfyUI\custom_nodes\ComfyUI-Venice-API\requirements.txt
```
</details>
### Configuration
1. Get your API key from [venice.ai](https://venice.ai)
2. Add your API key in ComfyUI settings > VeniceAI
### Parameters for Generate Image (Venice) node
| Parameter | Type | Range | Default | Description |
|-----------------|---------|------------|---------|-----------------------------------------|
| prompt | string | 1-1500 | "A flying cat made of lettuce" | Main generation prompt |
| negative_prompt | string | 0-1500 | "" | Elements to avoid |
| width | integer | 0-1280? | 1024 | Image width |
| height | integer | 0-1280? | 1024 | Image height |
| batch_size | integer | 1-4 | 1 | Number of Images to gen in a single run |
| steps | integer | 1-30 or 50 | 20 | Number of generation steps |
| cfg/guidance | float | 0-20.0 | 3.0 | Guidance scale |
| style_preset | string | N/A | none | The Style preset to apply |
| hide_watermark | boolean | N/A | true | Whether to hide watermark (NSFW = false)|
| safe_mode | boolean | N/A | false | Whether to blur NSFW images |
| seed | integer | -999999999 to 999999999 | -1 | Generation seed |
### License
MIT License - see [LICENSE](LICENSE) file for details.
### Credits
- This project is adapted from [ComfyUI-FLUX-TOGETHER-API](https://github.com/BZcreativ/ComfyUI-FLUX-TOGETHER-API)
- venice.ai for providing the generative AI services and API
- [ComfyUI-FLUX-TOGETHER-API](https://github.com/BZcreativ/ComfyUI-FLUX-TOGETHER-API) for their work
- ComfyUI team for the amazing framework
### Author
Created by [BZcreativ](https://github.com/BZcreativ)
venice.ai rewrite by [DraconicDragon](https://github.com/DraconicDragon)
### Contributing
Contributions are welcome! Feel free to submit a Pull Request.
### Example
todo
For detailed usage instructions, see [USAGE.md](USAGE.md) (not reworked)
WIP dev branch changing things woohee
see [TODO](TODO.md) for some sparse todo stuff
+7
View File
@@ -0,0 +1,7 @@
# TODO
- [ ] Support quote video gen button label for price estimates
- [ ] Convert to node schema v3
- [ ] Use DynamicCombo for all applicable nodes
- [ ] Combine DynamicCombo with other limits such as prompt length
- [ ] Check for TODOs in code files
-128
View File
@@ -1,128 +0,0 @@
# dont follow stuff thats in here i dont think its correct
# ComfyUI venice.ai API Node Usage Guide
### Disclaimer: I just made this on a whim because someone wanted something similar to Together.AI custom nodes but use venice instead and I don't have access to any API keys for any of the mentioned services.
## Setup
1. Ensure you have a venice.ai account and API key
2. Configure your API key in `config.ini`
3. Install all required dependencies (should be done automatically by comfy?)
## Node Configuration
### Input Parameters
#### Required Parameters:
- **Prompt** (String)
- Your main generation prompt
- Be specific and detailed for best results if using Flux
- **Negative Prompt** (String)
- Elements you want to avoid in the generation
- Leave empty if not needed | Will be ignored if flux-dev or flux-dev-uncensored is selected as model
- **Steps** (Integer)
- Range: 1-30
- Default: 20
- **Width** (Integer)
- Range: 512-2048
- Default: 1024
- Must be a multiple of 32
- Common values: 512, 768, 1024 (1MP), 1440 (Flux | 2MP)
- **Height** (Integer)
- Range: 512-2048
- Default: 1024
- Must be a multiple of 32
- Common values: 512, 768, 1024 (1MP), 1440 (Flux | 2MP)
- **Seed** (Integer)
- Range: 0 to max 64-bit integer
- Default: -1 | Random
- Reuse a seed with same prompt to reproduce an image
- **CFG (Guidance Scale)** (Float)
- Range: 0.1-15.0
- Default: 3.5
- Recommended range: 5.0-10.0 (SDXL) | ~3.5 (Flux)
### Output
The node outputs a single image tensor compatible with other ComfyUI nodes.
## Best Practices
1. **Prompt Engineering**
- Be specific and detailed in your prompts
- Use descriptive adjectives
- Include style references when needed
2. **Performance**
- Start with lower step counts (20-30) for testing
- Increase steps for final generations
- Use reasonable image dimensions (1024x1024 is standard)
3. **Error Handling/Troubleshooting**
- Check console for error messages
- Verify API key is correctly configured
- Ensure parameters are within valid ranges
## Common Workflows
### Basic Image Generation
1. Add Together API Node to workspace
2. Connect to a Load Image node
3. Configure prompt and basic parameters
4. Execute workflow
### Advanced Usage
1. Combine with other ComfyUI nodes
2. Use seed control for consistent results
3. Experiment with guidance scale for style control
## Troubleshooting
### Common Issues
1. **API Key Errors**
- Verify key in config.ini
- Check API key validity
- Ensure proper formatting
2. **Generation Errors**
- Verify parameter ranges
- Check prompt length
- Monitor API rate limits
3. **Image Quality Issues**
- Adjust step count
- Modify guidance scale
- Refine prompt
## Examples
### Basic Prompt Example
```
A beautiful landscape with mountains and lakes, cinematic lighting, high detail
```
### Advanced Prompt Example
```
A stunning mountain landscape at sunset, volumetric lighting,
golden hour, ultra detailed, professional photography,
8k resolution, artistic composition
```
### Negative Prompt Example
```
blur, haze, low quality, distortion, bad composition,
oversaturated, unrealistic lighting
```
## Support
For issues and feature requests, please use the GitHub issue tracker.
+22 -30
View File
@@ -1,39 +1,31 @@
import importlib
import logging
import os
from .pyserver import (
get_key_from_jssetting, # noqa: F401
update_characters, # noqa: F401
update_models, # noqa: F401
update_styles, # noqa: F401
)
from comfy_api.latest import ComfyExtension, io
node_list = [
# "things_n_stuff_node",
"gen_image_node",
# "gen_image_inpaint_node",
"gen_text_node",
"gen_text_advanced_node",
"gen_text_venice_params_node",
"i2i_enhance_upscale_node",
"gen_speech_node",
"util_nodes",
]
from .nodes.gen_image_node import GenerateImage
from .nodes.gen_speech_node import GenerateSpeech
from .nodes.gen_text_node import GenerateTextAdvanced
from .nodes.gen_video_from_text_node import GenerateVideoFromText
from .nodes.i2i_enhance_upscale import I2IEnhanceUpscale
from .nodes.test_node import DCTestNode
NODE_CLASS_MAPPINGS = {}
NODE_DISPLAY_NAME_MAPPINGS = {}
for module_name in node_list:
try:
imported_module = importlib.import_module(f".nodes.{module_name}", __name__)
NODE_CLASS_MAPPINGS.update(imported_module.NODE_CLASS_MAPPINGS)
NODE_DISPLAY_NAME_MAPPINGS.update(imported_module.NODE_DISPLAY_NAME_MAPPINGS)
except ImportError as e:
logging.warning(f"Could not import module '{module_name}': {e}")
class VeniceExtension(ComfyExtension):
async def get_node_list(self) -> list[type[io.ComfyNode]]:
return [
DCTestNode,
GenerateImage,
GenerateSpeech,
GenerateTextAdvanced,
GenerateVideoFromText,
I2IEnhanceUpscale,
]
async def comfy_entrypoint() -> VeniceExtension:
return VeniceExtension()
WEB_DIRECTORY = os.path.join(os.path.dirname(__file__), "js")
__all__ = ["NODE_CLASS_MAPPINGS", "NODE_DISPLAY_NAME_MAPPINGS", "WEB_DIRECTORY"]
__all__ = ["WEB_DIRECTORY"]
Binary file not shown.

Before

Width:  |  Height:  |  Size: 21 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 438 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 12 KiB

+12 -2
View File
@@ -1,3 +1,5 @@
import os
API_ENDPOINTS = {
"list_models": "/models", # response type is list of strings
"list_styles": "/image/styles", #
@@ -6,9 +8,17 @@ API_ENDPOINTS = {
"upscale_image": "/image/upscale", # NOTE: apparently doesnt even work yet? idk; response type is image/png file, content type is multipart/form-data
"text_generate": "/chat/completions", # has much info, text response is in choices: content, can have multiple choices apparently but dosnt seem to be utilized
"speech_generate": "/audio/speech", # type: file (audio/aac; audio/mpeg; audio/wav.. etc)
"video_queue": "/video/queue", #
"video_quote": "/video/quote", # price estimate, takes same payload as video_queue
"video_retrieve": "/video/retrieve", # get video file by job id
"list_api_keys": "/api_keys",
}
VENICEAI_BASE_URL = "https://api.venice.ai/api/v1"
# unused right now
headers = {"User-Agent": "ComfyUI-Venice-API/1.0 (by draconicdragon on github)"}
# request hygiene
USER_AGENT = "ComfyUI-Venice-API/1.0 (by draconicdragon on github)"
os.environ["VENICE_CLIENT_DRY_RUN"] = "0"
os.environ["VENICE_CLIENT_DEBUG"] = "1"
-125
View File
@@ -1,125 +0,0 @@
import { api } from "../../scripts/api.js";
import { app } from "../../scripts/app.js";
// Helper function to fetch data and assign to widget
async function fetchAndAssignWidget(widget, url, dataKey, logMsg, errorMsg) {
if (!widget) return;
try {
console.log(`(VeniceAI.NodeSpawn) ${logMsg}`);
const response = await api.fetchApi(url);
if (!response.ok) {
throw new Error(`HTTP error: ${response.status} ${response.statusText}`);
}
const rawText = await response.text();
let data;
try {
data = JSON.parse(rawText);
} catch (jsonError) {
throw new Error(`Failed to parse JSON: ${jsonError.message}. Raw response: ${rawText}`);
}
widget.options.values = data[dataKey];
if (widget.onChange) {
widget.onChange();
}
this.setDirtyCanvas(true);
} catch (error) {
console.error(`(VeniceAI.NodeSpawn) ${errorMsg}:`, error);
alert(`(VeniceAI.NodeSpawn) ${errorMsg}:\n${error}`);
}
}
app.registerExtension({
name: "VeniceAI.NodeSpawn",
async beforeRegisterNodeDef(nodeType, nodeData) {
if (nodeData.name === "GenerateImage_VENICE" || nodeData.name === "InpaintImage_VENICE") {
const originalOnNodeCreated = nodeType.prototype.onNodeCreated;
nodeType.prototype.onNodeCreated = async function () {
if (originalOnNodeCreated) {
originalOnNodeCreated.apply(this);
}
await fetchAndAssignWidget.call(
this,
this.widgets.find(w => w.name === "model"),
"/veniceai/get_models_list",
"image_models",
"Trying to fetch image models...",
"Failed to fetch image models"
);
await fetchAndAssignWidget.call(
this,
this.widgets.find(w => w.name === "style_preset"),
"/veniceai/get_styles_list",
"data",
"Trying to fetch styles...",
"Failed to fetch styles"
);
};
}
if (nodeData.name === "GenerateText_VENICE" || nodeData.name === "GenerateTextAdvanced_VENICE") {
const originalOnNodeCreated = nodeType.prototype.onNodeCreated;
nodeType.prototype.onNodeCreated = async function () {
if (originalOnNodeCreated) {
originalOnNodeCreated.apply(this);
}
await fetchAndAssignWidget.call(
this,
this.widgets.find(w => w.name === "model"),
"/veniceai/get_models_list",
"text_models",
"Trying to fetch text models...",
"Failed to fetch text models"
);
};
}
if (nodeData.name === "GenerateTextVeniceParameters_VENICE") {
const originalOnNodeCreated = nodeType.prototype.onNodeCreated;
nodeType.prototype.onNodeCreated = async function () {
if (originalOnNodeCreated) {
originalOnNodeCreated.apply(this);
}
await fetchAndAssignWidget.call(
this,
this.widgets.find(w => w.name === "character_slug"),
"/veniceai/get_characters_list",
"characters",
"Trying to fetch character slugs...",
"Failed to fetch character slugs"
);
};
}
if (nodeData.name === "GenerateSpeech_VENICE") {
const originalOnNodeCreated = nodeType.prototype.onNodeCreated;
nodeType.prototype.onNodeCreated = async function () {
if (originalOnNodeCreated) {
originalOnNodeCreated.apply(this);
}
await fetchAndAssignWidget.call(
this,
this.widgets.find(w => w.name === "model"),
"/veniceai/get_models_list",
"tts_models",
"Trying to fetch tts models...",
"Failed to fetch tts models"
);
await fetchAndAssignWidget.call(
this,
this.widgets.find(w => w.name === "voice"),
"/veniceai/get_models_list",
"tts_voices",
"Trying to fetch tts voices...",
"Failed to fetch tts voices"
);
}
}
}
});
+5 -2
View File
@@ -5,11 +5,15 @@ app.registerExtension({
name: "VeniceAI.Settings",
settings: [
{
category: ["VeniceAI", "API Key", "VeniceAI API Key"],
id: "VeniceAI.apikey",
name: "VeniceAI API Key",
type: "text",
defaultValue: "your_venice_api_key_here",
tooltip: "Enter your VeniceAI API Bearer Token Key here",
tooltip: "Enter your VeniceAI API Key/Bearer Token here",
attrs: {
type: "password",
},
onChange: async (newVal) => {
api.fetchApi("/veniceai/save_apikey", {
method: "POST",
@@ -19,4 +23,3 @@ app.registerExtension({
},
],
});
+5 -43
View File
@@ -5,51 +5,13 @@ app.registerExtension({
name: "VeniceAI.Startup",
async setup() {
// Load saved value from server on startup
console.log("(VeniceAI.Startup) Fetching VeniceAI API key from config file...");
const api_key_response = await api.fetchApi("/veniceai/get_apikey");
const savedKey = await api_key_response.json();
// update the settings UI
app.extensionManager.setting.set("VeniceAI.apikey", savedKey.apikey);
try {
// update the model list
console.log("(VeniceAI.Startup) Updating model list...");
//alert("fetching model list")
const response = await api.fetchApi("/veniceai/update_models_list");
const data = await response.json();
//alert(`response status ${JSON.stringify(data)}`);
if (data.error) {
alert(`${data.message}`);
console.log(`(VeniceAI.Startup) ${data.message}`);
}
else{
// update the style list if not model list error
console.log("(VeniceAI.Startup) Updating styles list...");
//alert("fetching styles list")
const response_s = await api.fetchApi("/veniceai/update_styles_list");
//alert(`response status ${await response_s.text()}`);
const data_s = await response_s.json();
if (data_s.error) {
alert(`${data_s.message}`);
console.log(`(VeniceAI.Startup) ${data_s.message}`);
}
// update the characters list
console.log("(VeniceAI.Startup) Updating characters list...");
const response_c = await api.fetchApi("/veniceai/update_characters_list");
const data_c = await response_c.json();
if (data_c.error) {
alert(`${data_c.message}`);
console.log(`(VeniceAI.Startup) ${data_c.message}`);
}
}
console.log("(VeniceAI.Startup) Fetching VeniceAI API key from config file...");
const api_key_response = await api.fetchApi("/veniceai/get_apikey");
const savedKey = await api_key_response.json();
app.extensionManager.setting.set("VeniceAI.apikey", savedKey.apikey);
} catch (error) {
// Handle any unexpected errors
alert(`(VeniceAI.Startup) Unexpected Error: ${error.message}`);
console.error("(VeniceAI.Startup) Unexpected Error:", error);
console.error("(VeniceAI.Startup) Failed to load Venice API key", error);
}
},
});
+89
View File
@@ -0,0 +1,89 @@
from __future__ import annotations
import logging
from typing import Any, Dict, Iterable, Sequence, Tuple
from ..venice_catalog import get_characters, get_models, get_styles
log = logging.getLogger(__name__)
def _safe_values(
loader,
key: str,
fallback: Sequence[str] = ("This list is unavailable; check logs for details?",),
) -> Tuple[str, ...]:
"""
Safely retrieves a sequence of string values from a loader payload keyed by `key`.
Attempts to call the provided `loader` callable to obtain a payload, logging failures
and falling back to the provided default values. Extracts and normalizes the value
associated with `key`, ensuring it is returned as a tuple of strings. If the extracted
value is missing, empty, or otherwise falsy, the fallback values are returned instead.
"""
try:
payload = loader()
except Exception as exc:
log.debug("Failed to load %s catalog: %s", key, exc)
return tuple(fallback)
raw = payload.get(key)
if raw is None:
return tuple(fallback)
if isinstance(raw, Iterable) and not isinstance(raw, (str, bytes)):
values = tuple(str(item) for item in raw if item)
else:
values = tuple(str(raw)) if raw else ()
return values or tuple(fallback)
def image_model_choices() -> Tuple[str, ...]:
return _safe_values(lambda: get_models(), "image_models")
def text2video_model_choices() -> Tuple[str, ...]:
return _safe_values(lambda: get_models(), "text2video_models")
def image2video_model_choices() -> Tuple[str, ...]:
return _safe_values(lambda: get_models(), "image2video_models")
def video_model_specs() -> Dict[str, Dict[str, Any]]:
"""Return a by-id mapping of video model specs with constraints for DynamicCombo use."""
models = get_models()
return models.get("video_models_by_id", {})
def image_model_specs() -> Dict[str, Dict[str, Any]]:
"""Return a by-id mapping of image model specs and constraints for UI validation."""
models = get_models()
return models.get("image_models_by_id", {})
def text_model_specs() -> Dict[str, Dict[str, Any]]:
"""Return a by-id mapping of text model specs and constraints for UI validation."""
models = get_models()
return models.get("text_models_by_id", {})
def text_model_choices() -> Tuple[str, ...]:
return _safe_values(lambda: get_models(), "text_models")
def tts_model_choices() -> Tuple[str, ...]:
return _safe_values(lambda: get_models(), "tts_models")
def tts_voice_choices() -> Tuple[str, ...]:
return _safe_values(lambda: get_models(), "tts_voices")
def style_choices() -> Tuple[str, ...]:
return _safe_values(lambda: get_styles(), "data")
def character_choices() -> Tuple[str, ...]:
return _safe_values(lambda: get_characters(), "characters")
+1 -13
View File
@@ -2,9 +2,7 @@ import base64
import io
import logging
import numpy as np
import torch # type: ignore
import torchvision.transforms as transforms # type: ignore
import torchvision.transforms as transforms
from PIL import Image
@@ -33,13 +31,3 @@ class GenerateImageBase:
except Exception as e:
raise Exception(f"Error processing image result: {str(e)}") from e
def create_blank_image(self):
blank_img = Image.new("RGB", (64, 64), color="black")
img_array = np.array(blank_img).astype(np.float32) / 255.0
img_tensor = torch.from_numpy(img_array)[None,]
return (img_tensor,)
def check_multiple_of_32(self, width, height):
if width % 32 != 0 or height % 32 != 0:
raise ValueError(f"Width {width} and height {height} must be multiples of 32.")
View File
+288 -132
View File
@@ -1,168 +1,324 @@
import logging
import os
import re
from typing import Any, Dict
import requests
import torch # type: ignore
import torch
from ..globals import API_ENDPOINTS, VENICEAI_BASE_URL
from comfy_api.latest import io
from ..globals import API_ENDPOINTS
from ..nodes.catalog_utils import image_model_specs, style_choices
from ..nodes.gen_image_base import GenerateImageBase
from ..nodes.utils import ensure_multiple_of, ensure_prompt_length
from ..venice_client import client
LOG = logging.getLogger(__name__)
class GenerateImage(GenerateImageBase):
class GenerateImage(io.ComfyNode):
_processor = GenerateImageBase()
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"model": (
"COMBO",
{
"default": "flux-dev",
"tooltip": "Model to use for image generation, if this just says flux-dev or C O M B O then something failed oopsie.",
},
),
"prompt": ("STRING", {"default": "A flying cat made of lettuce", "multiline": True}),
"neg_prompt": (
"STRING",
{
"placeholder": "Negative Prompt. (Ignored for Flux based models.)\nBad composition, rating_explicit, Text, signature, lowres, faded image, out of focus, cropped, out of frame, vacant scene, bad quality, worst quality,",
"multiline": True,
"tooltip": "Negative prompt. This is ignored when using flux-dev or flux-dev-uncensored",
},
),
"width": (
"INT",
{
"default": 1024,
"min": 0,
"max": 2048, # limit is 1280 but i dont want to restrict this in case of future updates, https://docs.venice.ai/api-reference/endpoint/image/generate#body-height
"step": 16,
"tooltip": "Must be a multiple of 32. Maximum allowed by venice.ai at time of writing is 1280",
},
),
"height": (
"INT",
{
"default": 1024,
"min": 0,
"max": 2048,
"step": 16,
"tooltip": "Must be a multiple of 32. Maximum allowed by venice.ai at time of writing is 1280",
},
),
"batch_size": ("INT", {"default": 1, "min": 1, "max": 4}),
"steps": (
"INT",
{
"default": 20,
"min": 1,
"max": 50,
"tooltip": (
"Number of inference steps. The following models have reduced max steps from "
"the global max: venice-sd35: 30 max steps, hidream: 50 max steps, fluently-xl: 50 max steps, "
"flux-dev: 30 max steps, flux-dev-uncensored: 30 max steps, getphat-flux: 50 max steps, "
"lustify-sdxl: 50 max steps, pony-realism: 50 max steps, stable-diffusion-3.5: 30 max steps, "
"juggernaut-xi: 50 max steps."
),
},
),
"guidance": ("FLOAT", {"default": 3.0, "min": 0.0, "max": 20.0, "step": 0.05}),
# "lora_strength": ("INT", {"default": 50, "min": 0, "max": 100}), # check docs idk how to work this yet
"style_preset": ("COMBO", {"default": "none"}),
"hide_watermark": (
"BOOLEAN",
{
"default": True,
"tooltip": "Whether to hide the Venice watermark. Venice may ignore this parameter for certain generated content (mainl NSFW seems like).",
},
),
"safe_mode": (
"BOOLEAN",
{
"default": False,
"tooltip": "Whether to use safe mode. If enabled, this will blur images that are classified as having adult content.",
},
),
# "format": (["png", "jpeg", "webp"], {"default": "png",}),
},
"optional": {
"seed": ("INT", {"default": -1, "min": -0x3B9AC9FF, "max": 0x3B9AC9FF})
}, # 0xffffffffffffffff is 64 bit integer limit, current hex is 999999999, venice max
}
def _image_specs(cls, require: bool = False) -> Dict[str, Dict[str, Any]]:
specs = image_model_specs() or {}
if require and not specs:
raise ValueError(
"No Venice image model specs available; refresh the catalog in VeniceAI settings and retry."
)
return specs
def generate(
self,
@staticmethod
def _style_options() -> tuple[str, ...]:
options = list(style_choices())
if not options:
return ("none_available",)
return tuple(options)
@staticmethod
def _option_input_id(model_id: str, field: str) -> str:
sanitized = "".join(ch if ch.isalnum() or ch in {"_", "-"} else "_" for ch in model_id)
return f"{sanitized}__{field}"
@classmethod
def _get_option_value(cls, model_payload: Dict[str, Any], model_id: str, field: str) -> Any:
candidates = (
cls._option_input_id(model_id, field),
field,
f"{field}__{model_id}",
f"{model_id}__{field}",
)
for key in candidates:
if key in model_payload:
return model_payload.get(key)
return None
@classmethod
def _resolve_option_value(cls, model_payload: Dict[str, Any], model_id: str, field: str, default: Any) -> Any:
value = cls._get_option_value(model_payload, model_id, field)
return default if value is None else value
@staticmethod
def _coerce_positive_int(value: Any) -> int | None:
try:
candidate = int(value)
except (TypeError, ValueError):
return None
return candidate if candidate > 0 else None
@staticmethod
def _width_height_divisor(constraints: Dict[str, Any]) -> int:
divisor = constraints.get("widthHeightDivisor")
if isinstance(divisor, int) and divisor > 0:
return divisor
return 16
@classmethod
def _steps_limits(cls, constraints: Dict[str, Any]) -> tuple[int, int]:
steps = constraints.get("steps") or {}
default = cls._coerce_positive_int(steps.get("default"))
max_value = cls._coerce_positive_int(steps.get("max"))
default = default if default is not None else 20
max_steps = max_value if max_value is not None else 50
if default > max_steps:
max_steps = default
return default, max_steps
@classmethod
def _model_option_inputs(
cls,
model_id: str,
width_divisor: int,
steps_default: int,
steps_max: int,
) -> list[io.Input]:
return [
io.Int.Input(
cls._option_input_id(model_id, "width"),
display_name="width",
default=1024,
min=0,
max=2048,
step=width_divisor,
tooltip="Target width for the generated image; stepping is tied to the model's `widthHeightDivisor`. Defaults to 16",
),
io.Int.Input(
cls._option_input_id(model_id, "height"),
display_name="height",
default=1024,
min=0,
max=2048,
step=width_divisor,
tooltip="Target height for the generated image; stepping is tied to the model's `widthHeightDivisor`. Defaults to 16",
),
io.Int.Input(
cls._option_input_id(model_id, "steps"),
display_name="steps",
default=steps_default,
min=1,
max=steps_max,
tooltip="Number of inference steps. Model constraints can reduce the range and have different defaults.",
),
]
@classmethod
def _build_model_options(cls) -> list[io.DynamicCombo.Option]:
specs = cls._image_specs(require=False)
options: list[io.DynamicCombo.Option] = []
def _sorted_model_items() -> list[tuple[str, Dict[str, Any]]]:
return sorted(specs.items())
for model_id, spec in _sorted_model_items():
if (
model_id == "nano-banana"
): # todo: implement ui for nano-banana, might be able to use code from video node
continue
constraints = spec.get("constraints") or {}
width_divisor = cls._width_height_divisor(constraints)
steps_default, steps_max = cls._steps_limits(constraints)
option_inputs = cls._model_option_inputs(
model_id,
width_divisor,
steps_default,
steps_max,
)
options.append(io.DynamicCombo.Option(model_id, option_inputs))
if not options:
option_inputs = cls._model_option_inputs(
"none_available",
16,
20,
50,
)
options.append(io.DynamicCombo.Option("none_available", option_inputs))
return options
@staticmethod
def _prompt_limit_from_spec(spec: Dict[str, Any] | None, default: int = 1500) -> int:
if not spec:
return default
constraints = spec.get("constraints") or {}
limit = constraints.get("promptCharacterLimit")
if isinstance(limit, int) and limit > 0:
return limit
try:
normalized = int(limit)
except (TypeError, ValueError):
return default
return normalized if normalized > 0 else default
@classmethod
def define_schema(cls) -> io.Schema:
model_options = cls._build_model_options()
style_options = cls._style_options()
return io.Schema(
node_id="GenerateImage_VENICE",
display_name="Generate Image (Venice)",
category="venice.ai",
inputs=[
io.String.Input(
"prompt",
default="A flying cat made of lettuce",
multiline=True,
placeholder="Positive Prompt. Example: A flying cat made of lettuce",
tooltip="The text prompt to guide the image generation. Character limit depends on model (usually around 1500-7500 characters).",
),
io.String.Input(
"neg_prompt",
default="",
multiline=True,
placeholder="Negative Prompt. Example: low quality, vacant scene",
tooltip=(
"Negative prompt (ignored for models that do not support CFG - z-image-turbo, flux-dev, etc.). "
"Character limit depends on model (usually around 1500-7500 characters)."
),
),
io.DynamicCombo.Input(
"model",
options=model_options,
tooltip="Select a Venice image model to auto-populate valid parameters",
),
io.Int.Input(
"batch_size",
default=1,
min=1,
max=4,
tooltip="Number of images to generate in a single batch (sequential requests, does not use variants api (yet?)).",
),
io.Float.Input(
"guidance",
default=6.0,
min=0.0,
max=20.0,
step=0.05,
tooltip=(
"CFG scale (SDXL based models work well with 6.0, most newer ones work with 3-4. "
"Closed Source models may ignore this setting and distilled models too, such as z-image-turbo or flux-dev and similar)."
),
),
io.Combo.Input(
"style_preset",
options=list(style_options),
default=style_options[0],
tooltip="Venice.ai style preset to apply to the generated image.",
),
io.Boolean.Input(
"hide_watermark",
default=True,
tooltip="Hide the Venice watermark when possible.",
),
io.Boolean.Input(
"safe_mode",
default=False,
tooltip="Enable safe mode (blurs NSFW content).",
),
io.Int.Input(
"seed",
optional=True,
default=42,
min=-0x3B9AC9FF,
max=0x3B9AC9FF,
tooltip="Seed for reproducibility.",
),
],
outputs=[io.Image.Output(id="image", display_name="Image")],
)
@classmethod
def execute(
cls,
model,
prompt,
neg_prompt,
width,
height,
batch_size,
steps,
guidance,
# lora_strength,
batch_size,
style_preset,
hide_watermark,
safe_mode,
# format,
seed=-1,
):
if prompt == "" or len(prompt) > 1500:
raise ValueError("VeniceAI Generate Image Node: Prompt cannot be empty or above 1500 characters")
if len(neg_prompt) > 1500:
raise ValueError("VeniceAI Generate Image Node: Negative prompt cannot be above 1500 characters")
if re.match(r"^flux.*", model):
logging.info(f"VeniceAPI INFO: Ignoring negative prompt for {model}.")
neg_prompt = ""
) -> io.NodeOutput:
if not isinstance(model, dict) or "model" not in model:
raise ValueError("Model selection is required")
images_tensor = () # empty tuple for tensors
model_id = model.get("model")
specs = cls._image_specs(require=True)
spec = specs.get(model_id)
if not spec:
raise ValueError("Selected model is missing from the Venice catalog; refresh the catalog and try again.")
constraints = spec.get("constraints") or {}
prompt_limit = cls._prompt_limit_from_spec(spec)
width_height_divisor = cls._width_height_divisor(constraints)
steps_default, _steps_max = cls._steps_limits(constraints)
width = int(cls._resolve_option_value(model, model_id, "width", 1024))
height = int(cls._resolve_option_value(model, model_id, "height", 1024))
steps = int(cls._resolve_option_value(model, model_id, "steps", steps_default))
guidance = float(guidance)
batch_size = int(batch_size)
style_options = cls._style_options()
if style_preset not in style_options:
style_preset = style_options[0]
hide_watermark = bool(hide_watermark)
safe_mode = bool(safe_mode)
seed = seed
ensure_multiple_of(width, height, multiple=width_height_divisor)
ensure_prompt_length(prompt, prompt_limit, "Prompt")
ensure_prompt_length(neg_prompt, prompt_limit, "Negative Prompt", allow_empty=True)
seed_value = -1 if seed is None else int(seed)
images_tensor = ()
try:
self.check_multiple_of_32(width, height) # todo: make this be validate node instead
headers = {"Authorization": f"Bearer {os.getenv('VENICEAI_API_KEY')}", "Content-Type": "application/json"}
url = VENICEAI_BASE_URL + API_ENDPOINTS["image_generate"]
payload = {
"model": model,
"model": model_id,
"prompt": prompt,
"negative_prompt": neg_prompt,
# "lora_strength": lora_strength,
"style_preset": style_preset,
"height": height,
"width": width,
"steps": steps,
"cfg_scale": guidance,
"seed": seed,
"seed": seed_value,
"return_binary": False,
"hide_watermark": hide_watermark,
"safe_mode": safe_mode,
"format": "png", # hardcoded because, change to format var and uncomment related stuff above if want dynamic
"format": "png",
"embed_exif_metadata": True,
}
if style_preset == "none":
del payload["style_preset"]
if style_preset in ("none", "none_available"):
payload.pop("style_preset", None)
for i in range(batch_size):
payload["seed"] = seed + i
response = requests.request("POST", url, json=payload, headers=headers)
if response.status_code != 200:
raise requests.exceptions.HTTPError(
f"HTTP error: {response.status_code}, Response: {response.text}"
)
images_tensor += self.process_result(response.json())
payload["seed"] = seed_value + i
response_json = client.post_json(API_ENDPOINTS["image_generate"], payload)
images_tensor += cls._processor.process_result(response_json)
merged = torch.cat(images_tensor, dim=0)
return (merged,)
return io.NodeOutput(merged)
except Exception as e:
raise Exception(f"Error processing image result: {str(e)}") from e
NODE_CLASS_MAPPINGS = {
"GenerateImage_VENICE": GenerateImage,
}
NODE_DISPLAY_NAME_MAPPINGS = {
"GenerateImage_VENICE": "Generate Image (Venice)",
}
except Exception as exc:
raise Exception(f"Error processing image result: {str(exc)}") from exc
+186 -131
View File
@@ -1,169 +1,224 @@
import logging
import os
import subprocess
import tempfile
import requests
import torch # type: ignore
import torchaudio # type: ignore
import torch
from ..globals import API_ENDPOINTS, VENICEAI_BASE_URL
from comfy_api.latest import io
try:
import imageio_ffmpeg
except ImportError: # pragma: no cover
imageio_ffmpeg = None
from ..globals import API_ENDPOINTS
from ..nodes.catalog_utils import tts_model_choices, tts_voice_choices
from ..nodes.utils import ensure_prompt_length
from ..venice_client import client
try:
from torchaudio import sox_io_backend
except ImportError: # pragma: no cover - default backend may not be available everywhere
sox_io_backend = None
try:
from torchaudio import soundfile_backend
except ImportError: # pragma: no cover
soundfile_backend = None
class GenerateSpeech:
LOG = logging.getLogger(__name__)
class GenerateSpeech(io.ComfyNode):
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"model": (
"COMBO",
{
"default": "tts-kokoro",
},
def _model_options(cls) -> list[str]:
options = list(tts_model_choices())
return options or ["none_available"]
@classmethod
def _voice_options(cls) -> list[str]:
options = list(tts_voice_choices())
return options or ["none_available"]
@staticmethod
def _ffmpeg_decode(temp_path: str) -> tuple[torch.Tensor, int]:
if imageio_ffmpeg is None:
LOG.error("imageio-ffmpeg is not installed; ffmpeg fallback unavailable")
raise RuntimeError("imageio-ffmpeg is not installed; install it to enable ffmpeg fallback.")
try:
ffmpeg = imageio_ffmpeg.get_ffmpeg_exe()
except Exception as exc:
LOG.error("Unable to download ffmpeg via imageio-ffmpeg: %s", exc)
raise RuntimeError("Failed to download ffmpeg via imageio-ffmpeg") from exc
LOG.info("Decoding Venice audio via ffmpeg executable %s", ffmpeg)
try:
process = subprocess.run(
[
ffmpeg,
"-hide_banner",
"-loglevel",
"error",
"-i",
temp_path,
"-acodec",
"pcm_f32le",
"-f",
"f32le",
"-ac",
"1",
"-ar",
"44100",
"-",
],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
check=True,
)
except subprocess.CalledProcessError as exc:
LOG.error("ffmpeg failed to decode %s: %s", temp_path, exc.stderr.decode(errors="ignore"))
raise RuntimeError(f"ffmpeg failed to decode {temp_path}: {exc.stderr.decode(errors='ignore')}") from exc
audio_data = torch.frombuffer(process.stdout, dtype=torch.float32).clone()
audio_data = audio_data.reshape(-1, 1).transpose(0, 1)
return audio_data, 44100
@staticmethod
def _load_with_torchaudio_backends(temp_path: str, response_format: str) -> tuple[torch.Tensor, int]:
backends = [sox_io_backend, soundfile_backend]
errors: list[str] = []
for backend in backends:
if backend is None:
continue
backend_name = getattr(backend, "__name__", "torchaudio_backend")
try:
return backend.load(temp_path)
except Exception as exc:
error = str(exc)
errors.append(error)
LOG.warning("torchaudio backend %s failed to load %s: %s", backend_name, response_format, error)
if errors:
raise RuntimeError(f"Failed to load audio format '{response_format}' with all methods: {', '.join(errors)}")
raise RuntimeError(
f"Failed to load audio format '{response_format}' because no torchaudio backend is available."
)
@classmethod
def define_schema(cls) -> io.Schema:
response_formats = ["mp3", "opus", "aac", "flac", "wav", "pcm"]
model_options = cls._model_options()
voice_options = cls._voice_options()
return io.Schema(
node_id="GenerateSpeech_VENICE",
display_name="Generate Speech (Venice)",
category="venice.ai",
inputs=[
io.Combo.Input(
"model",
options=model_options,
default=model_options[0],
tooltip="Model to use for text-to-speech",
),
"input": (
"STRING",
{
"default": "",
"multiline": True,
"tooltip": ("The text to generate audio for. The maximum length is 4096 characters."),
},
io.String.Input(
"input",
default="",
multiline=True,
placeholder="Text to speak",
tooltip="The text prompt used for speech generation (max 4096 chars)",
),
"response_format": (
[
# todo: some dont work because idk implementing would be ass
"mp3",
# "opus",
# "aac",
# "flac",
"wav",
"pcm",
],
{
"default": "mp3",
"tooltip": (
"mp3: widely supported, lossy; "
# "opus: very good quality at low bitrate; "
# "aac: lossy, good for streaming; "
# "flac: lossless compressed audio; "
"wav: lossless raw audio; "
"pcm: uncompressed raw audio."
),
},
io.Combo.Input(
"response_format",
options=response_formats,
default=response_formats[0],
tooltip="Audio format to request from the Venice TTS API",
),
"speed": (
"FLOAT",
{
"default": 1.0,
"min": 0.25,
"max": 4,
"step": 0.01,
"tooltip": (
"The text to image style to apply during prompt enhancement. "
"Does best with short descriptive prompts, like gold, marble or angry, menacing."
),
},
io.Float.Input(
"speed",
default=1.0,
min=0.25,
max=4.0,
step=0.01,
tooltip="Playback speed multiplier (1.0 = normal speed)",
),
# "streaming": (
# "BOOLEAN",
# {
# "default": False,
# "tooltip": (
# "Should the content stream back sentence by sentence or be processed and returned as a complete audio file."
#
# ),
# },
# ),
"voice": (
"COMBO",
{
"default": "af_sky - tts-kokoro",
},
io.Combo.Input(
"voice",
options=voice_options,
default=voice_options[0],
tooltip="Voice preset to use for the TTS model",
),
}
}
],
outputs=[io.Audio.Output(id="audio", display_name="audio")],
)
RETURN_TYPES = ("AUDIO",)
RETURN_NAMES = ("audio",)
FUNCTION = "gen_speech"
CATEGORY = "venice.ai"
@classmethod
def execute(cls, model, input, response_format, speed, voice) -> io.NodeOutput:
ensure_prompt_length(input, 4096, label="TTS input")
EXPERIMENTAL = True
def gen_speech(self, model, input, response_format, speed, voice):
if len(input) > 4096 or len(input) == 0:
raise ValueError("Generate Speech (Venice) Input exceeds the max length of 4096 characters or is empty.")
url = VENICEAI_BASE_URL + API_ENDPOINTS["speech_generate"]
# remove everything from voice string after and including the hyphen " - blabla"
voice = voice.split(" - ")[0] if " - " in voice else voice
# Prepare JSON payload
# todo: currently models' voices show up as "model-name - voice_name"
# todo: this can be in dynamiccombo so the single combo dropdown is not cluttered with all the voices of all selectable models
normalized_voice = voice.split(" - ")[-1].strip() if " - " in voice else voice
payload = {
"model": model,
"input": input,
"speed": speed,
"voice": voice,
"voice": normalized_voice,
"response_format": response_format,
"streaming": False,
}
headers = {"Authorization": f"Bearer {os.getenv('VENICEAI_API_KEY')}", "Content-Type": "application/json"}
# Send request
try:
response = requests.post(url, json=payload, headers=headers)
response.raise_for_status()
except requests.exceptions.RequestException as e:
raise RuntimeError(f"Generate Speech (Venice) API request failed: {str(e)}")
response = client.request(
"POST",
API_ENDPOINTS["speech_generate"],
json=payload,
headers={"Content-Type": "application/json"},
)
except requests.exceptions.RequestException as exc:
raise RuntimeError(f"Generate Speech (Venice) API request failed: {str(exc)}")
# Convert the audio response to a ComfyUI-compatible tensor and return it
if not response.content or len(response.content) == 0:
if not response.content:
raise RuntimeError("No audio data received from Venice API.")
# Save to temporary file and load with torchaudio for better format support
with tempfile.NamedTemporaryFile(suffix=f".{response_format}", delete=False) as temp_file:
temp_file.write(response.content)
temp_file_path = temp_file.name
temp_path = None
waveform = None
sample_rate = None
try:
# Load audio using torchaudio from the temporary file
waveform, sample_rate = torchaudio.load(temp_file_path)
except Exception as e:
# Fallback: try different approaches for problematic formats
# Save to temp file and load with torchaudio for better format support
with tempfile.NamedTemporaryFile(suffix=f".{response_format}", delete=False) as temp_file:
temp_file.write(response.content)
temp_path = temp_file.name
try:
if response_format == "pcm":
# For PCM, we need to handle it as raw audio data
# Assume 16-bit PCM, mono, 16kHz (adjust as needed based on API response)
waveform, sample_rate = cls._load_with_torchaudio_backends(temp_path, response_format)
except RuntimeError as audio_exc:
LOG.warning("torchaudio decoding failed for %s: %s", response_format, audio_exc)
if response_format == "pcm" and response.content:
LOG.info("Falling back to raw PCM interpretation for %s", response_format)
audio_data = torch.frombuffer(response.content, dtype=torch.int16).float() / 32768.0
waveform = audio_data.unsqueeze(0) # Add channel dimension
sample_rate = 16000 # Default sample rate, adjust if needed
waveform = audio_data.unsqueeze(0)
sample_rate = 16000 # Default sample rate, might need change?
else:
# For other formats, try loading without specifying format
waveform, sample_rate = torchaudio.load(temp_file_path, format=None)
except Exception as e2:
raise RuntimeError(
f"Failed to load audio format '{response_format}' with all methods. Errors: {str(e)}, {str(e2)}"
)
LOG.info("Attempting ffmpeg fallback for %s audio", response_format)
waveform, sample_rate = cls._ffmpeg_decode(temp_path)
finally:
# Clean up temporary file
try:
os.unlink(temp_file_path)
except:
pass
if temp_path:
try:
os.unlink(temp_path)
except OSError:
pass
if waveform is None or sample_rate is None:
raise RuntimeError("Unable to decode Venice speech response.")
# Ensure shape is [B, C, T] (batch size 1)
if waveform.dim() == 2:
waveform = waveform.unsqueeze(0) # [1, C, T]
elif waveform.dim() == 1:
waveform = waveform.unsqueeze(0).unsqueeze(0) # [1, 1, T]
return ({"waveform": waveform, "sample_rate": sample_rate},)
NODE_CLASS_MAPPINGS = {
"GenerateSpeech_VENICE": GenerateSpeech,
}
NODE_DISPLAY_NAME_MAPPINGS = {
"GenerateSpeech_VENICE": "Generate Speech [BETA] (Venice)",
}
audio_value: io.Audio.Type = {"waveform": waveform, "sample_rate": sample_rate}
return io.NodeOutput(audio_value)
-354
View File
@@ -1,354 +0,0 @@
import base64
import io
import os
import numpy as np
import requests
from PIL import Image
from ..globals import API_ENDPOINTS, VENICEAI_BASE_URL
class GenerateTextAdvanced:
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"model": ("COMBO", {"default": "llama-3.1-405b", "tooltip": ("The model to use for text generation.")}),
"prompt": (
"STRING",
{
"default": "",
"multiline": True,
"tooltip": ("The prompt to generate text from. Ask, command or chat with the model."),
},
),
"system_prompt": (
"STRING",
{
"default": "",
"multiline": True,
"tooltip": ("Optional system prompt to guide the model's behavior."),
},
),
"enable_system_prompt": (
"BOOLEAN",
{
"default": True,
"tooltip": ("Enable or disable system prompt being passed on."),
},
),
"frequency_penalty": (
"FLOAT",
{
"default": 0.0,
"min": -2.0,
"max": 2.0,
"step": 0.05,
"tooltip": (
"Positive values penalize new tokens based on their existing frequency in the text so far, "
"decreasing the model's likelihood to repeat the same line verbatim."
),
},
),
"presence_penalty": (
"FLOAT",
{
"default": 0.0,
"min": -2.0,
"max": 2.0,
"step": 0.05,
"tooltip": (
"Positive values penalize new tokens based on whether they appear in the text so far, "
"increasing the model's likelihood to talk about new topics."
),
},
),
"repetition_penalty": (
"FLOAT",
{
"default": 1.2,
"min": 0.0,
"max": 2.0,
"step": 0.05,
"tooltip": ("1.0 means no penalty. Values > 1.0 discourage repetition."),
},
),
"max_temp": (
"FLOAT",
{
"default": 1.5,
"min": 0.0,
"max": 2.0,
"step": 0.05,
"tooltip": ("Maximum temperature value for dynamic temperature scaling."),
},
),
"min_temp": (
"FLOAT",
{
"default": 0.1,
"min": 0.0,
"max": 2.0,
"step": 0.05,
"tooltip": ("Minimum temperature value for dynamic temperature scaling."),
},
),
"max_completion_tokens": (
"INT",
{
"default": 420,
"min": 1,
"max": 131072,
"step": 1,
"tooltip": (
"An upper bound for the number of tokens that can be generated for "
"a completion, including visible output tokens and reasoning tokens."
),
},
),
"temperature": (
"FLOAT",
{
"default": 0.5,
"min": 0.0,
"max": 2.0,
"step": 0.05,
"tooltip": (
"Higher values like 0.8 will make the output more random, "
"while lower values like 0.2 will make it more focused and deterministic. "
"We generally recommend altering this or top_p but not both."
),
},
),
"top_k": (
"INT",
{
"default": 40,
"min": 0,
"tooltip": ("The number of highest probability vocabulary tokens to keep for top-k-filtering."),
},
),
"top_p": (
"FLOAT",
{
"default": 0.8,
"min": 0.0,
"max": 2.0,
"step": 0.01,
"tooltip": (
"An alternative to sampling with temperature, called nucleus sampling, "
"where the model considers the results of the tokens with top_p probability mass. "
"So 0.1 means only the tokens comprising the top 10% probability mass are considered."
),
},
),
"min_p": (
"FLOAT",
{
"default": 0.05,
"min": 0.0,
"max": 1.0,
"step": 0.01,
"tooltip": (
"Sets a minimum probability threshold for token selection. "
"Tokens with probabilities below this value are filtered out."
),
},
),
# "stop": ("STRING", {"default": "", "tooltip": "Up to 4 sequences where the API will stop generating further tokens. Defaults to null.", "placeholder": "stop: [\"\\n\"]"}),
# "stop_token_ids": ("STRING", {"default": "", "tooltip": "Array of token IDs where the API will stop generating further tokens. Example: [151643, 151645]", "placeholder": "151643, 151645, ..."}),
"enable_vision": (
"BOOLEAN",
{
"default": False,
"tooltip": (
"Enable or disable vision tasks. "
"Requires image_for_vision input to be populated and "
"for the LLM to actually support vision tasks to process."
),
},
),
},
"optional": {
"venice_parameters": (
"STRING",
{
"forceInput": True,
"tooltip": (
"Optional input. "
"Use the Textgen Parameters (Venice) node to use "
"extra, venice specific parameters for text generation."
),
},
),
"image_for_vision": (
"IMAGE",
{
"tooltip": (
"Optional input. "
"Add an image for vision-supported LLMs to process. "
"Will only be processed if 'enable_vision' is 'True' and "
"if the LLM actually supports vision tasks."
),
},
),
},
}
RETURN_TYPES = ("STRING",)
RETURN_NAMES = ("response",)
FUNCTION = "generate_text"
CATEGORY = "venice.ai"
DESCRIPTION = (
"Text Generation node that makes use of Venice.AI's text generation API. "
"Use Textgen Parameters (Venice) node to pass on extra Venice.AI specific parameters."
"Does not have chat history context. "
)
def generate_text(
# region params
self,
model,
prompt,
system_prompt,
enable_system_prompt,
frequency_penalty,
presence_penalty,
repetition_penalty,
max_temp,
min_temp,
max_completion_tokens,
temperature,
top_k,
top_p,
min_p,
enable_vision,
**kwargs,
# endregion
):
url = VENICEAI_BASE_URL + API_ENDPOINTS["text_generate"]
user_content = []
venice_parameters = kwargs.get("venice_parameters", None)
image_for_vision = kwargs.get("image_for_vision", None)
if image_for_vision is not None and enable_vision:
# Convert tensor to PIL Image
image_tensor = image_for_vision[0] # shape: (H, W, 3)
image_np = image_tensor.cpu().numpy() # Still in (H, W, 3)
image_np = (image_np * 255).astype(np.uint8) # Scale from [0, 1] to [0, 255] if needed
pil_image = Image.fromarray(image_np)
# Resize image to meet constraints
original_width, original_height = pil_image.size
aspect_ratio = original_width / original_height
# Determine target dimensions
if original_width > original_height:
target_width = 1024
target_height = int(target_width / aspect_ratio)
if target_height < 256:
target_height = 256
target_width = int(target_height * aspect_ratio)
else:
target_height = 1024
target_width = int(target_height * aspect_ratio)
if target_width < 256:
target_width = 256
target_height = int(target_width / aspect_ratio)
# Round dimensions to multiples of 14
def round_down_to_multiple(value, multiple):
return (value // multiple) * multiple
target_width = round_down_to_multiple(target_width, 14)
target_height = round_down_to_multiple(target_height, 14)
# Ensure minimum dimension is 256 after rounding
if min(target_width, target_height) < 256:
if target_width < target_height:
target_width = ((256 + 13) // 14) * 14
target_height = round_down_to_multiple(int(target_width / aspect_ratio), 14)
else:
target_height = ((256 + 13) // 14) * 14
target_width = round_down_to_multiple(int(target_height * aspect_ratio), 14)
pil_image = pil_image.resize((target_width, target_height), Image.LANCZOS) # type: ignore
# Convert to base64 and check size
buffered = io.BytesIO()
pil_image.save(buffered, format="PNG")
img_base64 = base64.b64encode(buffered.getvalue()).decode("utf-8")
# Resize further if base64 exceeds 4.5MB
while len(img_base64) > 4500000:
scaling_factor = (4500000 / len(img_base64)) ** 0.5
new_width = int(target_width * scaling_factor)
new_height = int(target_height * scaling_factor)
new_width = max(round_down_to_multiple(new_width, 14), 256)
new_height = max(round_down_to_multiple(new_height, 14), 256)
pil_image = pil_image.resize((new_width, new_height), Image.LANCZOS) # type: ignore
target_width, target_height = new_width, new_height
buffered = io.BytesIO()
pil_image.save(buffered, format="PNG")
img_base64 = base64.b64encode(buffered.getvalue()).decode("utf-8")
user_content.extend(
[
{"type": "text", "text": prompt},
{"type": "image_url", "image_url": {"url": f"data:image/png;base64,{img_base64}"}},
]
)
else:
user_content.append({"type": "text", "text": prompt})
if not enable_system_prompt:
system_prompt = ""
messages = [{"role": "system", "content": system_prompt}]
messages.append({"role": "user", "content": user_content})
payload = {
"model": model,
"messages": messages,
"frequency_penalty": frequency_penalty,
"presence_penalty": presence_penalty,
"repetition_penalty": repetition_penalty,
"max_temp": max_temp,
"min_temp": min_temp,
"max_completion_tokens": max_completion_tokens,
"temperature": temperature,
"top_k": top_k,
"top_p": top_p,
"min_p": min_p,
}
if venice_parameters is not None:
payload["venice_parameters"] = venice_parameters
headers = {"Authorization": f"Bearer {os.getenv('VENICEAI_API_KEY')}", "Content-Type": "application/json"}
response = requests.post(url, json=payload, headers=headers)
if response.status_code != 200:
raise requests.exceptions.HTTPError(f"HTTP error: {response.status_code}, Response: {response.text}")
json_response = response.json()
try:
content = json_response["choices"][0]["message"]["content"]
except (KeyError, IndexError, TypeError) as e:
raise ValueError(f"Unexpected API response format: {json_response}") from e
return (content,)
NODE_CLASS_MAPPINGS = {
"GenerateTextAdvanced_VENICE": GenerateTextAdvanced,
}
NODE_DISPLAY_NAME_MAPPINGS = {
"GenerateTextAdvanced_VENICE": "Generate Text Advanced BETA (Venice)",
}
+466 -129
View File
@@ -1,161 +1,498 @@
import base64
import io
import os
import json
import logging
from typing import Any, Dict, Iterable
import numpy as np
import requests
from PIL import Image
from comfy_api.latest import io
from ..globals import API_ENDPOINTS, VENICEAI_BASE_URL
from ..globals import API_ENDPOINTS
from ..nodes.catalog_utils import character_choices, text_model_specs
from ..nodes.utils import encode_tensor_for_vision
from ..venice_client import client
logger = logging.getLogger(__name__)
class GenerateText:
class GenerateTextAdvanced(io.ComfyNode):
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"model": (
"COMBO",
{
"default": "llama-3.3-70b",
},
def _option_input_id(cls, model_id: str, field: str) -> str:
return f"{model_id}__{field}"
@staticmethod
def _constraint_default(value: Any, fallback: float) -> float:
if isinstance(value, dict):
default = value.get("default")
else:
default = value
if isinstance(default, (int, float)):
return float(default)
return fallback
@classmethod
def _get_option_value(cls, model_payload: Dict[str, Any], model_id: str, field: str) -> Any:
candidates = (
cls._option_input_id(model_id, field),
field,
f"{field}__{model_id}",
f"{model_id}__{field}",
)
for key in candidates:
if key in model_payload:
return model_payload.get(key)
return None
@classmethod
def _text_specs(cls) -> Dict[str, Dict[str, Any]]:
specs = text_model_specs() or {}
if not specs:
raise ValueError("No Venice text model specs available")
return specs
@staticmethod
def _normalize_stop_tokens(value: str | Iterable[str] | None) -> list[str]:
tokens: list[str] = []
if not value:
return tokens
segments: Iterable[str] = value.splitlines() if isinstance(value, str) else value
for segment in segments:
for raw_token in str(segment).split(","):
trimmed = raw_token.strip()
if trimmed:
tokens.append(trimmed)
return tokens
@classmethod
def _build_model_options(cls) -> list[io.DynamicCombo.Option]:
specs = cls._text_specs()
options: list[io.DynamicCombo.Option] = []
for model_id, spec in sorted(specs.items(), key=lambda item: item[0]):
constraints = spec.get("constraints") or {}
capabilities = spec.get("capabilities") or {}
temperature_default = cls._constraint_default(constraints.get("temperature"), 0.5)
top_p_default = cls._constraint_default(constraints.get("top_p"), 0.8)
option_inputs: list[io.Input] = [
io.Float.Input(
cls._option_input_id(model_id, "temperature"),
display_name="temperature",
default=temperature_default,
min=0.0,
max=2.0,
step=0.01,
tooltip="Sampling temperature (per-model default taken from the catalog).",
),
"system_prompt": ("STRING", {"default": "", "multiline": True}),
"prompt": ("STRING", {"default": "", "multiline": True}),
"frequency_penalty": ("FLOAT", {"default": 1.5, "min": 0.0, "max": 2.0, "step": 0.1}),
"presence_penalty": ("FLOAT", {"default": 1.5, "min": 0.0, "max": 2.0, "step": 0.1}),
"temperature": ("FLOAT", {"default": 0.5, "min": 0.0, "max": 2.0, "step": 0.1}),
"top_p": ("FLOAT", {"default": 0.9, "min": 0.0, "max": 1.0, "step": 0.1}),
"enable_vision": ("BOOLEAN", {"default": False}),
},
"optional": {
"image_for_vision": ("IMAGE",),
},
}
io.Float.Input(
cls._option_input_id(model_id, "top_p"),
display_name="top_p",
default=top_p_default,
min=0.0,
max=1.0,
step=0.01,
tooltip="Nucleus sampling probability (per-model default taken from the catalog).",
),
]
RETURN_TYPES = ("STRING",)
RETURN_NAMES = ("response",)
FUNCTION = "generate_text"
CATEGORY = "venice.ai"
if capabilities.get("supportsVision"):
option_inputs.append(
io.Boolean.Input(
cls._option_input_id(model_id, "enable_vision"),
display_name="enable_vision",
default=False,
tooltip="Enable vision inputs when the model supports vision.",
)
)
def generate_text(
self,
if capabilities.get("supportsReasoning"):
option_inputs.append(
io.Boolean.Input(
cls._option_input_id(model_id, "reasoning"),
display_name="reasoning",
default=True,
tooltip="Toggle reasoning capabilities for this model.",
)
)
option_inputs.extend(cls._venice_parameter_inputs(model_id, capabilities))
options.append(io.DynamicCombo.Option(model_id, option_inputs))
return options
@classmethod
def _venice_parameter_inputs(cls, model_id: str, capabilities: Dict[str, Any]) -> list[io.Input]:
character_options = ["none", *character_choices()]
inputs: list[io.Input] = [
io.Combo.Input(
cls._option_input_id(model_id, "vp_character_slug"),
display_name="vp_character_slug",
options=character_options,
default=character_options[0],
optional=True,
tooltip="Select a Venice character slug (public ID) from the catalog.",
),
]
if capabilities.get("supportsReasoning"):
inputs.extend(
[
io.Boolean.Input(
cls._option_input_id(model_id, "vp_strip_thinking_response"),
display_name="vp_strip_thinking_response",
default=False,
tooltip="Strip thinking blocks from the response on reasoning models.",
),
io.Boolean.Input(
cls._option_input_id(model_id, "vp_disable_thinking"),
display_name="vp_disable_thinking",
default=False,
tooltip="Disable thinking blocks for supported reasoning models.",
),
]
)
if capabilities.get("supportsWebSearch"):
inputs.extend(
[
io.Combo.Input(
cls._option_input_id(model_id, "vp_enable_web_search"),
display_name="vp_enable_web_search",
options=["auto", "off", "on"],
default="off",
optional=True,
tooltip="Set to auto/off/on to control Venice web search for this request.",
),
io.Boolean.Input(
cls._option_input_id(model_id, "vp_enable_web_scraping"),
display_name="vp_enable_web_scraping",
default=False,
tooltip="Enable Venice web scraping for URLs found in the latest user message.",
),
io.Boolean.Input(
cls._option_input_id(model_id, "vp_enable_web_citations"),
display_name="vp_enable_web_citations",
default=False,
tooltip="Request citations when web search returns sources.",
),
]
)
inputs.append(
io.Boolean.Input(
cls._option_input_id(model_id, "vp_include_venice_system_prompt"),
display_name="vp_include_venice_system_prompt",
default=False,
tooltip="Include Venice-supplied system prompts alongside your own.",
)
)
return inputs
@classmethod
def define_schema(cls) -> io.Schema:
model_options = cls._build_model_options()
return io.Schema(
node_id="GenerateTextAdvanced_VENICE",
display_name="Generate Text Advanced (Venice)",
category="venice.ai",
inputs=[
io.String.Input(
"prompt",
default="",
multiline=True,
tooltip="The prompt to generate text from. Ask, command or chat with the model.",
),
io.String.Input(
"system_prompt",
default="",
multiline=True,
tooltip="Optional system prompt to guide the model's behavior.",
),
io.Image.Input(
"vision_image",
display_name="vision_image",
optional=True,
tooltip="Optional image for vision-capable models. Enable the vision toggle to send it.",
),
io.DynamicCombo.Input(
"model",
options=model_options,
tooltip="The model to use for text generation.",
),
io.Float.Input(
"frequency_penalty",
default=0.0,
min=-2.0,
max=2.0,
step=0.05,
tooltip=(
"Positive values penalize new tokens based on their existing frequency in the text so far, "
"decreasing the model's likelihood to repeat the same line verbatim."
),
),
io.Float.Input(
"presence_penalty",
default=0.0,
min=-2.0,
max=2.0,
step=0.05,
tooltip=(
"Positive values penalize new tokens based on whether they appear in the text so far, "
"increasing the model's likelihood to talk about new topics."
),
),
io.Float.Input(
"repetition_penalty",
default=1.2,
min=0.0,
max=2.0,
step=0.05,
tooltip="1.0 means no penalty. Values > 1.0 discourage repetition.",
),
io.Float.Input(
"max_temp",
default=1.5,
min=0.0,
max=2.0,
step=0.05,
tooltip="Maximum temperature value for dynamic temperature scaling.",
),
io.Float.Input(
"min_temp",
default=0.1,
min=0.0,
max=2.0,
step=0.05,
tooltip="Minimum temperature value for dynamic temperature scaling.",
),
io.Int.Input(
"max_completion_tokens",
default=420,
min=1,
max=131072,
step=1,
tooltip=(
"An upper bound for the number of tokens that can be generated for "
"a completion, including visible output tokens and reasoning tokens."
),
),
io.Int.Input(
"top_k",
default=40,
min=0,
tooltip="The number of highest probability vocabulary tokens to keep for top-k-filtering.",
),
io.Float.Input(
"min_p",
default=0.05,
min=0.0,
max=1.0,
step=0.01,
tooltip=(
"Sets a minimum probability threshold for token selection. "
"Tokens with probabilities below this value are filtered out."
),
),
io.String.Input(
"stop_tokens",
default="",
tooltip="Optional comma- or newline-separated tokens to stop generation on (requires at least one).",
),
io.Int.Input(
"seed",
default=42,
min=1,
tooltip="Seed for Venice randomness; must be 1 or greater.",
),
io.Boolean.Input(
"enable_system_prompt",
default=True,
tooltip="Enable or disable system prompt being passed on.",
),
io.Boolean.Input(
"debug_append_response",
default=False,
tooltip="Append the raw Venice response after three newlines for debugging.",
),
],
outputs=[io.String.Output(id="response", display_name="response")],
)
@classmethod
def execute(
cls,
model,
system_prompt,
prompt,
system_prompt,
vision_image,
frequency_penalty,
presence_penalty,
temperature,
top_p,
enable_vision,
**kwargs,
):
url = VENICEAI_BASE_URL + API_ENDPOINTS["text_generate"]
repetition_penalty,
max_temp,
min_temp,
max_completion_tokens,
top_k,
min_p,
stop_tokens,
seed,
enable_system_prompt,
debug_append_response,
) -> io.NodeOutput:
if isinstance(model, str):
model = {"model": model}
if not isinstance(model, dict) or "model" not in model:
raise ValueError("Model selection is required")
model_id = model.get("model")
specs = cls._text_specs()
spec = specs.get(model_id)
if not spec:
raise ValueError("Selected model is missing from the Venice catalog; refresh the catalog and try again.")
constraints = spec.get("constraints") or {}
capabilities = spec.get("capabilities") or {}
temperature_value = cls._get_option_value(model, model_id, "temperature")
if temperature_value is None:
temperature_value = cls._constraint_default(constraints.get("temperature"), 0.5)
try:
temperature_value = float(temperature_value)
except (TypeError, ValueError):
temperature_value = 0.5
top_p_value = cls._get_option_value(model, model_id, "top_p")
if top_p_value is None:
top_p_value = cls._constraint_default(constraints.get("top_p"), 0.8)
try:
top_p_value = float(top_p_value)
except (TypeError, ValueError):
top_p_value = 0.8
reasoning_value = cls._get_option_value(model, model_id, "reasoning")
reasoning_enabled = (
bool(reasoning_value) if reasoning_value is not None else bool(capabilities.get("supportsReasoning"))
)
reasoning_effort_value = "medium"
vision_enabled = (
bool(cls._get_option_value(model, model_id, "enable_vision"))
if capabilities.get("supportsVision")
else False
)
vision_tensor = None
if vision_image is not None:
candidates = vision_image if isinstance(vision_image, (list, tuple)) else (vision_image,)
for candidate in candidates:
if candidate is not None:
vision_tensor = candidate
break
normalized_stop_tokens = cls._normalize_stop_tokens(stop_tokens)
try:
seed_value = int(seed)
except (TypeError, ValueError):
seed_value = 42
if seed_value < 1:
seed_value = 1
venice_parameters: Dict[str, Any] = {}
def _set_bool(field: str, key: str) -> None:
value = cls._get_option_value(model, model_id, field)
if value is not None:
venice_parameters[key] = bool(value)
slug_value = cls._get_option_value(model, model_id, "vp_character_slug")
if isinstance(slug_value, str):
trimmed = slug_value.strip()
if trimmed and trimmed.lower() not in {"", "none"}:
venice_parameters["character_slug"] = trimmed
if capabilities.get("supportsReasoning"):
_set_bool("vp_strip_thinking_response", "strip_thinking_response")
_set_bool("vp_disable_thinking", "disable_thinking")
if capabilities.get("supportsWebSearch"):
web_search = cls._get_option_value(model, model_id, "vp_enable_web_search")
if isinstance(web_search, str):
trimmed = web_search.strip()
if trimmed:
venice_parameters["enable_web_search"] = trimmed
elif web_search is not None:
venice_parameters["enable_web_search"] = str(web_search)
_set_bool("vp_enable_web_scraping", "enable_web_scraping")
_set_bool("vp_enable_web_citations", "enable_web_citations")
_set_bool("vp_include_venice_system_prompt", "include_venice_system_prompt")
if vision_tensor is not None and (not capabilities.get("supportsVision") or not vision_enabled):
logger.warning(
"Vision image provided but model %s does not support vision, or enable_vision is disabled",
model_id,
)
if vision_enabled and vision_tensor is None:
raise ValueError("Vision input is enabled but no image was provided")
user_content = []
image_for_vision = kwargs.get("image_for_vision", None)
if image_for_vision is not None and enable_vision:
# Convert tensor to PIL Image
image_tensor = image_for_vision[0] # shape: (H, W, 3)
image_np = image_tensor.cpu().numpy() # Still in (H, W, 3)
image_np = (image_np * 255).astype(np.uint8) # Scale from [0, 1] to [0, 255] if needed
pil_image = Image.fromarray(image_np)
# Resize image to meet constraints
original_width, original_height = pil_image.size
aspect_ratio = original_width / original_height
# Determine target dimensions
if original_width > original_height:
target_width = 1024
target_height = int(target_width / aspect_ratio)
if target_height < 256:
target_height = 256
target_width = int(target_height * aspect_ratio)
else:
target_height = 1024
target_width = int(target_height * aspect_ratio)
if target_width < 256:
target_width = 256
target_height = int(target_width / aspect_ratio)
# Round dimensions to multiples of 14
def round_down_to_multiple(value, multiple):
return (value // multiple) * multiple
target_width = round_down_to_multiple(target_width, 14)
target_height = round_down_to_multiple(target_height, 14)
# Ensure minimum dimension is 256 after rounding
if min(target_width, target_height) < 256:
if target_width < target_height:
target_width = ((256 + 13) // 14) * 14
target_height = round_down_to_multiple(int(target_width / aspect_ratio), 14)
else:
target_height = ((256 + 13) // 14) * 14
target_width = round_down_to_multiple(int(target_height * aspect_ratio), 14)
pil_image = pil_image.resize((target_width, target_height), Image.LANCZOS) # type: ignore
# Convert to base64 and check size
buffered = io.BytesIO()
pil_image.save(buffered, format="PNG")
img_base64 = base64.b64encode(buffered.getvalue()).decode("utf-8")
# Resize further if base64 exceeds 4.5MB
while len(img_base64) > 4500000:
scaling_factor = (4500000 / len(img_base64)) ** 0.5
new_width = int(target_width * scaling_factor)
new_height = int(target_height * scaling_factor)
new_width = max(round_down_to_multiple(new_width, 14), 256)
new_height = max(round_down_to_multiple(new_height, 14), 256)
pil_image = pil_image.resize((new_width, new_height), Image.LANCZOS) # type: ignore
target_width, target_height = new_width, new_height
buffered = io.BytesIO()
pil_image.save(buffered, format="PNG")
img_base64 = base64.b64encode(buffered.getvalue()).decode("utf-8")
if vision_enabled and vision_tensor is not None:
encoded_image = encode_tensor_for_vision(vision_tensor)
user_content.extend(
[
{"type": "text", "text": prompt},
{"type": "image_url", "image_url": {"url": f"data:image/png;base64,{img_base64}"}},
{"type": "image_url", "image_url": {"url": encoded_image}},
]
)
else:
user_content.append({"type": "text", "text": prompt})
if not enable_system_prompt:
system_prompt = ""
messages = [{"role": "system", "content": system_prompt}]
messages.append({"role": "user", "content": user_content})
payload = {
"model": model,
"model": model_id,
"messages": messages,
"frequency_penalty": frequency_penalty,
"logprobs": False, # unused, not supported by all models
"top_logprobs": 0, # x >= 0
"max_completion_tokens": max_completion_tokens,
"max_temp": max_temp,
"min_p": min_p,
"min_temp": min_temp,
"n": 1, # basically batch size
"presence_penalty": presence_penalty,
"temperature": temperature,
"top_p": top_p,
"repetition_penalty": repetition_penalty,
"seed": seed_value,
"stream": False,
"temperature": temperature_value,
"top_k": top_k,
"top_p": top_p_value,
"parallel_tool_calls": True,
}
if reasoning_enabled:
payload["reasoning"] = {"mode": reasoning_effort_value}
payload["reasoning_effort"] = reasoning_effort_value
if normalized_stop_tokens:
payload["stop"] = normalized_stop_tokens
if venice_parameters:
payload["venice_parameters"] = venice_parameters
headers = {"Authorization": f"Bearer {os.getenv('VENICEAI_API_KEY')}", "Content-Type": "application/json"}
response = requests.post(url, json=payload, headers=headers)
json_response = client.post_json(API_ENDPOINTS["text_generate"], payload)
try:
choice = json_response["choices"][0]
logger.info(
"Venice LLM finish_reason=%s native_finish_reason=%s stop_reason=%s",
choice.get("finish_reason"),
choice.get("native_finish_reason"),
choice.get("stop_reason"),
)
content = choice["message"]["content"]
except (KeyError, IndexError, TypeError) as exc:
raise ValueError(f"Unexpected API response format: {json_response}") from exc
if debug_append_response:
try:
raw_dump = json.dumps(json_response, indent=2)
except (TypeError, ValueError):
raw_dump = str(json_response)
content = f"{content}\n\n\n{raw_dump}"
if response.status_code != 200:
raise requests.exceptions.HTTPError(f"HTTP error: {response.status_code}, Response: {response.text}")
json_response = response.json()
content = json_response["choices"][0]["message"]["content"]
# print(content)
return (content,)
NODE_CLASS_MAPPINGS = {
"GenerateText_VENICE": GenerateText,
}
NODE_DISPLAY_NAME_MAPPINGS = {
"GenerateText_VENICE": "Generate Text (Venice)",
}
return io.NodeOutput(content)
-118
View File
@@ -1,118 +0,0 @@
class GenerateTextVeniceParameters:
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"character_slug": (
"COMBO",
{
"default": "strawberry-the-cat",
"tooltip": ("The character slug of a public Venice character."),
},
),
"enable_character": (
"BOOLEAN",
{
"default": False,
"tooltip": ("Enable or disable character parameter being passed on."),
},
),
"strip_thinking_response": (
"BOOLEAN",
{
"default": False,
"tooltip": (
"Strip <think></think> blocks from the response. "
"Applicable only to reasoning / thinking models. "
"Also available to use as a model feature suffix. "
"Defaults to false."
),
},
),
"disable_thinking": (
"BOOLEAN",
{
"default": False,
"tooltip": (
"On supported reasoning models, will disable thinking and "
"strip the <think></think> blocks from the response. "
"Defaults to false."
),
},
),
"web_search": (
["auto", "on", "off"],
{
"default": "auto",
"tooltip": (
"Auto will enable it based on the model's discretion. "
"On will force web search on the request. "
"Citations will be returned either in the first chunk of a "
"streaming result, or in the non streaming response."
"Defaults to off. "
),
},
),
"enable_web_citations": (
"BOOLEAN",
{
"default": False,
"tooltip": (
"When web search is enabled, this will request that the LLM cite "
"its sources using a [REF]0[/REF] format. "
"Defaults to false."
),
},
), # NOTE: include_search_results_in_stream not implemented
"use_venice_system_prompt": (
"BOOLEAN",
{
"default": False,
"tooltip": (
"Whether to include the Venice supplied system prompts "
"alongside specified system prompts. "
"Defaults to true."
),
},
),
},
}
RETURN_TYPES = ("STRING",)
RETURN_NAMES = ("venice_parameters",)
FUNCTION = "pass_on_params"
CATEGORY = "venice.ai"
DESCRIPTION = (
"Passes on parameters unique parameters to Venice's API implementation for Venice text generation nodes."
)
def pass_on_params(
self,
character_slug,
enable_character,
strip_thinking_response,
disable_thinking,
web_search,
enable_web_citations,
use_venice_system_prompt,
):
venice_params = {
"strip_thinking_response": strip_thinking_response,
"disable_thinking": disable_thinking,
"enable_web_search": web_search,
"enable_web_citations": enable_web_citations,
"include_venice_system_prompt": use_venice_system_prompt,
}
if enable_character:
venice_params["character_slug"] = character_slug
return (venice_params,)
NODE_CLASS_MAPPINGS = {
"GenerateTextVeniceParameters_VENICE": GenerateTextVeniceParameters,
}
NODE_DISPLAY_NAME_MAPPINGS = {
"GenerateTextVeniceParameters_VENICE": "Textgen Parameters (Venice)",
}
+241
View File
@@ -0,0 +1,241 @@
import logging
from typing import Any, Dict, Iterable
from comfy_api.latest import InputImpl, io
from ..nodes.catalog_utils import video_model_specs
from ..nodes.utils import encode_tensor_for_vision, ensure_prompt_length
from ..nodes.video_utils import poll_video_until_ready, queue_video_job
LOG = logging.getLogger(__name__)
class GenerateVideoFromText(io.ComfyNode):
@staticmethod
def _option_input_id(model_id: str, field: str) -> str:
sanitized = "".join(ch if ch.isalnum() or ch in {"_", "-"} else "_" for ch in model_id)
return f"{sanitized}__{field}"
@staticmethod
def _constraint_values(value: Iterable | None) -> list[str]:
if not value:
return []
if isinstance(value, (str, bytes)):
normalized = str(value).strip()
return [normalized] if normalized else []
return [str(item) for item in value if item]
@classmethod
def _video_specs(cls) -> Dict[str, Dict[str, Any]]:
specs = video_model_specs() or {}
if not specs:
raise ValueError(
"No Venice video model specs available; refresh the catalog in VeniceAI settings and retry."
)
return specs
@classmethod
def _get_option_value(cls, model_payload: Dict[str, Any], model_id: str, field: str) -> Any:
candidates = (
cls._option_input_id(model_id, field),
field,
f"{field}__{model_id}",
f"{model_id}__{field}",
)
for key in candidates:
if key in model_payload:
return model_payload.get(key)
return None
@classmethod
def _build_model_options(cls) -> list[io.DynamicCombo.Option]:
specs = cls._video_specs()
options: list[io.DynamicCombo.Option] = []
def _sorted_models_by_group(group: str) -> list[tuple[str, Dict[str, Any]]]:
return sorted(
(
(model_id, spec)
for model_id, spec in specs.items()
if spec.get("constraints", {}).get("model_type") == group
),
key=lambda item: item[0],
)
ordered_specs = [
# didnt know about this, is same as [] + []
*_sorted_models_by_group("text-to-video"),
*_sorted_models_by_group("image-to-video"),
]
for model_id, spec in ordered_specs:
constraints = spec.get("constraints") or {}
aspect_ratios = cls._constraint_values(constraints.get("aspect_ratios"))
resolutions = cls._constraint_values(constraints.get("resolutions"))
durations = cls._constraint_values(constraints.get("durations"))
audio_default = bool(constraints.get("audio")) if constraints.get("audio") is not None else False
option_inputs: list[io.Input] = []
if constraints.get("model_type") == "image-to-video":
option_inputs.append(
io.Image.Input(
cls._option_input_id(model_id, "image"),
display_name="image",
tooltip="Source image for image-to-video models",
)
)
if durations:
option_inputs.append(
io.Combo.Input(
id=cls._option_input_id(model_id, "duration"),
display_name="duration",
options=durations,
default=durations[0],
tooltip="Duration allowed by the selected model",
)
)
if aspect_ratios:
option_inputs.append(
io.Combo.Input(
cls._option_input_id(model_id, "aspect_ratio"),
display_name="aspect_ratio",
options=aspect_ratios,
default=aspect_ratios[0],
tooltip="Aspect ratios allowed by the selected model",
)
)
if resolutions:
option_inputs.append(
io.Combo.Input(
cls._option_input_id(model_id, "resolution"),
display_name="resolution",
options=resolutions,
default=resolutions[0],
tooltip="Resolutions allowed by the selected model",
)
)
if constraints.get("audio_configurable"):
option_inputs.append(
io.Boolean.Input(
cls._option_input_id(model_id, "audio"),
display_name="audio",
default=audio_default,
tooltip="Generate audio (only when the model allows toggling)",
)
)
options.append(io.DynamicCombo.Option(model_id, option_inputs))
return options
@classmethod
def define_schema(cls) -> io.Schema:
model_options = cls._build_model_options()
return io.Schema(
node_id="TextToVideo_VENICE",
display_name="Generate Video from Text (Venice)",
category="venice.ai",
inputs=[
io.DynamicCombo.Input(
"model",
options=model_options,
tooltip="Select a Venice video model to auto-populate valid parameters",
),
io.String.Input(
"prompt",
default="A cat made of lettuce flying through space",
placeholder="Positive Prompt. Example: A cat made of lettuce flying through space",
tooltip="Text prompt to generate the video from",
multiline=True,
),
io.String.Input(
"negative_prompt",
default="low resolution, error, worst quality, low quality, defects",
placeholder="Negative Prompt",
tooltip="Negative prompt to avoid elements in the video",
multiline=True,
),
],
outputs=[
io.Video.Output(id="video", display_name="Video"),
],
)
@classmethod
def execute(
cls,
model,
prompt,
negative_prompt,
) -> io.NodeOutput:
ensure_prompt_length(prompt, 2500, label="Prompt")
ensure_prompt_length(negative_prompt, 2500, label="Negative Prompt", allow_empty=True)
if not isinstance(model, dict) or "model" not in model:
raise ValueError("Model selection is required")
model_id = model.get("model")
specs = cls._video_specs()
spec = specs.get(model_id)
if not spec:
raise ValueError("Selected model is missing from the Venice catalog; refresh the catalog and try again.")
constraints = spec.get("constraints") or {}
durations = cls._constraint_values(constraints.get("durations"))
aspect_ratios = cls._constraint_values(constraints.get("aspect_ratios"))
resolutions = cls._constraint_values(constraints.get("resolutions"))
duration = cls._get_option_value(model, model_id, "duration")
if durations:
if duration is None:
raise ValueError(f"Model {model_id} requires a duration selection")
if duration not in durations:
raise ValueError(f"Duration '{duration}' is not supported by model {model_id}")
aspect_ratio = cls._get_option_value(model, model_id, "aspect_ratio")
if aspect_ratios:
if aspect_ratio is None:
raise ValueError(f"Model {model_id} requires an aspect ratio selection")
if aspect_ratio not in aspect_ratios:
raise ValueError(f"Aspect ratio '{aspect_ratio}' is not supported by model {model_id}")
resolution = cls._get_option_value(model, model_id, "resolution")
if resolutions:
if resolution is None:
raise ValueError(f"Model {model_id} requires a resolution selection")
if resolution not in resolutions:
raise ValueError(f"Resolution '{resolution}' is not supported by model {model_id}")
audio_configurable = bool(constraints.get("audio_configurable"))
audio_default = bool(constraints.get("audio")) if constraints.get("audio") is not None else False
audio_value = cls._get_option_value(model, model_id, "audio") if audio_configurable else None
audio = audio_value if audio_value is not None else audio_default
payload = {
"model": model_id,
"prompt": prompt,
"negative_prompt": negative_prompt,
}
if durations:
payload["duration"] = duration
if aspect_ratios:
payload["aspect_ratio"] = aspect_ratio
if resolutions:
payload["resolution"] = resolution
if audio is not None:
payload["audio"] = audio
if constraints.get("model_type") == "image-to-video":
image = cls._get_option_value(model, model_id, "image")
if image is None:
raise ValueError(f"Model {model_id} requires an input image")
payload["image_url"] = encode_tensor_for_vision(image)
model_id_resp, queue_id = queue_video_job(payload)
video_path, _ = poll_video_until_ready(model=model_id_resp, queue_id=queue_id)
return io.NodeOutput(InputImpl.VideoFromFile(video_path))
+155
View File
@@ -0,0 +1,155 @@
import base64
import io as python_io
import logging
import requests
from PIL import Image
from torchvision.transforms import ToPILImage, ToTensor # type: ignore
from comfy_api.latest import io
from ..globals import API_ENDPOINTS
from ..nodes.utils import ensure_prompt_length
from ..venice_client import client
LOG = logging.getLogger(__name__)
class I2IEnhanceUpscale(io.ComfyNode):
@classmethod
def define_schema(cls) -> io.Schema:
return io.Schema(
node_id="I2IEnhanceUpscale_VENICE",
display_name="Img2Img Enhance + Upscale (Venice)",
category="venice.ai",
inputs=[
io.Image.Input(
"image",
tooltip="Image tensor to enhance or upscale",
),
io.Float.Input(
"scale",
default=2.0,
min=1.0,
max=4.0,
step=0.01,
tooltip=(
"Scale factor for upscaling the image. Valid values are 1, 2, 3, or 4.\n"
"If set to 1, the image will not be upscaled but enhanced, 'enhance' must be set to 'True'."
),
),
io.Boolean.Input(
"enhance",
default=False,
tooltip=(
"Whether to enhance the image using Venice's image engine during upscaling.\n"
"Must be set to 'True' if scale is set to 1."
),
),
io.Float.Input(
"enhance_creativity",
default=0.5,
min=0.0,
max=1.0,
step=0.01,
tooltip=(
"Higher values let the enhancement AI change the image more. "
"Setting this to 1 effectively creates an entirely new image."
),
),
io.String.Input(
"enhance_prompt",
default="",
multiline=True,
placeholder="Prompt for enhance. Example: gold, graffiti, minimalistic",
tooltip=(
"The text to image style to apply during prompt enhancement. "
"Does best with short descriptive prompts, like gold, marble or angry, menacing."
),
),
io.Float.Input(
"replication",
default=0.1,
min=0.0,
max=1.0,
step=0.01,
tooltip=(
"How strongly lines and noise in the base image are preserved. "
"Higher values are noisier but less plastic/AI 'generated'/hallucinated"
),
),
],
outputs=[io.Image.Output(id="image", display_name="Image")],
)
@classmethod
def execute(
cls,
image,
scale,
enhance,
enhance_creativity,
enhance_prompt,
replication,
) -> io.NodeOutput:
ensure_prompt_length(enhance_prompt, 1500, "Enhance prompt", allow_empty=True)
if scale == 1 and not enhance:
raise ValueError("Upscale Image (Venice) 'enhance' must be set to 'True' if scale is 1.")
if scale == 4:
LOG.info(
(
"Upscale Image (Venice) A scale of 4 with large images will result "
"in the scale being dynamically set (by venice) to ensure the "
"final image stays within the maximum size limits."
)
)
# Convert tensor to PIL Image
try:
# Get first image from batch
img_tensor = image[0].detach().cpu() # Shape: (H, W, C)
# Ensure RGB format by taking first 3 channels
if img_tensor.shape[-1] > 3:
img_tensor = img_tensor[:, :, :3]
# Convert to CHW format and create PIL Image
pil_image = ToPILImage()(img_tensor.permute(2, 0, 1))
except Exception as exc:
raise ValueError(f"Upscale Image (Venice) Failed to convert tensor to PIL image: {str(exc)}")
# Convert image to base64
byte_io = python_io.BytesIO()
pil_image.save(byte_io, format="PNG")
byte_io.seek(0)
image_base64 = base64.b64encode(byte_io.read()).decode("utf-8")
payload = {
"image": image_base64,
"scale": scale,
"enhance": enhance,
"enhanceCreativity": enhance_creativity,
"enhancePrompt": enhance_prompt,
"replication": replication,
}
response = None
try:
response = client.request(
"POST",
API_ENDPOINTS["upscale_image"],
json=payload,
headers={"Content-Type": "application/json"},
)
except requests.exceptions.RequestException as exc:
raise RuntimeError(f"Upscale Image (Venice) API request failed: {str(exc)}")
try:
upscaled_image = Image.open(python_io.BytesIO(response.content))
tensor = ToTensor()(upscaled_image) # Converts to (C, H, W)
tensor = tensor.permute(1, 2, 0) # Convert to (H, W, C)
tensor = tensor.unsqueeze(0) # Add batch dimension (1, H, W, C)
except Exception as exc:
raise ValueError(f"Upscale Image (Venice) Failed to process response image: {str(exc)}")
return io.NodeOutput(tensor)
-164
View File
@@ -1,164 +0,0 @@
import base64
import io
import logging
import os
import requests
from PIL import Image
from torchvision.transforms import ToPILImage, ToTensor # type: ignore
from ..globals import API_ENDPOINTS, VENICEAI_BASE_URL
class I2IEnhanceUpscale:
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"image": ("IMAGE",),
"scale": (
"FLOAT",
{
"default": 2,
"min": 1,
"max": 4,
"step": 0.01,
"tooltip": (
"Scale factor for upscaling the image. Valid values are 1, 2, 3, or 4.\n"
"If set to 1, the image will not be upscaled but enhanced, 'enhanced setting must be set to 'True'."
),
},
),
"enhance": (
"BOOLEAN",
{
"default": False,
"tooltip": (
"Whether to enhance the image using Venice's image engine during upscaling.\n"
"Must be set to 'True' if scale is set to 1."
),
},
),
"enhance_creativity": (
"FLOAT",
{
"default": 0.5,
"min": 0.0,
"max": 1.0,
"step": 0.01,
"tooltip": (
"Higher values let the enhancement AI change the image more. "
"Setting this to 1 effectively creates an entirely new image."
),
},
),
"enhance_prompt": (
"STRING",
{
"default": "",
"placeholder": "gold, graffiti, minimalistic",
"multiline": True,
"tooltip": (
"The text to image style to apply during prompt enhancement. "
"Does best with short descriptive prompts, like gold, marble or angry, menacing."
),
},
),
"replication": (
"FLOAT",
{
"default": 0.1,
"min": 0.0,
"max": 1.0,
"step": 0.01,
"tooltip": (
"How strongly lines and noise in the base image are preserved. "
"Higher values are noisier but less plastic/AI 'generated'/hallucinated"
),
},
),
}
}
RETURN_TYPES = ("IMAGE",)
RETURN_NAMES = ("image",)
FUNCTION = "i2i_enhance_upscale"
CATEGORY = "venice.ai"
def i2i_enhance_upscale(self, image, scale, enhance, enhance_creativity, enhance_prompt, replication):
url = VENICEAI_BASE_URL + API_ENDPOINTS["upscale_image"]
if len(enhance_prompt) > 1500:
raise ValueError("Upscale Image (Venice) enhance_prompt cannot be above 1500 characters")
if scale == 1:
raise ValueError("Upscale Image (Venice) 'enhance' must be set to 'True' if scale is 1.")
if scale == 4:
logging.info(
(
"Upscale Image (Venice) A scale of 4 with large images will result "
"in the scale being dynamically set (by venice) to ensure the "
"final image stays within the maximum size limits."
)
)
# if not api_key:
# raise ValueError("VENICEAI_API_KEY environment variable not set")
# Convert tensor to PIL Image
try:
# Get first image from batch
img_tensor = image[0].detach().cpu() # Shape: (H, W, C)
# Ensure RGB format by taking first 3 channels
if img_tensor.shape[-1] > 3:
img_tensor = img_tensor[:, :, :3]
# Convert to CHW format and create PIL Image
pil_image = ToPILImage()(img_tensor.permute(2, 0, 1))
except Exception as e:
raise ValueError(f"Upscale Image (Venice) Failed to convert tensor to PIL image: {str(e)}")
# Convert image to base64
byte_io = io.BytesIO()
pil_image.save(byte_io, format="PNG")
byte_io.seek(0)
image_base64 = base64.b64encode(byte_io.read()).decode("utf-8")
# Prepare JSON payload
payload = {
"image": image_base64,
"scale": scale,
"enhance": enhance,
"enhanceCreativity": enhance_creativity,
"enhancePrompt": enhance_prompt,
"replication": replication,
}
headers = {"Authorization": f"Bearer {os.getenv('VENICEAI_API_KEY')}", "Content-Type": "application/json"}
# Send request
try:
response = requests.post(url, json=payload, headers=headers)
response.raise_for_status()
except requests.exceptions.RequestException as e:
raise RuntimeError(f"Upscale Image (Venice) API request failed: {str(e)}")
# Convert response to tensor
try:
upscaled_image = Image.open(io.BytesIO(response.content))
tensor = ToTensor()(upscaled_image) # Converts to (C, H, W)
tensor = tensor.permute(1, 2, 0) # Convert to (H, W, C)
tensor = tensor.unsqueeze(0) # Add batch dimension (1, H, W, C)
except Exception as e:
raise ValueError(f"Upscale Image (Venice) Failed to process response image: {str(e)}")
return (tensor,)
NODE_CLASS_MAPPINGS = {
"I2IEnhanceUpscale_VENICE": I2IEnhanceUpscale,
}
NODE_DISPLAY_NAME_MAPPINGS = {
"I2IEnhanceUpscale_VENICE": "Img2Img Enhance + Upscale (Venice)",
}
-185
View File
@@ -1,185 +0,0 @@
# Todo: chat history/memory/context for LLM
# todo: inpainting
# todo: LLM characters
# todo: use variants api (currently in beta)
# import base64
# import io
# import os
# import numpy as np
# import requests
# import torch
# from PIL import Image
# from ..globals import API_ENDPOINTS, VENICEAI_BASE_URL
# from ..nodes.gen_image_base import GenerateImageBase
# todo: https://docs.comfy.org/custom-nodes/backend/more_on_inputs#dynamically-created-inputs
# for model list update?
# region inpaint
# class InpaintImage(GenerateImageBase):
# @classmethod
# def INPUT_TYPES(cls):
# return {
# "required": {
# "image": ("IMAGE",),
# "model": (
# "COMBO",
# {"default": "flux-dev"},
# ),
# "prompt": ("STRING", {"default": "A flying cat made of lettuce", "multiline": True}),
# "neg_prompt": (
# "STRING",
# {
# "placeholder": "Bad composition, rating_explicit, Text, signature, lowres, lowres, low details, faded image, out of focus, cropped, clipped, cut-off, out of frame, deserted scene, empty scene, vacant scene, desolate scene, sparse décor, bad quality, worst quality,",
# "multiline": True,
# "tooltip": "Negative prompt. This is ignored when using flux-dev or flux-dev-uncensored",
# },
# ),
# "width": (
# "INT",
# {
# "default": 1024,
# "min": 256,
# "max": 2048, # limit is 1280 but i dont want to restrict this in case of future updates, https://docs.venice.ai/api-reference/endpoint/image/generate#body-height
# "step": 32,
# "tooltip": "Must be a multiple of 32. Maximum allowed by venice.ai is 1280",
# },
# ),
# "height": (
# "INT",
# {
# "default": 1024,
# "min": 256,
# "max": 2048,
# "step": 32,
# "tooltip": "Must be a multiple of 32. Maximum allowed by venice.ai is 1280",
# },
# ),
# "batch_size": ("INT", {"default": 1, "min:": 1, "max": 4}),
# "steps": ("INT", {"default": 20, "min": 1, "max": 30}),
# "guidance": ("FLOAT", {"default": 3.0, "min": 0.1, "max": 15.0}),
# "style_preset": ("COMBO", {"default": "none"}),
# "hide_watermark": ("BOOLEAN", {"default": True}),
# "inpaint_strength": ("INT", {"default": 50, "min": 0, "max": 100}),
# },
# "optional": {"seed": ("INT", {"default": -1})},
# }
# def generate_image(
# self,
# image,
# model,
# prompt,
# neg_prompt,
# width,
# height,
# batch_size,
# steps,
# guidance,
# style_preset,
# hide_watermark,
# inpaint_strength,
# seed=-1,
# ):
# images_tensor = () # empty tuple for tensors
# if model in ["flux-dev", "flux-dev-uncensored"]:
# print(f"VeniceAPI INFO: Ignoring negative prompt for {model}.")
# neg_prompt = ""
# try:
# self.check_multiple_of_32(width, height) # todo: make this be validate node instead
# # Convert input image tensor to base64
# if image is None or image.size(0) == 0:
# raise ValueError("Input image is required for inpainting")
# # Process first image in the batch
# img_tensor = image[0].cpu() # Convert to CPU tensor
# np_image = img_tensor.numpy()
# np_image = (np_image * 255).astype(np.uint8) # Convert to 0-255 range
# # Create PIL Image and convert to base64
# pil_image = Image.fromarray(np_image, "RGB")
# buffered = io.BytesIO()
# pil_image.save(buffered, format="PNG")
# img_base64 = base64.b64encode(buffered.getvalue()).decode("utf-8")
# source_image_base64 = f"data:image/png;base64,{img_base64}"
# headers = {"Authorization": f"Bearer {os.getenv('VENICEAI_API_KEY')}", "Content-Type": "application/json"}
# url = VENICEAI_BASE_URL + API_ENDPOINTS["image_generate"]
# payload = {
# "model": model,
# "prompt": prompt,
# "negative_prompt": neg_prompt,
# "style_preset": style_preset,
# "height": height,
# "width": width,
# "steps": steps,
# "cfg_scale": guidance,
# "seed": seed,
# "return_binary": False,
# "hide_watermark": hide_watermark,
# "format": "png",
# "inpaint": {
# "strength": inpaint_strength,
# "source_image_base64": source_image_base64,
# "mask": {
# "image_prompt": "Generate a high-resolution image...",
# "object_target": "rabbit's face",
# "inferred_object": "rabbit's face wearing round silver spectacles",
# },
# },
# }
# if style_preset == "none":
# del payload["style_preset"]
# for i in range(batch_size):
# payload["seed"] = seed + i
# response = requests.request("POST", url, json=payload, headers=headers)
# if response.status_code != 200:
# raise requests.exceptions.HTTPError(
# f"HTTP error: {response.status_code}, Response: {response.text}"
# )
# images_tensor += self.process_result(response.json())
# merged = torch.cat(images_tensor, dim=0)
# return (merged,)
# except Exception as e:
# raise Exception(f"Error processing image result: {str(e)}") from e
# endregion
# region text gen
# endregion
# region upscale img
# endregion
# NODE_CLASS_MAPPINGS = {
# "InpaintImage_VENICE": InpaintImage,
# }
# NODE_DISPLAY_NAME_MAPPINGS = {
# "InpaintImage_VENICE": "Inpaint Image (Venice)",
# }
-27
View File
@@ -1,27 +0,0 @@
class CharCountTextBox:
@classmethod
def INPUT_TYPES(s):
return {
"required": {
"input_text": ("STRING", {"default": "", "multiline": True}),
}
}
CATEGORY = "venice.ai"
RETURN_TYPES = ("STRING",)
RETURN_NAMES = ("text",)
FUNCTION = "return_same_text"
def return_same_text(self, input_text):
return {"ui": {"text": input_text}, "result": (input_text,)}
NODE_CLASS_MAPPINGS = {
"CharCountTextBox": CharCountTextBox,
}
NODE_DISPLAY_NAME_MAPPINGS = {
"CharCountTextBox": "Textbox w/ char count",
}
+136
View File
@@ -0,0 +1,136 @@
from __future__ import annotations
import base64
import io
import numpy as np
import torch
from PIL import Image
from requests import RequestException
from ..globals import API_ENDPOINTS
from ..venice_client import VeniceAPIError, client
from ..venice_config import config as venice_config
def _round_down(value: int, multiple: int) -> int:
return (value // multiple) * multiple
def ensure_prompt_length(text: str, maximum: int, label: str, *, allow_empty: bool = False) -> None:
if not text:
if allow_empty:
return
raise ValueError(f"{label} is required and cannot be empty")
if len(text) > maximum:
raise ValueError(f"{label} exceeds the maximum length of {maximum} characters")
def encode_tensor_for_vision(image_tensor: torch.Tensor, *, max_encoded_bytes: int = 4_500_000) -> str:
"""
Encode a vision tensor into a Base64-encoded PNG string suitable for transmission.
Parameters
----------
image_tensor : torch.Tensor
A height×width×(3 or 4) tensor representing an image in the range [0, 1].
max_encoded_bytes : int, optional
Maximum allowed length of the Base64 payload. Defaults to 4,500,000 bytes.
Returns
-------
str
A data URI containing the PNG image encoded in Base64.
Raises
------
ValueError
If the tensor does not have 3 dimensions or the last dimension is not 3 or 4.
"""
tensor = image_tensor.detach().cpu()
if tensor.ndim == 4 and tensor.shape[0] == 1:
tensor = tensor[0]
if tensor.ndim == 3 and tensor.shape[-1] not in {3, 4} and tensor.shape[0] in {3, 4}:
tensor = tensor.permute(1, 2, 0)
if tensor.ndim != 3 or tensor.shape[-1] not in {3, 4}:
raise ValueError("Vision images must be height×width×(3 or 4 channels)")
if tensor.shape[-1] > 3:
tensor = tensor[:, :, :3]
array = (tensor.numpy() * 255).clip(0, 255).astype(np.uint8)
pil_image = Image.fromarray(array, "RGB")
original_width, original_height = pil_image.size
aspect_ratio = original_width / original_height
if original_width > original_height:
target_width = 1024
target_height = int(target_width / aspect_ratio)
if target_height < 256:
target_height = 256
target_width = int(target_height * aspect_ratio)
else:
target_height = 1024
target_width = int(target_height * aspect_ratio)
if target_width < 256:
target_width = 256
target_height = int(target_width / aspect_ratio)
target_width = _round_down(target_width, 14)
target_height = _round_down(target_height, 14)
if min(target_width, target_height) < 256:
if target_width < target_height:
target_width = _round_down(256 + 13, 14)
target_height = _round_down(int(target_width / aspect_ratio), 14)
else:
target_height = _round_down(256 + 13, 14)
target_width = _round_down(int(target_height * aspect_ratio), 14)
pil_image = pil_image.resize((target_width, target_height), Image.LANCZOS)
buffered = io.BytesIO()
pil_image.save(buffered, format="PNG")
img_base64 = base64.b64encode(buffered.getvalue()).decode("utf-8")
while len(img_base64) > max_encoded_bytes:
scaling_factor = (max_encoded_bytes / len(img_base64)) ** 0.5
new_width = max(_round_down(int(target_width * scaling_factor), 14), 256)
new_height = max(_round_down(int(target_height * scaling_factor), 14), 256)
pil_image = pil_image.resize((new_width, new_height), Image.LANCZOS)
target_width, target_height = new_width, new_height
buffered = io.BytesIO()
pil_image.save(buffered, format="PNG")
img_base64 = base64.b64encode(buffered.getvalue()).decode("utf-8")
return f"data:image/png;base64,{img_base64}"
def ensure_multiple_of(width: int, height: int, *, multiple: int = 32) -> None:
bad_dimensions = []
if width % multiple != 0:
bad_dimensions.append(f"width ({width})")
if height % multiple != 0:
bad_dimensions.append(f"height ({height})")
if bad_dimensions:
dimensions = " and ".join(bad_dimensions)
raise ValueError(f"{dimensions} must be multiples of {multiple}")
# unused right now, might be useful, or not
def ensure_api_key_valid() -> None:
key = venice_config.apikey.strip()
if not key:
raise ValueError("VeniceAI API key is missing; set it in the VeniceAI settings first.")
try:
client.get_json(API_ENDPOINTS["list_api_keys"])
except VeniceAPIError as exc:
raise ValueError("Unable to validate the VeniceAI API key.", exc) from exc
except RequestException as exc:
raise ValueError("Unable to reach VeniceAI while validating the API key.", exc) from exc
return None
+130
View File
@@ -0,0 +1,130 @@
import logging
import tempfile
import time
from typing import Optional, Tuple
from comfy.utils import ProgressBar # type: ignore
from ..globals import API_ENDPOINTS
from ..venice_client import VeniceAPIError, client
LOG = logging.getLogger(__name__)
# Defaults
POLL_INTERVAL_SECONDS = 5
MAX_POLLS = 100 # MAX_POLLS * POLL_INTERVAL_SECONDS = x minutes total wait time
PROGRESS_BAR_TOTAL = 100
def queue_video_job(payload: dict) -> Tuple[str, str]:
"""
Enqueue a Venice video generation job and return (model, queue_id).
Raises VeniceAPIError on failure.
"""
resp = client.post_json(API_ENDPOINTS["video_queue"], payload)
model = resp.get("model") or payload.get("model") or ""
queue_id = resp.get("queue_id")
if not queue_id:
raise VeniceAPIError(f"Video queue response missing queue_id: {resp}")
LOG.info("Venice video queued: model=%s queue_id=%s", model, queue_id)
return model, queue_id
def _guess_suffix(content_type: str) -> str:
if "mp4" in content_type:
return ".mp4"
if "webm" in content_type:
return ".webm"
if "quicktime" in content_type or "mov" in content_type:
return ".mov"
return ".bin"
def poll_video_until_ready(
*,
model: str,
queue_id: str,
progress_bar: Optional[ProgressBar] = None,
delete_on_completion: bool = True,
max_polls: int = MAX_POLLS,
poll_interval: float = POLL_INTERVAL_SECONDS,
) -> Tuple[str, str]:
"""
Poll Venice /video/retrieve until the video is ready.
Returns (video_filepath, queue_id). Raises VeniceAPIError on failure/timeout.
"""
pbar = progress_bar or ProgressBar(PROGRESS_BAR_TOTAL)
last_progress = 0
for attempt in range(max_polls):
time.sleep(poll_interval)
retrieve_payload = {
"model": model,
"queue_id": queue_id,
"delete_media_on_completion": delete_on_completion,
}
resp = client.request(
"POST",
API_ENDPOINTS["video_retrieve"],
json=retrieve_payload,
headers={
"Content-Type": "application/json",
"Accept": "*/*", # allow binary video or JSON status
},
)
ctype = (resp.headers or {}).get("Content-Type", "").lower()
is_status_json = ctype.startswith("application/json") or ctype.startswith("text/")
if is_status_json:
data = resp.json()
status = (data.get("status") or "UNKNOWN").upper()
exec_dur = float(data.get("execution_duration") or 0)
avg_exec = float(data.get("average_execution_time") or 0)
LOG.debug(
"Venice video status: status=%s queue_id=%s exec_ms=%s avg_ms=%s attempt=%s/%s",
status,
queue_id,
exec_dur,
avg_exec,
attempt + 1,
max_polls,
)
if avg_exec and avg_exec > 0:
progress_value = min(int(exec_dur / avg_exec * PROGRESS_BAR_TOTAL), PROGRESS_BAR_TOTAL)
else:
progress_value = min(last_progress + 1, PROGRESS_BAR_TOTAL)
delta = progress_value - last_progress
if delta > 0:
pbar.update(delta)
last_progress = progress_value
if status not in {"PROCESSING", "QUEUED"}:
raise VeniceAPIError(f"Video job reported unexpected status '{status}' for queue_id {queue_id}")
continue
# Got binary content (video)
suffix = _guess_suffix(ctype)
with tempfile.NamedTemporaryFile(delete=False, suffix=suffix) as tmp_file:
tmp_file.write(resp.content)
video_path = tmp_file.name
LOG.debug(
"Venice video ready: queue_id=%s saved_to=%s content_type=%s",
queue_id,
video_path,
ctype,
)
if last_progress < PROGRESS_BAR_TOTAL:
pbar.update(PROGRESS_BAR_TOTAL - last_progress)
return str(video_path), queue_id
raise VeniceAPIError(
f"Timed out waiting for Venice video. queue_id={queue_id} after {max_polls * poll_interval:.0f}s"
)
+1
View File
@@ -0,0 +1 @@
from . import routes # noqa: F401
-47
View File
@@ -1,47 +0,0 @@
import json
import os
from pathlib import Path
from aiohttp import web
from server import PromptServer
routes = PromptServer.instance.routes
CONFIG_FILE = Path(__file__).parent.parent / "veniceai_config.json"
def ensure_config_file_exists():
# Create the config file with a default value if it doesn't exist
if not CONFIG_FILE.exists():
default_key = "your_venice_api_key_here" # Set a default value (e.g., empty string)
set_venice_key(default_key)
def set_venice_key(apikey):
os.environ["VENICEAI_API_KEY"] = apikey
with open(CONFIG_FILE, "w") as f:
json.dump({"apikey": apikey}, f)
@routes.post("/veniceai/save_apikey")
async def post_key_server(request):
data = await request.json()
set_venice_key(data.get("apikey", ""))
return web.json_response({})
@routes.get("/veniceai/get_apikey")
async def get_key_server(request):
return web.json_response({"apikey": os.getenv("VENICEAI_API_KEY")})
ensure_config_file_exists()
if CONFIG_FILE.exists():
with open(CONFIG_FILE) as f:
saved_key = json.load(f).get("apikey", "")
set_venice_key(saved_key)
+89
View File
@@ -0,0 +1,89 @@
import asyncio
from aiohttp import web
from server import PromptServer # type: ignore
from ..venice_catalog import (
get_characters,
get_models,
get_styles,
refresh_characters,
refresh_models,
refresh_styles,
)
from ..venice_config import config as venice_config
routes = PromptServer.instance.routes
def _error_response(message: str) -> web.Response:
return web.json_response({"message": message, "error": True})
@routes.post("/veniceai/save_apikey")
async def save_key_server(request: web.Request) -> web.Response:
payload = await request.json()
venice_config.save_apikey(payload.get("apikey", ""))
return web.json_response({"message": "API key saved", "error": False})
@routes.get("/veniceai/get_apikey")
async def get_key_server(_: web.Request) -> web.Response:
return web.json_response({"apikey": venice_config.apikey})
# @routes.get("/veniceai/update_models_list")
# async def update_models_list_server(request: web.Request) -> web.Response:
# model_type = request.rel_url.query.get("type")
# try:
# payload = await asyncio.to_thread(refresh_models, model_type)
# except Exception as exc:
# return _error_response(str(exc))
# return web.json_response({"message": "Model list updated", "error": False, "data": payload})
# @routes.get("/veniceai/get_models_list")
# async def get_models_list_server(request: web.Request) -> web.Response:
# model_type = request.rel_url.query.get("type")
# try:
# payload = await asyncio.to_thread(get_models, model_type)
# except Exception as exc:
# return _error_response(str(exc))
# return web.json_response(payload)
# @routes.get("/veniceai/update_styles_list")
# async def update_styles_list_server(_: web.Request) -> web.Response:
# try:
# payload = await asyncio.to_thread(refresh_styles)
# except Exception as exc:
# return _error_response(str(exc))
# return web.json_response({"message": "Styles updated", "error": False, "data": payload})
# @routes.get("/veniceai/get_styles_list")
# async def get_styles_list_server(_: web.Request) -> web.Response:
# try:
# payload = await asyncio.to_thread(get_styles)
# except Exception as exc:
# return _error_response(str(exc))
# return web.json_response(payload)
# @routes.get("/veniceai/update_characters_list")
# async def update_characters_list_server(_: web.Request) -> web.Response:
# try:
# payload = await asyncio.to_thread(refresh_characters)
# except Exception as exc:
# return _error_response(str(exc))
# return web.json_response({"message": "Characters updated", "error": False, "data": payload})
# @routes.get("/veniceai/get_characters_list")
# async def get_characters_list_server(_: web.Request) -> web.Response:
# try:
# payload = await asyncio.to_thread(get_characters)
# except Exception as exc:
# return _error_response(str(exc))
# return web.json_response(payload)
-56
View File
@@ -1,56 +0,0 @@
import json
import os
from pathlib import Path
import requests
from aiohttp import web
from server import PromptServer # type: ignore
from ..globals import API_ENDPOINTS, VENICEAI_BASE_URL
routes = PromptServer.instance.routes
script_dir = Path(__file__).resolve().parent
data_dir = script_dir.parent / "data"
data_dir.mkdir(exist_ok=True)
characters_list_path = data_dir / "characters_list.json"
@routes.get("/veniceai/update_characters_list")
async def update_characters_list_server(request):
try:
headers = {"Authorization": f"Bearer {os.getenv('VENICEAI_API_KEY')}"}
url = f"{VENICEAI_BASE_URL}{API_ENDPOINTS['characters']}"
response = requests.get(url, headers=headers)
response.raise_for_status() # Raises HTTPError for bad responses
response_data = response.json()
# remove "object" key from response_data
response_data.pop("object", None)
response_data["data"] = sorted(response_data.get("data", []), key=lambda item: item.get("slug", ""))
with open(characters_list_path, "w") as f:
json.dump(response_data, f, indent=4)
except Exception as e:
return (f"Unexpected error: {e}", True)
response = (response_data, False)
return web.json_response({"message": response[0], "error": response[1]})
@routes.get("/veniceai/get_characters_list")
async def get_local_characters_list(request):
with open(characters_list_path, "r") as f:
characters_list_json = json.load(f)
characters = []
for item in characters_list_json.get("data", []):
if isinstance(item, dict) and item.get("slug"):
characters.append(item.get("slug"))
return web.json_response({"characters": characters})
-155
View File
@@ -1,155 +0,0 @@
import json
import os
import re
from pathlib import Path
import requests
from aiohttp import web
from server import PromptServer # type: ignore
from ..globals import API_ENDPOINTS, VENICEAI_BASE_URL
routes = PromptServer.instance.routes
script_dir = Path(__file__).resolve().parent
data_dir = script_dir.parent / "data"
data_dir.mkdir(exist_ok=True)
all_model_list_path = data_dir / "all_model_list.json"
async def fetch_models_list():
try:
headers = {"Authorization": f"Bearer {os.getenv('VENICEAI_API_KEY')}"}
url = f"{VENICEAI_BASE_URL}{API_ENDPOINTS['list_models']}"
params = {"type": "all"}
response = requests.get(url, headers=headers, params=params)
response.raise_for_status() # Raises HTTPError for bad responses
response_data = response.json()
# params = {"type": "embedding"}
# r2 = requests.get(url, headers=headers, params=params)
# r2.raise_for_status() # Raises HTTPError for bad responses
# r2_data = r2.json()
# # concatenate response_data and r2_data
# if isinstance(response_data, list) and isinstance(r2_data, list):
# response_data = response_data + r2_data
# except requests.exceptions.HTTPError as http_err:
# print(f"HTTP error occurred: {http_err}")
# except requests.exceptions.RequestException as req_err:
# print(f"Request error occurred: {req_err}")
# except (ValueError, KeyError) as data_err:
# print(f"Data parsing error: {data_err}")
except Exception as e:
return (f"Unexpected error: {e}", True)
return (response_data, False)
async def humanize_name(model_id: str) -> str:
# Handle number-letter combinations without space
model_id = re.sub(r"(\d+)([a-zA-Z]+)", lambda m: f"{m.group(1)}{m.group(2).upper()}", model_id)
# Replace remaining separators with spaces and title case
model_id = re.sub(r"[-_]", " ", model_id).title()
humanized = model_id.replace("xl", "XL").replace("vl", "VL").replace("sd", "SD").replace("llama", "LLaMA")
return humanized
@routes.get("/veniceai/update_models_list")
async def update_models_list_server(request):
# img_model_json = await fetch_model_list("image")
# txt_model_json = await fetch_model_list("text")
# merged_json = {"object": "list", "data": img_model_json.get("data", []) + txt_model_json.get("data", [])}
response = await fetch_models_list()
if response[1]: # true = error happened
return response
merged_json = response[0]
# Validating response structure
if not isinstance(merged_json, dict) or "data" not in merged_json:
raise ValueError("Unexpected API response format")
# Create lookup dictionary and enhanced JSON
enhanced_data = []
model_dict = {}
# Process all models for enhanced JSON
for model in merged_json["data"]:
model_id = model["id"]
model_spec = model.get("model_spec", {})
# Format traits excluding function_calling_default
traits = [
trait.replace("_", " ").title()
for trait in model_spec.get("traits", [])
if trait != "function_calling_default"
]
# Create humanized description
parts = [await humanize_name(model_id)]
if "availableContextTokens" in model_spec:
parts.append(f"ctx: {model_spec['availableContextTokens']}")
if traits:
parts.append(" | ".join(traits))
# Add to enhanced data
enhanced_model = model.copy()
enhanced_model["humanized"] = " | ".join(parts)
enhanced_data.append(enhanced_model)
model_dict[model_id] = enhanced_model
# Save enhanced JSON
enhanced_json = {"object": "list", "data": enhanced_data}
with open(all_model_list_path, "w") as f:
json.dump(enhanced_json, f, indent=2)
response = ("Model list update success", False)
return web.json_response({"message": response[0], "error": response[1]})
# NOTE: routes are frozen and don't update with ComfyUI-HotReloadHack
@routes.get("/veniceai/get_models_list")
async def get_local_models_list_server(request):
with open(all_model_list_path, "r") as f:
model_list_json = json.load(f)
data = model_list_json["data"]
# Create final sorted lists of names/ids
img_models = [m["id"] for m in data if m.get("type") == "image"]
txt_models = [m["id"] for m in data if m.get("type") == "text"]
tts_models = [m["id"] for m in data if m.get("type") == "tts"]
# Collect voices with humanized model name
tts_voices = []
for m in data:
if m.get("type") == "tts":
voices = m.get("model_spec", {}).get("voices", [])
# humanized = m.get("humanized", m.get("id", ""))
humanized = m.get("id", "")
tts_voices.extend([f"{humanized} - {voice}" for voice in voices])
# if tts_voices:
# print(f"TTS Voices: {', '.join(tts_voices[:100])}")
data = {
"image_models": sorted(img_models),
"text_models": sorted(txt_models),
"tts_models": sorted(tts_models),
"tts_voices": sorted(tts_voices),
"model_list_json": model_list_json,
}
return web.json_response(data)
-58
View File
@@ -1,58 +0,0 @@
import json
import os
from pathlib import Path
import requests
from aiohttp import web
from server import PromptServer # type: ignore
from ..globals import API_ENDPOINTS, VENICEAI_BASE_URL
routes = PromptServer.instance.routes
script_dir = Path(__file__).resolve().parent
data_dir = script_dir.parent / "data"
data_dir.mkdir(exist_ok=True)
styles_list_path = data_dir / "styles_list.json"
@routes.get("/veniceai/update_styles_list")
async def update_styles_list_server(request):
try:
headers = {"Authorization": f"Bearer {os.getenv('VENICEAI_API_KEY')}"}
url = f"{VENICEAI_BASE_URL}{API_ENDPOINTS['list_styles']}"
response = requests.get(url, headers=headers)
response.raise_for_status() # Raises HTTPError for bad responses
response_data = response.json()
# remove "object" key from response_data
response_data.pop("object", None)
response_data["data"] = sorted(response_data.get("data", []))
response_data["data"].insert(0, "none")
with open(styles_list_path, "w") as f:
json.dump(response_data, f, indent=2)
# except requests.exceptions.HTTPError as http_err:
# print(f"HTTP error occurred: {http_err}")
# except requests.exceptions.RequestException as req_err:
# print(f"Request error occurred: {req_err}")
# except (ValueError, KeyError) as data_err:
# print(f"Data parsing error: {data_err}")
except Exception as e:
return (f"Unexpected error: {e}", True)
response = (response_data, False)
return web.json_response({"message": response[0], "error": response[1]})
@routes.get("/veniceai/get_styles_list")
async def get_local_styles_list(requests):
with open(styles_list_path, "r") as f:
styles_list_json = json.load(f)
return web.json_response(styles_list_json)
+4 -1
View File
@@ -1,4 +1,7 @@
requests>=2.31.0
Pillow>=10.0.0
numpy>=1.24.0
configparser>=5.3.0
configparser>=5.3.0
torchaudio
torchvision
imageio-ffmpeg
+343
View File
@@ -0,0 +1,343 @@
import json
import logging
import os
import time
from pathlib import Path
from typing import Any, Callable, Dict, Iterable, List, Optional
from .venice_client import VeniceAPIError, VeniceClient
DATA_DIR = Path(__file__).with_name("data")
DATA_DIR.mkdir(exist_ok=True)
class CatalogStore:
def __init__(
self,
name: str,
default_factory: Optional[Callable[[], Dict[str, Any]]] = None,
legacy_name: Optional[str] = None,
) -> None:
self.file = DATA_DIR / name
self._default_factory = default_factory or (lambda: {"object": "list", "data": []})
self._legacy_name = legacy_name
def load(self) -> Dict[str, Any]:
if not self.file.exists():
return self._load_legacy() or self._default_factory()
try:
with self.file.open("r", encoding="utf-8") as fp:
return json.load(fp)
except (ValueError, json.JSONDecodeError):
return self._load_legacy() or self._default_factory()
def save(self, payload: Dict[str, Any]) -> None:
with self.file.open("w", encoding="utf-8") as fp:
json.dump(payload, fp, indent=2)
def _load_legacy(self) -> Dict[str, Any] | None:
if not self._legacy_name:
return None
legacy_file = self.file.with_name(self._legacy_name)
if not legacy_file.exists():
return None
try:
with legacy_file.open("r", encoding="utf-8") as fp:
return json.load(fp)
except (ValueError, json.JSONDecodeError):
return None
def filter_by_type(self, payload: Dict[str, Any], *types: str) -> List[Dict[str, Any]]:
data = payload.get("data") or []
return [entry for entry in data if entry.get("type") in types]
_model_store = CatalogStore("models_list.json", legacy_name="all_model_list.json")
_styles_store = CatalogStore(
"styles_list.json",
lambda: {"data": ["none"], "object": "list"},
legacy_name="styles_list.json",
)
_characters_store = CatalogStore(
"characters_list.json",
lambda: {"data": [], "object": "list"},
legacy_name="characters_list.json",
)
_client = VeniceClient()
# to prevent spamming api.
# can be bypassed by setting force_refresh=True in get_* calls
_CACHE_TTL = float(os.environ.get("VENICE_CATALOG_TTL", "900"))
_models_last_refresh = 0.0
_styles_last_refresh = 0.0
_characters_last_refresh = 0.0
LOG = logging.getLogger(__name__)
def _enrich_models(raw: Dict[str, Any]) -> Dict[str, Any]:
return raw.copy()
def _extract_video_models(payload: Dict[str, Any]) -> Dict[str, Any]:
"""
Extract Venice video models and expose a by-id mapping of their constraints.
The mapping is intentionally lightweight (id, name, model_type, constraints) so downstream
callers (nodes) can build DynamicCombo inputs without re-parsing the raw payload.
"""
data = payload.get("data", []) or []
by_id: Dict[str, Any] = {}
for model in data:
if model.get("type") != "video":
continue
model_id = model.get("id")
if not model_id:
continue
model_spec = model.get("model_spec") or {}
constraints = model_spec.get("constraints") or {}
model_type = constraints.get("model_type")
if not isinstance(constraints, dict):
constraints = {}
entry = {
"id": model_id,
"name": model_spec.get("name"),
"model_type": model_type,
"constraints": {
"aspect_ratios": constraints.get("aspect_ratios") or [],
"resolutions": constraints.get("resolutions") or [],
"durations": constraints.get("durations") or [],
"audio": constraints.get("audio"),
"audio_configurable": constraints.get("audio_configurable"),
"model_type": model_type,
},
"raw": model,
}
by_id[model_id] = entry
return by_id
def _extract_image_models(payload: Dict[str, Any]) -> Dict[str, Any]:
data = payload.get("data", []) or []
by_id: Dict[str, Any] = {}
for model in data:
if model.get("type") != "image":
continue
model_id = model.get("id")
if not model_id:
continue
model_spec = model.get("model_spec") or {}
constraints = model_spec.get("constraints") or {}
if not isinstance(constraints, dict):
constraints = {}
entry = {
"id": model_id,
"name": model_spec.get("name"),
"constraints": dict(constraints),
"raw": model,
}
by_id[model_id] = entry
return by_id
def _extract_text_models(payload: Dict[str, Any]) -> Dict[str, Any]:
data = payload.get("data", []) or []
by_id: Dict[str, Any] = {}
for model in data:
if model.get("type") != "text":
continue
model_id = model.get("id")
if not model_id:
continue
model_spec = model.get("model_spec") or {}
constraints = model_spec.get("constraints") or {}
capabilities = model_spec.get("capabilities") or {}
if not isinstance(constraints, dict):
constraints = {}
if not isinstance(capabilities, dict):
capabilities = {}
entry = {
"id": model_id,
"name": model_spec.get("name"),
"constraints": dict(constraints),
"capabilities": dict(capabilities),
"raw": model,
}
by_id[model_id] = entry
return by_id
def _should_refresh(last_refresh: float) -> bool:
if _CACHE_TTL <= 0:
return False
return (time.monotonic() - last_refresh) > _CACHE_TTL
def _record_models_refresh() -> None:
global _models_last_refresh
_models_last_refresh = time.monotonic()
def _record_styles_refresh() -> None:
global _styles_last_refresh
_styles_last_refresh = time.monotonic()
def _record_characters_refresh() -> None:
global _characters_last_refresh
_characters_last_refresh = time.monotonic()
def _should_attempt_refresh(payload: Dict[str, Any], should_refresh: bool) -> bool:
if not should_refresh:
return False
if _client.dry_run and payload.get("data"):
LOG.debug("Dry-run mode skipping catalog refresh because cached data is available.")
return False
return True
def refresh_models(model_type: str = "all") -> Dict[str, Any]:
try:
raw = _client.list_models(model_type or "all")
except VeniceAPIError as exc:
LOG.error("Failed to refresh Venice model list: %s", exc)
raise
enriched = _enrich_models(raw)
if not enriched.get("data"):
raise VeniceAPIError("Venice returned an empty model catalog")
_model_store.save(enriched)
_record_models_refresh()
return enriched
def refresh_styles() -> Dict[str, Any]:
try:
raw = _client.list_styles()
except VeniceAPIError as exc:
LOG.error("Failed to refresh Venice styles list: %s", exc)
raise
data = raw.get("data")
if not data:
raise VeniceAPIError("Venice returned an empty styles catalog")
sorted_data = sorted(data)
sorted_data.insert(0, "none")
payload = {"object": raw.get("object", "list"), "data": sorted_data}
_styles_store.save(payload)
_record_styles_refresh()
return payload
def refresh_characters() -> Dict[str, Any]:
try:
raw = _client.list_characters()
except VeniceAPIError as exc:
LOG.error("Failed to refresh Venice characters list: %s", exc)
raise
items = raw.get("data")
if not items:
raise VeniceAPIError("Venice returned an empty characters catalog")
sorted_items = sorted(items, key=lambda item: item.get("slug", "")) if isinstance(items, Iterable) else []
payload = {"object": raw.get("object", "list"), "data": sorted_items}
_characters_store.save(payload)
_record_characters_refresh()
return payload
def get_models(model_type: Optional[str] = None, *, force_refresh: bool = False) -> Dict[str, Any]:
"""
Return model metadata from the Venice catalog, optionally filtered by type.
Parameters
----------
model_type : Optional[str]
If provided, filters models by the requested type (e.g., "image", "text", "tts") and
returns only that subset; when omitted, the response contains categorized lists
of model IDs, available TTS voices, and the raw payload.
Returns
-------
Dict[str, Any]
The filtered or fully categorized model information, keyed by categories such as
"image_models", "text_models", "tts_models", "tts_voices", and "model_list_json".
When a model_type filter is applied, returns a single "models" key with matching entries.
"""
payload = _model_store.load()
should_refresh = force_refresh or not payload.get("data") or _should_refresh(_models_last_refresh)
if _should_attempt_refresh(payload, should_refresh):
payload = refresh_models()
if model_type:
return {"models": _model_store.filter_by_type(payload, model_type)}
video_models_by_id = _extract_video_models(payload)
image_models_by_id = _extract_image_models(payload)
text_models_by_id = _extract_text_models(payload)
# todo: maybe dataclass is better for this
filtered = {
# todo: the *_by_id might be enough so this stuff below can be removed
"image_models": sorted([m.get("id") for m in payload.get("data", []) if m.get("type") == "image"]),
"text_models": sorted([m.get("id") for m in payload.get("data", []) if m.get("type") == "text"]),
# todo: voices should be linked to models like in models json
"tts_models": sorted([m.get("id") for m in payload.get("data", []) if m.get("type") == "tts"]),
"tts_voices": sorted(
[
f"{m.get('id', '')} - {voice}"
for m in payload.get("data", [])
if m.get("type") == "tts"
for voice in (m.get("model_spec", {}).get("voices") or [])
]
),
"text2video_models": sorted(
[
model_id
for model_id, spec in video_models_by_id.items()
if spec.get("constraints", {}).get("model_type") == "text-to-video"
]
),
"image2video_models": sorted(
[
model_id
for model_id, spec in video_models_by_id.items()
if spec.get("constraints", {}).get("model_type") == "image-to-video"
]
),
"video_models_by_id": video_models_by_id,
"image_models_by_id": image_models_by_id,
"model_list_json": payload,
"text_models_by_id": text_models_by_id,
}
return filtered
def get_styles(*, force_refresh: bool = False) -> Dict[str, Any]:
payload = _styles_store.load()
should_refresh = force_refresh or not payload.get("data") or _should_refresh(_styles_last_refresh)
if _should_attempt_refresh(payload, should_refresh):
payload = refresh_styles()
return payload
def get_characters(*, force_refresh: bool = False) -> Dict[str, Any]:
payload = _characters_store.load()
should_refresh = force_refresh or not payload.get("data") or _should_refresh(_characters_last_refresh)
if _should_attempt_refresh(payload, should_refresh):
payload = refresh_characters()
data = payload.get("data", [])
characters = [item.get("slug") for item in data if isinstance(item, dict) and item.get("slug")]
return {"characters": characters}
+156
View File
@@ -0,0 +1,156 @@
from __future__ import annotations
import logging
import os
from typing import Any, Dict, Mapping, Optional
import requests
from .globals import API_ENDPOINTS, USER_AGENT, VENICEAI_BASE_URL
from .venice_config import config as venice_config
LOG = logging.getLogger(__name__)
if os.environ.get("VENICE_CLIENT_DEBUG", "").lower() in {"1", "true"}:
LOG.setLevel(logging.DEBUG)
class VeniceAPIError(Exception):
pass
class DummyResponse:
def __init__(self, payload: dict | None = None, status_code: int = 200) -> None:
self._payload = payload or {}
self.status_code = status_code
def raise_for_status(self) -> None:
return None
def json(self) -> dict:
return self._payload
@property
def text(self) -> str:
return ""
def __getattr__(self, item: str) -> Any:
return None
class VeniceClient:
def __init__(self) -> None:
self._session = requests.Session()
self._session.headers.update({"User-Agent": USER_AGENT})
self._dry_run = os.environ.get("VENICE_CLIENT_DRY_RUN", "").lower() in {"1", "true"}
@property
def dry_run(self) -> bool:
return self._dry_run
def _ensure_api_key(self) -> str:
key = venice_config.apikey.strip()
if not key:
raise VeniceAPIError(
"VeniceAI API key is missing. Set it via the VeniceAI settings before using the nodes."
)
return key
def _build_headers(self, extra: Optional[Mapping[str, str]] = None) -> Dict[str, str]:
headers: Dict[str, str] = {"Authorization": f"Bearer {self._ensure_api_key()}"}
if extra:
headers.update(extra)
return headers
def _friendly_status_hint(self, status_code: int) -> str:
if status_code == 401:
return "The API key may be invalid, expired, or lack permissions."
if status_code == 404:
return "The requested VeniceAI endpoint was not found. Ensure the node and catalog data are up to date."
if status_code == 429:
return "Request rate limits were hit. Wait a moment before retrying."
if 500 <= status_code < 600:
return "VeniceAI appears to be experiencing server issues; try again in a bit."
return "Check your request parameters and ensure your API key is valid."
def _friendly_network_hint(self) -> str:
return "Unable to reach VeniceAI. Confirm your internet connection and that api.venice.ai is reachable."
def request(self, method: str, endpoint: str, **kwargs: Any) -> requests.Response:
if self._dry_run:
logging.debug("VeniceClient dry run skipping %s %s", method, endpoint)
return DummyResponse()
url = VENICEAI_BASE_URL + endpoint
headers = kwargs.pop("headers", None)
kwargs.setdefault("timeout", 30)
try:
response = self._session.request(method, url, headers=self._build_headers(headers), **kwargs)
response.raise_for_status()
except requests.HTTPError as exc:
self._log_response(method, endpoint, exc.response)
hint = (
self._friendly_status_hint(exc.response.status_code)
if exc.response is not None
else "Unexpected response from VeniceAI."
)
message = f"Venice request failed ({method} {endpoint}): {exc}. \n{hint}"
LOG.debug("Venice request failed: %s %s %s", method, endpoint, exc)
raise VeniceAPIError(message) from exc
except requests.RequestException as exc:
message = f"{self._friendly_network_hint()} Details: {exc}"
LOG.debug("Venice network error: %s %s %s", method, endpoint, exc)
raise VeniceAPIError(message) from exc
self._log_response(method, endpoint, response)
return response
def _log_response(self, method: str, endpoint: str, response: Optional[requests.Response | DummyResponse]) -> None:
if not LOG.isEnabledFor(logging.DEBUG) or response is None:
return
headers = getattr(response, "headers", None)
header_snapshot: Dict[str, Any] = dict(headers) if headers else {}
content_type = (header_snapshot.get("Content-Type") or "").lower()
body = ""
readable_body = content_type.startswith("application/json") or content_type.startswith("text/")
if readable_body or not content_type:
try:
body = response.text or ""
except Exception as exc:
body = f"<unable to read body: {exc}>"
if len(body) > 2000:
body = body[:2000] + "...[truncated]"
else:
body = f"<{content_type} response body omitted>"
LOG.debug(
"Venice response %s %s status=%s headers=%s body=%s",
method,
endpoint,
getattr(response, "status_code", "<?>"),
header_snapshot,
body,
)
def post_json(self, endpoint: str, payload: Mapping[str, Any], **kwargs: Any) -> Dict[str, Any]:
headers = {"Content-Type": "application/json"}
headers.update(kwargs.pop("headers", {}))
response = self.request("POST", endpoint, json=payload, headers=headers, **kwargs)
return response.json()
def get_json(self, endpoint: str, params: Optional[Mapping[str, Any]] = None) -> Dict[str, Any]:
response = self.request("GET", endpoint, params=params)
return response.json()
def list_models(self, model_type: Optional[str] = None) -> Dict[str, Any]:
params = {"type": model_type} if model_type else None
return self.get_json(API_ENDPOINTS["list_models"], params=params)
def list_styles(self) -> Dict[str, Any]:
return self.get_json(API_ENDPOINTS["list_styles"])
def list_characters(self) -> Dict[str, Any]:
return self.get_json(API_ENDPOINTS["characters"])
client = VeniceClient()
+57
View File
@@ -0,0 +1,57 @@
from __future__ import annotations
import json
import os
from pathlib import Path
from typing import Any, Dict
DEFAULT_CONFIG: Dict[str, Any] = {"apikey": ""}
class VeniceConfig:
_instance: "VeniceConfig" | None = None
def __new__(cls) -> "VeniceConfig":
if cls._instance is None:
cls._instance = super().__new__(cls)
cls._instance._initialized = False
return cls._instance
def __init__(self) -> None:
if self._initialized:
return
self._initialized = True
self.path = Path(__file__).with_name("veniceai_config.json")
self.path.parent.mkdir(parents=True, exist_ok=True)
self._data = self._read()
self._sync_env()
def _read(self) -> Dict[str, Any]:
if not self.path.exists():
self._write(DEFAULT_CONFIG)
return DEFAULT_CONFIG.copy()
try:
with self.path.open("r", encoding="utf-8") as fp:
return json.load(fp)
except (ValueError, json.JSONDecodeError):
self._write(DEFAULT_CONFIG)
return DEFAULT_CONFIG.copy()
def _write(self, data: Dict[str, Any]) -> None:
with self.path.open("w", encoding="utf-8") as fp:
json.dump(data, fp, indent=2)
def _sync_env(self) -> None:
os.environ.setdefault("VENICEAI_API_KEY", self._data.get("apikey", ""))
@property
def apikey(self) -> str:
return self._data.get("apikey", "") or ""
def save_apikey(self, key: str) -> None:
self._data["apikey"] = key
self._write(self._data)
os.environ["VENICEAI_API_KEY"] = key
config = VeniceConfig()
-310
View File
@@ -1,310 +0,0 @@
{
"last_node_id": 22,
"last_link_id": 6,
"nodes": [
{
"id": 20,
"type": "SaveImage",
"pos": {
"0": 645,
"1": 514
},
"size": {
"0": 315,
"1": 270
},
"flags": {},
"order": 4,
"mode": 0,
"inputs": [
{
"name": "images",
"type": "IMAGE",
"link": 6
}
],
"outputs": [],
"properties": {},
"widgets_values": [
"ComfyUIPro"
]
},
{
"id": 15,
"type": "SaveImage",
"pos": {
"0": 641,
"1": 913
},
"size": {
"0": 315,
"1": 270
},
"flags": {},
"order": 6,
"mode": 0,
"inputs": [
{
"name": "images",
"type": "IMAGE",
"link": 5
}
],
"outputs": [],
"properties": {},
"widgets_values": [
"ComfyUIP11"
]
},
{
"id": 5,
"type": "SaveImage",
"pos": {
"0": 624,
"1": 91
},
"size": {
"0": 315,
"1": 270
},
"flags": {},
"order": 5,
"mode": 0,
"inputs": [
{
"name": "images",
"type": "IMAGE",
"link": 4
}
],
"outputs": [],
"properties": {},
"widgets_values": [
"ComfyUIDev"
]
},
{
"id": 9,
"type": "FluxPro_TOGETHER",
"pos": {
"0": 116,
"1": 513
},
"size": {
"0": 400,
"1": 292
},
"flags": {},
"order": 0,
"mode": 0,
"inputs": [],
"outputs": [
{
"name": "IMAGE",
"type": "IMAGE",
"links": [
6
]
}
],
"properties": {
"Node name for S&R": "FluxPro_TOGETHER"
},
"widgets_values": [
"The image showcases a striking 3D CGI representation of a stylized angry muscle Doberman, characterized by a sleek black and silver coat that glistens under the soft lighting, enhancing its muscular physique. The dog is prominently positioned in the center of the composition against a stark black background, which accentuates its fierce expression and well-defined features. Its piercing eyes are shielded by trendy green sunglasses, adding a fashionable flair that contrasts with the dark tones of its fur. The meticulous detailing highlights the texture of the Doberman’s coat, revealing subtle variations in color that suggest a glossy sheen. Surrounding the Doberman, hints of a countryside landscape can be inferred in the background, possibly indicating an advertising theme, although it remains blurred and understated to maintain focus on the dog itself. The overall quality of the image is exceptional, marked by ultra-detailed craftsmanship that embodies a masterpiece in high-resolution, making it a captivating visual spectacle.\n\n",
1024,
1024,
28,
true,
"2",
2.5,
2,
324,
"randomize"
]
},
{
"id": 6,
"type": "FluxDev_TOGETHER",
"pos": {
"0": 112,
"1": 91
},
"size": {
"0": 400,
"1": 268
},
"flags": {},
"order": 1,
"mode": 0,
"inputs": [],
"outputs": [
{
"name": "IMAGE",
"type": "IMAGE",
"links": [
4
]
}
],
"properties": {
"Node name for S&R": "FluxDev_TOGETHER"
},
"widgets_values": [
"The image showcases a striking 3D CGI representation of a stylized angry muscle Doberman, characterized by a sleek black and silver coat that glistens under the soft lighting, enhancing its muscular physique. The dog is prominently positioned in the center of the composition against a stark black background, which accentuates its fierce expression and well-defined features. Its piercing eyes are shielded by trendy green sunglasses, adding a fashionable flair that contrasts with the dark tones of its fur. The meticulous detailing highlights the texture of the Doberman’s coat, revealing subtle variations in color that suggest a glossy sheen. Surrounding the Doberman, hints of a countryside landscape can be inferred in the background, possibly indicating an advertising theme, although it remains blurred and understated to maintain focus on the dog itself. The overall quality of the image is exceptional, marked by ultra-detailed craftsmanship that embodies a masterpiece in high-resolution, making it a captivating visual spectacle.\n\n",
1024,
1024,
4,
true,
"1",
2.5,
675,
"randomize"
]
},
{
"id": 14,
"type": "FluxPro11_TOGETHER",
"pos": {
"0": 115,
"1": 912
},
"size": {
"0": 400,
"1": 244
},
"flags": {},
"order": 2,
"mode": 0,
"inputs": [],
"outputs": [
{
"name": "IMAGE",
"type": "IMAGE",
"links": [
5
]
}
],
"properties": {
"Node name for S&R": "FluxPro11_TOGETHER"
},
"widgets_values": [
"The image showcases a striking 3D CGI representation of a stylized angry muscle Doberman, characterized by a sleek black and silver coat that glistens under the soft lighting, enhancing its muscular physique. The dog is prominently positioned in the center of the composition against a stark black background, which accentuates its fierce expression and well-defined features. Its piercing eyes are shielded by trendy green sunglasses, adding a fashionable flair that contrasts with the dark tones of its fur. The meticulous detailing highlights the texture of the Doberman’s coat, revealing subtle variations in color that suggest a glossy sheen. Surrounding the Doberman, hints of a countryside landscape can be inferred in the background, possibly indicating an advertising theme, although it remains blurred and understated to maintain focus on the dog itself. The overall quality of the image is exceptional, marked by ultra-detailed craftsmanship that embodies a masterpiece in high-resolution, making it a captivating visual spectacle.\n",
1024,
1024,
true,
1,
"2",
1148,
"randomize"
]
},
{
"id": 22,
"type": "Fast Groups Muter (rgthree)",
"pos": {
"0": -142,
"1": 36
},
"size": {
"0": 226.8000030517578,
"1": 130
},
"flags": {},
"order": 3,
"mode": 0,
"inputs": [],
"outputs": [
{
"name": "OPT_CONNECTION",
"type": "*",
"links": null
}
],
"properties": {
"matchColors": "",
"matchTitle": "",
"showNav": true,
"sort": "position",
"customSortAlphabet": "",
"toggleRestriction": "default"
}
}
],
"links": [
[
4,
6,
0,
5,
0,
"IMAGE"
],
[
5,
14,
0,
15,
0,
"IMAGE"
],
[
6,
9,
0,
20,
0,
"IMAGE"
]
],
"groups": [
{
"title": "Flux Pro",
"bounding": [
102,
440,
872,
376
],
"color": "#8A8",
"font_size": 24,
"flags": {}
},
{
"title": "Flux Dev",
"bounding": [
103,
11,
871,
407
],
"color": "#A88",
"font_size": 24,
"flags": {}
},
{
"title": "Flux Pro 1.1",
"bounding": [
100,
834,
879,
373
],
"color": "#444",
"font_size": 24,
"flags": {}
}
],
"config": {},
"extra": {
"ds": {
"scale": 0.8264462809917354,
"offset": [
442.1956952821454,
170.1053343944367
]
}
},
"version": 0.4
}