From 929c3e9e1edbeb709ac0f51aa5eaf5bc2b062689 Mon Sep 17 00:00:00 2001 From: Roland Tannous Date: Wed, 4 Mar 2026 06:42:37 +0000 Subject: [PATCH 01/10] fix: cast URL image columns to HF Image() type in VLM conversion --- studio/backend/utils/datasets/format_conversion.py | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/studio/backend/utils/datasets/format_conversion.py b/studio/backend/utils/datasets/format_conversion.py index 6436e7a82a..4e6a6d9919 100644 --- a/studio/backend/utils/datasets/format_conversion.py +++ b/studio/backend/utils/datasets/format_conversion.py @@ -254,8 +254,15 @@ def convert_to_vlm_format( list: List of dicts with 'messages' field """ from PIL import Image + from datasets import Image as datasets_Image from .vlm_processing import generate_smart_vlm_instruction + # Cast string image columns (URLs or local paths) to HF Image() type + # so HuggingFace handles downloading, decoding, and caching transparently. + sample_value = next(iter(dataset))[image_column] + if isinstance(sample_value, str): + dataset = dataset.cast_column(image_column, datasets_Image()) + # Generate smart instruction if not provided if instruction is None: instruction_info = generate_smart_vlm_instruction( From 2b704221f7add28d0796229c77ce3256a917093b Mon Sep 17 00:00:00 2001 From: Roland Tannous Date: Wed, 4 Mar 2026 06:42:48 +0000 Subject: [PATCH 02/10] fix: abort training pipeline on dataset conversion failure --- studio/backend/core/training/trainer.py | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/studio/backend/core/training/trainer.py b/studio/backend/core/training/trainer.py index 5ce273f43c..f5b2f18245 100644 --- a/studio/backend/core/training/trainer.py +++ b/studio/backend/core/training/trainer.py @@ -482,6 +482,14 @@ class UnslothTrainer: print("Stopped during dataset formatting\n") return None + # Abort if dataset formatting/conversion failed + if not dataset_info.get("success", True): + errors = dataset_info.get("errors", []) + error_msg = "; ".join(errors) if errors else "Dataset formatting failed" + logger.error(f"Dataset conversion failed: {error_msg}") + self._update_progress(error=error_msg) + return None + self._update_progress(status_message=f"Dataset formatted and ready for training") print(f"Dataset formatted successfully\n") From 8039eebcd5fb7a0158a5c94d55a742da344e3795 Mon Sep 17 00:00:00 2001 From: Roland Tannous Date: Wed, 4 Mar 2026 07:39:35 +0000 Subject: [PATCH 03/10] test: add URL image loading comparison script --- studio/tests/test_url_image_loading.py | 132 +++++++++++++++++++++++++ 1 file changed, 132 insertions(+) create mode 100644 studio/tests/test_url_image_loading.py diff --git a/studio/tests/test_url_image_loading.py b/studio/tests/test_url_image_loading.py new file mode 100644 index 0000000000..4899bab8e3 --- /dev/null +++ b/studio/tests/test_url_image_loading.py @@ -0,0 +1,132 @@ +""" +Reproduce: VLM URL image loading with HF datasets. +Tests cast_column(Image()) vs manual download approaches. +Dataset: google-research-datasets/conceptual_captions (subset: labeled) +""" +from datasets import load_dataset, Image as datasets_Image, Dataset +from PIL import Image as PILImage +from io import BytesIO +from itertools import islice +import time + +DATASET = "google-research-datasets/conceptual_captions" +SUBSET = "labeled" +SPLIT = "train" +N_SAMPLES = 20 # small slice for testing + +print("=" * 60) +print("Loading dataset (streaming, first N samples)...") +print("=" * 60) +ds = load_dataset(DATASET, name=SUBSET, split=SPLIT, streaming=True) +rows = list(islice(ds, N_SAMPLES)) +dataset = Dataset.from_list(rows) + +print(f"Loaded {len(dataset)} samples") +print(f"Columns: {dataset.column_names}") +print(f"First image_url: {dataset[0]['image_url'][:100]}...") +print() + +# ─── Test 1: cast_column(Image()) — what we tried ─── +print("=" * 60) +print("TEST 1: cast_column(Image()) approach") +print("=" * 60) +try: + ds_cast = dataset.cast_column("image_url", datasets_Image()) + success, fail = 0, 0 + t0 = time.time() + for i, sample in enumerate(ds_cast): + try: + img = sample["image_url"] + if img is not None: + print(f" [{i}] OK — {img.size} {img.mode}") + success += 1 + else: + print(f" [{i}] None returned") + fail += 1 + except Exception as e: + print(f" [{i}] FAILED: {type(e).__name__}: {str(e)[:80]}") + fail += 1 + elapsed = time.time() - t0 + print(f"\nResult: {success} ok, {fail} failed, {elapsed:.1f}s") +except Exception as e: + print(f"CRASHED during iteration: {type(e).__name__}: {str(e)[:120]}") +print() + +# ─── Test 2: Manual download with requests.Session ─── +print("=" * 60) +print("TEST 2: requests.Session() approach") +print("=" * 60) +try: + import requests + session = requests.Session() + success, fail = 0, 0 + t0 = time.time() + for i, sample in enumerate(dataset): + url = sample["image_url"] + try: + resp = session.get(url, timeout=10) + resp.raise_for_status() + img = PILImage.open(BytesIO(resp.content)).convert("RGB") + print(f" [{i}] OK — {img.size} {img.mode}") + success += 1 + except Exception as e: + print(f" [{i}] FAILED: {type(e).__name__}: {str(e)[:80]}") + fail += 1 + elapsed = time.time() - t0 + print(f"\nResult: {success} ok, {fail} failed, {elapsed:.1f}s") +except Exception as e: + print(f"CRASHED: {type(e).__name__}: {str(e)[:120]}") +print() + +# ─── Test 3: urllib (stdlib) ─── +print("=" * 60) +print("TEST 3: urllib approach (stdlib)") +print("=" * 60) +try: + from urllib.request import urlopen, Request + success, fail = 0, 0 + t0 = time.time() + for i, sample in enumerate(dataset): + url = sample["image_url"] + try: + req = Request(url, headers={"User-Agent": "Mozilla/5.0"}) + with urlopen(req, timeout=10) as resp: + img = PILImage.open(BytesIO(resp.read())).convert("RGB") + print(f" [{i}] OK — {img.size} {img.mode}") + success += 1 + except Exception as e: + print(f" [{i}] FAILED: {type(e).__name__}: {str(e)[:80]}") + fail += 1 + elapsed = time.time() - t0 + print(f"\nResult: {success} ok, {fail} failed, {elapsed:.1f}s") +except Exception as e: + print(f"CRASHED: {type(e).__name__}: {str(e)[:120]}") +print() + +# ─── Test 4: fsspec directly with expand=True ─── +print("=" * 60) +print("TEST 4: fsspec.open() with expand=True") +print("=" * 60) +try: + import fsspec + success, fail = 0, 0 + t0 = time.time() + for i, sample in enumerate(dataset): + url = sample["image_url"] + try: + with fsspec.open(url, "rb", expand=True) as f: + img = PILImage.open(BytesIO(f.read())).convert("RGB") + print(f" [{i}] OK — {img.size} {img.mode}") + success += 1 + except Exception as e: + print(f" [{i}] FAILED: {type(e).__name__}: {str(e)[:80]}") + fail += 1 + elapsed = time.time() - t0 + print(f"\nResult: {success} ok, {fail} failed, {elapsed:.1f}s") +except Exception as e: + print(f"CRASHED: {type(e).__name__}: {str(e)[:120]}") + +print() +print("=" * 60) +print("DONE — compare success rates and timing above") +print("=" * 60) From fdc23f4a43d0f77d9ba10ba88aabd88532ff09a6 Mon Sep 17 00:00:00 2001 From: Roland Tannous Date: Wed, 4 Mar 2026 07:50:55 +0000 Subject: [PATCH 04/10] fix: use fsspec for URL image downloads with per-sample error handling --- .../utils/datasets/format_conversion.py | 50 +++++++++++++------ 1 file changed, 36 insertions(+), 14 deletions(-) diff --git a/studio/backend/utils/datasets/format_conversion.py b/studio/backend/utils/datasets/format_conversion.py index 4e6a6d9919..37bd3c103c 100644 --- a/studio/backend/utils/datasets/format_conversion.py +++ b/studio/backend/utils/datasets/format_conversion.py @@ -254,15 +254,8 @@ def convert_to_vlm_format( list: List of dicts with 'messages' field """ from PIL import Image - from datasets import Image as datasets_Image from .vlm_processing import generate_smart_vlm_instruction - # Cast string image columns (URLs or local paths) to HF Image() type - # so HuggingFace handles downloading, decoding, and caching transparently. - sample_value = next(iter(dataset))[image_column] - if isinstance(sample_value, str): - dataset = dataset.cast_column(image_column, datasets_Image()) - # Generate smart instruction if not provided if instruction is None: instruction_info = generate_smart_vlm_instruction( @@ -288,12 +281,17 @@ def convert_to_vlm_format( def _convert_single_sample(sample): """Convert a single sample to VLM format.""" - # Get image (might be PIL Image or path) + # Get image (might be PIL Image, local path, or URL) image_data = sample[image_column] - # Handle image paths if isinstance(image_data, str): - image_data = Image.open(image_data).convert("RGB") + if image_data.startswith(("http://", "https://")): + import fsspec + from io import BytesIO + with fsspec.open(image_data, "rb", expand=True) as f: + image_data = Image.open(BytesIO(f.read())).convert("RGB") + else: + image_data = Image.open(image_data).convert("RGB") # Get text text_data = sample[text_column] @@ -324,11 +322,35 @@ def convert_to_vlm_format( # Return dict with messages return {"messages": messages} - # Use list comprehension and return the LIST directly - print(f"🔄 Converting {len(dataset)} samples to VLM format...") - converted_list = [_convert_single_sample(sample) for sample in dataset] + # Convert samples, skipping any with broken/unreachable images + total = len(dataset) + print(f"🔄 Converting {total} samples to VLM format...") + converted_list = [] + failed_count = 0 + for sample in dataset: + try: + converted_list.append(_convert_single_sample(sample)) + except Exception as e: + failed_count += 1 - print(f"✅ Converted {len(converted_list)} samples") + if failed_count > 0: + fail_rate = failed_count / total + print(f"⚠️ Skipped {failed_count}/{total} ({fail_rate:.0%}) samples with broken/unreachable images") + + if fail_rate >= 0.3: + raise ValueError( + f"{fail_rate:.0%} of images failed to download ({failed_count}/{total}). " + "This dataset has too many broken or unreachable image URLs to be usable for training. " + "Consider using a dataset with embedded images instead." + ) + + if len(converted_list) == 0: + raise ValueError( + f"All {total} samples failed during VLM conversion — no usable images found. " + "This dataset may contain only image URLs that are no longer accessible." + ) + + print(f"✅ Converted {len(converted_list)}/{total} samples") # Return list, NOT Dataset return converted_list From 50885a7aa3cb376c1f59c1f3128575b08c18d55e Mon Sep 17 00:00:00 2001 From: Roland Tannous Date: Wed, 4 Mar 2026 08:05:40 +0000 Subject: [PATCH 05/10] fix: add early probe to fail fast on datasets with too many broken image URLs --- .../utils/datasets/format_conversion.py | 32 +++++++++++++------ 1 file changed, 23 insertions(+), 9 deletions(-) diff --git a/studio/backend/utils/datasets/format_conversion.py b/studio/backend/utils/datasets/format_conversion.py index 37bd3c103c..f9da8b1a1e 100644 --- a/studio/backend/utils/datasets/format_conversion.py +++ b/studio/backend/utils/datasets/format_conversion.py @@ -322,28 +322,42 @@ def convert_to_vlm_format( # Return dict with messages return {"messages": messages} - # Convert samples, skipping any with broken/unreachable images + # Convert samples, skipping any with broken/unreachable images. + # For URL-based datasets, check the first PROBE_SIZE samples early to + # fail fast if too many images are broken, before downloading millions. + PROBE_SIZE = 5000 + MAX_FAIL_RATE = 0.3 + total = len(dataset) + has_urls = isinstance(next(iter(dataset))[image_column], str) + probe_needed = has_urls and total > PROBE_SIZE + print(f"🔄 Converting {total} samples to VLM format...") converted_list = [] failed_count = 0 - for sample in dataset: + + for i, sample in enumerate(dataset): try: converted_list.append(_convert_single_sample(sample)) except Exception as e: failed_count += 1 + # Early exit check after probing the first batch + if probe_needed and (i + 1) == PROBE_SIZE: + fail_rate = failed_count / PROBE_SIZE + if fail_rate >= MAX_FAIL_RATE: + raise ValueError( + f"{fail_rate:.0%} of the first {PROBE_SIZE} images failed to download " + f"({failed_count}/{PROBE_SIZE}). " + "This dataset has too many broken or unreachable image URLs. " + "Consider using a dataset with embedded images instead." + ) + print(f"✅ Probe passed: {failed_count}/{PROBE_SIZE} ({fail_rate:.0%}) failures in first batch, continuing...") + if failed_count > 0: fail_rate = failed_count / total print(f"⚠️ Skipped {failed_count}/{total} ({fail_rate:.0%}) samples with broken/unreachable images") - if fail_rate >= 0.3: - raise ValueError( - f"{fail_rate:.0%} of images failed to download ({failed_count}/{total}). " - "This dataset has too many broken or unreachable image URLs to be usable for training. " - "Consider using a dataset with embedded images instead." - ) - if len(converted_list) == 0: raise ValueError( f"All {total} samples failed during VLM conversion — no usable images found. " From f59eaad212789ba1cc4da15274ee9d2c6fc680a7 Mon Sep 17 00:00:00 2001 From: Roland Tannous Date: Wed, 4 Mar 2026 13:30:27 +0000 Subject: [PATCH 06/10] feat: add tqdm progress bar to VLM conversion and download benchmark test --- .../utils/datasets/format_conversion.py | 9 +++- studio/tests/test_url_download_benchmark.py | 54 +++++++++++++++++++ 2 files changed, 62 insertions(+), 1 deletion(-) create mode 100644 studio/tests/test_url_download_benchmark.py diff --git a/studio/backend/utils/datasets/format_conversion.py b/studio/backend/utils/datasets/format_conversion.py index f9da8b1a1e..e784e8e645 100644 --- a/studio/backend/utils/datasets/format_conversion.py +++ b/studio/backend/utils/datasets/format_conversion.py @@ -332,20 +332,26 @@ def convert_to_vlm_format( has_urls = isinstance(next(iter(dataset))[image_column], str) probe_needed = has_urls and total > PROBE_SIZE + from tqdm import tqdm + print(f"🔄 Converting {total} samples to VLM format...") converted_list = [] failed_count = 0 - for i, sample in enumerate(dataset): + pbar = tqdm(dataset, total=total, desc="Converting VLM samples", unit="sample") + for i, sample in enumerate(pbar): try: converted_list.append(_convert_single_sample(sample)) except Exception as e: failed_count += 1 + pbar.set_postfix(ok=len(converted_list), failed=failed_count, refresh=False) + # Early exit check after probing the first batch if probe_needed and (i + 1) == PROBE_SIZE: fail_rate = failed_count / PROBE_SIZE if fail_rate >= MAX_FAIL_RATE: + pbar.close() raise ValueError( f"{fail_rate:.0%} of the first {PROBE_SIZE} images failed to download " f"({failed_count}/{PROBE_SIZE}). " @@ -353,6 +359,7 @@ def convert_to_vlm_format( "Consider using a dataset with embedded images instead." ) print(f"✅ Probe passed: {failed_count}/{PROBE_SIZE} ({fail_rate:.0%}) failures in first batch, continuing...") + pbar.close() if failed_count > 0: fail_rate = failed_count / total diff --git a/studio/tests/test_url_download_benchmark.py b/studio/tests/test_url_download_benchmark.py new file mode 100644 index 0000000000..e29a8e05b6 --- /dev/null +++ b/studio/tests/test_url_download_benchmark.py @@ -0,0 +1,54 @@ +""" +Benchmark: fsspec URL image download throughput at different dataset sizes. +Dataset: google-research-datasets/conceptual_captions (subset: labeled) + +Tests sizes: 100, 200, 300, 500, 1000, 1500, 2000 +Reports: time, success/fail rate, throughput (images/sec) +""" +from datasets import load_dataset, Dataset +from PIL import Image as PILImage +from io import BytesIO +from itertools import islice +import fsspec +import time + +DATASET = "google-research-datasets/conceptual_captions" +SUBSET = "labeled" +SPLIT = "train" +SIZES = [100, 200, 300, 500, 1000, 1500, 2000] + +# Load the max we need in one go +max_size = max(SIZES) +print(f"Loading {max_size} samples from {DATASET} (streaming)...") +ds = load_dataset(DATASET, name=SUBSET, split=SPLIT, streaming=True) +rows = list(islice(ds, max_size)) +full_dataset = Dataset.from_list(rows) +print(f"Loaded {len(full_dataset)} samples") +print(f"Columns: {full_dataset.column_names}") +print() + +print(f"{'Size':>6} | {'Time':>8} | {'OK':>6} | {'Fail':>6} | {'Fail%':>6} | {'img/s':>7}") +print("-" * 55) + +for size in SIZES: + dataset = full_dataset.select(range(size)) + success, fail = 0, 0 + t0 = time.time() + + for sample in dataset: + url = sample["image_url"] + try: + with fsspec.open(url, "rb", expand=True) as f: + img = PILImage.open(BytesIO(f.read())).convert("RGB") + success += 1 + except Exception: + fail += 1 + + elapsed = time.time() - t0 + fail_pct = (fail / size) * 100 + throughput = success / elapsed if elapsed > 0 else 0 + + print(f"{size:>6} | {elapsed:>7.1f}s | {success:>6} | {fail:>6} | {fail_pct:>5.1f}% | {throughput:>6.1f}/s") + +print() +print("Done.") From 195c1a3ce38459422880665d6dd33c97e32242e8 Mon Sep 17 00:00:00 2001 From: Roland Tannous Date: Wed, 4 Mar 2026 14:30:11 +0000 Subject: [PATCH 07/10] test: add parallel download benchmark with ThreadPoolExecutor --- studio/tests/test_url_parallel_benchmark.py | 79 +++++++++++++++++++++ 1 file changed, 79 insertions(+) create mode 100644 studio/tests/test_url_parallel_benchmark.py diff --git a/studio/tests/test_url_parallel_benchmark.py b/studio/tests/test_url_parallel_benchmark.py new file mode 100644 index 0000000000..a0d160136c --- /dev/null +++ b/studio/tests/test_url_parallel_benchmark.py @@ -0,0 +1,79 @@ +""" +Benchmark: parallel fsspec URL image downloads with ThreadPoolExecutor. +Tests different worker counts to find optimal parallelism. +Dataset: google-research-datasets/conceptual_captions (subset: labeled) +""" +from datasets import load_dataset, Dataset +from PIL import Image as PILImage +from io import BytesIO +from itertools import islice +from concurrent.futures import ThreadPoolExecutor, as_completed +import fsspec +import time +import os + +DATASET = "google-research-datasets/conceptual_captions" +SUBSET = "labeled" +SPLIT = "train" +N_SAMPLES = 500 + +# safe_num_proc formula from studio/backend/utils/hardware/hardware.py +cpu_count = os.cpu_count() +safe_workers = max(1, cpu_count // 3) +print(f"CPU count: {cpu_count}, safe_num_proc: {safe_workers}") + +WORKER_COUNTS = [1, 4, 8, 16, 32, safe_workers] +# Deduplicate and sort +WORKER_COUNTS = sorted(set(WORKER_COUNTS)) + +print(f"Loading {N_SAMPLES} samples from {DATASET} (streaming)...") +ds = load_dataset(DATASET, name=SUBSET, split=SPLIT, streaming=True) +rows = list(islice(ds, N_SAMPLES)) +dataset = Dataset.from_list(rows) +urls = [row["image_url"] for row in dataset] +print(f"Loaded {len(urls)} URLs") +print() + + +def download_single(url): + """Download a single image URL using fsspec. Returns PIL image or raises.""" + with fsspec.open(url, "rb", expand=True) as f: + img = PILImage.open(BytesIO(f.read())).convert("RGB") + return img + + +print(f"{'Workers':>8} | {'Time':>8} | {'OK':>6} | {'Fail':>6} | {'Fail%':>6} | {'img/s':>7} | {'Speedup':>8}") +print("-" * 70) + +baseline_throughput = None + +for n_workers in WORKER_COUNTS: + success, fail = 0, 0 + t0 = time.time() + + with ThreadPoolExecutor(max_workers=n_workers) as pool: + futures = {pool.submit(download_single, url): url for url in urls} + for future in as_completed(futures): + try: + img = future.result(timeout=30) + success += 1 + except Exception: + fail += 1 + + elapsed = time.time() - t0 + fail_pct = (fail / N_SAMPLES) * 100 + throughput = success / elapsed if elapsed > 0 else 0 + + if baseline_throughput is None: + baseline_throughput = throughput + speedup = throughput / baseline_throughput if baseline_throughput > 0 else 0 + + label = f"{n_workers}" + if n_workers == safe_workers: + label += "*" # mark the safe_num_proc value + + print(f"{label:>8} | {elapsed:>7.1f}s | {success:>6} | {fail:>6} | {fail_pct:>5.1f}% | {throughput:>6.1f}/s | {speedup:>7.1f}x") + +print() +print("* = safe_num_proc value") +print("Done.") From 9ca45826d4ebdb9972df255696b69e3d45d9846a Mon Sep 17 00:00:00 2001 From: Roland Tannous Date: Wed, 4 Mar 2026 23:40:38 +0000 Subject: [PATCH 08/10] feat: parallel URL image probe with time estimate and progress reporting - Add 200-sample parallel probe using ThreadPoolExecutor + safe_num_proc to estimate download speed and failure rate before full conversion - Abort with clear error if >=30% of probe images fail to download - Show estimated download time in the training overlay modal - Parallel batch conversion for URL-based datasets (vs sequential for local) - Add warning field to /check-format response for URL-based image datasets - Display URL warning in dataset preview dialog (amber banner) - Thread progress_callback from trainer through format_and_template_dataset to convert_to_vlm_format for real-time status updates --- studio/backend/core/training/trainer.py | 1 + studio/backend/models/datasets.py | 1 + studio/backend/routes/datasets.py | 16 ++ .../backend/utils/datasets/dataset_utils.py | 3 + .../utils/datasets/format_conversion.py | 167 +++++++++++++++--- .../sections/dataset-preview-dialog.tsx | 7 + .../src/features/training/types/datasets.ts | 1 + 7 files changed, 169 insertions(+), 27 deletions(-) diff --git a/studio/backend/core/training/trainer.py b/studio/backend/core/training/trainer.py index f5b2f18245..f2e43f76f4 100644 --- a/studio/backend/core/training/trainer.py +++ b/studio/backend/core/training/trainer.py @@ -475,6 +475,7 @@ class UnslothTrainer: format_type=format_type, dataset_name=dataset_source, custom_format_mapping=custom_format_mapping, + progress_callback=self._update_progress, ) # Check if stopped during formatting diff --git a/studio/backend/models/datasets.py b/studio/backend/models/datasets.py index 81adef7577..18f6ec224b 100644 --- a/studio/backend/models/datasets.py +++ b/studio/backend/models/datasets.py @@ -34,3 +34,4 @@ class CheckFormatResponse(BaseModel): detected_text_column: Optional[str] = None preview_samples: Optional[List[Dict]] = None total_rows: Optional[int] = None + warning: Optional[str] = None diff --git a/studio/backend/routes/datasets.py b/studio/backend/routes/datasets.py index a53e223015..4a475ff8c2 100644 --- a/studio/backend/routes/datasets.py +++ b/studio/backend/routes/datasets.py @@ -211,6 +211,21 @@ def check_format( else: preview_samples = _serialize_preview_rows(preview_slice) + # Lightweight URL-based image detection for VLM datasets + warning = None + image_col = result.get("detected_image_column") + if image_col and image_col in (result.get("columns") or []): + try: + sample_val = preview_slice[0][image_col] + if isinstance(sample_val, str) and sample_val.startswith(("http://", "https://")): + warning = ( + "This dataset contains image URLs instead of embedded images. " + "Images will be downloaded during training, which may be slow for large datasets." + ) + logger.info(f"URL-based image column detected: {image_col}") + except Exception: + pass + return CheckFormatResponse( requires_manual_mapping=result["requires_manual_mapping"], detected_format=result["detected_format"], @@ -222,6 +237,7 @@ def check_format( detected_text_column=result.get("detected_text_column"), preview_samples=preview_samples, total_rows=total_rows, + warning=warning, ) except HTTPException: diff --git a/studio/backend/utils/datasets/dataset_utils.py b/studio/backend/utils/datasets/dataset_utils.py index 3a4d54f93f..9e1f54a75c 100644 --- a/studio/backend/utils/datasets/dataset_utils.py +++ b/studio/backend/utils/datasets/dataset_utils.py @@ -593,6 +593,7 @@ def format_and_template_dataset( aliases_for_assistant=["gpt", "assistant", "output",], batch_size=1000, num_proc=None, + progress_callback=None, ): """ Convenience function that combines format_dataset and apply_chat_template_to_dataset. @@ -638,6 +639,7 @@ def format_and_template_dataset( text_column=user_vlm_text_column, image_column=user_vlm_image_column, dataset_name=dataset_name, + progress_callback=progress_callback, ) warnings.append(f"Applied user VLM mapping: image='{user_vlm_image_column}', text='{user_vlm_text_column}'") @@ -734,6 +736,7 @@ def format_and_template_dataset( text_column=vlm_text_column, image_column=vlm_image_column, dataset_name=dataset_name, + progress_callback=progress_callback, ) if vlm_instruction: diff --git a/studio/backend/utils/datasets/format_conversion.py b/studio/backend/utils/datasets/format_conversion.py index e784e8e645..c5c9a4d6e7 100644 --- a/studio/backend/utils/datasets/format_conversion.py +++ b/studio/backend/utils/datasets/format_conversion.py @@ -238,24 +238,51 @@ def convert_alpaca_to_chatml(dataset, batch_size=1000, num_proc=None): return dataset.map(_convert, **dataset_map_kwargs) +def _format_eta(seconds): + """Format seconds into a human-readable ETA string.""" + if seconds < 60: + return f"{seconds:.0f}s" + elif seconds < 3600: + m, s = divmod(int(seconds), 60) + return f"{m}m {s}s" + else: + h, remainder = divmod(int(seconds), 3600) + m, _ = divmod(remainder, 60) + return f"{h}h {m}m" + + def convert_to_vlm_format( dataset, instruction=None, text_column="text", image_column="image", dataset_name=None, + progress_callback=None, ): """ Converts simple {image, text} format to VLM messages format. Returns a LIST, not a HuggingFace Dataset (to preserve PIL Images). + For URL-based image datasets, runs a 200-sample parallel probe first to + estimate download speed and failure rate, then reports time estimate or + warning through progress_callback before proceeding with the full conversion. + + Args: + progress_callback: Optional callable(status_message=str) to report + progress to the training overlay. + Returns: list: List of dicts with 'messages' field """ from PIL import Image from .vlm_processing import generate_smart_vlm_instruction + def _notify(msg): + """Send status update to the training overlay if callback is available.""" + if progress_callback: + progress_callback(status_message=msg) + # Generate smart instruction if not provided if instruction is None: instruction_info = generate_smart_vlm_instruction( @@ -322,48 +349,133 @@ def convert_to_vlm_format( # Return dict with messages return {"messages": messages} - # Convert samples, skipping any with broken/unreachable images. - # For URL-based datasets, check the first PROBE_SIZE samples early to - # fail fast if too many images are broken, before downloading millions. - PROBE_SIZE = 5000 - MAX_FAIL_RATE = 0.3 - total = len(dataset) has_urls = isinstance(next(iter(dataset))[image_column], str) - probe_needed = has_urls and total > PROBE_SIZE + # ── URL probe: 200 samples with parallel workers to estimate speed + failure rate ── + PROBE_SIZE = 200 + MAX_FAIL_RATE = 0.3 + + if has_urls and total > PROBE_SIZE: + import time + from concurrent.futures import ThreadPoolExecutor, as_completed + from utils.hardware import safe_num_proc + + num_workers = safe_num_proc() + _notify(f"Probing {PROBE_SIZE} image URLs with {num_workers} workers...") + print(f"🔍 Probing {PROBE_SIZE}/{total} image URLs with {num_workers} workers...") + + probe_samples = [dataset[i] for i in range(PROBE_SIZE)] + probe_ok = 0 + probe_fail = 0 + probe_start = time.time() + + with ThreadPoolExecutor(max_workers=num_workers) as executor: + futures = {executor.submit(_convert_single_sample, s): s for s in probe_samples} + for future in as_completed(futures): + try: + future.result() + probe_ok += 1 + except Exception: + probe_fail += 1 + + probe_elapsed = time.time() - probe_start + probe_total = probe_ok + probe_fail + fail_rate = probe_fail / probe_total if probe_total > 0 else 0 + throughput = probe_total / probe_elapsed if probe_elapsed > 0 else 0 + + if fail_rate >= MAX_FAIL_RATE: + msg = ( + f"⚠️ {fail_rate:.0%} of the first {PROBE_SIZE} images failed to download " + f"({probe_fail}/{probe_total}). " + "This dataset has too many broken or unreachable image URLs. " + "Consider using a dataset with embedded images instead." + ) + print(msg) + _notify(msg) + raise ValueError(msg) + + # Estimate total time for remaining samples + remaining = total - PROBE_SIZE + estimated_seconds = remaining / throughput if throughput > 0 else 0 + eta_str = _format_eta(estimated_seconds) + + info_msg = ( + f"Downloading {total:,} images ({num_workers} workers, ~{throughput:.1f} img/s). " + f"Estimated time: ~{eta_str}" + ) + if probe_fail > 0: + info_msg += f" | {fail_rate:.0%} broken URLs will be skipped" + + print(f"✅ Probe passed: {probe_ok}/{probe_total} ok, {probe_fail} failed ({fail_rate:.0%}), {throughput:.1f} img/s") + print(f"⏱️ Estimated time for {total:,} samples: ~{eta_str}") + _notify(info_msg) + + # ── Full conversion with progress ── from tqdm import tqdm print(f"🔄 Converting {total} samples to VLM format...") converted_list = [] failed_count = 0 - pbar = tqdm(dataset, total=total, desc="Converting VLM samples", unit="sample") - for i, sample in enumerate(pbar): - try: - converted_list.append(_convert_single_sample(sample)) - except Exception as e: - failed_count += 1 + if has_urls: + # Parallel conversion for URL-based datasets + import time + from concurrent.futures import ThreadPoolExecutor, as_completed + from utils.hardware import safe_num_proc - pbar.set_postfix(ok=len(converted_list), failed=failed_count, refresh=False) + num_workers = safe_num_proc() + batch_size = 500 + start_time = time.time() - # Early exit check after probing the first batch - if probe_needed and (i + 1) == PROBE_SIZE: - fail_rate = failed_count / PROBE_SIZE - if fail_rate >= MAX_FAIL_RATE: - pbar.close() - raise ValueError( - f"{fail_rate:.0%} of the first {PROBE_SIZE} images failed to download " - f"({failed_count}/{PROBE_SIZE}). " - "This dataset has too many broken or unreachable image URLs. " - "Consider using a dataset with embedded images instead." - ) - print(f"✅ Probe passed: {failed_count}/{PROBE_SIZE} ({fail_rate:.0%}) failures in first batch, continuing...") - pbar.close() + for batch_start in range(0, total, batch_size): + batch_end = min(batch_start + batch_size, total) + batch_samples = [dataset[i] for i in range(batch_start, batch_end)] + + with ThreadPoolExecutor(max_workers=num_workers) as executor: + futures = {executor.submit(_convert_single_sample, s): i for i, s in enumerate(batch_samples)} + batch_results = [None] * len(batch_samples) + for future in as_completed(futures): + idx = futures[future] + try: + batch_results[idx] = future.result() + except Exception: + failed_count += 1 + + converted_list.extend(r for r in batch_results if r is not None) + + # Progress update every batch + elapsed = time.time() - start_time + done = batch_end + rate = done / elapsed if elapsed > 0 else 0 + remaining_time = (total - done) / rate if rate > 0 else 0 + eta_str = _format_eta(remaining_time) + progress_msg = f"Downloading images: {done:,}/{total:,} ({done*100//total}%) | ~{eta_str} remaining | {failed_count} skipped" + print(f" [{done}/{total}] {rate:.1f} img/s, {failed_count} failed, ETA {eta_str}") + _notify(progress_msg) + else: + # Sequential conversion for local/embedded images (fast, no I/O bottleneck) + pbar = tqdm(dataset, total=total, desc="Converting VLM samples", unit="sample") + for sample in pbar: + try: + converted_list.append(_convert_single_sample(sample)) + except Exception: + failed_count += 1 + pbar.set_postfix(ok=len(converted_list), failed=failed_count, refresh=False) + pbar.close() if failed_count > 0: fail_rate = failed_count / total print(f"⚠️ Skipped {failed_count}/{total} ({fail_rate:.0%}) samples with broken/unreachable images") + # For datasets that skipped the probe (small URL datasets), check fail rate now + if has_urls and fail_rate >= MAX_FAIL_RATE: + msg = ( + f"⚠️ {fail_rate:.0%} of images failed to download ({failed_count}/{total}). " + "This dataset has too many broken or unreachable image URLs. " + "Consider using a dataset with embedded images instead." + ) + _notify(msg) + raise ValueError(msg) if len(converted_list) == 0: raise ValueError( @@ -372,6 +484,7 @@ def convert_to_vlm_format( ) print(f"✅ Converted {len(converted_list)}/{total} samples") + _notify(f"Converted {len(converted_list):,}/{total:,} images successfully") # Return list, NOT Dataset return converted_list diff --git a/studio/frontend/src/features/studio/sections/dataset-preview-dialog.tsx b/studio/frontend/src/features/studio/sections/dataset-preview-dialog.tsx index 00a738bc64..516f3612c0 100644 --- a/studio/frontend/src/features/studio/sections/dataset-preview-dialog.tsx +++ b/studio/frontend/src/features/studio/sections/dataset-preview-dialog.tsx @@ -340,6 +340,13 @@ export function DatasetPreviewDialog({ /> + {data.warning && ( +
+ + {data.warning} +
+ )} + {mappingEnabled && ( Date: Thu, 5 Mar 2026 06:06:47 +0000 Subject: [PATCH 09/10] fix: remove benchmark scripts from git tracking These are standalone benchmark scripts that were force-added despite being gitignored. They have no test functions and run network calls at module level, which breaks pytest collection in CI. --- studio/tests/test_url_download_benchmark.py | 54 -------- studio/tests/test_url_image_loading.py | 132 -------------------- studio/tests/test_url_parallel_benchmark.py | 79 ------------ 3 files changed, 265 deletions(-) delete mode 100644 studio/tests/test_url_download_benchmark.py delete mode 100644 studio/tests/test_url_image_loading.py delete mode 100644 studio/tests/test_url_parallel_benchmark.py diff --git a/studio/tests/test_url_download_benchmark.py b/studio/tests/test_url_download_benchmark.py deleted file mode 100644 index e29a8e05b6..0000000000 --- a/studio/tests/test_url_download_benchmark.py +++ /dev/null @@ -1,54 +0,0 @@ -""" -Benchmark: fsspec URL image download throughput at different dataset sizes. -Dataset: google-research-datasets/conceptual_captions (subset: labeled) - -Tests sizes: 100, 200, 300, 500, 1000, 1500, 2000 -Reports: time, success/fail rate, throughput (images/sec) -""" -from datasets import load_dataset, Dataset -from PIL import Image as PILImage -from io import BytesIO -from itertools import islice -import fsspec -import time - -DATASET = "google-research-datasets/conceptual_captions" -SUBSET = "labeled" -SPLIT = "train" -SIZES = [100, 200, 300, 500, 1000, 1500, 2000] - -# Load the max we need in one go -max_size = max(SIZES) -print(f"Loading {max_size} samples from {DATASET} (streaming)...") -ds = load_dataset(DATASET, name=SUBSET, split=SPLIT, streaming=True) -rows = list(islice(ds, max_size)) -full_dataset = Dataset.from_list(rows) -print(f"Loaded {len(full_dataset)} samples") -print(f"Columns: {full_dataset.column_names}") -print() - -print(f"{'Size':>6} | {'Time':>8} | {'OK':>6} | {'Fail':>6} | {'Fail%':>6} | {'img/s':>7}") -print("-" * 55) - -for size in SIZES: - dataset = full_dataset.select(range(size)) - success, fail = 0, 0 - t0 = time.time() - - for sample in dataset: - url = sample["image_url"] - try: - with fsspec.open(url, "rb", expand=True) as f: - img = PILImage.open(BytesIO(f.read())).convert("RGB") - success += 1 - except Exception: - fail += 1 - - elapsed = time.time() - t0 - fail_pct = (fail / size) * 100 - throughput = success / elapsed if elapsed > 0 else 0 - - print(f"{size:>6} | {elapsed:>7.1f}s | {success:>6} | {fail:>6} | {fail_pct:>5.1f}% | {throughput:>6.1f}/s") - -print() -print("Done.") diff --git a/studio/tests/test_url_image_loading.py b/studio/tests/test_url_image_loading.py deleted file mode 100644 index 4899bab8e3..0000000000 --- a/studio/tests/test_url_image_loading.py +++ /dev/null @@ -1,132 +0,0 @@ -""" -Reproduce: VLM URL image loading with HF datasets. -Tests cast_column(Image()) vs manual download approaches. -Dataset: google-research-datasets/conceptual_captions (subset: labeled) -""" -from datasets import load_dataset, Image as datasets_Image, Dataset -from PIL import Image as PILImage -from io import BytesIO -from itertools import islice -import time - -DATASET = "google-research-datasets/conceptual_captions" -SUBSET = "labeled" -SPLIT = "train" -N_SAMPLES = 20 # small slice for testing - -print("=" * 60) -print("Loading dataset (streaming, first N samples)...") -print("=" * 60) -ds = load_dataset(DATASET, name=SUBSET, split=SPLIT, streaming=True) -rows = list(islice(ds, N_SAMPLES)) -dataset = Dataset.from_list(rows) - -print(f"Loaded {len(dataset)} samples") -print(f"Columns: {dataset.column_names}") -print(f"First image_url: {dataset[0]['image_url'][:100]}...") -print() - -# ─── Test 1: cast_column(Image()) — what we tried ─── -print("=" * 60) -print("TEST 1: cast_column(Image()) approach") -print("=" * 60) -try: - ds_cast = dataset.cast_column("image_url", datasets_Image()) - success, fail = 0, 0 - t0 = time.time() - for i, sample in enumerate(ds_cast): - try: - img = sample["image_url"] - if img is not None: - print(f" [{i}] OK — {img.size} {img.mode}") - success += 1 - else: - print(f" [{i}] None returned") - fail += 1 - except Exception as e: - print(f" [{i}] FAILED: {type(e).__name__}: {str(e)[:80]}") - fail += 1 - elapsed = time.time() - t0 - print(f"\nResult: {success} ok, {fail} failed, {elapsed:.1f}s") -except Exception as e: - print(f"CRASHED during iteration: {type(e).__name__}: {str(e)[:120]}") -print() - -# ─── Test 2: Manual download with requests.Session ─── -print("=" * 60) -print("TEST 2: requests.Session() approach") -print("=" * 60) -try: - import requests - session = requests.Session() - success, fail = 0, 0 - t0 = time.time() - for i, sample in enumerate(dataset): - url = sample["image_url"] - try: - resp = session.get(url, timeout=10) - resp.raise_for_status() - img = PILImage.open(BytesIO(resp.content)).convert("RGB") - print(f" [{i}] OK — {img.size} {img.mode}") - success += 1 - except Exception as e: - print(f" [{i}] FAILED: {type(e).__name__}: {str(e)[:80]}") - fail += 1 - elapsed = time.time() - t0 - print(f"\nResult: {success} ok, {fail} failed, {elapsed:.1f}s") -except Exception as e: - print(f"CRASHED: {type(e).__name__}: {str(e)[:120]}") -print() - -# ─── Test 3: urllib (stdlib) ─── -print("=" * 60) -print("TEST 3: urllib approach (stdlib)") -print("=" * 60) -try: - from urllib.request import urlopen, Request - success, fail = 0, 0 - t0 = time.time() - for i, sample in enumerate(dataset): - url = sample["image_url"] - try: - req = Request(url, headers={"User-Agent": "Mozilla/5.0"}) - with urlopen(req, timeout=10) as resp: - img = PILImage.open(BytesIO(resp.read())).convert("RGB") - print(f" [{i}] OK — {img.size} {img.mode}") - success += 1 - except Exception as e: - print(f" [{i}] FAILED: {type(e).__name__}: {str(e)[:80]}") - fail += 1 - elapsed = time.time() - t0 - print(f"\nResult: {success} ok, {fail} failed, {elapsed:.1f}s") -except Exception as e: - print(f"CRASHED: {type(e).__name__}: {str(e)[:120]}") -print() - -# ─── Test 4: fsspec directly with expand=True ─── -print("=" * 60) -print("TEST 4: fsspec.open() with expand=True") -print("=" * 60) -try: - import fsspec - success, fail = 0, 0 - t0 = time.time() - for i, sample in enumerate(dataset): - url = sample["image_url"] - try: - with fsspec.open(url, "rb", expand=True) as f: - img = PILImage.open(BytesIO(f.read())).convert("RGB") - print(f" [{i}] OK — {img.size} {img.mode}") - success += 1 - except Exception as e: - print(f" [{i}] FAILED: {type(e).__name__}: {str(e)[:80]}") - fail += 1 - elapsed = time.time() - t0 - print(f"\nResult: {success} ok, {fail} failed, {elapsed:.1f}s") -except Exception as e: - print(f"CRASHED: {type(e).__name__}: {str(e)[:120]}") - -print() -print("=" * 60) -print("DONE — compare success rates and timing above") -print("=" * 60) diff --git a/studio/tests/test_url_parallel_benchmark.py b/studio/tests/test_url_parallel_benchmark.py deleted file mode 100644 index a0d160136c..0000000000 --- a/studio/tests/test_url_parallel_benchmark.py +++ /dev/null @@ -1,79 +0,0 @@ -""" -Benchmark: parallel fsspec URL image downloads with ThreadPoolExecutor. -Tests different worker counts to find optimal parallelism. -Dataset: google-research-datasets/conceptual_captions (subset: labeled) -""" -from datasets import load_dataset, Dataset -from PIL import Image as PILImage -from io import BytesIO -from itertools import islice -from concurrent.futures import ThreadPoolExecutor, as_completed -import fsspec -import time -import os - -DATASET = "google-research-datasets/conceptual_captions" -SUBSET = "labeled" -SPLIT = "train" -N_SAMPLES = 500 - -# safe_num_proc formula from studio/backend/utils/hardware/hardware.py -cpu_count = os.cpu_count() -safe_workers = max(1, cpu_count // 3) -print(f"CPU count: {cpu_count}, safe_num_proc: {safe_workers}") - -WORKER_COUNTS = [1, 4, 8, 16, 32, safe_workers] -# Deduplicate and sort -WORKER_COUNTS = sorted(set(WORKER_COUNTS)) - -print(f"Loading {N_SAMPLES} samples from {DATASET} (streaming)...") -ds = load_dataset(DATASET, name=SUBSET, split=SPLIT, streaming=True) -rows = list(islice(ds, N_SAMPLES)) -dataset = Dataset.from_list(rows) -urls = [row["image_url"] for row in dataset] -print(f"Loaded {len(urls)} URLs") -print() - - -def download_single(url): - """Download a single image URL using fsspec. Returns PIL image or raises.""" - with fsspec.open(url, "rb", expand=True) as f: - img = PILImage.open(BytesIO(f.read())).convert("RGB") - return img - - -print(f"{'Workers':>8} | {'Time':>8} | {'OK':>6} | {'Fail':>6} | {'Fail%':>6} | {'img/s':>7} | {'Speedup':>8}") -print("-" * 70) - -baseline_throughput = None - -for n_workers in WORKER_COUNTS: - success, fail = 0, 0 - t0 = time.time() - - with ThreadPoolExecutor(max_workers=n_workers) as pool: - futures = {pool.submit(download_single, url): url for url in urls} - for future in as_completed(futures): - try: - img = future.result(timeout=30) - success += 1 - except Exception: - fail += 1 - - elapsed = time.time() - t0 - fail_pct = (fail / N_SAMPLES) * 100 - throughput = success / elapsed if elapsed > 0 else 0 - - if baseline_throughput is None: - baseline_throughput = throughput - speedup = throughput / baseline_throughput if baseline_throughput > 0 else 0 - - label = f"{n_workers}" - if n_workers == safe_workers: - label += "*" # mark the safe_num_proc value - - print(f"{label:>8} | {elapsed:>7.1f}s | {success:>6} | {fail:>6} | {fail_pct:>5.1f}% | {throughput:>6.1f}/s | {speedup:>7.1f}x") - -print() -print("* = safe_num_proc value") -print("Done.") From c171573a8f1e4c243a26e6a99cf031afee8bc6aa Mon Sep 17 00:00:00 2001 From: Roland Tannous Date: Thu, 5 Mar 2026 06:10:10 +0000 Subject: [PATCH 10/10] fix: check for http(s) prefix instead of bare string type for URL detection --- studio/backend/utils/datasets/format_conversion.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/studio/backend/utils/datasets/format_conversion.py b/studio/backend/utils/datasets/format_conversion.py index c5c9a4d6e7..41a9617857 100644 --- a/studio/backend/utils/datasets/format_conversion.py +++ b/studio/backend/utils/datasets/format_conversion.py @@ -350,7 +350,8 @@ def convert_to_vlm_format( return {"messages": messages} total = len(dataset) - has_urls = isinstance(next(iter(dataset))[image_column], str) + first_image = next(iter(dataset))[image_column] + has_urls = isinstance(first_image, str) and first_image.startswith(("http://", "https://")) # ── URL probe: 200 samples with parallel workers to estimate speed + failure rate ── PROBE_SIZE = 200