diff --git a/README.md b/README.md index 4194a4f..49889a3 100644 --- a/README.md +++ b/README.md @@ -7,9 +7,10 @@ A Custom Node that integrates [Fal.ai's](https://fal.ai) powerful image editing ## ✨ Features * **Multi-Model Support:** Access multiple editing models from a single node. -* **Multi-Image Input:** Support for up to 5 input images for context-aware editing (model dependent). +* **Multi-Image Input:** Support for up to 10 input images for context-aware editing (model dependent). * **Flexible Output:** Control aspect ratios, resolution, and batch size. -* **Cloud Inference:** Offloads heavy GPU processing to Fal.ai's cloud infrastructure. +* **Asynchronous Queue:** Every model uses fal.ai's durable queue with status polling and automatic provider-side retries. +* **Bounded Concurrency:** Submit up to 8 independent runs in parallel while preserving output order. ## 🚀 Supported Models @@ -18,6 +19,7 @@ A Custom Node that integrates [Fal.ai's](https://fal.ai) powerful image editing | **Nano Banana** | `nano_banana` | Fast, efficient editing. | | **Nano Banana Pro** | `nano_banana_pro` | Enhanced quality version of Nano Banana. | | **Nano Banana 2** | `nano_banana_2` | New generation model with web search and thinking options. | +| **GPT Image 2 Edit** | `gpt_image_2_edit` | OpenAI image editing with quality and mask controls. | | **Grok Imagine Edit** | `grok_imagine_edit` | xAI image edit model, up to 3 input images. | | **Seedream 4.5** | `seedream_4.5` | ByteDance's model. Optimized for high-res editing. | | **Seedream 5 Pro Edit** | `seedream_5` | ByteDance Seedream 5.0 Pro region-precise editing. | @@ -46,15 +48,16 @@ A Custom Node that integrates [Fal.ai's](https://fal.ai) powerful image editing You must have a **Fal.ai API Key** to use this node. 1. Go to [fal.ai](https://fal.ai/dashboard/keys) and generate an API Key. -2. Enter this key into the `fal_key` widget on the node. +2. Set `FAL_KEY` in the environment running ComfyUI, or enter the key in the `fal_key` widget. The environment variable takes priority. ## 🎛️ Usage ### Inputs * **prompt**: (Required) A text description of how to edit the image. * **model**: Select the specific AI model to use. -* **fal_key**: Your API key. -* **image1 - image5**: Connect the images you want to edit. +* **fal_key**: Optional UI fallback for your API key when `FAL_KEY` is not set. +* **image1 - image10**: Connect the images you want to edit. +* **mask**: Optional GPT Image 2 edit mask. Other models ignore it. ### Settings * **width / height**: Output resolution. @@ -66,6 +69,8 @@ You must have a **Fal.ai API Key** to use this node. * **aspect_ratio**: Set target aspect ratio (e.g., 16:9, 1:1). * **enable_web_search**: Enables web search context for supported models. * **thinking_level**: `off`, `minimal`, or `high` for supported models. +* **concurrent_runs**: Number of independent fal.ai queue jobs to submit in parallel. Each run requests its own `num_images`; outputs are returned in run order. Models with an explicit seed increment it for each additional run. +* **queue_timeout**: Maximum wait per queue job in seconds. Timed-out jobs receive a best-effort cancellation request. ### ⚠️ Model Specific Constraints @@ -106,7 +111,8 @@ Different models have different requirements implemented in the node: ## 🛠️ Troubleshooting -* **"No images returned from API"**: Check your `fal_key` balance or ensure your prompt is not triggering safety filters. +* **Queue failures**: The node reports the failed run number and fal request ID. Check account balance, model access, input validation, and safety filters. +* **Queue timeout**: Increase `queue_timeout` when fal.ai is busy or when using slower high-resolution models. * **Resolution Errors**: If using Seedream, ensure your resolution is set high enough (min 1920x1920). If using Flux, ensure it is within 512-2048. *Star this repo if it helps your workflow! 🚀* diff --git a/nano_seed.py b/nano_seed.py index 1c6a28e..23e8914 100644 --- a/nano_seed.py +++ b/nano_seed.py @@ -1,4 +1,8 @@ import os +import re +import time +from concurrent.futures import ThreadPoolExecutor, as_completed + import torch import requests import base64 @@ -6,6 +10,12 @@ import numpy as np from io import BytesIO from PIL import Image +FAL_RUN_URL_PREFIX = "https://fal.run/" +FAL_QUEUE_URL_PREFIX = "https://queue.fal.run/" +FAL_HTTP_TIMEOUT_SECONDS = 60 +FAL_QUEUE_POLL_INTERVAL_SECONDS = 1.0 +FAL_MAX_CONCURRENT_RUNS = 8 + SEEDREAM_5_AUTO_IMAGE_SIZE_BY_RESOLUTION = { "seedream_5": { "0.5K": "auto_1K", @@ -129,6 +139,213 @@ def pil2data_uri(pil_image): img_str = base64.b64encode(buffer.getvalue()).decode() return f"data:image/png;base64,{img_str}" + +def _safe_error_detail(detail): + detail = re.sub( + r"data:[^,\s\"']+,[A-Za-z0-9+/=]+", + "data:[redacted]", + str(detail or ""), + ).strip() + if len(detail) > 1000: + return f"{detail[:1000]}..." + return detail + + +def _response_json(response, action): + if not 200 <= response.status_code < 300: + detail = _safe_error_detail(response.text) + suffix = f": {detail}" if detail else "" + raise ValueError(f"{action} failed with HTTP {response.status_code}{suffix}") + + try: + return response.json() + except ValueError as error: + raise ValueError(f"{action} returned invalid JSON.") from error + + +def _queue_endpoint(run_url): + if not run_url.startswith(FAL_RUN_URL_PREFIX): + raise ValueError(f"Unsupported fal.ai endpoint: {run_url}") + return f"{FAL_QUEUE_URL_PREFIX}{run_url[len(FAL_RUN_URL_PREFIX):]}" + + +def _queue_operation_url(value, fallback): + if not value: + return fallback + if value.startswith("http://") or value.startswith("https://"): + return value + return f"{FAL_QUEUE_URL_PREFIX.rstrip('/')}/{value.lstrip('/')}" + + +def _cancel_queue_request(cancel_url, headers): + try: + requests.put(cancel_url, headers=headers, timeout=10) + except requests.RequestException: + pass + + +def _submit_fal_queue_request(run_url, payload, headers, queue_timeout, run_index, run_count): + queue_url = _queue_endpoint(run_url) + try: + submit_response = requests.post( + queue_url, + json=payload, + headers=headers, + timeout=FAL_HTTP_TIMEOUT_SECONDS, + ) + except requests.RequestException as error: + raise ValueError(f"fal.ai queue submission failed: {error}") from error + + submit_data = _response_json(submit_response, "fal.ai queue submission") + request_id = submit_data.get("request_id") + if not request_id: + raise ValueError("fal.ai queue submission returned no request_id.") + + request_base_url = f"{queue_url}/requests/{request_id}" + status_url = _queue_operation_url( + submit_data.get("status_url"), + f"{request_base_url}/status", + ) + response_url = _queue_operation_url( + submit_data.get("response_url"), + request_base_url, + ) + cancel_url = _queue_operation_url( + submit_data.get("cancel_url"), + f"{request_base_url}/cancel", + ) + deadline = time.monotonic() + queue_timeout + run_label = f"{run_index + 1}/{run_count}" + print(f"NanoSeed: queued async run {run_label} ({request_id}).") + + while True: + if time.monotonic() >= deadline: + _cancel_queue_request(cancel_url, headers) + raise TimeoutError( + f"fal.ai queue run {run_label} timed out after {queue_timeout} seconds " + f"(request_id: {request_id})." + ) + + try: + status_response = requests.get( + status_url, + headers=headers, + timeout=FAL_HTTP_TIMEOUT_SECONDS, + ) + except requests.RequestException: + time.sleep(FAL_QUEUE_POLL_INTERVAL_SECONDS) + continue + + if status_response.status_code in {408, 425, 429, 500, 502, 503, 504}: + time.sleep(FAL_QUEUE_POLL_INTERVAL_SECONDS) + continue + + status_data = _response_json(status_response, "fal.ai queue status") + status = status_data.get("status") + if status == "COMPLETED": + error_message = _safe_error_detail(status_data.get("error")) + if error_message: + error_type = status_data.get("error_type") + error_suffix = f" ({error_type})" if error_type else "" + raise ValueError( + f"fal.ai queue run {run_label} failed{error_suffix}: {error_message}" + ) + response_url = _queue_operation_url( + status_data.get("response_url"), + response_url, + ) + break + if status in {"FAILED", "CANCELLED", "CANCELED"}: + error_message = _safe_error_detail(status_data.get("error")) or "Unknown queue error" + raise ValueError(f"fal.ai queue run {run_label} failed: {error_message}") + if status not in {"IN_QUEUE", "IN_PROGRESS"}: + raise ValueError( + f"fal.ai queue run {run_label} returned unknown status: {status!r}" + ) + + time.sleep(FAL_QUEUE_POLL_INTERVAL_SECONDS) + + while True: + if time.monotonic() >= deadline: + raise TimeoutError( + f"fal.ai result retrieval timed out for run {run_label} " + f"(request_id: {request_id})." + ) + + try: + result_response = requests.get( + response_url, + headers=headers, + timeout=FAL_HTTP_TIMEOUT_SECONDS, + ) + except requests.RequestException: + time.sleep(FAL_QUEUE_POLL_INTERVAL_SECONDS) + continue + + if result_response.status_code in {202, 408, 425, 429, 500, 502, 503, 504}: + time.sleep(FAL_QUEUE_POLL_INTERVAL_SECONDS) + continue + + result = _response_json(result_response, "fal.ai queue result") + if result.get("error") and not result.get("images"): + error_message = _safe_error_detail(result["error"]) + raise ValueError(f"fal.ai queue run {run_label} failed: {error_message}") + print(f"NanoSeed: completed async run {run_label} ({request_id}).") + return result + + +def _run_fal_queue_requests(run_url, payload, headers, concurrent_runs, queue_timeout): + if not 1 <= concurrent_runs <= FAL_MAX_CONCURRENT_RUNS: + raise ValueError( + f"concurrent_runs must be between 1 and {FAL_MAX_CONCURRENT_RUNS}." + ) + if queue_timeout < 1: + raise ValueError("queue_timeout must be at least 1 second.") + + def run_request(run_index): + run_payload = dict(payload) + run_payload["sync_mode"] = False + run_headers = dict(headers) + run_headers["X-Fal-Request-Timeout"] = str(queue_timeout) + if run_index and isinstance(run_payload.get("seed"), int): + run_payload["seed"] = (run_payload["seed"] + run_index) % (2**32) + return _submit_fal_queue_request( + run_url, + run_payload, + run_headers, + queue_timeout, + run_index, + concurrent_runs, + ) + + if concurrent_runs == 1: + return [run_request(0)] + + ordered_results = [None] * concurrent_runs + failures = [] + with ThreadPoolExecutor(max_workers=concurrent_runs) as executor: + future_to_index = { + executor.submit(run_request, run_index): run_index + for run_index in range(concurrent_runs) + } + for future in as_completed(future_to_index): + run_index = future_to_index[future] + try: + ordered_results[run_index] = future.result() + except Exception as error: + failures.append((run_index, error)) + + if failures: + details = "; ".join( + f"run {run_index + 1}: {error}" + for run_index, error in sorted(failures) + ) + raise ValueError( + f"{len(failures)} of {concurrent_runs} fal.ai async runs failed: {details}" + ) + + return ordered_results + # Main node class class NanoSeedEdit: @classmethod @@ -161,6 +378,8 @@ class NanoSeedEdit: "quality": (["low", "medium", "high"], {"default": "high"}), "enable_web_search": ("BOOLEAN", {"default": False}), "thinking_level": (["off", "minimal", "high"], {"default": "off"}), + "concurrent_runs": ("INT", {"default": 1, "min": 1, "max": 8, "step": 1}), + "queue_timeout": ("INT", {"default": 900, "min": 60, "max": 3600, "step": 30}), } } @@ -173,7 +392,8 @@ class NanoSeedEdit: def edit_image(self, prompt, model, fal_key, image1=None, image2=None, image3=None, image4=None, image5=None, image6=None, image7=None, image8=None, image9=None, image10=None, mask=None, width=0, height=0, num_images=1, num_inference_steps=28, seed=0, aspect_ratio="auto", resolution="1K", - quality="high", enable_web_search=False, thinking_level="off", acceleration="none"): # Hardcoded to none, kept for compatibility + quality="high", enable_web_search=False, thinking_level="off", acceleration="none", + concurrent_runs=1, queue_timeout=900): # Hardcoded to none, kept for compatibility env_fal_key = (os.environ.get("FAL_KEY") or "").strip() ui_fal_key = (fal_key or "").strip() if ui_fal_key == "your_fal_key_here": @@ -229,10 +449,10 @@ class NanoSeedEdit: payload = { "prompt": prompt, "image_urls": img_data_uris, - "num_images": min(num_images, 4), - "aspect_ratio": aspect_ratio, - "output_format": "png", - "sync_mode": True, + "num_images": min(num_images, 4), + "aspect_ratio": aspect_ratio, + "output_format": "png", + "sync_mode": False, } elif model == "nano_banana_pro": url = "https://fal.run/fal-ai/nano-banana-pro/edit" @@ -240,10 +460,10 @@ class NanoSeedEdit: "prompt": prompt, "image_urls": img_data_uris, "num_images": min(num_images, 4), - "aspect_ratio": aspect_ratio, - "resolution": resolution, - "output_format": "png", - "sync_mode": True, + "aspect_ratio": aspect_ratio, + "resolution": resolution, + "output_format": "png", + "sync_mode": False, } elif model == "nano_banana_2": url = "https://fal.run/fal-ai/nano-banana-2/edit" @@ -258,7 +478,7 @@ class NanoSeedEdit: "enable_web_search": enable_web_search, "limit_generations": True, "safety_tolerance": "6", - "sync_mode": True, + "sync_mode": False, } if thinking_level != "off": payload["thinking_level"] = thinking_level @@ -270,7 +490,7 @@ class NanoSeedEdit: "quality": quality, "num_images": num_images, "output_format": "png", - "sync_mode": True, + "sync_mode": False, } if custom_size: payload["image_size"] = {"width": width, "height": height} @@ -282,20 +502,20 @@ class NanoSeedEdit: url = "https://fal.run/xai/grok-imagine-image/edit" payload = { "prompt": prompt, - "image_urls": img_data_uris[:3], - "num_images": min(num_images, 4), - "output_format": "png", - "sync_mode": True, + "image_urls": img_data_uris[:3], + "num_images": min(num_images, 4), + "output_format": "png", + "sync_mode": False, } elif model == "seedream_4.5": url = "https://fal.run/fal-ai/bytedance/seedream/v4.5/edit" payload = { "prompt": prompt, "image_urls": img_data_uris, - "num_images": min(num_images, 6), - "seed": seed, + "num_images": min(num_images, 6), + "seed": seed, "enable_safety_checker": False, - "sync_mode": True, + "sync_mode": False, } if custom_size: if not (1920 <= width <= 4096 and 1920 <= height <= 4096): @@ -315,7 +535,7 @@ class NanoSeedEdit: "image_size": seedream_5_image_size(aspect_ratio, resolution, model), "num_images": min(num_images, 6), "enable_safety_checker": False, - "sync_mode": True, + "sync_mode": False, } if model in ["seedream_5", "seedream_5_pro"]: payload["output_format"] = "png" @@ -331,8 +551,8 @@ class NanoSeedEdit: "guidance_scale": 4.0, "num_inference_steps": num_inference_steps, "enable_safety_checker": False, - "output_format": "png", - "sync_mode": True, + "output_format": "png", + "sync_mode": False, "acceleration": acceleration, } if custom_size: @@ -361,7 +581,7 @@ class NanoSeedEdit: "num_inference_steps": inference_steps, "enable_safety_checker": False, "output_format": "png", - "sync_mode": True, + "sync_mode": False, } if model != "flux_2_klein_9b_edit": payload["guidance_scale"] = 2.5 @@ -376,39 +596,56 @@ class NanoSeedEdit: payload["image_size"] = {"width": width, "height": height} - # API call - headers = { - "Authorization": f"Key {fal_key}", - "Content-Type": "application/json", - } - response = requests.post(url, json=payload, headers=headers) - if response.status_code != 200: - raise ValueError(f"API error: {response.text}") - - api_result = response.json() - if "images" not in api_result or len(api_result["images"]) == 0: - raise ValueError("No images returned from API") - - all_edited_tensors = [] - - # Process up to num_images - for img_info in api_result["images"][:num_images]: - img_data = img_info.get("data_uri") or img_info.get("url") - if not img_data: - continue - - if img_data.startswith("data:"): - _, encoded = img_data.split(",", 1) - pil_edited = Image.open(BytesIO(base64.b64decode(encoded))) - else: - img_resp = requests.get(img_data) - if img_resp.status_code != 200: - raise ValueError("Failed to download generated image") - pil_edited = Image.open(BytesIO(img_resp.content)) - - tensor_edited = pil2tensor(pil_edited) - if tensor_edited is not None: - all_edited_tensors.append(tensor_edited) + headers = { + "Authorization": f"Key {fal_key}", + "Content-Type": "application/json", + } + api_results = _run_fal_queue_requests( + url, + payload, + headers, + concurrent_runs, + queue_timeout, + ) + + all_edited_tensors = [] + + for run_index, api_result in enumerate(api_results): + output_images = api_result.get("images") or [] + if not output_images and api_result.get("image"): + output_images = [api_result["image"]] + if not output_images: + raise ValueError( + f"No images returned from fal.ai async run {run_index + 1}." + ) + + for img_info in output_images[:num_images]: + img_data = img_info.get("data_uri") or img_info.get("url") + if not img_data: + continue + + if img_data.startswith("data:"): + _, encoded = img_data.split(",", 1) + pil_edited = Image.open(BytesIO(base64.b64decode(encoded))) + else: + try: + img_resp = requests.get( + img_data, + timeout=FAL_HTTP_TIMEOUT_SECONDS, + ) + except requests.RequestException as error: + raise ValueError( + f"Failed to download generated image: {error}" + ) from error + if img_resp.status_code != 200: + raise ValueError( + f"Failed to download generated image (HTTP {img_resp.status_code})." + ) + pil_edited = Image.open(BytesIO(img_resp.content)) + + tensor_edited = pil2tensor(pil_edited) + if tensor_edited is not None: + all_edited_tensors.append(tensor_edited) # Stack output if all_edited_tensors: