Fleet-1 / fleet_train.py

Publish Fleet-1 continued multilingual LoRA, evaluation, and Apache-2.0 license

0cc5816 verified about 21 hours ago

21.1 kB

"""Fleet-1: continue Sky-Kim's Whistle adapter on balanced FLEURS subsets."""
import argparse, hashlib, io, json, os, shutil, sys, time
from pathlib import Path
from dataclasses import replace
import numpy as np
from huggingface_hub import HfApi, hf_hub_download, get_token, login

SOURCE = 'Sky-Kim/whistle-finetuned-ko'
REVISION = 'b8a4f3188a828a0e87e2713b2d3e83d71d893068'
CODE_REVISION = 'ee2ffc67fe0f9537d4928e158440c7d228a63100'
LANGS = dict(en='en_us', de='de_de', fr='fr_fr', es='es_419', it='it_it', nl='nl_nl', pl='pl_pl', ko='ko_kr')
SEED, SECONDS, TOKENS, EPOCHS, LR = 42, 12, 160, 2, 2e-5
ROOT = Path(os.environ.get('FLEET_RUN_DIR', 'fleet-1-run')).resolve()
ROOT.mkdir(parents=True, exist_ok=True)
PUB = ROOT / 'publish'
PUB.mkdir(exist_ok=True)
SMOKE = os.environ.get('FLEET_SMOKE') == '1'
ACTIVE = {'en': 'en_us'} if SMOKE else LANGS
COUNTS = {'train': 2, 'validation': 1, 'test': 1} if SMOKE else {'train': 400, 'validation': 50, 'test': 100}

def write_json(path, data):
    path = Path(path)
    tmp = path.with_suffix(path.suffix + '.tmp')
    tmp.write_text(json.dumps(data, ensure_ascii=False, indent=2) + '\n')
    tmp.replace(path)

def source_file(name):
    return hf_hub_download(SOURCE, name, revision=REVISION)

def adapter_and_tokenizer():
    from cactus.model.checkpoints import read_adapter
    from cactus.model.whistle_finetune import WhistleTokenizer
    a = read_adapter(source_file('adapter/whistle_ko_lora.safetensors'))
    assert set(LANGS).issubset(a['languages']), a['languages']
    assert a['pieces'] and a['rows'] is not None
    return a, WhistleTokenizer(a['pieces'], a['languages'])

def auth():
    if not get_token():
        print('Enter a Hugging Face WRITE token. It stays on this Mac; do not paste it in chat.')
        login(add_to_git_credential=False)
    api = HfApi()
    who = api.whoami()
    repo = who['name'] + '/Fleet-1'
    marker = ROOT / 'destination.json'
    if marker.exists() and json.loads(marker.read_text())['repo'] != repo:
        raise RuntimeError('The signed-in account changed. Use the original account for this run.')
    if api.repo_exists(repo_id=repo) and not marker.exists():
        raise RuntimeError(f'{repo} already exists. This launcher will not overwrite an unrelated model.')
    # Reserve the requested public destination now, checking write access before training.
    api.create_repo(repo_id=repo, private=False, exist_ok=True)
    write_json(marker, {'repo': repo})
    print('Training will publish to https://huggingface.co/' + repo)

def prepare():
    from datasets import load_dataset, Audio
    import soundfile as sf
    import soxr
    print('Fetching the pinned Korean adapter and tokenizer...', flush=True)
    _, tokenizer = adapter_and_tokenizer()
    # Download deployment file early, to surface unavailable source files before training.
    print('Fetching the source deployment model...', flush=True)
    source_file('whistle.cact')
    seen = set()
    info = HfApi().dataset_info('google/fleurs', revision='refs/convert/parquet')
    version_file = ROOT / 'dataset_revision.json'
    if not version_file.exists():
        write_json(version_file, {'revision': info.sha})
    revision = json.loads(version_file.read_text())['revision']
    files = HfApi().list_repo_files('google/fleurs', repo_type='dataset', revision=revision)
    for split, target in COUNTS.items():
        for lang, config in ACTIVE.items():
            manifest = ROOT / f'{split}-{lang}.json'
            if manifest.exists():
                for item in json.loads(manifest.read_text()):
                    seen.add((lang, item['text_key']))
                continue
            paths = sorted(p for p in files if p.startswith(f'{config}/{split}/') and p.endswith('.parquet'))
            if not paths:
                raise RuntimeError(f'No parquet files for {config}/{split}')
            urls = [f'https://huggingface.co/datasets/google/fleurs/resolve/{revision}/{p}' for p in paths]
            print(f'Downloading {split} {lang}: {target} clips (remote parquet reads can take several minutes)', flush=True)
            data = load_dataset('parquet', data_files={split: urls}, split=split,
                                streaming=True, batch_size=8).cast_column('audio', Audio(decode=False))
            # Select deterministically in dataset order; shuffle examples during training.
            records = []
            for index, row in enumerate(data):
                text = (row.get('raw_transcription') or row.get('transcription') or '').strip()
                key = ' '.join(text.casefold().split())
                if not text or (lang, key) in seen or len(tokenizer.encode(text)) + 3 > TOKENS + 1:
                    continue
                if not 0 < row.get('num_samples', 1) <= SECONDS * 16000:
                    continue
                audio = row['audio']
                wav, rate = sf.read(io.BytesIO(audio['bytes']) if audio.get('bytes') else audio['path'],
                                    dtype='float32', always_2d=True)
                wav = wav.mean(axis=1)
                if rate != 16000:
                    wav = soxr.resample(wav, rate, 16000)
                if not 0 < len(wav) <= SECONDS * 16000:
                    continue
                folder = ROOT / 'audio' / split / lang
                folder.mkdir(parents=True, exist_ok=True)
                path = folder / f'{index:06d}.wav'
                sf.write(path, wav, 16000, subtype='PCM_16')
                records.append({'audio': str(path.relative_to(ROOT)), 'text': text, 'language': lang,
                                'row_index': index, 'id': row.get('id'), 'text_key': key,
                                'sha256': hashlib.sha256(path.read_bytes()).hexdigest()})
                seen.add((lang, key))
                if len(records) == target:
                    break
            if len(records) != target:
                raise RuntimeError(f'{lang}/{split}: only {len(records)} eligible clips; need {target}.')
            write_json(manifest, records)
            print(f'Saved {manifest.name}: {len(records)} clips', flush=True)
    print('Data ready. Test audio is never used for training or checkpoint selection.', flush=True)

def records(split):
    return [r for lang in ACTIVE for r in json.loads((ROOT / f'{split}-{lang}.json').read_text())]

def train():
    import jax
    import jax.numpy as jnp
    import optax
    import soundfile as sf
    from cactus.model import whistle_finetune as wf
    from cactus.model.checkpoints import write_adapter
    from cactus.model.whistle import Whistle
    from cactus.model.quantize import configure_deploy, cq_ste_params, WEIGHT_BITS
    if (ROOT / 'trained.json').exists():
        print('Completed training found; skipping training.'); return
    adapter, tokenizer = adapter_and_tokenizer()
    base_path = str(ROOT / 'checkpoints' / 'whistle.safetensors')
    params, cfg = wf.load_base(base_path)
    cfg_train = replace(cfg, max_tokens=TOKENS)
    tuned = replace(cfg_train, text=tokenizer.text, languages=tuple(adapter['languages']))
    params = jax.tree.map(lambda x: jnp.asarray(x, dtype=jnp.float32), params)
    current = {'lora': {tuple(k.split('/')): {n: jnp.asarray(v) for n,v in pair.items()}
                        for k,pair in adapter['lora'].items()}, 'rows': jnp.asarray(adapter['rows'])}
    configure_deploy(act_bits=8, kv_bits=8)
    model = Whistle(tuned)
    training, validation = records('train'), records('validation')
    n_epochs = 1 if SMOKE else EPOCHS
    total = len(training) * n_epochs
    schedule = optax.warmup_cosine_decay_schedule(0., LR, max(1, total // 20), total)
    optimizer = optax.chain(optax.clip_by_global_norm(1.), optax.adamw(schedule))
    opt_state = optimizer.init(current)

    def arrays(row):
        wav, rate = sf.read(ROOT / row['audio'], dtype='float32')
        assert rate == 16000 and wav.ndim == 1
        padded = np.zeros((1, SECONDS * 16000), np.float32)
        padded[0, :len(wav)] = wav
        ids = [cfg.bos, tuned.tag(row['language']), *tokenizer.encode(row['text']), cfg.eos]
        token_ids = np.full((1, TOKENS + 1), cfg.pad, np.int32)
        token_ids[0, :len(ids)] = ids
        return padded, np.array([len(wav)], np.int32), token_ids

    def loss_fn(weights, wav, lengths, ids):
        merged = cq_ste_params(wf.tuned_params(params, cfg, tokenizer, weights['lora'],
                                              adapter['scale'], weights['rows']), WEIGHT_BITS)
        logits = model.apply({'params': merged}, wav, lengths, ids[:, :-1], quant=True)
        targets = ids[:, 1:]
        mask = targets != cfg.pad
        ce = optax.softmax_cross_entropy_with_integer_labels(logits, targets)
        return (ce * mask).sum() / jnp.maximum(mask.sum(), 1.)

    @jax.jit
    def step(weights, state, wav, lengths, ids):
        loss, grads = jax.value_and_grad(loss_fn)(weights, wav, lengths, ids)
        updates, state = optimizer.update(grads, state, weights)
        return optax.apply_updates(weights, updates), state, loss

    evaluate = jax.jit(loss_fn)
    checkpoint = ROOT / 'resume.npz'
    meta = {'step': 0, 'best_loss': None, 'best_epoch': None, 'history': []}
    _, tree = jax.tree.flatten((current, opt_state))
    if checkpoint.exists():
        with np.load(checkpoint, allow_pickle=False) as saved:
            leaves = [jnp.asarray(saved[f'a{i}']) for i in range(len(jax.tree.leaves((current, opt_state))))]
            current, opt_state = jax.tree.unflatten(tree, leaves)
            meta = json.loads(str(saved['metadata']))
        print(f"Resuming step {meta['step']}/{total}", flush=True)

    def save_state():
        arrays_ = {f'a{i}': np.asarray(v) for i,v in enumerate(jax.tree.leaves((current, opt_state)))}
        with open(ROOT / 'resume.tmp', 'wb') as handle:
            np.savez(handle, metadata=np.array(json.dumps(meta)), **arrays_)
        (ROOT / 'resume.tmp').replace(checkpoint)

    def save_adapter():
        output = {**adapter, 'base': base_path, 'seed': SEED,
                  'lora': {'/'.join(k): {n: np.asarray(v) for n,v in pair.items()}
                           for k,pair in current['lora'].items()}, 'rows': np.asarray(current['rows'])}
        write_adapter(str(PUB / 'fleet_lora.tmp.safetensors'), output)
        (PUB / 'fleet_lora.tmp.safetensors').replace(PUB / 'fleet_lora.safetensors')

    start = time.monotonic()
    initial_step = meta['step']
    print(f'CPU training: batch 1, {SECONDS}s maximum, {total} steps. First JAX compilation may take minutes.', flush=True)
    for epoch in range(n_epochs):
        if meta['step'] >= (epoch + 1) * len(training):
            continue
        order = np.random.default_rng(SEED + epoch).permutation(len(training))
        offset = max(0, meta['step'] - epoch * len(training))
        for position in range(offset, len(order)):
            row = training[int(order[position])]
            current, opt_state, loss = step(current, opt_state, *arrays(row))
            loss_value = float(loss)
            if not np.isfinite(loss_value):
                raise RuntimeError('Nonfinite training loss; publication stopped.')
            meta['step'] += 1
            if meta['step'] % 10 == 0 or meta['step'] == initial_step + 1:
                elapsed = time.monotonic() - start
                remaining = elapsed / (meta['step'] - initial_step) * (total - meta['step']) / 3600
                print(f"Step {meta['step']}/{total}, loss {loss_value:.4f}, rough remaining training {remaining:.1f}h", flush=True)
            # Epoch boundaries are saved after validation, so a resumed run cannot skip validation.
            if meta['step'] % 100 == 0 and position + 1 < len(order):
                save_state()
        per_language = {}
        for lang in ACTIVE:
            values = [float(evaluate(current, *arrays(r))) for r in validation if r['language'] == lang]
            per_language[lang] = float(np.mean(values))
        score = float(np.mean(list(per_language.values())))
        if not np.isfinite(score):
            raise RuntimeError('Nonfinite validation loss; publication stopped.')
        meta['history'].append({'epoch': epoch + 1, 'macro_validation_loss': score, 'by_language': per_language})
        if meta['best_loss'] is None or score < meta['best_loss']:
            meta['best_loss'], meta['best_epoch'] = score, epoch + 1
            save_adapter()
        save_state()
        print(f'Epoch {epoch + 1}: validation loss {score:.4f}', flush=True)
    write_json(ROOT / 'trained.json', meta)


def export():
    import jax.numpy as jnp
    from cactus.model import whistle_finetune as wf
    from cactus.model.checkpoints import read_adapter
    adapter = read_adapter(str(PUB / 'fleet_lora.safetensors'))
    params, cfg = wf.load_base(str(ROOT / 'checkpoints' / 'whistle.safetensors'))
    tokenizer = wf.WhistleTokenizer(adapter['pieces'], adapter['languages'])
    lora = {tuple(k.split('/')): {n: jnp.asarray(v) for n,v in pair.items()} for k,pair in adapter['lora'].items()}
    merged = wf.tuned_params(params, cfg, tokenizer, lora, adapter['scale'], adapter['rows'])
    wf.build(merged, cfg, tokenizer, str(PUB / 'whistle.cact'))
    print('Exported Fleet-1 as whistle.cact', flush=True)


def benchmark():
    from cactus.agent.whistle import Whistle
    from cactus.model.whistle_finetune import errors, _normalizer
    report = {}
    for name, weights in [('source', source_file('whistle.cact')), ('Fleet-1', str(PUB / 'whistle.cact'))]:
        result_file = ROOT / f'scores-{name}.json'
        if result_file.exists():
            report[name] = json.loads(result_file.read_text()); continue
        model = Whistle(weights=weights)
        scores = {}
        for lang in ACTIVE:
            edits = units = correct_language = count = 0
            for row in [r for r in records('test') if r['language'] == lang]:
                result = model.transcribe(str(ROOT / row['audio']))
                # Always CER for Korean, even if a particular reference contains Latin characters only.
                if lang == 'ko':
                    import editdistance
                    norm = _normalizer(lang)
                    ref, hyp = norm(row['text']).replace(' ', ''), norm(result['text']).replace(' ', '')
                    e, n = editdistance.eval(ref, hyp), len(ref)
                else:
                    e, n = errors(row['text'], result['text'], lang)
                edits += e; units += n; count += 1
                correct_language += result.get('language') == lang
            if not units:
                raise RuntimeError(f'No reference units for {lang}')
            scores[lang] = {'metric': 'CER' if lang == 'ko' else 'WER', 'error_rate': edits / units,
                            'edits': edits, 'reference_units': units, 'clips': count,
                            'language_accuracy': correct_language / count}
            print(name, lang, scores[lang], flush=True)
        write_json(result_file, scores)
        report[name] = scores
    report['all_eight_improved'] = len(ACTIVE) == 8 and all(
        report['Fleet-1'][l]['error_rate'] < report['source'][l]['error_rate'] for l in ACTIVE)
    write_json(PUB / 'evaluation.json', report)


def publish():
    if SMOKE:
        raise RuntimeError('Smoke runs cannot be published.')
    report = json.loads((PUB / 'evaluation.json').read_text())
    meta = json.loads((ROOT / 'trained.json').read_text())
    destination = json.loads((ROOT / 'destination.json').read_text())['repo']
    api = HfApi()
    if destination.split('/')[0] != api.whoami()['name']:
        raise RuntimeError('Hugging Face account changed; publication stopped.')
    shutil.copyfile(source_file('LICENSE'), PUB / 'LICENSE')
    shutil.copyfile(source_file('README.md'), PUB / 'UPSTREAM_MODEL_CARD.md')
    for file in api.list_repo_files(SOURCE, revision=REVISION):
        if Path(file).name.upper().startswith('NOTICE'):
            target = PUB / file
            target.parent.mkdir(parents=True, exist_ok=True)
            shutil.copyfile(source_file(file), target)
    cfg = json.loads(Path(source_file('config.json')).read_text())
    cfg.update(name_or_path=destination, base_model=SOURCE)
    cfg['finetune'] = {'method': 'continued LoRA', 'rank': 32, 'alpha': 64,
                       'epochs': meta['best_epoch'], 'adapter': 'fleet_lora.safetensors'}
    write_json(PUB / 'config.json', cfg)
    table = '\n'.join(f"| {lang} | {report['source'][lang]['metric']} | {report['source'][lang]['error_rate']:.2%} | {report['Fleet-1'][lang]['error_rate']:.2%} |"
                       for lang in LANGS)
    status = ('All eight sampled error rates decreased.' if report['all_eight_improved'] else
              'Improvement across all eight languages was NOT demonstrated. Inspect the per-language results; some may regress.')
    card = f'''---
license: apache-2.0
base_model: {SOURCE}
language: [en, de, fr, es, it, nl, pl, ko]
pipeline_tag: automatic-speech-recognition
datasets: [google/fleurs]
tags: [whistle, cactus, lora, speech-to-text]
---
# Fleet-1
Experimental continuation of [{SOURCE}](https://huggingface.co/{SOURCE}), retaining its trained Korean vocabulary, language tags, and LoRA weights. Original architecture and pretrained weights: Cactus Compute's Whistle.
{status}
## Evaluation
Automatic language detection; same native Cactus engine for both models. Lower is better.
| Language | Metric | Source | Fleet-1 |
|---|---|---:|---:|
{table}
These are 100 short FLEURS test clips per language, filtered to at most {SECONDS} seconds and {TOKENS-2} text tokens, not full benchmark scores. They do not establish statistical significance or accuracy on long, noisy, or conversational recordings. Test clips were not used for this run's training or checkpoint selection. The upstream model was previously evaluated on FLEURS and trained on some FLEURS training clips. Prior training exposure in the original Cactus model is not fully known.
## Training
400 clips per language from FLEURS train; 50 per language from validation. Original splits are kept separate; repeated normalized transcripts within a language are excluded across selected splits. First eligible unique rows in dataset order; epoch order shuffled with seed {SEED}. Existing adapter is continued with a fresh AdamW optimizer, learning rate {LR}, batch 1, two epochs, gradient clipping 1, warmup/cosine schedule, W4/A8 quantization-aware training on CPU. Best trained epoch by macro-average validation loss: {meta['best_epoch']}. Training selection is not based on test scores. No guarantee of improvement over the source.
Source revision: `{REVISION}`. Training code revision: `{CODE_REVISION}`. Exact examples and settings: `training_manifest.json`. Language-detection scores: `evaluation.json`.
## Usage
```python
import cactus
from huggingface_hub import hf_hub_download
path = hf_hub_download("{destination}", "whistle.cact")
print(cactus.transcribe("clip.wav", weights=path))
```
Use 16 kHz mono WAV audio and the Cactus revision recorded above. This is a Cactus model, not a Transformers AutoModel checkpoint.
## License and attribution
Model: Apache License 2.0; original LICENSE retained verbatim. This derivative modifies the source adapter through further multilingual training. See `UPSTREAM_MODEL_CARD.md` for the source author's training details, attribution, and citation to Whistle (Mroz et al., 2026, Cactus Compute).
Training/evaluation data: [FLEURS](https://huggingface.co/datasets/google/fleurs), [FLEURS: Few-shot Learning Evaluation of Universal Representations of Speech](https://arxiv.org/abs/2205.12446), Conneau et al. (2022), [CC BY 4.0](https://creativecommons.org/licenses/by/4.0/). Audio was converted to 16 kHz mono PCM and subsetted. Source-model data also includes [Zeroth-Korean](https://www.openslr.org/40/), CC BY 4.0. No training audio is included in this model repository.
'''
    (PUB / 'README.md').write_text(card)
    write_json(PUB / 'training_manifest.json', {
        'source': SOURCE, 'source_revision': REVISION, 'code_revision': CODE_REVISION,
        'dataset_revision': json.loads((ROOT / 'dataset_revision.json').read_text())['revision'],
        'seed': SEED, 'epochs': EPOCHS, 'lr': LR, 'batch_size': 1, 'seconds': SECONDS, 'tokens': TOKENS,
        'training': meta, 'splits': {s: records(s) for s in COUNTS}})
    shutil.copyfile(__file__, PUB / 'fleet_train.py')
    lock = Path(__file__).with_name('requirements.lock')
    if lock.exists(): shutil.copyfile(lock, PUB / 'requirements.lock')
    api.upload_folder(repo_id=destination, folder_path=str(PUB), repo_type='model',
                      commit_message='Publish Fleet-1 continued multilingual LoRA, evaluation, and Apache-2.0 license')
    print('\nPublished: https://huggingface.co/' + destination, flush=True)

if __name__ == '__main__':
    parser = argparse.ArgumentParser()
    parser.add_argument('stage', choices=['auth','prepare','train','export','benchmark','publish'])
    args = parser.parse_args()
    globals()[args.stage]()

	"""Fleet-1: continue Sky-Kim's Whistle adapter on balanced FLEURS subsets."""
	import argparse, hashlib, io, json, os, shutil, sys, time
	from pathlib import Path
	from dataclasses import replace
	import numpy as np
	from huggingface_hub import HfApi, hf_hub_download, get_token, login

	SOURCE = 'Sky-Kim/whistle-finetuned-ko'
	REVISION = 'b8a4f3188a828a0e87e2713b2d3e83d71d893068'
	CODE_REVISION = 'ee2ffc67fe0f9537d4928e158440c7d228a63100'
	LANGS = dict(en='en_us', de='de_de', fr='fr_fr', es='es_419', it='it_it', nl='nl_nl', pl='pl_pl', ko='ko_kr')
	SEED, SECONDS, TOKENS, EPOCHS, LR = 42, 12, 160, 2, 2e-5
	ROOT = Path(os.environ.get('FLEET_RUN_DIR', 'fleet-1-run')).resolve()
	ROOT.mkdir(parents=True, exist_ok=True)
	PUB = ROOT / 'publish'
	PUB.mkdir(exist_ok=True)
	SMOKE = os.environ.get('FLEET_SMOKE') == '1'
	ACTIVE = {'en': 'en_us'} if SMOKE else LANGS
	COUNTS = {'train': 2, 'validation': 1, 'test': 1} if SMOKE else {'train': 400, 'validation': 50, 'test': 100}

	def write_json(path, data):
	path = Path(path)
	tmp = path.with_suffix(path.suffix + '.tmp')
	tmp.write_text(json.dumps(data, ensure_ascii=False, indent=2) + '\n')
	tmp.replace(path)

	def source_file(name):
	return hf_hub_download(SOURCE, name, revision=REVISION)

	def adapter_and_tokenizer():
	from cactus.model.checkpoints import read_adapter
	from cactus.model.whistle_finetune import WhistleTokenizer
	a = read_adapter(source_file('adapter/whistle_ko_lora.safetensors'))
	assert set(LANGS).issubset(a['languages']), a['languages']
	assert a['pieces'] and a['rows'] is not None
	return a, WhistleTokenizer(a['pieces'], a['languages'])

	def auth():
	if not get_token():
	print('Enter a Hugging Face WRITE token. It stays on this Mac; do not paste it in chat.')
	login(add_to_git_credential=False)
	api = HfApi()
	who = api.whoami()
	repo = who['name'] + '/Fleet-1'
	marker = ROOT / 'destination.json'
	if marker.exists() and json.loads(marker.read_text())['repo'] != repo:
	raise RuntimeError('The signed-in account changed. Use the original account for this run.')
	if api.repo_exists(repo_id=repo) and not marker.exists():
	raise RuntimeError(f'{repo} already exists. This launcher will not overwrite an unrelated model.')
	# Reserve the requested public destination now, checking write access before training.
	api.create_repo(repo_id=repo, private=False, exist_ok=True)
	write_json(marker, {'repo': repo})
	print('Training will publish to https://huggingface.co/' + repo)

	def prepare():
	from datasets import load_dataset, Audio
	import soundfile as sf
	import soxr
	print('Fetching the pinned Korean adapter and tokenizer...', flush=True)
	_, tokenizer = adapter_and_tokenizer()
	# Download deployment file early, to surface unavailable source files before training.
	print('Fetching the source deployment model...', flush=True)
	source_file('whistle.cact')
	seen = set()
	info = HfApi().dataset_info('google/fleurs', revision='refs/convert/parquet')
	version_file = ROOT / 'dataset_revision.json'
	if not version_file.exists():
	write_json(version_file, {'revision': info.sha})
	revision = json.loads(version_file.read_text())['revision']
	files = HfApi().list_repo_files('google/fleurs', repo_type='dataset', revision=revision)
	for split, target in COUNTS.items():
	for lang, config in ACTIVE.items():
	manifest = ROOT / f'{split}-{lang}.json'
	if manifest.exists():
	for item in json.loads(manifest.read_text()):
	seen.add((lang, item['text_key']))
	continue
	paths = sorted(p for p in files if p.startswith(f'{config}/{split}/') and p.endswith('.parquet'))
	if not paths:
	raise RuntimeError(f'No parquet files for {config}/{split}')
	urls = [f'https://huggingface.co/datasets/google/fleurs/resolve/{revision}/{p}' for p in paths]
	print(f'Downloading {split} {lang}: {target} clips (remote parquet reads can take several minutes)', flush=True)
	data = load_dataset('parquet', data_files={split: urls}, split=split,
	streaming=True, batch_size=8).cast_column('audio', Audio(decode=False))
	# Select deterministically in dataset order; shuffle examples during training.
	records = []
	for index, row in enumerate(data):
	text = (row.get('raw_transcription') or row.get('transcription') or '').strip()
	key = ' '.join(text.casefold().split())
	if not text or (lang, key) in seen or len(tokenizer.encode(text)) + 3 > TOKENS + 1:
	continue
	if not 0 < row.get('num_samples', 1) <= SECONDS * 16000:
	continue
	audio = row['audio']
	wav, rate = sf.read(io.BytesIO(audio['bytes']) if audio.get('bytes') else audio['path'],
	dtype='float32', always_2d=True)
	wav = wav.mean(axis=1)
	if rate != 16000:
	wav = soxr.resample(wav, rate, 16000)
	if not 0 < len(wav) <= SECONDS * 16000:
	continue
	folder = ROOT / 'audio' / split / lang
	folder.mkdir(parents=True, exist_ok=True)
	path = folder / f'{index:06d}.wav'
	sf.write(path, wav, 16000, subtype='PCM_16')
	records.append({'audio': str(path.relative_to(ROOT)), 'text': text, 'language': lang,
	'row_index': index, 'id': row.get('id'), 'text_key': key,
	'sha256': hashlib.sha256(path.read_bytes()).hexdigest()})
	seen.add((lang, key))
	if len(records) == target:
	break
	if len(records) != target:
	raise RuntimeError(f'{lang}/{split}: only {len(records)} eligible clips; need {target}.')
	write_json(manifest, records)
	print(f'Saved {manifest.name}: {len(records)} clips', flush=True)
	print('Data ready. Test audio is never used for training or checkpoint selection.', flush=True)

	def records(split):
	return [r for lang in ACTIVE for r in json.loads((ROOT / f'{split}-{lang}.json').read_text())]

	def train():
	import jax
	import jax.numpy as jnp
	import optax
	import soundfile as sf
	from cactus.model import whistle_finetune as wf
	from cactus.model.checkpoints import write_adapter
	from cactus.model.whistle import Whistle
	from cactus.model.quantize import configure_deploy, cq_ste_params, WEIGHT_BITS
	if (ROOT / 'trained.json').exists():
	print('Completed training found; skipping training.'); return
	adapter, tokenizer = adapter_and_tokenizer()
	base_path = str(ROOT / 'checkpoints' / 'whistle.safetensors')
	params, cfg = wf.load_base(base_path)
	cfg_train = replace(cfg, max_tokens=TOKENS)
	tuned = replace(cfg_train, text=tokenizer.text, languages=tuple(adapter['languages']))
	params = jax.tree.map(lambda x: jnp.asarray(x, dtype=jnp.float32), params)
	current = {'lora': {tuple(k.split('/')): {n: jnp.asarray(v) for n,v in pair.items()}
	for k,pair in adapter['lora'].items()}, 'rows': jnp.asarray(adapter['rows'])}
	configure_deploy(act_bits=8, kv_bits=8)
	model = Whistle(tuned)
	training, validation = records('train'), records('validation')
	n_epochs = 1 if SMOKE else EPOCHS
	total = len(training) * n_epochs
	schedule = optax.warmup_cosine_decay_schedule(0., LR, max(1, total // 20), total)
	optimizer = optax.chain(optax.clip_by_global_norm(1.), optax.adamw(schedule))
	opt_state = optimizer.init(current)

	def arrays(row):
	wav, rate = sf.read(ROOT / row['audio'], dtype='float32')
	assert rate == 16000 and wav.ndim == 1
	padded = np.zeros((1, SECONDS * 16000), np.float32)
	padded[0, :len(wav)] = wav
	ids = [cfg.bos, tuned.tag(row['language']), *tokenizer.encode(row['text']), cfg.eos]
	token_ids = np.full((1, TOKENS + 1), cfg.pad, np.int32)
	token_ids[0, :len(ids)] = ids
	return padded, np.array([len(wav)], np.int32), token_ids

	def loss_fn(weights, wav, lengths, ids):
	merged = cq_ste_params(wf.tuned_params(params, cfg, tokenizer, weights['lora'],
	adapter['scale'], weights['rows']), WEIGHT_BITS)
	logits = model.apply({'params': merged}, wav, lengths, ids[:, :-1], quant=True)
	targets = ids[:, 1:]
	mask = targets != cfg.pad
	ce = optax.softmax_cross_entropy_with_integer_labels(logits, targets)
	return (ce * mask).sum() / jnp.maximum(mask.sum(), 1.)

	@jax.jit
	def step(weights, state, wav, lengths, ids):
	loss, grads = jax.value_and_grad(loss_fn)(weights, wav, lengths, ids)
	updates, state = optimizer.update(grads, state, weights)
	return optax.apply_updates(weights, updates), state, loss

	evaluate = jax.jit(loss_fn)
	checkpoint = ROOT / 'resume.npz'
	meta = {'step': 0, 'best_loss': None, 'best_epoch': None, 'history': []}
	_, tree = jax.tree.flatten((current, opt_state))
	if checkpoint.exists():
	with np.load(checkpoint, allow_pickle=False) as saved:
	leaves = [jnp.asarray(saved[f'a{i}']) for i in range(len(jax.tree.leaves((current, opt_state))))]
	current, opt_state = jax.tree.unflatten(tree, leaves)
	meta = json.loads(str(saved['metadata']))
	print(f"Resuming step {meta['step']}/{total}", flush=True)

	def save_state():
	arrays_ = {f'a{i}': np.asarray(v) for i,v in enumerate(jax.tree.leaves((current, opt_state)))}
	with open(ROOT / 'resume.tmp', 'wb') as handle:
	np.savez(handle, metadata=np.array(json.dumps(meta)), **arrays_)
	(ROOT / 'resume.tmp').replace(checkpoint)

	def save_adapter():
	output = {**adapter, 'base': base_path, 'seed': SEED,
	'lora': {'/'.join(k): {n: np.asarray(v) for n,v in pair.items()}
	for k,pair in current['lora'].items()}, 'rows': np.asarray(current['rows'])}
	write_adapter(str(PUB / 'fleet_lora.tmp.safetensors'), output)
	(PUB / 'fleet_lora.tmp.safetensors').replace(PUB / 'fleet_lora.safetensors')

	start = time.monotonic()
	initial_step = meta['step']
	print(f'CPU training: batch 1, {SECONDS}s maximum, {total} steps. First JAX compilation may take minutes.', flush=True)
	for epoch in range(n_epochs):
	if meta['step'] >= (epoch + 1) * len(training):
	continue
	order = np.random.default_rng(SEED + epoch).permutation(len(training))
	offset = max(0, meta['step'] - epoch * len(training))
	for position in range(offset, len(order)):
	row = training[int(order[position])]
	current, opt_state, loss = step(current, opt_state, *arrays(row))
	loss_value = float(loss)
	if not np.isfinite(loss_value):
	raise RuntimeError('Nonfinite training loss; publication stopped.')
	meta['step'] += 1
	if meta['step'] % 10 == 0 or meta['step'] == initial_step + 1:
	elapsed = time.monotonic() - start
	remaining = elapsed / (meta['step'] - initial_step) * (total - meta['step']) / 3600
	print(f"Step {meta['step']}/{total}, loss {loss_value:.4f}, rough remaining training {remaining:.1f}h", flush=True)
	# Epoch boundaries are saved after validation, so a resumed run cannot skip validation.
	if meta['step'] % 100 == 0 and position + 1 < len(order):
	save_state()
	per_language = {}
	for lang in ACTIVE:
	values = [float(evaluate(current, *arrays(r))) for r in validation if r['language'] == lang]
	per_language[lang] = float(np.mean(values))
	score = float(np.mean(list(per_language.values())))
	if not np.isfinite(score):
	raise RuntimeError('Nonfinite validation loss; publication stopped.')
	meta['history'].append({'epoch': epoch + 1, 'macro_validation_loss': score, 'by_language': per_language})
	if meta['best_loss'] is None or score < meta['best_loss']:
	meta['best_loss'], meta['best_epoch'] = score, epoch + 1
	save_adapter()
	save_state()
	print(f'Epoch {epoch + 1}: validation loss {score:.4f}', flush=True)
	write_json(ROOT / 'trained.json', meta)


	def export():
	import jax.numpy as jnp
	from cactus.model import whistle_finetune as wf
	from cactus.model.checkpoints import read_adapter
	adapter = read_adapter(str(PUB / 'fleet_lora.safetensors'))
	params, cfg = wf.load_base(str(ROOT / 'checkpoints' / 'whistle.safetensors'))
	tokenizer = wf.WhistleTokenizer(adapter['pieces'], adapter['languages'])
	lora = {tuple(k.split('/')): {n: jnp.asarray(v) for n,v in pair.items()} for k,pair in adapter['lora'].items()}
	merged = wf.tuned_params(params, cfg, tokenizer, lora, adapter['scale'], adapter['rows'])
	wf.build(merged, cfg, tokenizer, str(PUB / 'whistle.cact'))
	print('Exported Fleet-1 as whistle.cact', flush=True)


	def benchmark():
	from cactus.agent.whistle import Whistle
	from cactus.model.whistle_finetune import errors, _normalizer
	report = {}
	for name, weights in [('source', source_file('whistle.cact')), ('Fleet-1', str(PUB / 'whistle.cact'))]:
	result_file = ROOT / f'scores-{name}.json'
	if result_file.exists():
	report[name] = json.loads(result_file.read_text()); continue
	model = Whistle(weights=weights)
	scores = {}
	for lang in ACTIVE:
	edits = units = correct_language = count = 0
	for row in [r for r in records('test') if r['language'] == lang]:
	result = model.transcribe(str(ROOT / row['audio']))
	# Always CER for Korean, even if a particular reference contains Latin characters only.
	if lang == 'ko':
	import editdistance
	norm = _normalizer(lang)
	ref, hyp = norm(row['text']).replace(' ', ''), norm(result['text']).replace(' ', '')
	e, n = editdistance.eval(ref, hyp), len(ref)
	else:
	e, n = errors(row['text'], result['text'], lang)
	edits += e; units += n; count += 1
	correct_language += result.get('language') == lang
	if not units:
	raise RuntimeError(f'No reference units for {lang}')
	scores[lang] = {'metric': 'CER' if lang == 'ko' else 'WER', 'error_rate': edits / units,
	'edits': edits, 'reference_units': units, 'clips': count,
	'language_accuracy': correct_language / count}
	print(name, lang, scores[lang], flush=True)
	write_json(result_file, scores)
	report[name] = scores
	report['all_eight_improved'] = len(ACTIVE) == 8 and all(
	report['Fleet-1'][l]['error_rate'] < report['source'][l]['error_rate'] for l in ACTIVE)
	write_json(PUB / 'evaluation.json', report)


	def publish():
	if SMOKE:
	raise RuntimeError('Smoke runs cannot be published.')
	report = json.loads((PUB / 'evaluation.json').read_text())
	meta = json.loads((ROOT / 'trained.json').read_text())
	destination = json.loads((ROOT / 'destination.json').read_text())['repo']
	api = HfApi()
	if destination.split('/')[0] != api.whoami()['name']:
	raise RuntimeError('Hugging Face account changed; publication stopped.')
	shutil.copyfile(source_file('LICENSE'), PUB / 'LICENSE')
	shutil.copyfile(source_file('README.md'), PUB / 'UPSTREAM_MODEL_CARD.md')
	for file in api.list_repo_files(SOURCE, revision=REVISION):
	if Path(file).name.upper().startswith('NOTICE'):
	target = PUB / file
	target.parent.mkdir(parents=True, exist_ok=True)
	shutil.copyfile(source_file(file), target)
	cfg = json.loads(Path(source_file('config.json')).read_text())
	cfg.update(name_or_path=destination, base_model=SOURCE)
	cfg['finetune'] = {'method': 'continued LoRA', 'rank': 32, 'alpha': 64,
	'epochs': meta['best_epoch'], 'adapter': 'fleet_lora.safetensors'}
	write_json(PUB / 'config.json', cfg)
	table = '\n'.join(f"\| {lang} \| {report['source'][lang]['metric']} \| {report['source'][lang]['error_rate']:.2%} \| {report['Fleet-1'][lang]['error_rate']:.2%} \|"
	for lang in LANGS)
	status = ('All eight sampled error rates decreased.' if report['all_eight_improved'] else
	'Improvement across all eight languages was NOT demonstrated. Inspect the per-language results; some may regress.')
	card = f'''---
	license: apache-2.0
	base_model: {SOURCE}
	language: [en, de, fr, es, it, nl, pl, ko]
	pipeline_tag: automatic-speech-recognition
	datasets: [google/fleurs]
	tags: [whistle, cactus, lora, speech-to-text]
	---
	# Fleet-1

	Experimental continuation of [{SOURCE}](https://huggingface.co/{SOURCE}), retaining its trained Korean vocabulary, language tags, and LoRA weights. Original architecture and pretrained weights: Cactus Compute's Whistle.

	{status}

	## Evaluation
	Automatic language detection; same native Cactus engine for both models. Lower is better.

	\| Language \| Metric \| Source \| Fleet-1 \|
	\|---\|---\|---:\|---:\|
	{table}

	These are 100 short FLEURS test clips per language, filtered to at most {SECONDS} seconds and {TOKENS-2} text tokens, not full benchmark scores. They do not establish statistical significance or accuracy on long, noisy, or conversational recordings. Test clips were not used for this run's training or checkpoint selection. The upstream model was previously evaluated on FLEURS and trained on some FLEURS training clips. Prior training exposure in the original Cactus model is not fully known.

	## Training
	400 clips per language from FLEURS train; 50 per language from validation. Original splits are kept separate; repeated normalized transcripts within a language are excluded across selected splits. First eligible unique rows in dataset order; epoch order shuffled with seed {SEED}. Existing adapter is continued with a fresh AdamW optimizer, learning rate {LR}, batch 1, two epochs, gradient clipping 1, warmup/cosine schedule, W4/A8 quantization-aware training on CPU. Best trained epoch by macro-average validation loss: {meta['best_epoch']}. Training selection is not based on test scores. No guarantee of improvement over the source.

	Source revision: `{REVISION}`. Training code revision: `{CODE_REVISION}`. Exact examples and settings: `training_manifest.json`. Language-detection scores: `evaluation.json`.

	## Usage
	```python
	import cactus
	from huggingface_hub import hf_hub_download
	path = hf_hub_download("{destination}", "whistle.cact")
	print(cactus.transcribe("clip.wav", weights=path))
	```
	Use 16 kHz mono WAV audio and the Cactus revision recorded above. This is a Cactus model, not a Transformers AutoModel checkpoint.

	## License and attribution
	Model: Apache License 2.0; original LICENSE retained verbatim. This derivative modifies the source adapter through further multilingual training. See `UPSTREAM_MODEL_CARD.md` for the source author's training details, attribution, and citation to Whistle (Mroz et al., 2026, Cactus Compute).

	Training/evaluation data: [FLEURS](https://huggingface.co/datasets/google/fleurs), [FLEURS: Few-shot Learning Evaluation of Universal Representations of Speech](https://arxiv.org/abs/2205.12446), Conneau et al. (2022), [CC BY 4.0](https://creativecommons.org/licenses/by/4.0/). Audio was converted to 16 kHz mono PCM and subsetted. Source-model data also includes [Zeroth-Korean](https://www.openslr.org/40/), CC BY 4.0. No training audio is included in this model repository.
	'''
	(PUB / 'README.md').write_text(card)
	write_json(PUB / 'training_manifest.json', {
	'source': SOURCE, 'source_revision': REVISION, 'code_revision': CODE_REVISION,
	'dataset_revision': json.loads((ROOT / 'dataset_revision.json').read_text())['revision'],
	'seed': SEED, 'epochs': EPOCHS, 'lr': LR, 'batch_size': 1, 'seconds': SECONDS, 'tokens': TOKENS,
	'training': meta, 'splits': {s: records(s) for s in COUNTS}})
	shutil.copyfile(__file__, PUB / 'fleet_train.py')
	lock = Path(__file__).with_name('requirements.lock')
	if lock.exists(): shutil.copyfile(lock, PUB / 'requirements.lock')
	api.upload_folder(repo_id=destination, folder_path=str(PUB), repo_type='model',
	commit_message='Publish Fleet-1 continued multilingual LoRA, evaluation, and Apache-2.0 license')
	print('\nPublished: https://huggingface.co/' + destination, flush=True)

	if __name__ == '__main__':
	parser = argparse.ArgumentParser()
	parser.add_argument('stage', choices=['auth','prepare','train','export','benchmark','publish'])
	args = parser.parse_args()
	globals()[args.stage]()