"""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]()