This commit is contained in:
vladmandic
2025-12-08 18:00:10 +01:00
8 changed files with 2121 additions and 713 deletions
+34 -11
View File
@@ -18,32 +18,31 @@ from modules import shared, devices, paths, sd_models
# model_path = "deepseek-ai/deepseek-vl2-small"
vl_gpt = None
vl_chat_processor = None
loaded_repo = None
class fake_attrdict():
class AttrDict(dict): # dot notation access to dictionary attributes
class AttrDict(dict): # dot notation access to dictionary attributes
__getattr__ = dict.get
__setattr__ = dict.__setitem__
__delattr__ = dict.__delitem__
# def fake_is_flash_attn_2_available():
# return False
def predict(question, image, repo):
global vl_gpt, vl_chat_processor # pylint: disable=global-statement
def load(repo: str):
"""Load DeepSeek VL2 model (experimental)."""
global vl_gpt, vl_chat_processor, loaded_repo # pylint: disable=global-statement
if not shared.cmd_opts.experimental:
shared.log.error(f'Interrogate: type=vlm model="DeepSeek VL2" repo="{repo}" is experimental-only')
return ''
return False
folder = os.path.join(paths.script_path, 'repositories', 'deepseek-vl2')
if not os.path.exists(folder):
shared.log.error(f'Interrogate: type=vlm model="DeepSeek VL2" repo="{repo}" deepseek-vl2 repo not found')
return ''
if vl_gpt is None:
return False
if vl_gpt is None or loaded_repo != repo:
sys.modules['attrdict'] = fake_attrdict
from transformers.models.llama import modeling_llama
modeling_llama.LlamaFlashAttention2 = modeling_llama.LlamaAttention
_deekseek_vl = importlib.import_module('repositories.deepseek-vl2.deepseek_vl2')
importlib.import_module('repositories.deepseek-vl2.deepseek_vl2')
deekseek_vl_models = importlib.import_module('repositories.deepseek-vl2.deepseek_vl2.models')
vl_chat_processor = deekseek_vl_models.DeepseekVLV2Processor.from_pretrained(repo, cache_dir=shared.opts.hfcache_dir)
vl_gpt = AutoModelForCausalLM.from_pretrained(
@@ -51,7 +50,31 @@ def predict(question, image, repo):
trust_remote_code=True,
cache_dir=shared.opts.hfcache_dir,
)
vl_gpt = vl_gpt.to(device=devices.device, dtype=devices.dtype).eval()
vl_gpt.to(dtype=devices.dtype)
vl_gpt.eval()
loaded_repo = repo
shared.log.info(f'Interrogate: type=vlm model="DeepSeek VL2" repo="{repo}"')
sd_models.move_model(vl_gpt, devices.device)
return True
def unload():
"""Release DeepSeek VL2 model from GPU/memory."""
global vl_gpt, vl_chat_processor, loaded_repo # pylint: disable=global-statement
if vl_gpt is not None:
shared.log.debug(f'DeepSeek unload: model="{loaded_repo}"')
sd_models.move_model(vl_gpt, devices.cpu, force=True)
vl_gpt = None
vl_chat_processor = None
loaded_repo = None
devices.torch_gc(force=True)
else:
shared.log.debug('DeepSeek unload: no model loaded')
def predict(question, image, repo):
if not load(repo):
return ''
if len(question) < 2:
question = "Describe the image."
+29 -13
View File
@@ -57,26 +57,43 @@ llava_model: LlavaForConditionalGeneration = None
opts = JoyOptions()
@torch.no_grad()
def predict(question: str, image, vqa_model: str = None) -> str:
global llava_model, processor # pylint: disable=global-statement
opts.max_new_tokens = shared.opts.interrogate_vlm_max_length
if vqa_model is not None and opts.repo != vqa_model:
opts.repo = vqa_model
def load(repo: str = None):
"""Load JoyCaption model."""
global llava_model, processor # pylint: disable=global-statement
repo = repo or opts.repo
if llava_model is None or opts.repo != repo:
opts.repo = repo
llava_model = None
if llava_model is None:
shared.log.info(f'Interrogate: type=vlm model="JoyCaption" {str(opts)}')
processor = AutoProcessor.from_pretrained(opts.repo, max_pixels=1024*1024)
processor = AutoProcessor.from_pretrained(repo, max_pixels=1024*1024, cache_dir=shared.opts.hfcache_dir)
quant_args = model_quant.create_config(module='LLM')
llava_model = LlavaForConditionalGeneration.from_pretrained(
opts.repo,
repo,
torch_dtype=devices.dtype,
device_map="auto",
cache_dir=shared.opts.hfcache_dir,
**quant_args,
)
llava_model.eval()
sd_models.move_model(llava_model, devices.device)
def unload():
"""Release JoyCaption model from GPU/memory."""
global llava_model, processor # pylint: disable=global-statement
if llava_model is not None:
shared.log.debug(f'JoyCaption unload: model="{opts.repo}"')
sd_models.move_model(llava_model, devices.cpu, force=True)
llava_model = None
processor = None
devices.torch_gc(force=True)
else:
shared.log.debug('JoyCaption unload: no model loaded')
@torch.no_grad()
def predict(question: str, image, vqa_model: str = None) -> str:
opts.max_new_tokens = shared.opts.interrogate_vlm_max_length
load(vqa_model)
if len(question) < 2:
question = "Describe the image."
@@ -86,9 +103,8 @@ def predict(question: str, image, vqa_model: str = None) -> str:
{ "role": "user", "content": question },
]
convo_string = processor.apply_chat_template(convo, tokenize=False, add_generation_prompt=True)
inputs = processor(text=[convo_string], images=[image], return_tensors="pt").to(devices.device) # Process the inputs
inputs = processor(text=[convo_string], images=[image], return_tensors="pt").to(devices.device)
inputs['pixel_values'] = inputs['pixel_values'].to(devices.dtype)
sd_models.move_model(llava_model, devices.device)
with devices.inference_context():
generate_ids = llava_model.generate( # Generate the captions
**inputs,
+25 -8
View File
@@ -16,7 +16,7 @@ import torchvision.transforms.functional as TVF
import einops
from einops.layers.torch import Rearrange
import huggingface_hub
from modules import shared, devices
from modules import shared, devices, sd_models
model = None
@@ -1034,22 +1034,39 @@ def prepare_image(image: Image.Image, target_size: int) -> torch.Tensor:
return image_tensor
def predict(image: Image.Image):
global model, tags # pylint: disable=global-statement
def load():
"""Load JoyTag model."""
global model, tags # pylint: disable=global-statement
if model is None:
folder = huggingface_hub.snapshot_download(MODEL_REPO, cache_dir=shared.opts.hfcache_dir)
model = VisionModel.load_model(folder)
model = model.to(device=devices.device, dtype=devices.dtype)
model.to(dtype=devices.dtype)
model.eval()
with open(os.path.join(folder, 'top_tags.txt'), 'r', encoding='utf8') as f:
tags = [line.strip() for line in f.readlines() if line.strip()]
shared.log.info(f'Interrogate: type=vlm model="JoyCaption" repo="{MODEL_REPO}" tags={len(tags)}')
shared.log.info(f'Interrogate: type=vlm model="JoyTag" repo="{MODEL_REPO}" tags={len(tags)}')
sd_models.move_model(model, devices.device)
def unload():
"""Release JoyTag model from GPU/memory."""
global model, tags # pylint: disable=global-statement
if model is not None:
shared.log.debug('JoyTag unload')
sd_models.move_model(model, devices.cpu, force=True)
model = None
tags = None
devices.torch_gc(force=True)
else:
shared.log.debug('JoyTag unload: no model loaded')
def predict(image: Image.Image):
load()
image_tensor = prepare_image(image, model.image_size).unsqueeze(0).to(device=devices.device, dtype=devices.dtype)
model = model.to(devices.device)
with devices.inference_context():
preds = model({ 'image': image_tensor })
preds = model({'image': image_tensor})
tag_preds = preds['tags'].sigmoid().cpu()
model = model.to(devices.cpu)
scores = {tags[i]: tag_preds[0][i] for i in range(len(tags))}
if shared.opts.interrogate_score:
predicted_tags = [f'{tag}:{score:.2f}' for tag, score in scores.items() if score > THRESHOLD]
+410
View File
@@ -0,0 +1,410 @@
# Moondream 3 Preview VLM Implementation
# Source: https://huggingface.co/moondream/moondream3-preview
# Model: 9.3GB, gated (requires HuggingFace authentication)
# Architecture: Mixture-of-Experts (9B total params, 2B active)
import os
import re
import torch
import transformers
from PIL import Image
from modules import shared, devices, sd_models
from modules.interrogate import vqa_detection
# Debug logging - function-based to avoid circular import
debug_enabled = os.environ.get('SD_VQA_DEBUG', None) is not None
def debug(*args, **kwargs):
if debug_enabled:
shared.log.trace(*args, **kwargs)
# Global state
moondream3_model = None
loaded = None
image_cache = {} # Cache encoded images for reuse
def get_settings():
"""
Build settings dict for Moondream 3 API from global VQA options.
Moondream 3 accepts: temperature, top_p, max_tokens
"""
settings = {}
if shared.opts.interrogate_vlm_max_length > 0:
settings['max_tokens'] = shared.opts.interrogate_vlm_max_length
if shared.opts.interrogate_vlm_temperature > 0:
settings['temperature'] = shared.opts.interrogate_vlm_temperature
if shared.opts.interrogate_vlm_top_p > 0:
settings['top_p'] = shared.opts.interrogate_vlm_top_p
return settings if settings else None
def load_model(repo: str):
"""Load Moondream 3 model."""
global moondream3_model, loaded # pylint: disable=global-statement
if moondream3_model is None or loaded != repo:
shared.log.debug(f'Interrogate load: vlm="{repo}"')
moondream3_model = None
moondream3_model = transformers.AutoModelForCausalLM.from_pretrained(
repo,
trust_remote_code=True,
torch_dtype=devices.dtype,
cache_dir=shared.opts.hfcache_dir,
)
moondream3_model.eval()
# Initialize KV caches before moving to device (they're lazy by default)
if hasattr(moondream3_model, '_setup_caches'):
moondream3_model._setup_caches()
# Disable flex_attention decoding (can cause hangs due to torch.compile)
if hasattr(moondream3_model, 'model') and hasattr(moondream3_model.model, 'use_flex_decoding'):
moondream3_model.model.use_flex_decoding = False
loaded = repo
devices.torch_gc()
# Move model to active device
sd_models.move_model(moondream3_model, devices.device)
return moondream3_model
def encode_image(image: Image.Image, cache_key: str = None):
"""
Encode image for reuse across multiple queries.
Args:
image: PIL Image
cache_key: Optional cache key for storing encoded image
Returns:
Encoded image tensor
"""
if cache_key and cache_key in image_cache:
debug(f'VQA interrogate: handler=moondream3 using cached encoding for cache_key="{cache_key}"')
return image_cache[cache_key]
model = load_model(loaded)
with devices.inference_context():
encoded = model.encode_image(image)
if cache_key:
image_cache[cache_key] = encoded
debug(f'VQA interrogate: handler=moondream3 cached encoding cache_key="{cache_key}" cache_size={len(image_cache)}')
return encoded
def query(image: Image.Image, question: str, repo: str, stream: bool = False,
temperature: float = None, top_p: float = None, max_tokens: int = None,
use_cache: bool = False, reasoning: bool = True):
"""
Visual question answering with optional streaming.
Args:
image: PIL Image
question: Question about the image
repo: Model repository
stream: Enable streaming output (generator)
temperature: Sampling temperature (overrides global setting)
top_p: Nucleus sampling parameter (overrides global setting)
max_tokens: Maximum tokens to generate (overrides global setting)
use_cache: Use cached image encoding if available
Returns:
Answer dict or string (or generator if stream=True)
"""
model = load_model(repo)
# Build settings - per-call parameters override global settings
settings = get_settings() or {}
if temperature is not None:
settings['temperature'] = temperature
if top_p is not None:
settings['top_p'] = top_p
if max_tokens is not None:
settings['max_tokens'] = max_tokens
debug(f'VQA interrogate: handler=moondream3 method=query question="{question}" stream={stream} settings={settings}')
# Use cached encoding if requested
if use_cache:
cache_key = f"{id(image)}_{question}"
image_input = encode_image(image, cache_key)
else:
image_input = image
with devices.inference_context():
response = model.query(
image=image_input,
question=question,
stream=stream,
settings=settings if settings else None,
reasoning=reasoning
)
# Log response structure (for non-streaming)
if not stream:
if isinstance(response, dict):
debug(f'VQA interrogate: handler=moondream3 response_type=dict keys={list(response.keys())}')
if 'reasoning' in response:
reasoning_text = response['reasoning'].get('text', '')[:100] + '...' if len(response['reasoning'].get('text', '')) > 100 else response['reasoning'].get('text', '')
debug(f'VQA interrogate: handler=moondream3 reasoning="{reasoning_text}"')
if 'answer' in response:
debug(f'VQA interrogate: handler=moondream3 answer="{response["answer"]}"')
return response
def caption(image: Image.Image, repo: str, length: str = 'normal', stream: bool = False,
temperature: float = None, top_p: float = None, max_tokens: int = None):
"""
Generate image captions at different lengths.
Args:
image: PIL Image
repo: Model repository
length: Caption length - 'short', 'normal', or 'long'
stream: Enable streaming output (generator)
temperature: Sampling temperature (overrides global setting)
top_p: Nucleus sampling parameter (overrides global setting)
max_tokens: Maximum tokens to generate (overrides global setting)
Returns:
Caption dict or string (or generator if stream=True)
"""
model = load_model(repo)
# Build settings - per-call parameters override global settings
settings = get_settings() or {}
if temperature is not None:
settings['temperature'] = temperature
if top_p is not None:
settings['top_p'] = top_p
if max_tokens is not None:
settings['max_tokens'] = max_tokens
debug(f'VQA interrogate: handler=moondream3 method=caption length={length} stream={stream} settings={settings}')
with devices.inference_context():
response = model.caption(
image,
length=length,
stream=stream,
settings=settings if settings else None
)
# Log response structure (for non-streaming)
if not stream and isinstance(response, dict):
debug(f'VQA interrogate: handler=moondream3 response_type=dict keys={list(response.keys())}')
return response
def point(image: Image.Image, object_name: str, repo: str):
"""
Identify coordinates of all instances of a specific object in the image.
Args:
image: PIL Image
object_name: Name of object to locate
repo: Model repository
Returns:
List of (x, y) tuples with coordinates normalized to 0-1 range, or None if not found
Example: [(0.733, 0.442), (0.5, 0.6)] for 2 instances
"""
model = load_model(repo)
debug(f'VQA interrogate: handler=moondream3 method=point object_name="{object_name}"')
with devices.inference_context():
result = model.point(image, object_name)
debug(f'VQA interrogate: handler=moondream3 point_raw_result="{result}" type={type(result)}')
if isinstance(result, dict):
debug(f'VQA interrogate: handler=moondream3 point_raw_result_keys={list(result.keys())}')
points = vqa_detection.parse_points(result)
if points:
debug(f'VQA interrogate: handler=moondream3 point_result={len(points)} points found')
return points
debug('VQA interrogate: handler=moondream3 point_result=not found')
return None
def detect(image: Image.Image, object_name: str, repo: str, max_objects: int = 10):
"""
Detect all instances of a specific object with bounding boxes.
Args:
image: PIL Image
object_name: Name of object to detect
repo: Model repository
max_objects: Maximum number of objects to return
Returns:
List of detection dicts with keys:
- 'bbox': [x1, y1, x2, y2] normalized to 0-1
- 'label': Object label
- 'confidence': Detection confidence (0-1)
Returns empty list if no objects found.
"""
model = load_model(repo)
debug(f'VQA interrogate: handler=moondream3 method=detect object_name="{object_name}" max_objects={max_objects}')
with devices.inference_context():
result = model.detect(image, object_name)
debug(f'VQA interrogate: handler=moondream3 detect_raw_result="{result}" type={type(result)}')
if isinstance(result, dict):
debug(f'VQA interrogate: handler=moondream3 detect_raw_result_keys={list(result.keys())}')
detections = vqa_detection.parse_detections(result, object_name, max_objects)
debug(f'VQA interrogate: handler=moondream3 detect_result={len(detections)} objects found')
return detections
def predict(question: str, image: Image.Image, repo: str, model_name: str = None, thinking_mode: bool = False,
mode: str = None, stream: bool = False, use_cache: bool = False, **kwargs):
"""
Main entry point for Moondream 3 VQA - auto-detects mode from question.
Args:
question: The question/prompt (e.g., "caption", "where is the cat?", "describe this")
image: PIL Image
repo: Model repository
model_name: Display name for logging
thinking_mode: Enable reasoning mode for query
mode: Force specific mode ('query', 'caption', 'caption_short', 'caption_long', 'point', 'detect')
stream: Enable streaming output (for query/caption)
use_cache: Use cached image encoding (for query)
**kwargs: Additional parameters (max_objects for detect, etc.)
Returns:
Response string (detection data stored on VQA singleton instance.last_detection_data)
(or generator if stream=True for query/caption modes)
"""
debug(f'VQA interrogate: handler=moondream3 model_name="{model_name}" repo="{repo}" question="{question}" image_size={image.size if image else None} mode={mode} stream={stream}')
# Clean question
question = question.replace('<', '').replace('>', '').replace('_', ' ') if question else ''
# Auto-detect mode from question if not specified
if mode is None:
question_lower = question.lower()
# Caption detection
if question in ['CAPTION', 'caption'] or 'caption' in question_lower:
if 'more detailed' in question_lower or 'very long' in question_lower:
mode = 'caption_long'
elif 'detailed' in question_lower or 'long' in question_lower:
mode = 'caption_normal'
elif 'short' in question_lower or 'brief' in question_lower:
mode = 'caption_short'
else:
# Default caption mode (matches vqa.py legacy behavior)
if question == 'CAPTION':
mode = 'caption_short'
elif question == 'DETAILED CAPTION':
mode = 'caption_normal'
elif question == 'MORE DETAILED CAPTION':
mode = 'caption_long'
else:
mode = 'caption_normal'
# Point detection
elif 'where is' in question_lower or 'locate' in question_lower or 'find' in question_lower or 'point' in question_lower:
mode = 'point'
# Object detection
elif 'detect' in question_lower or 'bounding box' in question_lower or 'bbox' in question_lower:
mode = 'detect'
# Default to query
else:
mode = 'query'
debug(f'VQA interrogate: handler=moondream3 mode_selected={mode}')
# Dispatch to appropriate method
try:
if mode == 'caption_short':
response = caption(image, repo, length='short', stream=stream)
elif mode == 'caption_long':
response = caption(image, repo, length='long', stream=stream)
elif mode in ['caption', 'caption_normal']:
response = caption(image, repo, length='normal', stream=stream)
elif mode == 'point':
# Extract object name from question - case insensitive, preserve object names
object_name = question
for phrase in ['point at', 'where is', 'locate', 'find']:
object_name = re.sub(rf'\b{phrase}\b', '', object_name, flags=re.IGNORECASE)
object_name = re.sub(r'[?.!,]', '', object_name).strip()
object_name = re.sub(r'^\s*the\s+', '', object_name, flags=re.IGNORECASE)
debug(f'VQA interrogate: handler=moondream3 point_extracted_object="{object_name}"')
result = point(image, object_name, repo)
if result:
from modules.interrogate import vqa
vqa.get_instance().last_detection_data = {'points': result}
return vqa_detection.format_points_text(result)
return "Object not found"
elif mode == 'detect':
# Extract object name from question - case insensitive
object_name = question
for phrase in ['detect', 'find all', 'bounding box', 'bbox', 'find']:
object_name = re.sub(rf'\b{phrase}\b', '', object_name, flags=re.IGNORECASE)
object_name = re.sub(r'[?.!,]', '', object_name).strip()
object_name = re.sub(r'^\s*the\s+', '', object_name, flags=re.IGNORECASE)
if ' and ' in object_name.lower():
object_name = re.split(r'\s+and\s+', object_name, flags=re.IGNORECASE)[0].strip()
debug(f'VQA interrogate: handler=moondream3 detect_extracted_object="{object_name}"')
results = detect(image, object_name, repo, max_objects=kwargs.get('max_objects', 10))
if results:
from modules.interrogate import vqa
vqa.get_instance().last_detection_data = {'detections': results}
return vqa_detection.format_detections_text(results)
return "No objects detected"
else: # mode == 'query'
if len(question) < 2:
question = "Describe this image."
response = query(image, question, repo, stream=stream, use_cache=use_cache, reasoning=thinking_mode)
debug(f'VQA interrogate: handler=moondream3 response_before_clean="{response}"')
return response
except Exception as e:
from modules import errors
errors.display(e, 'Moondream3')
return f"Error: {str(e)}"
def clear_cache():
"""Clear image encoding cache."""
global image_cache # pylint: disable=global-statement
cache_size = len(image_cache)
image_cache.clear()
debug(f'VQA interrogate: handler=moondream3 cleared image cache cache_size_was={cache_size}')
shared.log.debug(f'Moondream3: Cleared image cache ({cache_size} entries)')
def unload():
"""Release Moondream 3 model from GPU/memory."""
global moondream3_model, loaded # pylint: disable=global-statement
if moondream3_model is not None:
shared.log.debug(f'Moondream3 unload: model="{loaded}"')
sd_models.move_model(moondream3_model, devices.cpu, force=True)
moondream3_model = None
loaded = None
clear_cache()
devices.torch_gc(force=True)
else:
shared.log.debug('Moondream3 unload: no model loaded')
+1344 -663
View File
File diff suppressed because it is too large Load Diff
+207
View File
@@ -0,0 +1,207 @@
# VQA Detection Utilities
# Parsing, formatting, and drawing functions for detection results (points, bboxes, gaze)
from PIL import Image, ImageDraw, ImageFont
from modules import shared
def parse_points(result) -> list:
"""Parse and validate point coordinates from model result.
Args:
result: Model output, typically dict with 'points' key or list of coordinates
Returns:
List of (x, y) tuples with coordinates clamped to 0-1 range.
"""
points = []
# Dict format: {'points': [{'x': 0.5, 'y': 0.5}, ...]}
if isinstance(result, dict) and 'points' in result:
points_list = result['points']
if points_list and len(points_list) > 0:
for point_data in points_list:
if isinstance(point_data, dict) and 'x' in point_data and 'y' in point_data:
x = max(0.0, min(1.0, float(point_data['x'])))
y = max(0.0, min(1.0, float(point_data['y'])))
points.append((x, y))
# Fallback for simple [x, y] format
elif isinstance(result, (list, tuple)) and len(result) == 2:
try:
x = max(0.0, min(1.0, float(result[0])))
y = max(0.0, min(1.0, float(result[1])))
points.append((x, y))
except (ValueError, TypeError):
pass
return points
def parse_detections(result, label: str, max_objects: int = None) -> list:
"""Parse and validate detection bboxes from model result.
Args:
result: Model output, typically dict with 'objects' key
label: Label to assign to detected objects
max_objects: Maximum number of objects to return (None for all)
Returns:
List of {'bbox': [x1,y1,x2,y2], 'label': str, 'confidence': float}
with coordinates clamped to 0-1 range.
"""
detections = []
if isinstance(result, dict) and 'objects' in result:
objects = result['objects']
if max_objects is not None:
objects = objects[:max_objects]
for obj in objects:
if all(k in obj for k in ['x_min', 'y_min', 'x_max', 'y_max']):
bbox = [
max(0.0, min(1.0, float(obj['x_min']))),
max(0.0, min(1.0, float(obj['y_min']))),
max(0.0, min(1.0, float(obj['x_max']))),
max(0.0, min(1.0, float(obj['y_max'])))
]
detections.append({
'bbox': bbox,
'label': label,
'confidence': obj.get('confidence', 1.0)
})
return detections
def format_points_text(points: list) -> str:
"""Format point coordinates as human-readable text.
Args:
points: List of (x, y) tuples with normalized coordinates
Returns:
Formatted text string describing the points.
"""
if not points:
return "Object not found"
if len(points) == 1:
return f"Found at: ({points[0][0]:.3f}, {points[0][1]:.3f})"
lines = [f"Found {len(points)} instances:"]
for i, (x, y) in enumerate(points, 1):
lines.append(f" {i}. ({x:.3f}, {y:.3f})")
return '\n'.join(lines)
def format_detections_text(detections: list, include_confidence: bool = True) -> str:
"""Format detections with bboxes as human-readable text.
Args:
detections: List of detection dicts with 'bbox', 'label', 'confidence'
include_confidence: Whether to include confidence scores in output
Returns:
Formatted text string describing the detections.
"""
if not detections:
return "No objects detected"
lines = []
for det in detections:
bbox = det['bbox']
label = det.get('label', 'object')
confidence = det.get('confidence', 1.0)
if include_confidence and confidence < 1.0:
lines.append(f"{label}: [{bbox[0]:.3f}, {bbox[1]:.3f}, {bbox[2]:.3f}, {bbox[3]:.3f}] (confidence: {confidence:.2f})")
else:
lines.append(f"{label}: [{bbox[0]:.3f}, {bbox[1]:.3f}, {bbox[2]:.3f}, {bbox[3]:.3f}]")
return '\n'.join(lines)
def calculate_eye_position(face_bbox: dict) -> tuple:
"""Calculate approximate eye position from face bounding box.
Args:
face_bbox: Dict with 'x_min', 'y_min', 'x_max', 'y_max' keys
Returns:
(eye_x, eye_y) tuple with normalized coordinates.
"""
eye_x = (face_bbox['x_min'] + face_bbox['x_max']) / 2
eye_y = face_bbox['y_min'] + (face_bbox['y_max'] - face_bbox['y_min']) * 0.3 # Approximate eye level
return (eye_x, eye_y)
def draw_bounding_boxes(image: Image.Image, detections: list, points: list = None) -> Image.Image:
"""
Draw bounding boxes and/or points on an image.
Args:
image: PIL Image to annotate
detections: List of detection dicts with format:
[{'label': str, 'bbox': [x1, y1, x2, y2], 'confidence': float}, ...]
where coordinates are normalized 0-1
points: Optional list of (x, y) tuples with normalized 0-1 coordinates
Returns:
Annotated PIL Image with boxes and labels drawn, or None if no annotations
"""
if not detections and not points:
return None
# Create a copy to avoid modifying original
annotated = image.copy()
draw = ImageDraw.Draw(annotated)
width, height = image.size
# Try to load a font, fall back to default if unavailable
try:
font_size = max(12, int(min(width, height) * 0.02))
font_path = shared.opts.font or "javascript/notosans-nerdfont-regular.ttf"
font = ImageFont.truetype(font_path, size=font_size)
except Exception:
font = ImageFont.load_default()
# Draw bounding boxes
if detections:
colors = ['#FF0000', '#00FF00', '#0000FF', '#FFFF00', '#FF00FF', '#00FFFF', '#FFA500', '#800080']
for idx, det in enumerate(detections):
bbox = det['bbox']
label = det.get('label', 'object')
confidence = det.get('confidence', 1.0)
# Convert normalized coordinates to pixel coordinates
x1 = int(bbox[0] * width)
y1 = int(bbox[1] * height)
x2 = int(bbox[2] * width)
y2 = int(bbox[3] * height)
# Choose color
color = colors[idx % len(colors)]
# Draw box
draw.rectangle([x1, y1, x2, y2], outline=color, width=max(2, int(min(width, height) * 0.003)))
# Draw label with background
label_text = f"{label} {confidence:.2f}" if confidence < 1.0 else label
bbox_font = draw.textbbox((x1, y1), label_text, font=font)
text_width = bbox_font[2] - bbox_font[0]
text_height = bbox_font[3] - bbox_font[1]
draw.rectangle([x1, y1 - text_height - 4, x1 + text_width + 4, y1], fill=color)
draw.text((x1 + 2, y1 - text_height - 2), label_text, fill='white', font=font)
# Draw points
if points:
point_radius = max(3, int(min(width, height) * 0.01))
for px, py in points:
x = int(px * width)
y = int(py * height)
# Draw point as a circle
draw.ellipse([x - point_radius, y - point_radius, x + point_radius, y + point_radius],
fill='#FF0000', outline='#FFFFFF', width=2)
return annotated
+6 -3
View File
@@ -656,12 +656,12 @@ options_templates.update(options_section(('interrogate', "Interrogate"), {
"interrogate_clip_max_length": OptionInfo(74, "CLiP: max length", gr.Slider, {"minimum": 1, "maximum": 512, "step": 1, "visible": False}),
"interrogate_clip_min_flavors": OptionInfo(2, "CLiP: min flavors", gr.Slider, {"minimum": 0, "maximum": 32, "step": 1, "visible": False}),
"interrogate_clip_max_flavors": OptionInfo(16, "CLiP: max flavors", gr.Slider, {"minimum": 0, "maximum": 32, "step": 1, "visible": False}),
"interrogate_clip_flavor_count": OptionInfo(16, "CLiP: intermediate flavors", gr.Slider, {"minimum": 0, "maximum": 32, "step": 1, "visible": False}),
"interrogate_clip_chunk_size": OptionInfo(1024, "CLiP: chunk size", gr.Slider, {"minimum": 256, "maximum": 4096, "step": 8, "visible": False}),
"interrogate_clip_flavor_count": OptionInfo(1024, "CLiP: intermediate flavors", gr.Slider, {"minimum": 256, "maximum": 4096, "step": 64, "visible": False}),
"interrogate_clip_chunk_size": OptionInfo(1024, "CLiP: chunk size", gr.Slider, {"minimum": 256, "maximum": 4096, "step": 64, "visible": False}),
"interrogate_vlm_sep": OptionInfo("<h2>VLM</h2>", "", gr.HTML),
"interrogate_vlm_model": OptionInfo(vlm_default, "VLM: default model", gr.Dropdown, {"choices": list(vlm_models)}),
"interrogate_vlm_prompt": OptionInfo(vlm_prompts[2], "VLM: default prompt", DropdownEditable, {"choices": vlm_prompts }),
"interrogate_vlm_prompt": OptionInfo(vlm_prompts[0], "VLM: default prompt", DropdownEditable, {"choices": vlm_prompts }),
"interrogate_vlm_system": OptionInfo(vlm_system, "VLM: default prompt"),
"interrogate_vlm_num_beams": OptionInfo(1, "VLM: num beams", gr.Slider, {"minimum": 1, "maximum": 16, "step": 1, "visible": False}),
"interrogate_vlm_max_length": OptionInfo(512, "VLM: max length", gr.Slider, {"minimum": 1, "maximum": 4096, "step": 1, "visible": False}),
@@ -669,6 +669,9 @@ options_templates.update(options_section(('interrogate', "Interrogate"), {
"interrogate_vlm_temperature": OptionInfo(0, "VLM: temperature", gr.Slider, {"minimum": 0, "maximum": 1.0, "step": 0.01, "visible": False}),
"interrogate_vlm_top_k": OptionInfo(0, "VLM: top-k", gr.Slider, {"minimum": 0, "maximum": 99, "step": 1, "visible": False}),
"interrogate_vlm_top_p": OptionInfo(0, "VLM: top-p", gr.Slider, {"minimum": 0, "maximum": 1.0, "step": 0.01, "visible": False}),
"interrogate_vlm_keep_prefill": OptionInfo(False, "VLM: keep prefill text in output", gr.Checkbox, {"visible": False}),
"interrogate_vlm_keep_thinking": OptionInfo(False, "VLM: keep reasoning trace in output", gr.Checkbox, {"visible": False}),
"interrogate_vlm_thinking_mode": OptionInfo(False, "VLM: enable thinking/reasoning mode", gr.Checkbox, {"visible": False}),
"deepbooru_sep": OptionInfo("<h2>DeepBooru</h2>", "", gr.HTML),
"deepbooru_score_threshold": OptionInfo(0.65, "DeepBooru: score threshold", gr.Slider, {"minimum": 0, "maximum": 1, "step": 0.01}),
+66 -15
View File
@@ -3,14 +3,41 @@ from modules import shared, ui_common, generation_parameters_copypaste
from modules.interrogate import openclip
def vlm_caption_wrapper(question, system_prompt, prompt, image, model_name, prefill, thinking_mode):
"""Wrapper for vqa.interrogate that handles annotated image display."""
from modules.interrogate import vqa
answer = vqa.interrogate(question, system_prompt, prompt, image, model_name, prefill, thinking_mode)
annotated_image = vqa.get_last_annotated_image()
if annotated_image is not None:
return answer, gr.update(value=annotated_image, visible=True)
return answer, gr.update(visible=False)
def update_vlm_prompts_for_model(model_name):
"""Update the task dropdown choices based on selected model."""
from modules.interrogate import vqa
prompts = vqa.get_prompts_for_model(model_name)
return gr.update(choices=prompts, value=prompts[0] if prompts else "Use Prompt")
def update_vlm_prompt_placeholder(question):
"""Update the prompt field placeholder based on selected task."""
from modules.interrogate import vqa
placeholder = vqa.get_prompt_placeholder(question)
return gr.update(placeholder=placeholder)
def update_vlm_params(*args):
vlm_max_tokens, vlm_num_beams, vlm_temperature, vlm_do_sample, vlm_top_k, vlm_top_p = args
vlm_max_tokens, vlm_num_beams, vlm_temperature, vlm_do_sample, vlm_top_k, vlm_top_p, vlm_keep_prefill, vlm_keep_thinking, vlm_thinking_mode = args
shared.opts.interrogate_vlm_max_length = int(vlm_max_tokens)
shared.opts.interrogate_vlm_num_beams = int(vlm_num_beams)
shared.opts.interrogate_vlm_temperature = float(vlm_temperature)
shared.opts.interrogate_vlm_do_sample = bool(vlm_do_sample)
shared.opts.interrogate_vlm_top_k = int(vlm_top_k)
shared.opts.interrogate_vlm_top_p = float(vlm_top_p)
shared.opts.interrogate_vlm_keep_prefill = bool(vlm_keep_prefill)
shared.opts.interrogate_vlm_keep_thinking = bool(vlm_keep_thinking)
shared.opts.interrogate_vlm_thinking_mode = bool(vlm_thinking_mode)
shared.opts.save(shared.config_filename)
@@ -36,14 +63,19 @@ def create_ui():
with gr.Tabs(elem_id="mode_caption"):
with gr.Tab("VLM Caption", elem_id="tab_vlm_caption"):
from modules.interrogate import vqa
current_vlm_model = shared.opts.interrogate_vlm_model or vqa.vlm_default
initial_prompts = vqa.get_prompts_for_model(current_vlm_model)
with gr.Row():
vlm_system = gr.Textbox(label="System prompt", value=vqa.vlm_system, lines=1, elem_id='vlm_system')
with gr.Row():
vlm_question = gr.Dropdown(label="Predefined question", allow_custom_value=False, choices=vqa.vlm_prompts, value=vqa.vlm_prompts[2], elem_id='vlm_question')
vlm_question = gr.Dropdown(label="Task", allow_custom_value=False, choices=initial_prompts, value=initial_prompts[0] if initial_prompts else "Use Prompt", elem_id='vlm_question')
with gr.Row():
vlm_prompt = gr.Textbox(label="Prompt", placeholder="optionally enter custom prompt", lines=2, elem_id='vlm_prompt')
vlm_prompt = gr.Textbox(label="Prompt", placeholder=vqa.get_prompt_placeholder(initial_prompts[0] if initial_prompts else "Use Prompt"), lines=2, elem_id='vlm_prompt')
with gr.Row(elem_id='interrogate_buttons_query'):
vlm_model = gr.Dropdown(list(vqa.vlm_models), value=vqa.vlm_default, label='VLM Model', elem_id='vlm_model')
vlm_model = gr.Dropdown(list(vqa.vlm_models), value=current_vlm_model, label='VLM Model', elem_id='vlm_model')
with gr.Row():
vlm_load_btn = gr.Button(value='Load', elem_id='vlm_load', variant='secondary')
vlm_unload_btn = gr.Button(value='Unload', elem_id='vlm_unload', variant='secondary')
with gr.Accordion(label='Advanced options', open=False, visible=True):
with gr.Row():
vlm_max_tokens = gr.Slider(label='VLM max tokens', value=shared.opts.interrogate_vlm_max_length, minimum=16, maximum=4096, step=1, elem_id='vlm_max_tokens')
@@ -54,12 +86,21 @@ def create_ui():
vlm_top_p = gr.Slider(label='Top-P', value=shared.opts.interrogate_vlm_top_p, minimum=0.0, maximum=1.0, step=0.01, elem_id='vlm_top_p')
with gr.Row():
vlm_do_sample = gr.Checkbox(label='Use sample', value=shared.opts.interrogate_vlm_do_sample, elem_id='vlm_do_sample')
vlm_max_tokens.change(fn=update_vlm_params, inputs=[vlm_max_tokens, vlm_num_beams, vlm_temperature, vlm_do_sample, vlm_top_k, vlm_top_p], outputs=[])
vlm_num_beams.change(fn=update_vlm_params, inputs=[vlm_max_tokens, vlm_num_beams, vlm_temperature, vlm_do_sample, vlm_top_k, vlm_top_p], outputs=[])
vlm_temperature.change(fn=update_vlm_params, inputs=[vlm_max_tokens, vlm_num_beams, vlm_temperature, vlm_do_sample, vlm_top_k, vlm_top_p], outputs=[])
vlm_do_sample.change(fn=update_vlm_params, inputs=[vlm_max_tokens, vlm_num_beams, vlm_temperature, vlm_do_sample, vlm_top_k, vlm_top_p], outputs=[])
vlm_top_k.change(fn=update_vlm_params, inputs=[vlm_max_tokens, vlm_num_beams, vlm_temperature, vlm_do_sample, vlm_top_k, vlm_top_p], outputs=[])
vlm_top_p.change(fn=update_vlm_params, inputs=[vlm_max_tokens, vlm_num_beams, vlm_temperature, vlm_do_sample, vlm_top_k, vlm_top_p], outputs=[])
vlm_thinking_mode = gr.Checkbox(label='Thinking Mode', value=shared.opts.interrogate_vlm_thinking_mode, elem_id='vlm_thinking_mode')
with gr.Row():
vlm_keep_thinking = gr.Checkbox(label='Keep Thinking Trace', value=shared.opts.interrogate_vlm_keep_thinking, elem_id='vlm_keep_thinking')
vlm_keep_prefill = gr.Checkbox(label='Keep Prefill', value=shared.opts.interrogate_vlm_keep_prefill, elem_id='vlm_keep_prefill')
with gr.Row():
vlm_prefill = gr.Textbox(label='Prefill Text', value='', lines=1, elem_id='vlm_prefill', placeholder='Optional prefill text for model to continue from')
vlm_max_tokens.change(fn=update_vlm_params, inputs=[vlm_max_tokens, vlm_num_beams, vlm_temperature, vlm_do_sample, vlm_top_k, vlm_top_p, vlm_keep_prefill, vlm_keep_thinking, vlm_thinking_mode], outputs=[])
vlm_num_beams.change(fn=update_vlm_params, inputs=[vlm_max_tokens, vlm_num_beams, vlm_temperature, vlm_do_sample, vlm_top_k, vlm_top_p, vlm_keep_prefill, vlm_keep_thinking, vlm_thinking_mode], outputs=[])
vlm_temperature.change(fn=update_vlm_params, inputs=[vlm_max_tokens, vlm_num_beams, vlm_temperature, vlm_do_sample, vlm_top_k, vlm_top_p, vlm_keep_prefill, vlm_keep_thinking, vlm_thinking_mode], outputs=[])
vlm_do_sample.change(fn=update_vlm_params, inputs=[vlm_max_tokens, vlm_num_beams, vlm_temperature, vlm_do_sample, vlm_top_k, vlm_top_p, vlm_keep_prefill, vlm_keep_thinking, vlm_thinking_mode], outputs=[])
vlm_top_k.change(fn=update_vlm_params, inputs=[vlm_max_tokens, vlm_num_beams, vlm_temperature, vlm_do_sample, vlm_top_k, vlm_top_p, vlm_keep_prefill, vlm_keep_thinking, vlm_thinking_mode], outputs=[])
vlm_top_p.change(fn=update_vlm_params, inputs=[vlm_max_tokens, vlm_num_beams, vlm_temperature, vlm_do_sample, vlm_top_k, vlm_top_p, vlm_keep_prefill, vlm_keep_thinking, vlm_thinking_mode], outputs=[])
vlm_keep_prefill.change(fn=update_vlm_params, inputs=[vlm_max_tokens, vlm_num_beams, vlm_temperature, vlm_do_sample, vlm_top_k, vlm_top_p, vlm_keep_prefill, vlm_keep_thinking, vlm_thinking_mode], outputs=[])
vlm_keep_thinking.change(fn=update_vlm_params, inputs=[vlm_max_tokens, vlm_num_beams, vlm_temperature, vlm_do_sample, vlm_top_k, vlm_top_p, vlm_keep_prefill, vlm_keep_thinking, vlm_thinking_mode], outputs=[])
vlm_thinking_mode.change(fn=update_vlm_params, inputs=[vlm_max_tokens, vlm_num_beams, vlm_temperature, vlm_do_sample, vlm_top_k, vlm_top_p, vlm_keep_prefill, vlm_keep_thinking, vlm_thinking_mode], outputs=[])
with gr.Accordion(label='Batch caption', open=False, visible=True):
with gr.Row():
vlm_batch_files = gr.File(label="Files", show_label=True, file_count='multiple', file_types=['image'], interactive=True, height=100, elem_id='vlm_batch_files')
@@ -118,6 +159,8 @@ def create_ui():
with gr.Column(variant='compact', elem_id='interrogate_output'):
with gr.Row(elem_id='interrogate_output_prompt'):
prompt = gr.Textbox(label="Answer", lines=12, placeholder="ai generated image description")
with gr.Row(elem_id='interrogate_output_image'):
output_image = gr.Image(type='pil', label="Annotated Image", interactive=False, visible=False, elem_id='interrogate_output_image_display')
with gr.Row(elem_id='interrogate_output_classes'):
medium = gr.Label(elem_id="interrogate_label_medium", label="Medium", num_top_classes=5, visible=False)
artist = gr.Label(elem_id="interrogate_label_artist", label="Artist", num_top_classes=5, visible=False)
@@ -127,11 +170,19 @@ def create_ui():
with gr.Row(elem_id='copy_buttons_interrogate'):
copy_interrogate_buttons = generation_parameters_copypaste.create_buttons(["txt2img", "img2img", "control", "extras"])
btn_clip_interrogate_img.click(openclip.interrogate_image, inputs=[image, clip_model, blip_model, clip_mode], outputs=[prompt])
btn_clip_analyze_img.click(openclip.analyze_image, inputs=[image, clip_model, blip_model], outputs=[medium, artist, movement, trending, flavor])
btn_clip_interrogate_batch.click(fn=openclip.interrogate_batch, inputs=[clip_batch_files, clip_batch_folder, clip_batch_str, clip_model, blip_model, clip_mode, clip_save_output, clip_save_append, clip_folder_recursive], outputs=[prompt])
btn_vlm_caption.click(fn=vqa.interrogate, inputs=[vlm_question, vlm_system, vlm_prompt, image, vlm_model], outputs=[prompt])
btn_vlm_caption_batch.click(fn=vqa.batch, inputs=[vlm_model, vlm_system, vlm_batch_files, vlm_batch_folder, vlm_batch_str, vlm_question, vlm_prompt, vlm_save_output, vlm_save_append, vlm_folder_recursive], outputs=[prompt])
btn_clip_interrogate_img.click(openclip.interrogate_image, inputs=[image, clip_model, blip_model, clip_mode], outputs=[prompt]).then(fn=lambda: gr.update(visible=False), inputs=[], outputs=[output_image])
btn_clip_analyze_img.click(openclip.analyze_image, inputs=[image, clip_model, blip_model], outputs=[medium, artist, movement, trending, flavor]).then(fn=lambda: gr.update(visible=False), inputs=[], outputs=[output_image])
btn_clip_interrogate_batch.click(fn=openclip.interrogate_batch, inputs=[clip_batch_files, clip_batch_folder, clip_batch_str, clip_model, blip_model, clip_mode, clip_save_output, clip_save_append, clip_folder_recursive], outputs=[prompt]).then(fn=lambda: gr.update(visible=False), inputs=[], outputs=[output_image])
btn_vlm_caption.click(fn=vlm_caption_wrapper, inputs=[vlm_question, vlm_system, vlm_prompt, image, vlm_model, vlm_prefill, vlm_thinking_mode], outputs=[prompt, output_image])
btn_vlm_caption_batch.click(fn=vqa.batch, inputs=[vlm_model, vlm_system, vlm_batch_files, vlm_batch_folder, vlm_batch_str, vlm_question, vlm_prompt, vlm_save_output, vlm_save_append, vlm_folder_recursive, vlm_prefill, vlm_thinking_mode], outputs=[prompt]).then(fn=lambda: gr.update(visible=False), inputs=[], outputs=[output_image])
# Dynamic UI updates based on selected model and task
vlm_model.change(fn=update_vlm_prompts_for_model, inputs=[vlm_model], outputs=[vlm_question])
vlm_question.change(fn=update_vlm_prompt_placeholder, inputs=[vlm_question], outputs=[vlm_prompt])
# Load/Unload model buttons
vlm_load_btn.click(fn=vqa.load_model, inputs=[vlm_model], outputs=[])
vlm_unload_btn.click(fn=vqa.unload_model, inputs=[], outputs=[])
for tabname, button in copy_interrogate_buttons.items():
generation_parameters_copypaste.register_paste_params_button(generation_parameters_copypaste.ParamBinding(paste_button=button, tabname=tabname, source_text_component=prompt, source_image_component=image,))