mirror of
https://github.com/unslothai/unsloth.git
synced 2026-08-15 03:43:55 +00:00
* 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>
180 lines
6 KiB
Python
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)
|