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 && (