nodes/EricRollei-Visual_Research
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Initial release: Eric Visual Research - agentic web research nodes for ComfyUI

Browse Source
This commit is contained in:
Eric Rollei
2026-04-28 22:42:55 -07:00
commit e127730be7
9 changed files with 1677 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+31
View File
@@ -0,0 +1,31 @@
# API keys - NEVER commit these
api_keys.ini
# Python cache
__pycache__/
*.py[cod]
*.pyo
*.pyd
.Python
# Virtual environments
.venv/
venv/
env/
# OS files
.DS_Store
Thumbs.db
desktop.ini
# Editor files
.vscode/
.idea/
*.swp
*.swo
# Temp files from gallery-dl downloads
gensearch_*/
# Test outputs
*.tmp
README.md
+165
View File
@@ -0,0 +1,165 @@
# Eric Visual Research
**Agentic web research nodes for ComfyUI image generation.**
Searches the web, retrieves visual references, and writes detailed grounded prompts.
**Platform agnostic** - outputs wire into any generation model: Flux, SDXL, Qwen-Image,
Hunyuan, or any node that accepts `STRING` and `IMAGE` inputs.
---
## Nodes
### Eric Gen-Searcher
Sends your prompt to a running Gen-Searcher-8B model, which autonomously performs
multi-hop web research - searching for text information, finding visual reference images,
and browsing web pages - then synthesizes everything into a rich, detailed generation prompt.
**Outputs:**
- `grounded_prompt` - detailed text prompt ready for any generation node
- `ref_image_1..4` - individual reference images at native aspect ratio
- `reference_images` - batch IMAGE tensor (all refs, letterboxed)
- `search_log` - human-readable record of searches performed
### Eric Reference Describer
Converts reference images into text descriptions using a VL model, then merges
those descriptions with the grounded prompt. Solves the "pixel contamination" problem
when using web thumbnails with edit-mode models.
**Outputs:**
- `combined_prompt` - grounded prompt enriched with VL image descriptions
- `descriptions_log` - what the VL model said about each image
---
## Typical Workflows
### With any text-to-image model
```
[EricGenSearcherNode]
→ grounded_prompt ──→ any text prompt input (Flux, SDXL, etc.)
→ ref_image_1..4 ──→ any image input (for reference/IP-adapter/etc.)
```
### With pure text-to-image (no image input, maximum quality)
```
[EricGenSearcherNode]
→ grounded_prompt ──┐
→ ref_image_1..4 ──→ [EricReferenceDescriber] → combined_prompt → any t2i node
```
### With Qwen-Image-Edit (reference-guided generation)
```
[EricGenSearcherNode]
→ grounded_prompt ──→ [EricQwenGroundedGenerate]
→ ref_image_1..4 ──→ [EricQwenGroundedGenerate]
```
---
## Requirements
- Python 3.10+ (included with ComfyUI)
- `torch` (included with ComfyUI)
- `Pillow` (included with ComfyUI)
- **SGLang server** running Gen-Searcher-8B - see [docs/SERVER_SETUP.md](docs/SERVER_SETUP.md)
- **Serper API key** - free at https://serper.dev (2,500 queries/month free)
- (Optional) **Jina API key** - free tier at https://jina.ai/reader/
- (Optional) **gallery-dl** - for downloading from Pinterest, Flickr, etc.
---
## Installation
### 1. Clone into ComfyUI custom_nodes
```bash
cd ComfyUI/custom_nodes
git clone https://github.com/EricRollei/Eric_Visual_Research
```
### 2. Configure API keys
```bash
cd Eric_Visual_Research
cp api_keys.ini.example api_keys.ini
# Edit api_keys.ini and add your Serper key
```
Or set environment variables:
```bash
export SERPER_API_KEY=your-key-here
export JINA_API_KEY=your-key-here # optional
```
### 3. Set up the Gen-Searcher server
See **[docs/SERVER_SETUP.md](docs/SERVER_SETUP.md)** for full instructions covering:
- WSL2 setup (Windows)
- CUDA toolkit installation
- SGLang installation
- Model download (~17GB)
- Starting the server
### 4. Restart ComfyUI
Nodes appear under **Eric/VisualResearch** in the node menu.
---
## Configuration Reference
Edit `api_keys.ini` in the package directory:
```ini
[api_keys]
serper = your-serper-api-key ; required
; jina = your-jina-api-key ; optional
[gallery_dl]
; exe = /path/to/gallery-dl ; optional, auto-detected
; config = /path/to/config.json ; optional
```
---
## Gen-Searcher Node Settings
| Setting | Default | Description |
|---|---|---|
| `agent_api_url` | `http://localhost:30000` | SGLang server URL |
| `max_hops` | 8 | Max search iterations (6-10 typical) |
| `ref_image_count` | 4 | Number of reference images to download |
| `enable_browse` | True | Allow reading full web pages via Jina |
| `use_instagram` | False | ⚠ Enable with dummy account only |
| `use_facebook` | False | ⚠ Enable with dummy account only |
---
## Reference Describer Node Settings
| Setting | Default | Description |
|---|---|---|
| `vl_api_url` | `http://localhost:30000` | VL model server URL |
| `description_focus` | `auto` | What to extract from each image |
| `synthesis_mode` | `synthesize` | How to combine descriptions |
| `max_image_dim` | 512 | Resolution sent to VL model |
---
## Credits
- **Gen-Searcher-8B** - [tulerfeng/Gen-Searcher](https://github.com/tulerfeng/Gen-Searcher)
(fine-tuned from Qwen3-VL-8B by the Gen-Searcher team)
- **Qwen3-VL** - Qwen Team, Alibaba
- **SGLang** - [sgl-project/sglang](https://github.com/sgl-project/sglang)
- **Serper** - https://serper.dev
- **Jina Reader** - https://jina.ai
Node implementation by **Eric Hiss** (GitHub: EricRollei)
---
## License
CC BY-NC 4.0 / Commercial dual license - see LICENSE.txt
__init__.py
+24
View File
@@ -0,0 +1,24 @@
# Copyright (c) 2026 Eric Hiss. All rights reserved.
# Licensed under the terms in LICENSE.txt (CC BY-NC 4.0 / Commercial dual license).
# https://github.com/EricRollei/Eric_Visual_Research
"""
Eric Visual Research - ComfyUI Node Package
Agentic web research for image generation.
Searches the web, retrieves visual references, and writes detailed
grounded prompts. Works with any downstream model - Flux, SDXL,
Qwen-Image, or any other IMAGE/STRING consumer.
Nodes:
Eric Gen-Searcher - multi-hop web research agent
Eric Reference Describer - convert reference images to text via VL model
Requires:
- SGLang server running Gen-Searcher-8B (see docs/SERVER_SETUP.md)
- Serper API key (https://serper.dev - 2,500 free queries/month)
- (Optional) Jina API key (https://jina.ai - free tier available)
"""
from .nodes import NODE_CLASS_MAPPINGS, NODE_DISPLAY_NAME_MAPPINGS
__all__ = ["NODE_CLASS_MAPPINGS", "NODE_DISPLAY_NAME_MAPPINGS"]
api_keys.ini.example
+35
View File
@@ -0,0 +1,35 @@
; ═══════════════════════════════════════════════════════════════════════
; Eric Visual Research - API Key Configuration
; ═══════════════════════════════════════════════════════════════════════
;
; Copy this file to "api_keys.ini" (remove the .example extension)
; and fill in your keys. This file is git-ignored.
;
; Environment variables are checked FIRST and take priority:
; SERPER_API_KEY, JINA_API_KEY
;
; ═══════════════════════════════════════════════════════════════════════
[api_keys]
; ── Serper (required for web + image search) ──────────────────────────
; Get your free key at https://serper.dev
; Free tier: 2,500 queries/month, no credit card required
serper = your-serper-api-key-here
; ── Jina Reader (optional, for web page browsing) ─────────────────────
; Get your key at https://jina.ai/reader
; Free tier available. Leave commented out to use anonymous free tier.
; jina = your-jina-api-key-here
; ═══════════════════════════════════════════════════════════════════════
; gallery-dl (optional, for social media image download)
; gallery-dl lets the node download images from Pinterest, Flickr,
; 500px, Reddit, Twitter/X, and other sites that block direct access.
;
; If not configured, the node will auto-detect gallery-dl from common
; install locations. You only need to set these if auto-detection fails.
; ═══════════════════════════════════════════════════════════════════════
; [gallery_dl]
; exe = C:\path\to\gallery-dl.exe
; config = C:\path\to\gallery-dl.conf
docs/SERVER_SETUP.md
+243
View File
@@ -0,0 +1,243 @@
# Eric Visual Research - Server Setup Guide
This package requires a running **Gen-Searcher-8B** model served via **SGLang**.
Gen-Searcher-8B is a fine-tuned Qwen3-VL-8B that has been specifically trained
to perform multi-hop web research for image generation - searching the web,
finding visual references, and writing detailed grounded prompts.
---
## Why SGLang? Why not LM Studio?
Gen-Searcher-8B is a **vision-language model** that makes **structured multi-turn
tool calls** (search, image_search, browse). LM Studio does not currently support
multi-turn tool calling for VL models. SGLang does, and is significantly faster.
---
## Why Linux / WSL?
SGLang is developed for Linux. On Windows, several components fail:
- `deep_gemm` (matrix kernel optimization) requires `nvcc` - the CUDA compiler -
which is not easily available on bare Windows
- SGLang's process management and async workers assume Linux process semantics
- FlashInfer (attention backend) requires CUDA 12.9+ for Blackwell GPUs on Linux;
the Windows support is incomplete
**WSL2 (Windows Subsystem for Linux)** gives you a full Ubuntu environment inside
Windows with direct CUDA GPU access via your existing Windows NVIDIA driver.
No dual-boot, no VM. Linux users skip WSL entirely.
---
## Prerequisites
- NVIDIA GPU with 20GB+ VRAM (for Gen-Searcher-8B)
- NVIDIA driver 560+ (already installed if you're using ComfyUI with recent CUDA)
- Windows 10/11 (build 19041+) for WSL2, or any Linux distro
- ~30GB free disk space (17GB model + Python env)
---
## Step 1 - Install WSL2 (Windows only)
Open **PowerShell as Administrator**:
```powershell
# Enable WSL2 features
dism.exe /online /enable-feature /featurename:Microsoft-Windows-Subsystem-Linux /all /norestart
dism.exe /online /enable-feature /featurename:VirtualMachinePlatform /all /norestart
wsl --set-default-version 2
# Install Ubuntu 24.04
winget install --id Canonical.Ubuntu.2404 -e
```
**Restart Windows.**
If you get error `0x80370114` after restart:
- Enter BIOS (F1/Del at startup)
- Find **SVM Mode** (AMD) or **VT-x** (Intel) under CPU/Security settings
- Enable it, save and exit
After restart, launch Ubuntu from Start Menu or run `ubuntu2404` in PowerShell.
Create a username (lowercase only, e.g. `eric`) and password when prompted.
---
## Step 2 - Install CUDA Toolkit in WSL
Inside the Ubuntu terminal:
```bash
# Add NVIDIA CUDA repository
wget https://developer.download.nvidia.com/compute/cuda/repos/ubuntu2404/x86_64/cuda-keyring_1.1-1_all.deb
sudo dpkg -i cuda-keyring_1.1-1_all.deb
sudo apt-get update
# Install CUDA toolkit (compiler + headers - does NOT reinstall the driver)
sudo apt-get install -y cuda-toolkit-12-8
# Add to PATH permanently
echo 'export PATH=/usr/local/cuda-12.8/bin:$PATH' >> ~/.bashrc
echo 'export CUDA_HOME=/usr/local/cuda-12.8' >> ~/.bashrc
echo 'export LD_LIBRARY_PATH=/usr/local/cuda-12.8/lib64:$LD_LIBRARY_PATH' >> ~/.bashrc
source ~/.bashrc
# Verify
nvcc --version
```
You should see `Cuda compilation tools, release 12.8`.
---
## Step 3 - Create Python Environment and Install SGLang
```bash
# Create a dedicated venv
python3 -m venv ~/sglang_env
source ~/sglang_env/bin/activate
pip install -U pip
# Install PyTorch with CUDA 12.8
pip install torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cu128
# Verify GPU is visible
python3 -c "import torch; print(torch.cuda.get_device_name(0))"
# Install SGLang
pip install sglang[all]
# Install HuggingFace tools
pip install huggingface_hub transformers accelerate
```
---
## Step 4 - Download Gen-Searcher-8B (~17GB)
Choose a destination. On Windows, your drives are accessible in WSL at `/mnt/c/`,
`/mnt/d/`, etc. Example using an H: drive:
```bash
source ~/sglang_env/bin/activate
hf download GenSearcher/Gen-Searcher-8B \
--local-dir /mnt/h/models/Gen-Searcher-8B
```
Supports automatic resume if the download is interrupted - just run again.
---
## Step 5 - Start the Server
Save this as `~/start_gen_searcher.sh`:
```bash
#!/bin/bash
source ~/sglang_env/bin/activate
export SGLANG_DISABLE_DEEP_GEMM=1
# CUDA_VISIBLE_DEVICES: use the GPU with 20GB+ VRAM
# 0 = first GPU, 1 = second GPU, etc.
# Run: python3 -c "import torch; [print(i, torch.cuda.get_device_name(i)) for i in range(torch.cuda.device_count())]"
# to find which device is which.
CUDA_VISIBLE_DEVICES=0 \
sglang serve \
--model-path /mnt/h/models/Gen-Searcher-8B \
--host 0.0.0.0 \
--port 30000 \
--trust-remote-code \
--dtype bfloat16 \
--mem-fraction-static 0.85 \
--context-length 16384 \
--attention-backend triton \
--sampling-backend pytorch \
--tool-call-parser qwen25
```
Run it:
```bash
chmod +x ~/start_gen_searcher.sh
~/start_gen_searcher.sh
```
**Note on `--attention-backend triton`:** Required for NVIDIA Blackwell (RTX 4090, RTX 5090,
RTX PRO 6000, etc.) with CUDA 12.8. FlashInfer requires CUDA 12.9+ for Blackwell.
If you have CUDA 12.9+ installed, you can remove this flag for ~20% better throughput.
Watch for: `The server is fired up and ready to roll!`
Keep this terminal open while using ComfyUI. The server runs in the foreground.
---
## Step 6 - Verify the Server
In a second terminal:
```bash
curl http://localhost:30000/health
curl http://localhost:30000/v1/models
```
Or in Python:
```python
import urllib.request, json
with urllib.request.urlopen("http://localhost:30000/v1/models") as r:
print(json.loads(r.read()))
```
---
## Step 7 - Configure API Keys
1. Copy `api_keys.ini.example` to `api_keys.ini` in the package directory
2. Get a free Serper API key at https://serper.dev (2,500 free queries/month)
3. Add your key:
```ini
[api_keys]
serper = your-actual-key-here
```
Alternatively, set environment variables:
```bash
export SERPER_API_KEY=your-key-here # on Linux/WSL
# or in ComfyUI's launch script on Windows
```
---
## Linux (Native) Users
Skip Steps 1-2 entirely. Start from Step 3.
If you have a different CUDA version, replace `cu128` with your version in the pip URL.
---
## Troubleshooting
**`0x80370114` on Windows:** Enable SVM/VT-x virtualization in BIOS. See Step 1.
**`CUDA device not found` in WSL:** Your NVIDIA driver must be 560+. Check with
`nvidia-smi` in PowerShell (not inside WSL - WSL uses the Windows driver).
**`FlashInfer requires GPUs with sm75 or higher`:** Add `--attention-backend triton`
to your serve command. This is needed for Blackwell GPUs with CUDA 12.8.
**`deep_gemm` errors on startup:** Add `export SGLANG_DISABLE_DEEP_GEMM=1` before
the serve command. deep_gemm is an optional optimization that requires nvcc to be
on the PATH at startup time.
**Server runs but ComfyUI can't connect:** WSL uses a different localhost from Windows.
Try `http://127.0.0.1:30000` instead of `http://localhost:30000` in the node.
If that fails, run `hostname -I` in WSL and use that IP.
**Tool calls not working (searches not running):** Ensure `--tool-call-parser qwen25`
is in your serve command. Without it, the model outputs tool calls as inline
`<tool_call>` tags which the node parses as a fallback - but the server-side parser
is more reliable.
nodes/__init__.py
+15
View File
@@ -0,0 +1,15 @@
# Eric Visual Research - Node Package
# Copyright (c) 2026 Eric Hiss. All rights reserved.
from .eric_gen_searcher_node import EricGenSearcherNode
from .eric_reference_describer import EricReferenceDescriber
NODE_CLASS_MAPPINGS = {
"Eric Gen-Searcher": EricGenSearcherNode,
"Eric Reference Describer": EricReferenceDescriber,
}
NODE_DISPLAY_NAME_MAPPINGS = {
"Eric Gen-Searcher": "Eric Gen-Searcher (Grounded Research)",
"Eric Reference Describer": "Eric Reference Describer (VL→Text)",
}
nodes/eric_gen_searcher_node.py
+779
View File
@@ -0,0 +1,779 @@
# Copyright (c) 2026 Eric Hiss. All rights reserved.
# Licensed under the terms in LICENSE.txt (CC BY-NC 4.0 / Commercial dual license).
# https://github.com/EricRollei/Eric_Visual_Research
"""
Eric Gen-Searcher Node
Agentic web research for grounded image generation using Gen-Searcher-8B.
Platform-agnostic: outputs STRING (grounded prompt) and IMAGE (reference images).
Works with any downstream generation model - Flux, SDXL, Qwen-Image, etc.
Author: Eric Hiss (GitHub: EricRollei)
"""
import configparser
import glob
import io
import json
import logging
import os
import re
import shutil
import subprocess
import tempfile
import urllib.error
import urllib.request
from pathlib import Path
from typing import Dict, List, Optional, Tuple
import torch
from PIL import Image
logger = logging.getLogger(__name__)
# ---------------------------------------------------------------------------
# Constants
# ---------------------------------------------------------------------------
_PACKAGE_ROOT = Path(__file__).resolve().parent.parent
_CONFIG_PATH = _PACKAGE_ROOT / "api_keys.ini"
_REF_MAX_DIM = 768
_CANDIDATE_MULTIPLIER = 3 # collect 3x slots so we have fallbacks when downloads fail
# gallery-dl paths - auto-detected from common locations, or set explicitly
# in api_keys.ini under [gallery_dl] section
def _find_gallery_dl() -> Tuple[str, str]:
"""Find gallery-dl executable and config from common install locations."""
# Check ini config first
if _CONFIG_PATH.exists():
cfg = configparser.ConfigParser()
cfg.read(str(_CONFIG_PATH), encoding="utf-8")
exe = cfg.get("gallery_dl", "exe", fallback="").strip()
conf = cfg.get("gallery_dl", "config", fallback="").strip()
if exe and os.path.exists(exe):
return exe, conf
# Common Windows locations
candidates_exe = [
# ComfyUI embedded python
r"A:\Comfy25\ComfyUI_windows_portable\python_embeded\Scripts\gallery-dl.exe",
r"A:\Comfy_Dec\python_embeded\Scripts\gallery-dl.exe",
# Standard Python installs
os.path.join(os.environ.get("APPDATA", ""), "Python", "Scripts", "gallery-dl.exe"),
os.path.join(os.environ.get("LOCALAPPDATA", ""), "Programs", "Python",
"Python312", "Scripts", "gallery-dl.exe"),
# PATH lookup
shutil.which("gallery-dl") or "",
]
candidates_conf = [
r"A:\Comfy25\ComfyUI_windows_portable\ComfyUI\custom_nodes\download_tools\configs\gallery-dl.conf",
os.path.join(os.path.expanduser("~"), ".config", "gallery-dl", "config.json"),
os.path.join(os.path.expanduser("~"), "gallery-dl.conf"),
]
exe = next((p for p in candidates_exe if p and os.path.exists(p)), "")
conf = next((p for p in candidates_conf if p and os.path.exists(p)), "")
return exe, conf
_GALLERY_DL_EXE, _GALLERY_DL_CONF = _find_gallery_dl()
# ---------------------------------------------------------------------------
# Domain classification
# ---------------------------------------------------------------------------
_BLOCKED_CDN_DOMAINS = {
"lookaside.instagram.com", "lookaside.fbsbx.com", "lookaside.facebook.com",
"scontent.fbcdn.net", "scontent.instagram.com",
"external.fwfw1-1.fna.fbcdn.net", "external.fwfw1-2.fna.fbcdn.net",
"t.co",
}
_GALLERY_DL_PAGE_DOMAINS = {
"instagram.com", "www.instagram.com",
"twitter.com", "www.twitter.com", "x.com", "www.x.com",
"pinterest.com", "www.pinterest.com", "pin.it",
"500px.com", "www.500px.com",
"flickr.com", "www.flickr.com",
"deviantart.com", "www.deviantart.com",
"bsky.app", "bluesky.com",
"reddit.com", "www.reddit.com", "old.reddit.com",
"tiktok.com", "www.tiktok.com", "vm.tiktok.com",
"facebook.com", "www.facebook.com", "fb.com",
"tumblr.com", "www.tumblr.com",
}
_INSTAGRAM_DOMAINS = {"instagram.com", "www.instagram.com"}
_FACEBOOK_DOMAINS = {
"facebook.com", "www.facebook.com", "fb.com",
"lookaside.fbsbx.com", "lookaside.facebook.com", "scontent.fbcdn.net",
}
def _domain_of(url: str) -> str:
try:
from urllib.parse import urlparse
return urlparse(url).netloc.lower()
except Exception:
return ""
def _matches_domain_set(url: str, domain_set: set) -> bool:
host = _domain_of(url)
return any(host == d or host.endswith("." + d) for d in domain_set)
def _is_blocked_cdn(url: str) -> bool:
return _matches_domain_set(url, _BLOCKED_CDN_DOMAINS)
def _is_gallery_dl_page(url: str) -> bool:
return _matches_domain_set(url, _GALLERY_DL_PAGE_DOMAINS)
# ---------------------------------------------------------------------------
# System prompt
# ---------------------------------------------------------------------------
_GEN_SEARCHER_SYSTEM = """You are Gen-Searcher, an expert AI assistant for knowledge-grounded image generation. Your task is to research the user's image generation prompt, gather visual references, and then write a final detailed image generation prompt.
You have access to three tools:
- search(query): Search the web for text information
- image_search(query): Search for reference images of a subject
- browse(url): Read the full content of a web page
## Research Strategy
- Aim for 3-6 searches - enough to be thorough, not exhaustive
- Prioritize visual details: colors, textures, shapes, proportions, distinctive features
- Stop when you have enough visual information to write a complete description
## CRITICAL - Final Output Format
When you have enough information, output ONLY a single paragraph of descriptive text.
- NO bullet points, NO lists, NO analysis, NO reasoning
- NO phrases like "I will...", "From my analysis...", "The user wants..."
- Just one flowing paragraph describing exactly what to generate
- Must be 150-300 words of vivid visual description
- Start directly with the subject, style, lighting, etc.
Example of correct output:
"A cinematic portrait of a young woman with long dark braided hair, wearing an oversized magenta Adidas hoodie with white logo detailing on the hood panels, standing on a red clay tennis court. The hoodie features drawstring closures and a relaxed athletic fit. She wears white tennis shorts and clean white sneakers. Soft afternoon light falls from the left, casting gentle shadows. The background shows blurred court lines and a net. Shot at eye level with a medium telephoto perspective, shallow depth of field. Photorealistic, high detail."
Do not explain your research. Do not list what you found. Just write the prompt."""
# ---------------------------------------------------------------------------
# Analysis output detection + cleanup
# ---------------------------------------------------------------------------
_ANALYSIS_MARKERS = [
"from my analysis", "from the search results", "i have gathered",
"i will describe", "i will focus", "i will make", "i will create",
"i will combine", "the user asks", "the user wants",
"i have enough information", "based on the search",
"based on the images", "based on the results", "let me ", "to summarize",
]
def _is_analysis_output(text: str) -> bool:
lower = text.lower()
if text.count("\n-") + text.count("\n*") + text.count("\n*") >= 3:
return True
return any(m in lower for m in _ANALYSIS_MARKERS)
def _cleanup_prompt(raw_output: str, api_url: str, model: str, messages: list) -> str:
print("[EricGenSearcher] Analysis output detected - requesting clean prompt...")
cleanup_messages = messages + [{
"role": "user",
"content": (
"Your previous response contained analysis and bullet points. "
"Now write ONLY the final image generation prompt - a single flowing paragraph "
"of 150-300 words describing exactly what to generate. "
"No bullets, no analysis, no 'I will...', no explanations. "
"Start directly with the subject description."
),
}]
try:
response = _chat_completion(api_url, cleanup_messages, None, model, max_tokens=1024)
cleaned = response["choices"][0]["message"].get("content", "").strip()
cleaned = re.sub(r"<think>.*?</think>", "", cleaned, flags=re.DOTALL).strip()
if cleaned and not _is_analysis_output(cleaned):
print(f"[EricGenSearcher] Cleanup succeeded ({len(cleaned.split())} words)")
return cleaned
except Exception as e:
print(f"[EricGenSearcher] Cleanup call failed: {e}")
return raw_output
# ---------------------------------------------------------------------------
# API key resolution - env vars > api_keys.ini
# ---------------------------------------------------------------------------
def _resolve_api_key(service: str) -> str:
env_map = {
"serper": ["SERPER_API_KEY"],
"jina": ["JINA_API_KEY"],
"deepseek": ["DEEPSEEK_API_KEY"],
"openai": ["OPENAI_API_KEY"],
}
for env_name in env_map.get(service, [f"{service.upper()}_API_KEY"]):
val = os.environ.get(env_name, "").strip()
if val:
return val
if _CONFIG_PATH.exists():
cfg = configparser.ConfigParser()
cfg.read(str(_CONFIG_PATH), encoding="utf-8")
if cfg.has_section("api_keys"):
val = cfg.get("api_keys", service, fallback="").strip()
if val:
return val
return ""
# ---------------------------------------------------------------------------
# Tool implementations
# ---------------------------------------------------------------------------
def _serper_search(query: str, api_key: str, num_results: int = 5) -> str:
if not api_key:
return f"[search unavailable: no Serper API key]\nQuery: {query}"
url = "https://google.serper.dev/search"
body = json.dumps({"q": query, "num": num_results}).encode()
req = urllib.request.Request(
url, data=body,
headers={"X-API-KEY": api_key, "Content-Type": "application/json"},
method="POST",
)
try:
with urllib.request.urlopen(req, timeout=15) as r:
data = json.loads(r.read())
except Exception as e:
return f"[search error: {e}]"
parts = [
f"**{i.get('title','')}**\n{i.get('snippet','')}\nURL: {i.get('link','')}"
for i in data.get("organic", [])[:num_results]
]
return "\n\n".join(parts) if parts else "[no search results]"
def _serper_image_search(query: str, api_key: str, num_results: int = 10) -> Tuple[str, List[Dict]]:
if not api_key:
return "[image_search unavailable: no Serper API key]", []
url = "https://google.serper.dev/images"
body = json.dumps({"q": query, "num": num_results}).encode()
req = urllib.request.Request(
url, data=body,
headers={"X-API-KEY": api_key, "Content-Type": "application/json"},
method="POST",
)
try:
with urllib.request.urlopen(req, timeout=15) as r:
data = json.loads(r.read())
except Exception as e:
return f"[image_search error: {e}]", []
images = data.get("images", [])[:num_results]
entries = [
{"img": img.get("imageUrl", ""), "page": img.get("link", "")}
for img in images if img.get("imageUrl")
]
lines = [f"- {img.get('title','')}: {img.get('imageUrl','')}" for img in images]
return f"Found {len(entries)} images for '{query}':\n" + "\n".join(lines), entries
def _jina_browse(url: str, api_key: str = "") -> str:
jina_url = f"https://r.jina.ai/{url}"
headers = {"Accept": "text/plain"}
if api_key:
headers["Authorization"] = f"Bearer {api_key}"
req = urllib.request.Request(jina_url, headers=headers, method="GET")
try:
with urllib.request.urlopen(req, timeout=20) as r:
content = r.read().decode("utf-8", errors="replace")
return content[:4000] + "\n[... truncated ...]" if len(content) > 4000 else content
except Exception as e:
return f"[browse error for {url}: {e}]"
# ---------------------------------------------------------------------------
# Image download - direct + gallery-dl fallback
# ---------------------------------------------------------------------------
def _download_direct(url: str, timeout: int = 10) -> Optional[Image.Image]:
try:
req = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0"}, method="GET")
with urllib.request.urlopen(req, timeout=timeout) as r:
return Image.open(io.BytesIO(r.read())).convert("RGB")
except Exception as e:
print(f"[EricGenSearcher] Direct download failed {url[:80]}: {e}")
return None
def _download_with_gallery_dl(page_url: str, timeout: int = 45) -> Optional[Image.Image]:
if not _GALLERY_DL_EXE or not os.path.exists(_GALLERY_DL_EXE):
print(f"[EricGenSearcher] gallery-dl not found")
return None
tmp_dir = tempfile.mkdtemp(prefix="gensearch_")
try:
cmd = [_GALLERY_DL_EXE, "--dest", tmp_dir, "--range", "1", "--no-skip", page_url]
if _GALLERY_DL_CONF and os.path.exists(_GALLERY_DL_CONF):
cmd = [_GALLERY_DL_EXE, "--config", _GALLERY_DL_CONF,
"--dest", tmp_dir, "--range", "1", "--no-skip", page_url]
result = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout)
found = []
for ext in ("*.jpg", "*.jpeg", "*.png", "*.webp", "*.gif", "*.avif"):
found.extend(glob.glob(os.path.join(tmp_dir, "**", ext), recursive=True))
if found:
found.sort(key=os.path.getmtime)
img = Image.open(found[0]).convert("RGB")
print(f"[EricGenSearcher] gallery-dl ✓ {os.path.basename(found[0])} "
f"({img.size[0]}×{img.size[1]})")
return img
stderr = result.stderr.strip()[:120] if result.stderr else ""
print(f"[EricGenSearcher] gallery-dl: no images from {page_url[:60]}"
f"{' - ' + stderr if stderr else ''}")
except subprocess.TimeoutExpired:
print(f"[EricGenSearcher] gallery-dl timeout ({timeout}s)")
except Exception as e:
print(f"[EricGenSearcher] gallery-dl exception: {e}")
finally:
shutil.rmtree(tmp_dir, ignore_errors=True)
return None
def _download_smart(
entry: Dict,
use_instagram: bool = False,
use_facebook: bool = False,
) -> Optional[Image.Image]:
img_url = entry.get("img", "")
page_url = entry.get("page", "")
check_url = page_url or img_url
if check_url:
if not use_instagram and _matches_domain_set(check_url, _INSTAGRAM_DOMAINS):
print(f"[EricGenSearcher] Skipping Instagram (disabled)")
return None
if not use_facebook and _matches_domain_set(check_url, _FACEBOOK_DOMAINS):
print(f"[EricGenSearcher] Skipping Facebook (disabled)")
return None
if img_url and not _is_blocked_cdn(img_url):
img = _download_direct(img_url)
if img:
return img
if page_url and _is_gallery_dl_page(page_url):
img = _download_with_gallery_dl(page_url)
if img:
return img
if img_url and _is_blocked_cdn(img_url):
img = _download_direct(img_url)
if img:
return img
return None
# ---------------------------------------------------------------------------
# Low-level API call (stdlib only - no requests dependency)
# ---------------------------------------------------------------------------
def _chat_completion(
api_url: str, messages: list, tools: Optional[list], model: str,
api_key: str = "", timeout: int = 120, max_tokens: int = 2048,
) -> dict:
url = api_url.rstrip("/")
if not url.endswith("/chat/completions"):
url = url.rstrip("/v1").rstrip("/") + "/v1/chat/completions"
payload: dict = {"model": model, "messages": messages,
"max_tokens": max_tokens, "stream": False}
if tools:
payload["tools"] = tools
payload["tool_choice"] = "auto"
headers = {"Content-Type": "application/json"}
if api_key:
headers["Authorization"] = f"Bearer {api_key}"
req = urllib.request.Request(
url, data=json.dumps(payload).encode(), headers=headers, method="POST"
)
try:
with urllib.request.urlopen(req, timeout=timeout) as r:
return json.loads(r.read())
except urllib.error.HTTPError as e:
raise RuntimeError(f"API error {e.code}: {e.read().decode()[:400]}") from e
except urllib.error.URLError as e:
raise RuntimeError(
f"Cannot connect to Gen-Searcher server at {api_url}.\n"
f"Is SGLang running? See docs/SERVER_SETUP.md\n"
f"Error: {e.reason}"
) from e
def _detect_model(api_url: str) -> str:
url = api_url.rstrip("/")
if url.endswith("/chat/completions"):
url = url.rsplit("/chat/completions", 1)[0]
if not url.endswith("/v1"):
url = url.rstrip("/") + "/v1"
try:
req = urllib.request.Request(url + "/models", method="GET")
with urllib.request.urlopen(req, timeout=8) as r:
models = json.loads(r.read()).get("data", [])
if models:
return models[0]["id"]
except Exception:
pass
return "Gen-Searcher-8B"
# ---------------------------------------------------------------------------
# Inline tool call parser (fallback for Qwen3 <tool_call> format)
# ---------------------------------------------------------------------------
def _parse_inline_tool_calls(content: str) -> list:
tool_calls = []
for i, match in enumerate(re.findall(r"<tool_call>\s*(.*?)\s*</tool_call>", content, re.DOTALL)):
try:
data = json.loads(match.strip())
name = data.get("name", "")
if not name:
continue
tool_calls.append({
"id": f"inline_{i}", "type": "function",
"function": {"name": name, "arguments": json.dumps(data.get("arguments", {}))},
})
print(f"[EricGenSearcher] Inline tool call: {name}({data.get('arguments',{})})")
except json.JSONDecodeError:
pass
return tool_calls
# ---------------------------------------------------------------------------
# Tool definitions
# ---------------------------------------------------------------------------
_TOOLS = [
{"type": "function", "function": {
"name": "search",
"description": "Search the web for text information about a topic, person, place, object, or event.",
"parameters": {"type": "object", "properties": {"query": {"type": "string"}}, "required": ["query"]},
}},
{"type": "function", "function": {
"name": "image_search",
"description": "Search for reference images. Returns image URLs showing what the subject looks like.",
"parameters": {"type": "object", "properties": {"query": {"type": "string"}}, "required": ["query"]},
}},
{"type": "function", "function": {
"name": "browse",
"description": "Read the full content of a web page URL for detailed visual information.",
"parameters": {"type": "object", "properties": {"url": {"type": "string"}}, "required": ["url"]},
}},
]
# ---------------------------------------------------------------------------
# Execute a single tool call
# ---------------------------------------------------------------------------
def _execute_tool(fn_name, fn_args, serper_key, jina_key,
enable_browse, log_lines, image_entries, max_candidates) -> str:
if fn_name == "search":
q = fn_args.get("query", "")
log_lines.append(f"[search] {q}")
print(f"[EricGenSearcher] search: {q}")
return _serper_search(q, serper_key)
elif fn_name == "image_search":
q = fn_args.get("query", "")
log_lines.append(f"[image_search] {q}")
print(f"[EricGenSearcher] image_search: {q}")
summary, entries = _serper_image_search(q, serper_key)
existing = {e["img"] for e in image_entries}
for entry in entries:
if len(image_entries) >= max_candidates:
break
if entry["img"] and entry["img"] not in existing:
image_entries.append(entry)
existing.add(entry["img"])
return summary
elif fn_name == "browse":
url = fn_args.get("url", "")
if enable_browse:
log_lines.append(f"[browse] {url}")
print(f"[EricGenSearcher] browse: {url}")
return _jina_browse(url, jina_key)
log_lines.append(f"[browse skipped] {url}")
return "[browse disabled]"
return f"[unknown tool: {fn_name}]"
# ---------------------------------------------------------------------------
# Agent loop
# ---------------------------------------------------------------------------
def _run_agent_loop(
prompt, api_url, model, serper_key, jina_key,
max_hops=8, enable_browse=True, max_ref_images=4,
) -> Tuple[str, List[Dict], str]:
max_candidates = max_ref_images * _CANDIDATE_MULTIPLIER
messages = [
{"role": "system", "content": _GEN_SEARCHER_SYSTEM},
{"role": "user", "content": prompt},
]
image_entries: List[Dict] = []
log_lines: List[str] = []
hops = 0
grounded = prompt
while hops < max_hops:
response = _chat_completion(api_url, messages, _TOOLS, model, max_tokens=2048)
choice = response["choices"][0]
message = choice["message"]
content = (message.get("content") or "").strip()
tool_calls = message.get("tool_calls") or []
if not tool_calls and content:
tool_calls = _parse_inline_tool_calls(content)
if not tool_calls:
grounded = re.sub(r"<think>.*?</think>", "", content, flags=re.DOTALL).strip()
grounded = re.sub(r"<tool_call>.*", "", grounded, flags=re.DOTALL).strip()
if not grounded:
grounded = prompt
if _is_analysis_output(grounded):
grounded = _cleanup_prompt(grounded, api_url, model, messages + [message])
break
messages.append(message)
using_inline = (message.get("tool_calls") is None and bool(tool_calls))
for tc in tool_calls:
fn_name = tc["function"]["name"]
try:
fn_args = json.loads(tc["function"].get("arguments", "{}"))
except json.JSONDecodeError:
fn_args = {}
result = _execute_tool(fn_name, fn_args, serper_key, jina_key,
enable_browse, log_lines, image_entries, max_candidates)
if using_inline:
messages.append({"role": "user", "content": f"<tool_response>\n{result}\n</tool_response>"})
else:
messages.append({"role": "tool", "tool_call_id": tc.get("id", f"c{hops}"), "content": result})
hops += 1
else:
print(f"[EricGenSearcher] max_hops={max_hops} reached, requesting final prompt")
messages.append({
"role": "user",
"content": (
"You have gathered enough information. Write ONLY the final image generation "
"prompt now - a single paragraph of 150-300 words. No bullets, no analysis, "
"no explanations. Start with the subject."
),
})
resp = _chat_completion(api_url, messages, None, model, max_tokens=1024)
content = resp["choices"][0]["message"].get("content", "").strip()
grounded = re.sub(r"<think>.*?</think>", "", content, flags=re.DOTALL).strip() or prompt
if _is_analysis_output(grounded):
grounded = _cleanup_prompt(grounded, api_url, model, messages)
search_log = "\n".join(log_lines) if log_lines else "(no searches performed)"
print(f"[EricGenSearcher] Done. {hops} hop(s), {len(image_entries)} candidates")
print(f"[EricGenSearcher] Prompt ({len(grounded.split())} words): {grounded[:120]}...")
return grounded, image_entries, search_log
# ---------------------------------------------------------------------------
# Image helpers - aspect-ratio preserving, no external dependencies
# ---------------------------------------------------------------------------
def _pil_to_tensor(img: Image.Image) -> torch.Tensor:
import numpy as np
return torch.from_numpy(np.array(img).astype(np.float32) / 255.0)
def _resize_preserve_aspect(img: Image.Image, max_dim: int = _REF_MAX_DIM) -> Image.Image:
w, h = img.size
if max(w, h) <= max_dim:
nw = max(32, (w // 32) * 32)
nh = max(32, (h // 32) * 32)
return img if (nw == w and nh == h) else img.resize((nw, nh), Image.LANCZOS)
if w >= h:
nw, nh = max_dim, max(32, int(h * max_dim / w))
else:
nh, nw = max_dim, max(32, int(w * max_dim / h))
nw = max(32, (nw // 32) * 32)
nh = max(32, (nh // 32) * 32)
return img.resize((nw, nh), Image.LANCZOS)
def _pad_to_size(tensor: torch.Tensor, th: int, tw: int) -> torch.Tensor:
h, w, c = tensor.shape
if h == th and w == tw:
return tensor
padded = torch.zeros(th, tw, c, dtype=tensor.dtype)
t, l = (th - h) // 2, (tw - w) // 2
padded[t:t+h, l:l+w, :] = tensor
return padded
def _build_reference_batch(tensors: List[torch.Tensor]) -> torch.Tensor:
if not tensors:
return torch.ones(1, _REF_MAX_DIM, _REF_MAX_DIM, 3) * 0.5
mh = max(t.shape[1] for t in tensors)
mw = max(t.shape[2] for t in tensors)
return torch.cat([_pad_to_size(t[0], mh, mw).unsqueeze(0) for t in tensors], dim=0)
# ---------------------------------------------------------------------------
# ComfyUI Node
# ---------------------------------------------------------------------------
class EricGenSearcherNode:
"""
Agentic web research for grounded image generation.
Sends your prompt to a running Gen-Searcher-8B model (via SGLang),
which autonomously searches the web for text information and visual
references, then returns a grounded prompt + reference images.
PLATFORM AGNOSTIC - outputs wire into any generation node:
grounded_prompt → any STRING prompt input (Flux, SDXL, Qwen-Image, etc.)
ref_image_1..4 → any IMAGE input or Eric Reference Describer
reference_images → batch IMAGE for multi-image nodes
See docs/SERVER_SETUP.md for SGLang server setup instructions.
See api_keys.ini.example for API key configuration.
"""
CATEGORY = "Eric/VisualResearch"
FUNCTION = "research"
RETURN_TYPES = ("STRING", "IMAGE", "STRING", "IMAGE", "IMAGE", "IMAGE", "IMAGE")
RETURN_NAMES = ("grounded_prompt", "reference_images", "search_log",
"ref_image_1", "ref_image_2", "ref_image_3", "ref_image_4")
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"prompt": ("STRING", {
"multiline": True,
"default": "a portrait of ...",
"tooltip": "Your image generation concept. Gen-Searcher researches it and writes a detailed grounded prompt.",
}),
"agent_api_url": ("STRING", {
"default": "http://localhost:30000",
"tooltip": (
"URL of the running SGLang server.\n"
"See docs/SERVER_SETUP.md for setup instructions.\n"
"Default: http://localhost:30000"
),
}),
},
"optional": {
"model_name": ("STRING", {
"default": "",
"tooltip": "Model name override. Leave blank to auto-detect from server.",
}),
"max_hops": ("INT", {
"default": 8, "min": 1, "max": 20, "step": 1,
"tooltip": "Maximum search iterations. 6-10 is typical.",
}),
"ref_image_count": ("INT", {
"default": 4, "min": 1, "max": 4, "step": 1,
"tooltip": "Number of reference images to download (1-4).",
}),
"enable_browse": ("BOOLEAN", {
"default": True,
"tooltip": "Allow reading full web pages via Jina for deeper detail. Disable for faster runs.",
}),
"fallback_on_error": ("BOOLEAN", {
"default": True,
"tooltip": "Return original prompt if server is unreachable rather than failing.",
}),
"use_instagram": ("BOOLEAN", {
"default": False,
"tooltip": (
"Allow gallery-dl to download from Instagram.\n"
"WARNING: Use only with a dedicated dummy account.\n"
"Instagram flags automated access aggressively."
),
}),
"use_facebook": ("BOOLEAN", {
"default": False,
"tooltip": (
"Allow gallery-dl to download from Facebook.\n"
"WARNING: Use only with a dedicated dummy account."
),
}),
},
}
@classmethod
def IS_CHANGED(cls, prompt, **kwargs):
return float("nan") # always re-run
def research(
self,
prompt: str,
agent_api_url: str = "http://localhost:30000",
model_name: str = "",
max_hops: int = 8,
ref_image_count: int = 4,
enable_browse: bool = True,
fallback_on_error: bool = True,
use_instagram: bool = False,
use_facebook: bool = False,
):
serper_key = _resolve_api_key("serper")
jina_key = _resolve_api_key("jina")
if not serper_key:
print(f"[EricGenSearcher] WARNING: No Serper key found.\n"
f" Add to {_CONFIG_PATH}: serper = your-key\n"
f" Or set env var SERPER_API_KEY\n"
f" Get a free key at https://serper.dev")
model = model_name.strip() if model_name.strip() else _detect_model(agent_api_url)
print(f"[EricGenSearcher] model={model} url={agent_api_url}")
print(f"[EricGenSearcher] prompt: {prompt[:100]}{'...' if len(prompt) > 100 else ''}")
if _GALLERY_DL_EXE:
platforms = ["Pinterest/Flickr/500px/Reddit/Twitter"]
if use_instagram: platforms.append("Instagram ⚠")
if use_facebook: platforms.append("Facebook ⚠")
print(f"[EricGenSearcher] gallery-dl: {', '.join(platforms)}")
else:
print(f"[EricGenSearcher] gallery-dl not found - direct download only")
placeholder = torch.ones(1, _REF_MAX_DIM, _REF_MAX_DIM, 3) * 0.5
try:
grounded, image_entries, search_log = _run_agent_loop(
prompt=prompt, api_url=agent_api_url, model=model,
serper_key=serper_key, jina_key=jina_key,
max_hops=max_hops, enable_browse=enable_browse,
max_ref_images=ref_image_count,
)
except RuntimeError as e:
if fallback_on_error:
print(f"[EricGenSearcher] ERROR (fallback): {e}")
return (prompt, placeholder, f"ERROR: {e}",
placeholder, placeholder, placeholder, placeholder)
raise
print(f"[EricGenSearcher] {len(image_entries)} candidates, need {ref_image_count}...")
good_images: List[torch.Tensor] = []
for entry in image_entries:
if len(good_images) >= ref_image_count:
break
img = _download_smart(entry, use_instagram=use_instagram, use_facebook=use_facebook)
if img is not None:
img = _resize_preserve_aspect(img, _REF_MAX_DIM)
ar = img.size[0] / img.size[1]
src = "gdl" if _is_blocked_cdn(entry.get("img", "")) else "direct"
print(f"[EricGenSearcher] ref {len(good_images)+1}: "
f"{img.size[0]}×{img.size[1]} AR={ar:.2f} [{src}]")
good_images.append(_pil_to_tensor(img).unsqueeze(0))
print(f"[EricGenSearcher] Downloaded {len(good_images)}/{ref_image_count}")
fill = good_images[-1] if good_images else placeholder
individual = good_images[:]
while len(individual) < 4:
individual.append(fill.clone())
ref_tensor = _build_reference_batch(individual[:ref_image_count])
print(f"[EricGenSearcher] batch shape: {ref_tensor.shape}")
return (grounded, ref_tensor, search_log,
individual[0], individual[1], individual[2], individual[3])
nodes/eric_reference_describer.py
+383
View File
@@ -0,0 +1,383 @@
# Copyright (c) 2026 Eric Hiss. All rights reserved.
# Licensed under the terms in LICENSE.txt (CC BY-NC 4.0 / Commercial dual license).
# https://github.com/EricRollei/Eric_Visual_Research
"""
Eric Reference Describer Node
Converts reference images + a grounded prompt into a pure text prompt
by using a running VL model (Qwen3-VL-8B via SGLang) to describe each
image's visual characteristics, then synthesizing everything together.
PLATFORM AGNOSTIC - the combined_prompt output wires into any generation
node: Flux, SDXL, Qwen-Image, Hunyuan, etc.
Problem this solves:
Edit-mode generation models (Qwen-Image-Edit, img2img) try to EDIT
reference images rather than generate something new. Web thumbnail
references contaminate the output with their pixel noise and artefacts.
Solution:
Use the VL model to READ reference images and extract their visual
characteristics as clean text. Feed that text to a pure text-to-image
model instead. No pixel contamination, full generation quality.
Typical workflow:
[EricGenSearcherNode]
→ grounded_prompt ─────────────────────────────────────────────────┐
→ ref_image_1..4 → [EricReferenceDescriber] → combined_prompt ───→ [any t2i node]
Author: Eric Hiss (GitHub: EricRollei)
"""
import base64
import io
import json
import urllib.error
import urllib.request
from typing import List, Optional, Tuple
import numpy as np
import torch
from PIL import Image
# ---------------------------------------------------------------------------
# Inline tensor → PIL (no external package dependencies)
# ---------------------------------------------------------------------------
def _tensor_to_pil(tensor: torch.Tensor) -> Image.Image:
"""Convert a ComfyUI float32 (H, W, C) tensor to a PIL RGB image."""
arr = tensor.detach().cpu().numpy()
arr = (arr * 255.0).clip(0, 255).astype(np.uint8)
return Image.fromarray(arr, "RGB")
# ---------------------------------------------------------------------------
# Focus prompts - what to ask the VL model about each reference image
# ---------------------------------------------------------------------------
_FOCUS_PROMPTS = {
"auto": (
"Examine this reference image carefully. Extract the key visual characteristics "
"that would be most useful for recreating or depicting this scene/subject in a new "
"AI-generated image. Focus on the most visually distinctive and important details: "
"appearance, clothing, setting, lighting. Be precise and concise - no more than "
"3-4 sentences. Prioritize specifics over generalities."
),
"full_scene": (
"Examine this reference image carefully. Provide a concise description of the "
"complete visual scene covering: (1) subject appearance and clothing, "
"(2) setting and environment, (3) lighting and color palette, "
"(4) composition and camera perspective. "
"Be precise and factual - no more than 4 sentences total. "
"This description will be used to guide image generation."
),
"subject_appearance": (
"Examine this reference image carefully. Describe ONLY the visual characteristics "
"of the main subject that would be needed to depict them accurately in a new "
"generated image. Cover: face shape, skin tone, hair color/texture/style, eye color, "
"build/physique, distinctive facial features. Be precise and concise - no more than "
"3 sentences. Do not describe the background, setting, or clothing."
),
"clothing_and_style": (
"Examine this reference image carefully. Describe ONLY the clothing, accessories, "
"and styling worn by the main subject. Cover: garment types, colors, textures, "
"fit/silhouette, visible logos or details, footwear, jewelry. Be precise and concise "
"- no more than 3 sentences. Do not describe the person's face or the background."
),
"setting_and_environment": (
"Examine this reference image carefully. Describe ONLY the setting, environment, "
"and lighting. Cover: location type, architectural elements, time of day, light "
"direction and quality, color palette, atmosphere/mood. Be precise and concise "
"- no more than 3 sentences. Do not describe any people in the image."
),
}
_VL_SYSTEM = (
"You are a precise visual analyst assisting with AI image generation. "
"When asked to describe images, be factual, specific, and concise. "
"Use clear visual language. Never use vague terms like 'beautiful' or 'stunning'. "
"Describe exactly what you see in concrete visual terms."
)
_SYNTHESIS_PROMPT = (
"You are writing the final image generation prompt for a high-quality AI image generator.\n\n"
"Here is the base prompt:\n{grounded_prompt}\n\n"
"Here are visual details extracted from reference images:\n{descriptions}\n\n"
"Write a single flowing paragraph that combines the base prompt with the most "
"important and relevant visual details from the references. "
"The paragraph should read as a natural, detailed image generation prompt. "
"Prioritize details that refine or add specificity to the base prompt. "
"Do NOT include contradictory information. Do NOT use bullet points or headers. "
"Output ONLY the final combined prompt paragraph, nothing else."
)
_APPEND_TEMPLATE = "{grounded_prompt}\n\nVisual reference details:\n{descriptions}"
# ---------------------------------------------------------------------------
# API helpers - stdlib only, no requests dependency
# ---------------------------------------------------------------------------
def _pil_to_base64_jpeg(img: Image.Image, max_dim: int = 512) -> str:
"""Resize to max_dim on longest side and return base64 JPEG string."""
w, h = img.size
if max(w, h) > max_dim:
if w >= h:
nw, nh = max_dim, max(1, int(h * max_dim / w))
else:
nh, nw = max_dim, max(1, int(w * max_dim / h))
img = img.resize((nw, nh), Image.LANCZOS)
buf = io.BytesIO()
img.convert("RGB").save(buf, format="JPEG", quality=85)
return base64.b64encode(buf.getvalue()).decode("utf-8")
def _api_call(api_url: str, payload: dict, timeout: int = 90) -> dict:
"""Single OpenAI-compatible chat/completions call."""
url = api_url.rstrip("/")
if not url.endswith("/chat/completions"):
url = url.rstrip("/v1").rstrip("/") + "/v1/chat/completions"
req = urllib.request.Request(
url,
data=json.dumps(payload).encode(),
headers={"Content-Type": "application/json"},
method="POST",
)
try:
with urllib.request.urlopen(req, timeout=timeout) as r:
return json.loads(r.read())
except urllib.error.HTTPError as e:
print(f"[EricRefDescriber] API error {e.code}: {e.read().decode()[:200]}")
return {}
except Exception as e:
print(f"[EricRefDescriber] API failed: {e}")
return {}
def _describe_image(img_b64: str, focus: str, api_url: str, model: str) -> str:
"""Ask the VL model to describe a single reference image."""
data = _api_call(api_url, {
"model": model,
"max_tokens": 256,
"stream": False,
"messages": [
{"role": "system", "content": _VL_SYSTEM},
{"role": "user", "content": [
{"type": "image_url",
"image_url": {"url": f"data:image/jpeg;base64,{img_b64}"}},
{"type": "text",
"text": _FOCUS_PROMPTS.get(focus, _FOCUS_PROMPTS["auto"])},
]},
],
}, timeout=60)
if data:
return data.get("choices", [{}])[0].get("message", {}).get("content", "").strip()
return ""
def _synthesize(grounded_prompt: str, descriptions: str, api_url: str, model: str) -> str:
"""Ask VL model to synthesize descriptions into a final prompt paragraph."""
data = _api_call(api_url, {
"model": model,
"max_tokens": 512,
"stream": False,
"messages": [{
"role": "user",
"content": _SYNTHESIS_PROMPT.format(
grounded_prompt=grounded_prompt,
descriptions=descriptions,
),
}],
}, timeout=90)
if data:
return data.get("choices", [{}])[0].get("message", {}).get("content", "").strip()
return ""
def _detect_model(api_url: str) -> str:
url = api_url.rstrip("/")
if url.endswith("/chat/completions"):
url = url.rsplit("/chat/completions", 1)[0]
if not url.endswith("/v1"):
url = url.rstrip("/") + "/v1"
try:
req = urllib.request.Request(url + "/models", method="GET")
with urllib.request.urlopen(req, timeout=8) as r:
models = json.loads(r.read()).get("data", [])
if models:
return models[0]["id"]
except Exception:
pass
return "Gen-Searcher-8B"
# ---------------------------------------------------------------------------
# ComfyUI Node
# ---------------------------------------------------------------------------
class EricReferenceDescriber:
"""
Convert reference images + a grounded prompt into a pure text prompt.
Uses a running VL model (via any OpenAI-compatible API - SGLang, vLLM,
LM Studio, OpenAI, etc.) to extract visual characteristics from each
reference image as text, then synthesizes a final combined prompt.
PLATFORM AGNOSTIC - combined_prompt wires into any generation node.
Recommended workflow:
[EricGenSearcherNode]
→ grounded_prompt ──────────────────────────────────────────────┐
→ ref_image_1..4 → [EricReferenceDescriber] → combined_prompt ─→ [any t2i node]
Why this helps:
Edit-mode models try to EDIT reference images rather than generate
something new, causing pixel contamination from web thumbnails.
This node extracts the visual KNOWLEDGE as clean text, giving the
generation model full information without pixel-level constraints.
"""
CATEGORY = "Eric/VisualResearch"
FUNCTION = "describe_and_combine"
RETURN_TYPES = ("STRING", "STRING")
RETURN_NAMES = ("combined_prompt", "descriptions_log")
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"grounded_prompt": ("STRING", {
"multiline": True,
"default": "",
"tooltip": "Base prompt - typically from EricGenSearcherNode.",
}),
"ref_image_1": ("IMAGE", {
"tooltip": "Primary reference image.",
}),
"vl_api_url": ("STRING", {
"default": "http://localhost:30000",
"tooltip": (
"OpenAI-compatible API URL of a VL model server.\n"
"Default: http://localhost:30000 (SGLang with Gen-Searcher-8B)\n"
"Also works with: LM Studio, vLLM, or any OpenAI-compatible endpoint\n"
"that supports vision/image inputs."
),
}),
},
"optional": {
"ref_image_2": ("IMAGE", {"tooltip": "Optional 2nd reference."}),
"ref_image_3": ("IMAGE", {"tooltip": "Optional 3rd reference."}),
"ref_image_4": ("IMAGE", {"tooltip": "Optional 4th reference."}),
"model_name": ("STRING", {
"default": "",
"tooltip": "Model name override. Leave blank to auto-detect.",
}),
"description_focus": (
["auto", "full_scene", "subject_appearance",
"clothing_and_style", "setting_and_environment"],
{
"default": "auto",
"tooltip": (
"What aspect of each reference image to focus on:\n\n"
"auto - model decides what's most important (recommended)\n"
"full_scene - subject + clothing + setting + lighting\n"
"subject_appearance - face, hair, build only\n"
"clothing_and_style - garments and accessories only\n"
"setting_and_environment - location and lighting only"
),
}
),
"synthesis_mode": (
["synthesize", "append"],
{
"default": "synthesize",
"tooltip": (
"synthesize (recommended):\n"
" The VL model weaves descriptions into the grounded prompt\n"
" as a single coherent paragraph. One extra API call.\n\n"
"append:\n"
" Descriptions are appended below the grounded prompt.\n"
" Faster, no extra API call, but less coherent."
),
}
),
"max_image_dim": ("INT", {
"default": 512, "min": 256, "max": 1024, "step": 128,
"tooltip": (
"Max image dimension when sending to the VL model.\n"
"512px is sufficient - VL models don't need full resolution\n"
"to understand content. Higher = slower, diminishing returns."
),
}),
},
}
@classmethod
def IS_CHANGED(cls, **kwargs):
return float("nan") # always re-run
def describe_and_combine(
self,
grounded_prompt: str,
ref_image_1: torch.Tensor,
vl_api_url: str = "http://localhost:30000",
ref_image_2: Optional[torch.Tensor] = None,
ref_image_3: Optional[torch.Tensor] = None,
ref_image_4: Optional[torch.Tensor] = None,
model_name: str = "",
description_focus: str = "auto",
synthesis_mode: str = "synthesize",
max_image_dim: int = 512,
) -> Tuple[str, str]:
model = model_name.strip() if model_name.strip() else _detect_model(vl_api_url)
print(f"[EricRefDescriber] model={model} url={vl_api_url}")
print(f"[EricRefDescriber] focus={description_focus} mode={synthesis_mode}")
# Collect images
raw_tensors = [ref_image_1, ref_image_2, ref_image_3, ref_image_4]
pil_images: List[Image.Image] = []
for t in raw_tensors:
if t is None:
break
src = t[0] if t.dim() == 4 else t
pil_images.append(_tensor_to_pil(src))
print(f"[EricRefDescriber] {len(pil_images)} reference image(s) to describe")
print(f"[EricRefDescriber] base prompt ({len(grounded_prompt.split())} words): "
f"{grounded_prompt[:100]}...")
# Describe each image
desc_lines: List[str] = []
for i, pil in enumerate(pil_images):
print(f"[EricRefDescriber] describing ref {i+1} ({pil.size[0]}×{pil.size[1]})...")
b64 = _pil_to_base64_jpeg(pil, max_image_dim)
desc = _describe_image(b64, description_focus, vl_api_url, model)
if desc:
desc_lines.append(f"Reference {i+1}: {desc}")
print(f"[EricRefDescriber] → {desc[:100]}...")
else:
print(f"[EricRefDescriber] → (no description returned)")
if not desc_lines:
print("[EricRefDescriber] WARNING: No descriptions - using grounded prompt as-is")
return (grounded_prompt, "(no descriptions obtained)")
descriptions_text = "\n".join(desc_lines)
# Combine
if synthesis_mode == "synthesize":
print("[EricRefDescriber] Synthesizing final prompt...")
combined = _synthesize(grounded_prompt, descriptions_text, vl_api_url, model)
if not combined:
print("[EricRefDescriber] Synthesis failed - falling back to append")
combined = _APPEND_TEMPLATE.format(
grounded_prompt=grounded_prompt, descriptions=descriptions_text)
else:
combined = _APPEND_TEMPLATE.format(
grounded_prompt=grounded_prompt, descriptions=descriptions_text)
print(f"[EricRefDescriber] combined prompt ({len(combined.split())} words): "
f"{combined[:120]}...")
return (combined, descriptions_text)
requirements.txt
+2
View File
@@ -0,0 +1,2 @@
Pillow
torch