unsloth/studio/backend/utils/datasets/raw_text.py
Sanat Bhargava 1fc8bf53c7
Add Hugging Face dataset streaming mode to Studio (#4946)
* Add HF dataset streaming mode to Studio

* Added default value for datasetStreaming in training-config-store.ts

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* Handle None max_steps for streaming validation

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* studio: fast-fail streaming validation and guard incompatible modes

Reject dataset_streaming at the API boundary when hf_dataset is empty,
the dataset is vision/audio, or max_steps is not set. Probe eval split
with get_dataset_split_names before the streaming load so typos fail
immediately instead of mid-training. Guard column_names=None after map
on iterables. Hide the UI toggle for non-text configurations and clear
the stale flag when config becomes incompatible.

* studio: add streaming dataset tests, iterable helper, and streaming template/format support (WIP)

Work-in-progress on top of feat/studio-dataset-streaming-mode (PR #4946):
- new test_training_streaming.py and iterable.py dataset helper
- streaming support in chat_templates.py and format_conversion.py
- additional streaming guards in trainer.py / models / routes
- frontend streaming wiring in params-section and training-config-store

Committed to preserve uncommitted work before merging latest main.

* studio: fix review-team findings for streaming + main merge

BLOCKER: streaming + raw-text/CPT crashed on len(IterableDataset). Guard it in the
start route (reject format_type=="raw" or training_type=="Continued Pretraining")
and in isStreamingSupported (datasetFormat !== "raw").

Also:
- models/training.py: validate hf_dataset/subset/split (charset+length, block ..//);
  cap dataset slice indices (le=1e9); note validator ordering
- chat_templates.py: guard _apply_custom_mapping .map() for streaming
- trainer.py: warn when packing+streaming
- training-config-store.ts: persist-migration bump to v11 (standalone datasetStreaming
  backfill); add isVisionModel to NON_PERSISTED; toast on silent streamingCompatiblePatch
  mutations in the 4 indirect setters
- tests: route rejections (max_steps, raw/cpt), slice cap, unsafe hf_dataset

* studio: enable raw-text/CPT dataset streaming + streaming UX polish

- raw_text: keep the lazy filter but skip len()-based row counting for
  IterableDatasets so raw-text / CPT can stream; guard the eval-size log
- routes/trainer: drop the raw/CPT streaming block; add a defensive
  not-streaming guard on the eval auto-split (train_test_split)
- dataset-section: streaming toggle is visible-but-disabled and lists the
  exact unmet requirement(s) in its tooltip; block embedding models
- training-start-overlay: show "streaming (no full download)" instead of a
  stuck download bar for streaming runs
- trim the streaming test suite to the high-value cases

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* studio: address streaming review (MLX/embedding guards, sliced eval split, rehydrate timing)

- routes: reject dataset_streaming for embedding training and on Apple Silicon
  (MLX); both loaders materialize the full dataset instead of streaming
- trainer: validate the base eval split name so streaming eval accepts HF slice
  syntax such as "validation[:1000]"
- training-config-store: defer the onRehydrateStorage setState to a microtask so
  it doesn't hit the store's TDZ during synchronous hydration
- test: streaming start rejects embedding models

* studio: harden HF dataset streaming (column_names, split slicing, empty/eval bounds, gating)

Address a deeper streaming review:
- raw_text: resolve_column_names() guards IterableDataset.column_names=None
  (from_generator / unresolved features) so raw-text and CPT streaming no longer
  raise TypeError before training
- models/routes: reject HF slice syntax in train_split/eval_split when streaming
  (load_dataset(streaming=True) raises "Bad split"); reject mixed sources
  (local/S3) and embedding/MLX streaming at the API, not just in the UI
- trainer: an empty post-slice/filter stream fails preflight with a clear message;
  streaming eval is capped (STREAMING_EVAL_MAX_SAMPLES) so each eval terminates;
  the manual-slice shortcut falls back to a regular load when train_split is sliced
- format_conversion: streaming conversions preflight the first mapped row so
  format errors surface before training, not mid-iteration
- frontend: block streaming on Apple Silicon; clear datasetStreaming when a
  dataset is detected as image/audio at start

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* studio: fix CI for streaming PR (lint blocker + no-torch sandbox + preflight test)

- trainer.py: drop unused `IterableDataset` import (hoist safety-net blocker).
- test_training_streaming.py: only select real classes (isinstance type) when
  locating the trainer class, so a MagicMock-stubbed global is never passed to
  object.__new__ (fixes TypeError on the Python 3.10-3.13 jobs).
- no-torch import sandboxes (test_e2e_no_torch_sandbox.py,
  test_studio_import_no_torch.py): teach the chat_templates/format_conversion
  exec stubs and the full-import-chain copy list about the new `.iterable`
  module so the AFTER/runtime cases import without torch again.

---------

Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>
Co-authored-by: Roland Tannous <115670425+rolandtannous@users.noreply.github.com>
Co-authored-by: Roland Tannous <rolandtannous@gravityq.ai>
Co-authored-by: Etherll <61019402+Etherll@users.noreply.github.com>
2026-06-22 17:48:18 +03:00

180 lines
6 KiB
Python

# SPDX-License-Identifier: AGPL-3.0-only
# Copyright 2026-present the Unsloth AI Inc. team. All rights reserved. See /studio/LICENSE.AGPL-3.0
"""Shared helpers for raw-text dataset preparation."""
from dataclasses import dataclass
from typing import Literal
from datasets import Dataset
@dataclass(frozen = True)
class RawTextNotice:
message: str
level: Literal["info", "warning"]
update_status: bool = False
@dataclass(frozen = True)
class RawTextPreparationResult:
dataset: Dataset
notices: list[RawTextNotice]
def resolve_column_names(dataset) -> list[str]:
"""Return the column names for *dataset*, guarding against None.
IterableDataset.column_names is None until HF datasets>=X materialises
it from the first batch; .map() also keeps it None. Resolution order:
1. dataset.column_names if truthy (regular Dataset or HF>=4.4)
2. keys of dataset.features if available
3. bounded first-row probe, consumes one element, safe on IterableDataset
because HF re-iterates from the generator on the next pass
4. [] as a last resort so callers never see None
"""
col_names = getattr(dataset, "column_names", None)
if col_names:
return list(col_names)
features = getattr(dataset, "features", None)
if features:
return list(features.keys())
try:
first_row = next(iter(dataset))
return list(first_row.keys())
except Exception:
return []
def _string_columns(dataset: Dataset) -> list[str]:
feature_map = getattr(dataset, "features", {}) or {}
string_cols: list[str] = []
for col in resolve_column_names(dataset):
feature = feature_map.get(col)
dtype = str(getattr(feature, "dtype", ""))
if dtype in {"string", "large_string"}:
string_cols.append(col)
return string_cols
def _split_scope(split_name: str | None) -> str:
return f"the {split_name} split" if split_name else "this dataset"
def _drop_invalid_text_rows(
dataset: Dataset, *, mode_title: str, split_scope: str
) -> tuple[Dataset, list[RawTextNotice]]:
# Lazy filter — drops rows whose 'text' is null/non-string before they reach
# the tokenizer. Works on both Dataset and streaming IterableDataset.
filtered_dataset = dataset.filter(lambda ex: isinstance(ex["text"], str))
# Streaming datasets (IterableDataset) have no __len__, so we can't count the
# dropped rows or verify the result is non-empty without consuming the whole
# stream. Keep the filter, skip only the len()-based diagnostics.
if not hasattr(dataset, "__len__"):
return filtered_dataset, [
RawTextNotice(
message = (
f"{mode_title}: streaming dataset — rows with null or "
f"non-string 'text' in {split_scope} are dropped on the fly."
),
level = "info",
)
]
dropped_rows = len(dataset) - len(filtered_dataset)
if not dropped_rows:
return filtered_dataset, []
if len(filtered_dataset) == 0:
raise ValueError(
f"{mode_title} training requires at least one string 'text' value "
f"in {split_scope}; all {dropped_rows} rows were null or non-string."
)
return filtered_dataset, [
RawTextNotice(
message = (
f"{mode_title}: dropped {dropped_rows:,} row(s) with null or "
f"non-string 'text' values from {split_scope}"
),
level = "warning",
update_status = True,
)
]
def prepare_raw_text_dataset(
dataset: Dataset,
*,
mode_label: str = "raw text",
split_name: str | None = None,
eos_token: str | None = None,
append_eos: bool = False,
) -> RawTextPreparationResult:
notices: list[RawTextNotice] = []
mode_title = mode_label.capitalize()
split_scope = _split_scope(split_name)
col_names = resolve_column_names(dataset)
if "text" not in col_names:
string_cols = _string_columns(dataset)
if not string_cols:
raise ValueError(
f"{mode_title} training requires a string 'text' column but none "
f"was found in {split_scope} (columns: {col_names})."
)
renamed_col = string_cols[0]
if len(string_cols) > 1:
notices.append(
RawTextNotice(
message = (
f"{mode_title}: dataset has {len(string_cols)} string "
f"columns ({string_cols}); auto-selecting '{renamed_col}' "
"as the training text. Rename the intended column to "
"'text' to override."
),
level = "warning",
update_status = True,
)
)
notices.append(
RawTextNotice(
message = (
f"{mode_title}: renaming column '{renamed_col}' -> 'text' " f"for {split_scope}"
),
level = "info",
)
)
dataset = dataset.rename_column(renamed_col, "text")
dataset, invalid_row_notices = _drop_invalid_text_rows(
dataset,
mode_title = mode_title,
split_scope = split_scope,
)
notices.extend(invalid_row_notices)
if append_eos:
if not eos_token:
notices.append(
RawTextNotice(
message = (
f"{mode_title}: tokenizer has no eos_token; skipping EOS "
"append. Model will not learn document boundaries."
),
level = "warning",
)
)
else:
def _append_eos(ex, _eos = eos_token):
text = ex["text"]
return {"text": text if text.endswith(_eos) else text + _eos}
dataset = dataset.map(_append_eos)
return RawTextPreparationResult(dataset = dataset, notices = notices)