"""BlueMagpie-TTS Gradio demo using the validated production inference profile.""" from __future__ import annotations from dataclasses import dataclass, replace from importlib import metadata as importlib_metadata import inspect import json import os import secrets import threading # Cross-process release reproducibility is a hard contract. Configure CUDA # before importing torch so cuBLAS and Mamba choose their deterministic paths. DETERMINISTIC_ALGORITHMS = True CUBLAS_WORKSPACE_CONFIG = ":4096:8" MAMBA_DETERMINISTIC = "1" CUDNN_BENCHMARK = False CUDNN_DETERMINISTIC = True CUDA_MATMUL_ALLOW_TF32 = False CUDNN_ALLOW_TF32 = False EXPECTED_TORCH_VERSION = "2.11.0+cu130" EXPECTED_TORCH_CUDA_VERSION = "13.0" EXPECTED_CUDNN_VERSION = 91900 EXPECTED_MAMBA_SSM_VERSION = "2.3.2.post1" EXPECTED_TRITON_VERSION = "3.6.0" if ( os.environ.setdefault( "CUBLAS_WORKSPACE_CONFIG", CUBLAS_WORKSPACE_CONFIG, ) != CUBLAS_WORKSPACE_CONFIG ): raise RuntimeError("CUBLAS_WORKSPACE_CONFIG conflicts with the release contract") if ( os.environ.setdefault( "MAMBA_DETERMINISTIC", MAMBA_DETERMINISTIC, ) != MAMBA_DETERMINISTIC ): raise RuntimeError("MAMBA_DETERMINISTIC conflicts with the release contract") import gradio as gr import librosa import numpy as np import torch from huggingface_hub import snapshot_download from transformers import PreTrainedTokenizerFast runtime_versions = { "torch": str(torch.__version__), "cuda": str(torch.version.cuda), "cudnn": torch.backends.cudnn.version(), "mamba-ssm": importlib_metadata.version("mamba-ssm"), "triton": importlib_metadata.version("triton"), } expected_runtime_versions = { "torch": EXPECTED_TORCH_VERSION, "cuda": EXPECTED_TORCH_CUDA_VERSION, "cudnn": EXPECTED_CUDNN_VERSION, "mamba-ssm": EXPECTED_MAMBA_SSM_VERSION, "triton": EXPECTED_TRITON_VERSION, } if runtime_versions != expected_runtime_versions: raise RuntimeError( "ZeroGPU runtime conflicts with the deterministic release contract: " f"expected={expected_runtime_versions!r} actual={runtime_versions!r}" ) torch.use_deterministic_algorithms(DETERMINISTIC_ALGORITHMS) torch.backends.cudnn.benchmark = CUDNN_BENCHMARK torch.backends.cudnn.deterministic = CUDNN_DETERMINISTIC torch.backends.cuda.matmul.allow_tf32 = CUDA_MATMUL_ALLOW_TF32 torch.backends.cudnn.allow_tf32 = CUDNN_ALLOW_TF32 from bluemagpie import BlueMagpieModel from production import ( StopHysteresisController, GenerationChunkSpec, NetworkFragmentProof, active_pace_correction_speed, apply_loudness_floor, canonicalize_asr_network_fragments, coalesce_text_chunks, count_network_endpoint_duration_units, contains_naturalized_url_spoken_form, contains_network_identifier, count_speech_units, effective_generation_cfg, email_domain_mail_generation_variant, endpoint_generation_plan, estimate_step_seconds, extract_windowed_speaker_embedding, fade_variable_internal_edges, finish_audio, join_audio_chunks_variable, inverse_network_url_rendering, match_chunk_rms, network_identifier_has_ambiguous_iri, network_protected_spoken_spans, naturalized_url_rendering_proof, normalize_spoken_forms, punctuation_pause_seconds, pcm16_audio_output, pcm16_verification_waveform, plan_generation_chunks, select_generation_cps, set_generation_seed, split_quality_text_for_tts, strip_unspoken_cjk_quotes_for_generation, target_pace_speed, ) from quality_runtime import ( BASE_GENERATION_POLICY, SAFE_DURATION_GENERATION_POLICY, VERIFICATION_WHISPER_MODEL_ID, VERIFICATION_WHISPER_REVISION, WHISPER_MODEL_ID, WHISPER_REVISION, CandidateGenerationContext, CandidateGenerationEvidence, CandidateVerification, CandidateObservation, ChunkCandidateArtifact, FinalOutputRejectedError, GenerationPolicy, LocalIndependentGateEvidence, NoQualifiedCandidateError, RELEASE_SPEAKER_TRIGGER_SECONDS, SEQUENCE_FALLBACK_MAX_LOCAL_BOUNDARY_SPEAKER_DROP, WholeWaveformVerificationCache, active_voiced_duration_seconds, candidate_gate_evidence, active_audio_rms_db, active_audio_median_f0_hz, endpoint_tail_energy_ratio, format_cascade_evidence_log, generation_cfg_for_candidate_offset, generation_policy_for_candidate_offset, intersect_local_semantic_verification, local_candidate_has_coverage_eligibility, prepare_candidate_audio, qualify_trajectory_with_joined_output, require_verified_final_output, release_speaker_evidence_from_audio, release_speaker_measurement_required, resolve_request_seed, run_coverage_adaptive_cascade, speaker_evidence_from_audio, squim_objective_evidence_from_audio, transcribe_verification_whisper, transcribe_whisper, trajectory_gate_evidence, verify_trajectory, ) GPU_LEASE_SECONDS = 120 try: import spaces gpu = spaces.GPU(duration=GPU_LEASE_SECONDS) DEVICE = "cuda" except ImportError: def gpu(function): return function DEVICE = "cuda" if torch.cuda.is_available() else "cpu" REPO_ID = "OpenFormosa/BlueMagpie-TTS" MODEL_REVISION = "aaf1a0878e37875382bb0e5c8a3a2ba43be67297" ECAPA_REPO_ID = "speechbrain/spkrec-ecapa-voxceleb" ECAPA_REVISION = "0f99f2d0ebe89ac095bcc5903c4dd8f72b367286" DEFAULT_CFG = 3.0 DEFAULT_STEPS = 10 TARGET_CPS = 4.0 ACTIVE_PACE_TARGET_CPS = 4.00 CLOSED_LOOP_ACTIVE_PACE_TARGET_CPS = 3.95 MIXED_CFG_PRIMARY = 3.0 MIXED_CFG_ALTERNATE = 2.0 MIXED_CFG_SCHEDULE = "row_ordinal_zero_and_even_primary_odd_alternate" MIN_ENDPOINT_CUE_UNITS = 6 SHORT_TEXT_CFG_MIN = 3.0 SHORT_TEXT_CFG_UNITS = 6 NETWORK_TEXT_CFG_MIN = 3.0 CHUNK_CHARS = 80 ONSET_CLAUSE_SEARCH_CHARS = 0 MIN_CHUNK_CHARS = 12 QUALITY_MEDIUM_TEXT_MAX_UNITS = 48 QUALITY_MEDIUM_CHUNK_UNITS = 24 CROSSFADE_MS = 80.0 CHUNK_EDGE_FADE_MS = 80.0 CHUNK_RMS_MATCH_DB = 4.0 FINAL_ENDPOINT_FADE_MS = 3.0 NETWORK_GENERATION_MIN_UNITS = 8 NETWORK_GENERATION_TARGET_UNITS = 32 NETWORK_GENERATION_MAX_UNITS = 36 NETWORK_REQUEST_ORDINARY_MAX_UNITS = 36 NETWORK_SPLIT_ATOMIC_EMAIL_AT_SEPARATOR = True NETWORK_INTERNAL_FADE_MS = 5.0 NETWORK_INTERNAL_SILENCE_MS = 400.0 SEMANTIC_CHUNK_MIN_SILENCE_MS = 250.0 NETWORK_REQUEST_SEMANTIC_CHUNK_MIN_SILENCE_MS = 350.0 STOP_THRESHOLD = 0.50 STOP_LATE_THRESHOLD = 0.05 STOP_LATE_START_RATIO = 0.75 STOP_LATE_FULL_RATIO = 0.95 STOP_CONSECUTIVE = 1 MIN_PACE_SPEED = 0.80 EXTREME_SHORT_MAX_UNITS = 2 EXTREME_SHORT_MAX_SPEED = 0.95 PACE_ONLY_FALLBACK_MIN_SPEED = 0.76 PACE_ONLY_FALLBACK_MIN_UNITS = 30 PACE_STRETCH_N_FFT = 1536 PACE_STRETCH_HOP_LENGTH = 384 NETWORK_PACE_STRETCH_N_FFT = 2048 NETWORK_PACE_STRETCH_HOP_LENGTH = 512 MAX_TEXT_CHARS = 360 QUALITY_MAX_GENERATED_CHUNKS = 32 QUALITY_MAX_GENERATED_TEXT_UNITS = 800 EMAIL_MAIL_FALLBACK_CANDIDATE_ORDINALS = frozenset((2, 5, 6)) BASE_CHUNK_TEXT_VARIANT = "base" EMAIL_DOMAIN_MAIL_CHUNK_TEXT_VARIANT = "email_domain_mail_v1" QUALITY_FINAL_ASR_MAX_NEW_TOKENS = 440 QUALITY_MAX_CER = 0.20 QUALITY_MAX_PACE_CPS = 4.30 QUALITY_PREFIX_SUFFIX_UNITS = 6 QUALITY_MIN_SPEAKER_SIMILARITY = 0.10 QUALITY_MAX_BOUNDARY_SPEAKER_DROP = 0.10 QUALITY_RELEASE_MIN_SPEAKER_SIMILARITY = 0.105 QUALITY_RELEASE_MAX_BOUNDARY_SPEAKER_DROP = 0.095 QUALITY_MIN_SQUIM_STOI = 0.60 QUALITY_MIN_SQUIM_PESQ = 1.12 QUALITY_PREFERRED_MIN_SQUIM_STOI = 0.72 QUALITY_PREFERRED_MIN_SQUIM_PESQ = 1.20 QUALITY_PREFERRED_SQUIM_MIN_DURATION_SECONDS = 1.50 QUALITY_PREFERRED_MIN_SPEAKER_SIMILARITY = 0.25 QUALITY_PREFERRED_MAX_BOUNDARY_SPEAKER_DROP = 0.05 QUALITY_MAX_SEQUENCE_PATHS = 3 SHORT_AUDIO_SPEAKER_GATE_SECONDS = 1.50 VERIFICATION_ASR_PROFILE = ( f"{VERIFICATION_WHISPER_MODEL_ID}@{VERIFICATION_WHISPER_REVISION}:" f"greedy:zh:short{QUALITY_PREFIX_SUFFIX_UNITS}:cer{QUALITY_MAX_CER}:" f"prefix{QUALITY_PREFIX_SUFFIX_UNITS}=0:" f"suffix{QUALITY_PREFIX_SUFFIX_UNITS}=0:tail=0:" f"max_tokens{QUALITY_FINAL_ASR_MAX_NEW_TOKENS}" ) print(f"[BlueMagpie] downloading model from {REPO_ID}@{MODEL_REVISION} ...") MODEL_DIR = snapshot_download(REPO_ID, revision=MODEL_REVISION) print(f"[BlueMagpie] caching speaker encoder from {ECAPA_REPO_ID}@{ECAPA_REVISION} ...") ECAPA_DIR = snapshot_download(ECAPA_REPO_ID, revision=ECAPA_REVISION) print(f"[BlueMagpie] caching quality ASR from {WHISPER_MODEL_ID}@{WHISPER_REVISION} ...") ASR_DIR = snapshot_download(WHISPER_MODEL_ID, revision=WHISPER_REVISION) print( "[BlueMagpie] caching independent final ASR from " f"{VERIFICATION_WHISPER_MODEL_ID}@{VERIFICATION_WHISPER_REVISION} ..." ) VERIFICATION_ASR_DIR = snapshot_download( VERIFICATION_WHISPER_MODEL_ID, revision=VERIFICATION_WHISPER_REVISION, ) tokenizer = PreTrainedTokenizerFast(tokenizer_file=os.path.join(MODEL_DIR, "tokenizer.json")) print(f"[BlueMagpie] loading model on device={DEVICE} ...") model = BlueMagpieModel.from_local(MODEL_DIR, tokenizer=tokenizer, training=False, device=DEVICE) SR = int(model.sample_rate) STEP_SECONDS = estimate_step_seconds(model, SR) METADATA: dict = {} try: with open(os.path.join(MODEL_DIR, "release_metadata.json"), encoding="utf-8") as handle: METADATA = json.load(handle) except (OSError, ValueError) as error: print(f"[BlueMagpie] release metadata unavailable: {error}") CHECKPOINT = str(METADATA.get("checkpoint", "release")) def _load_speakers() -> tuple[dict[str, torch.Tensor], str]: path = os.path.join(MODEL_DIR, "checkpoints", "speaker_centroids.pt") if not os.path.exists(path): raise RuntimeError("speaker_centroids.pt is missing from the model release") table = torch.load(path, map_location="cpu", weights_only=True) speaker_ids = [str(value) for value in table["speaker_ids"]] centroids = table["centroids"] labels = {f"內建語者 {chr(65 + index)}": centroid for index, centroid in enumerate(centroids)} requested_id = ( METADATA.get("recommended_generation_defaults", {}).get("speaker_id") if isinstance(METADATA.get("recommended_generation_defaults"), dict) else None ) if requested_id not in speaker_ids and "female_voice" in speaker_ids: requested_id = "female_voice" default_index = speaker_ids.index(requested_id) if requested_id in speaker_ids else 0 return labels, f"內建語者 {chr(65 + default_index)}" SPEAKERS, DEFAULT_SPEAKER = _load_speakers() DEFAULT_CENTROID = SPEAKERS[DEFAULT_SPEAKER] # The pinned public package predates native stop hysteresis. Wrap only that # version; newer packages receive the same policy through native arguments. _GENERATE_PARAMETERS = set(inspect.signature(model._generate).parameters) _NATIVE_STOP_POLICY = {"stop_threshold", "stop_consecutive"}.issubset(_GENERATE_PARAMETERS) _STOP_CONTROLLER: StopHysteresisController | None = None if not _NATIVE_STOP_POLICY: _STOP_CONTROLLER = StopHysteresisController( model.stop_head, threshold=STOP_THRESHOLD, late_threshold=STOP_LATE_THRESHOLD, consecutive=STOP_CONSECUTIVE, late_start_ratio=STOP_LATE_START_RATIO, late_full_ratio=STOP_LATE_FULL_RATIO, ) model.stop_head = _STOP_CONTROLLER _GENERATION_LOCK = threading.Lock() _ECAPA_ENCODER = None _ECAPA_LOCK = threading.Lock() print( f"[BlueMagpie] ready checkpoint={CHECKPOINT} sample_rate={SR} " f"step_seconds={STEP_SECONDS} native_stop_policy={_NATIVE_STOP_POLICY}" ) def _get_ecapa_encoder(): global _ECAPA_ENCODER with _ECAPA_LOCK: if _ECAPA_ENCODER is None: import torchaudio # SpeechBrain 1.0.3 still probes this API during import, while the # ZeroGPU torchaudio build has removed it. Audio loading below is # handled by librosa, so an empty compatibility result is correct. if not hasattr(torchaudio, "list_audio_backends"): torchaudio.list_audio_backends = lambda: [] from speechbrain.inference.speaker import EncoderClassifier cache_dir = os.path.join(os.environ.get("HF_HOME", "/tmp"), "speechbrain", "ecapa") _ECAPA_ENCODER = EncoderClassifier.from_hparams( source=ECAPA_DIR, # The upstream hyperparams otherwise points back to the repo # and SpeechBrain 1.0.3 uses a removed Hub keyword. Keep every # weight fetch inside the already pinned local snapshot. overrides={"pretrained_path": ECAPA_DIR}, savedir=cache_dir, run_opts={"device": "cpu"}, ) return _ECAPA_ENCODER def _apply_speed( audio: np.ndarray, speed: float, *, network_conditioned: bool = False, ) -> np.ndarray: speed = float(speed or 1.0) if abs(speed - 1.0) < 1.0e-3: return audio n_fft = ( NETWORK_PACE_STRETCH_N_FFT if network_conditioned else PACE_STRETCH_N_FFT ) hop_length = ( NETWORK_PACE_STRETCH_HOP_LENGTH if network_conditioned else PACE_STRETCH_HOP_LENGTH ) return librosa.effects.time_stretch( np.asarray(audio, dtype=np.float32), rate=speed, n_fft=n_fft, hop_length=hop_length, ) @dataclass(frozen=True) class _ChunkGenerationOutcome: audio: np.ndarray stop_reason: str endpoint_energy_ratio: float generated_steps: int hard_stop_steps: int def _generate_chunk( text: str, centroid: torch.Tensor, *, cfg: float, steps: int, request_seed: int, policy: GenerationPolicy, network_conditioned: bool = False, ) -> _ChunkGenerationOutcome: generation_cps = ( policy.ascii_cps if network_conditioned else select_generation_cps( text, cjk_cps=policy.cjk_cps, ascii_cps=policy.ascii_cps, ) ) endpoint_duration_units = ( count_network_endpoint_duration_units(text) if network_conditioned else count_speech_units(text) ) model_text, expected_steps, hard_stop_steps = endpoint_generation_plan( text, generation_cps=generation_cps, step_seconds=STEP_SECONDS, margin_steps=policy.hard_stop_margin_steps, add_terminal_punctuation=count_speech_units(text) >= MIN_ENDPOINT_CUE_UNITS, duration_units=endpoint_duration_units, ) public_text_units = count_speech_units(text) short_headroom_max_units = int( getattr(policy, "short_headroom_max_units", 0) ) short_hard_stop_floor_steps = int( getattr(policy, "short_hard_stop_floor_steps", 0) ) short_headroom_floor_applied = bool( 0 < public_text_units <= short_headroom_max_units and hard_stop_steps < short_hard_stop_floor_steps ) if short_headroom_floor_applied: hard_stop_steps = short_hard_stop_floor_steps # Do not hold generation open to enforce pace. The model can finish the # requested text early; extending its latent sequence creates tail speech. min_len = 2 generation_cfg = effective_generation_cfg( text, cfg, short_text_unit_threshold=SHORT_TEXT_CFG_UNITS, short_text_min_cfg=SHORT_TEXT_CFG_MIN, ) set_generation_seed(request_seed) kwargs = { "target_text": model_text, "speaker_centroid": centroid, "cfg_value": generation_cfg, "inference_timesteps": int(steps), "min_len": min_len, "max_len": hard_stop_steps, "retry_badcase": False, "retry_badcase_max_times": 1, "retry_badcase_ratio_threshold": 6.0, } if _NATIVE_STOP_POLICY: kwargs["stop_threshold"] = STOP_THRESHOLD kwargs["stop_consecutive"] = STOP_CONSECUTIVE if "generation_seed" in _GENERATE_PARAMETERS: kwargs["generation_seed"] = request_seed if _STOP_CONTROLLER is not None: _STOP_CONTROLLER.begin( min_len, expected_steps=expected_steps, hard_stop_steps=hard_stop_steps, ) try: audio = model.generate(**kwargs) finally: if _STOP_CONTROLLER is not None: _STOP_CONTROLLER.end() stop_reason = ( _STOP_CONTROLLER.last_stop_reason if _STOP_CONTROLLER is not None else "native_stop" ) generated_steps = ( _STOP_CONTROLLER.last_generated_steps if _STOP_CONTROLLER is not None else hard_stop_steps ) if _STOP_CONTROLLER is not None: print( "[BlueMagpie] endpoint " f"expected_steps={expected_steps} hard_stop_steps={hard_stop_steps} " f"generated_steps={_STOP_CONTROLLER.last_generated_steps} " f"reason={_STOP_CONTROLLER.last_stop_reason} cfg={generation_cfg:.2f}" ) print( "[BlueMagpie] generation policy " f"name={policy.name} seed={request_seed} generation_cps={generation_cps:.2f} " f"duration_units={endpoint_duration_units} " f"duration_counter={'network_conservative' if network_conditioned else 'public'} " f"expected_steps={expected_steps} hard_stop_steps={hard_stop_steps} " f"min_len={min_len} " f"short_headroom_floor_applied={short_headroom_floor_applied}" ) audio = audio.detach().float().cpu().numpy().reshape(-1) def outcome(waveform: np.ndarray) -> _ChunkGenerationOutcome: normalized = np.asarray(waveform, dtype=np.float32).reshape(-1) return _ChunkGenerationOutcome( audio=normalized, stop_reason=stop_reason, endpoint_energy_ratio=endpoint_tail_energy_ratio(normalized, SR), generated_steps=generated_steps, hard_stop_steps=hard_stop_steps, ) pace_speed = target_pace_speed( audio.size, SR, text, target_cps=TARGET_CPS, min_speed=MIN_PACE_SPEED, ) combined_speed = pace_speed try: active_duration = active_voiced_duration_seconds(audio, SR) except (TypeError, ValueError, RuntimeError, OverflowError): # Invalid pace evidence must never turn into an unbounded correction. # Total-duration pace remains bounded and faces the ordinary local gate. pass else: active_speed = active_pace_correction_speed( active_duration, text, target_cps=ACTIVE_PACE_TARGET_CPS, prior_speed=1.0, min_total_speed=MIN_PACE_SPEED, ) combined_speed = min(pace_speed, active_speed) if combined_speed < 1.0: print( "[BlueMagpie] single-pass pace correction " f"active_duration={active_duration:.3f} total_rate={pace_speed:.6f} " f"active_rate={active_speed:.6f} combined_rate={combined_speed:.6f}" ) corrected = _apply_speed( audio, combined_speed, network_conditioned=network_conditioned, ) try: corrected_active_duration = active_voiced_duration_seconds(corrected, SR) except (TypeError, ValueError, RuntimeError, OverflowError): return outcome(corrected) rerender_speed = active_pace_correction_speed( corrected_active_duration, text, target_cps=CLOSED_LOOP_ACTIVE_PACE_TARGET_CPS, prior_speed=combined_speed, min_total_speed=MIN_PACE_SPEED, ) final_speed = combined_speed final_audio = corrected if rerender_speed < 1.0: final_speed = combined_speed * rerender_speed print( "[BlueMagpie] closed-loop pace rerender " f"active_duration={corrected_active_duration:.3f} " f"prior_rate={combined_speed:.6f} residual_rate={rerender_speed:.6f} " f"total_rate={final_speed:.6f}" ) # Re-render from the untouched model waveform so the returned samples # have exactly one pitch-preserving phase-vocoder pass. final_audio = _apply_speed( audio, final_speed, network_conditioned=network_conditioned, ) speech_units = count_speech_units(text) if ( not network_conditioned and speech_units <= EXTREME_SHORT_MAX_UNITS and final_speed > EXTREME_SHORT_MAX_SPEED ): final_speed = EXTREME_SHORT_MAX_SPEED print( "[BlueMagpie] extreme-short robustness rerender " f"units={speech_units} total_rate={final_speed:.6f}" ) final_audio = _apply_speed( audio, final_speed, network_conditioned=False, ) # Preserve every already-compliant waveform bit-for-bit. A lower stretch # floor is available only to long, ordinary chunks whose exact base output # would otherwise fail the frozen local pace gate. Re-rendering still uses # the untouched model waveform, so no candidate receives two vocoder passes. if ( not network_conditioned and speech_units >= PACE_ONLY_FALLBACK_MIN_UNITS ): try: final_active_duration = active_voiced_duration_seconds(final_audio, SR) except (TypeError, ValueError, RuntimeError, OverflowError): return outcome(final_audio) observed_cps = speech_units / final_active_duration if observed_cps > QUALITY_MAX_PACE_CPS: fallback_residual = active_pace_correction_speed( final_active_duration, text, target_cps=CLOSED_LOOP_ACTIVE_PACE_TARGET_CPS, prior_speed=final_speed, min_total_speed=PACE_ONLY_FALLBACK_MIN_SPEED, ) if fallback_residual < 1.0: fallback_speed = final_speed * fallback_residual print( "[BlueMagpie] pace-only fallback rerender " f"active_duration={final_active_duration:.3f} " f"observed_cps={observed_cps:.6f} " f"prior_rate={final_speed:.6f} " f"residual_rate={fallback_residual:.6f} " f"total_rate={fallback_speed:.6f}" ) return outcome( _apply_speed( audio, fallback_speed, network_conditioned=False, ) ) return outcome(final_audio) def _speaker_anchor_array(centroid: torch.Tensor) -> np.ndarray: anchor = torch.as_tensor(centroid).detach().float().cpu().numpy().reshape(-1) if anchor.size == 0 or not np.isfinite(anchor).all(): raise ValueError("speaker anchor is invalid") norm = float(np.linalg.norm(anchor)) if not np.isfinite(norm) or norm <= 1.0e-8: raise ValueError("speaker anchor has zero norm") return np.asarray(anchor / norm, dtype=np.float32) def _generate_trajectory( chunks: tuple[str, ...], centroid: torch.Tensor, *, cfg: float, steps: int, request_seed: int, policy: GenerationPolicy, network_cfg_min: float = NETWORK_TEXT_CFG_MIN, network_conditioned: tuple[bool, ...] | None = None, ) -> tuple[tuple[np.ndarray, ...], tuple[_ChunkGenerationOutcome, ...]]: scheduled_cfg = float(cfg) trajectory: list[np.ndarray] = [] endpoint_outcomes: list[_ChunkGenerationOutcome] = [] if network_conditioned is not None and len(network_conditioned) != len(chunks): raise ValueError("network provenance must align with generation chunks") for local_chunk_index, chunk in enumerate(chunks): network_chunk = ( bool(network_conditioned[local_chunk_index]) if network_conditioned is not None else bool(network_protected_spoken_spans(chunk)) ) network_floor_applied = bool( network_chunk and scheduled_cfg < float(network_cfg_min) ) network_adjusted_cfg = ( max(scheduled_cfg, float(network_cfg_min)) if network_chunk else scheduled_cfg ) effective_cfg = effective_generation_cfg( chunk, network_adjusted_cfg, short_text_unit_threshold=SHORT_TEXT_CFG_UNITS, short_text_min_cfg=SHORT_TEXT_CFG_MIN, ) short_floor_applied = effective_cfg > network_adjusted_cfg print( "[BlueMagpie] generation attempt " f"seed={request_seed} local_chunk_index={local_chunk_index} " f"policy={policy.name} scheduled_cfg={scheduled_cfg:.2f} " f"effective_cfg={effective_cfg:.2f} " f"network_floor_applied={network_floor_applied} " f"short_floor_applied={short_floor_applied}" ) outcome = _generate_chunk( chunk, centroid, cfg=network_adjusted_cfg, steps=steps, request_seed=request_seed, policy=policy, network_conditioned=network_chunk, ) trajectory.append(outcome.audio) endpoint_outcomes.append(outcome) return tuple(trajectory), tuple(endpoint_outcomes) def _attach_transition_f0( verification, prepared_waveforms: tuple[np.ndarray | None, ...], *, collect_transition_f0: bool, local_candidate_pool: bool, ): """Measure F0 only for rows that can enter multi-chunk sequence DP.""" if not collect_transition_f0: return verification if not local_candidate_pool: raise ValueError("transition F0 is only valid for local candidate pools") if ( len(prepared_waveforms) != len(verification.candidate_results) or len(prepared_waveforms) != len(verification.chunk_artifacts) ): raise RuntimeError("transition F0 evidence does not align") updated_artifacts: list[ChunkCandidateArtifact] = [] for waveform, result, artifact in zip( prepared_waveforms, verification.candidate_results, verification.chunk_artifacts, strict=True, ): median_f0_hz = None if waveform is not None and local_candidate_has_coverage_eligibility( result, max_local_boundary_speaker_drop=( SEQUENCE_FALLBACK_MAX_LOCAL_BOUNDARY_SPEAKER_DROP ), ): try: median_f0_hz = active_audio_median_f0_hz(waveform, SR) except (RuntimeError, TypeError, ValueError): median_f0_hz = None updated_artifacts.append( replace(artifact, median_f0_hz=median_f0_hz) ) return replace( verification, chunk_artifacts=tuple(updated_artifacts), ) def _verify_trajectory_audio( trajectory: tuple[np.ndarray, ...], chunks: tuple[str, ...], anchor: np.ndarray, playback_speed: float, asr_max_new_tokens: int = 128, *, release_speaker_gate: bool = False, transcriber=transcribe_whisper, semantic_only: bool = False, network_fragment_proofs: ( tuple[tuple[NetworkFragmentProof, ...], ...] | None ) = None, local_endpoint_roles: tuple[tuple[bool, bool], ...] | None = None, collect_transition_f0: bool = False, ): if len(trajectory) != len(chunks): return verify_trajectory(()) if network_fragment_proofs is not None and len(network_fragment_proofs) != len( chunks ): raise ValueError("network fragment proofs must align with local chunks") if local_endpoint_roles is not None and len(local_endpoint_roles) != len(chunks): raise ValueError("local endpoint roles must align with local chunks") encoder = None observations: list[CandidateObservation] = [] artifacts: list[ChunkCandidateArtifact] = [] prepared_waveforms: list[np.ndarray | None] = [] proof_rows = network_fragment_proofs or ((),) * len(chunks) for chunk, audio, fragment_proofs in zip( chunks, trajectory, proof_rows, strict=True, ): prepared = prepare_candidate_audio( audio, SR, transcriber=lambda waveform, sample_rate: transcriber( waveform, sample_rate, max_new_tokens=asr_max_new_tokens, ), ) if prepared is None: observations.append( CandidateObservation( target_text=chunk, transcript_text="", audio_duration_seconds=0.0, pace_cps=None, ) ) artifacts.append(ChunkCandidateArtifact()) prepared_waveforms.append(None) continue waveform = prepared.waveform prepared_waveforms.append(waveform) try: duration = active_voiced_duration_seconds(waveform, SR) except ValueError: duration = 0.0 transcript = prepared.transcript_text if fragment_proofs: fragment_evidence = canonicalize_asr_network_fragments( transcript, chunk, fragment_proofs, ) # A range-bound identifier mismatch is a hard local semantic # failure even when its contribution to whole-chunk CER is small. transcript = ( fragment_evidence.transcript_text if fragment_evidence.passed else "" ) speaker_similarity = None begin_similarity = None end_similarity = None speaker_embedding = None rms_db = None squim_stoi = None squim_pesq = None squim_si_sdr = None measure_speaker = ( not semantic_only and ( release_speaker_measurement_required(duration) if release_speaker_gate else duration >= SHORT_AUDIO_SPEAKER_GATE_SECONDS ) ) if measure_speaker: try: if encoder is None: encoder = _get_ecapa_encoder() speaker_measurement = ( release_speaker_evidence_from_audio if release_speaker_gate else speaker_evidence_from_audio ) evidence = speaker_measurement( waveform, SR, encoder, anchor, device="cpu", ) duration = evidence.active_duration_seconds speaker_similarity = evidence.similarity begin_similarity = evidence.begin_similarity end_similarity = evidence.end_similarity speaker_embedding = evidence.speaker_embedding rms_db = evidence.active_rms_db except ValueError: # A malformed/empty speaker measurement remains missing and is # rejected by the fail-closed gate for non-short candidates. pass if rms_db is None: try: rms_db = active_audio_rms_db(waveform) except ValueError: pass if not semantic_only and transcript: try: squim_evidence = squim_objective_evidence_from_audio( waveform, SR, ) squim_stoi = squim_evidence.stoi squim_pesq = squim_evidence.pesq squim_si_sdr = squim_evidence.si_sdr except ValueError: # Candidate-local invalid/non-finite acoustic evidence remains # missing and is rejected; runtime/hash failures propagate. pass observations.append( CandidateObservation( target_text=chunk, transcript_text=transcript, audio_duration_seconds=duration, speaker_similarity=speaker_similarity, begin_speaker_similarity=begin_similarity, end_speaker_similarity=end_similarity, pace_cps=( count_speech_units(chunk) / duration * float(playback_speed) if duration > 0.0 else None ), squim_stoi=squim_stoi, squim_pesq=squim_pesq, squim_si_sdr=squim_si_sdr, ) ) artifacts.append( ChunkCandidateArtifact( speaker_embedding=speaker_embedding, rms_db=rms_db, ) ) local_candidate_pool = not semantic_only and not release_speaker_gate indexed_gate_kwargs = None if local_candidate_pool and local_endpoint_roles is not None: indexed_gate_kwargs = tuple( { **({"max_prefix_cer": 0.0} if request_first else {}), **({"max_suffix_cer": 0.0} if request_last else {}), } for request_first, request_last in local_endpoint_roles ) verification = verify_trajectory( observations, chunk_artifacts=artifacts, candidate_gate_kwargs_by_index=indexed_gate_kwargs, short_text_units=6, short_text_max_cer=0.0, max_cer=QUALITY_MAX_CER, prefix_units=QUALITY_PREFIX_SUFFIX_UNITS, suffix_units=QUALITY_PREFIX_SUFFIX_UNITS, max_prefix_cer=(1.0 / 6.0 if local_candidate_pool else 0.0), max_suffix_cer=(1.0 / 6.0 if local_candidate_pool else 0.0), max_prefix_deletions=(0 if local_candidate_pool else None), max_suffix_deletions=(0 if local_candidate_pool else None), max_extra_tail_units=0, speaker_gate_enabled=not semantic_only, short_audio_seconds=( RELEASE_SPEAKER_TRIGGER_SECONDS if release_speaker_gate else SHORT_AUDIO_SPEAKER_GATE_SECONDS ), min_speaker_similarity=( QUALITY_RELEASE_MIN_SPEAKER_SIMILARITY if release_speaker_gate else QUALITY_MIN_SPEAKER_SIMILARITY ), max_boundary_speaker_drop=( QUALITY_RELEASE_MAX_BOUNDARY_SPEAKER_DROP if release_speaker_gate else QUALITY_MAX_BOUNDARY_SPEAKER_DROP ), max_pace_cps=None if semantic_only else QUALITY_MAX_PACE_CPS, squim_gate_enabled=not semantic_only, min_squim_stoi=QUALITY_MIN_SQUIM_STOI, min_squim_pesq=QUALITY_MIN_SQUIM_PESQ, ) return _attach_transition_f0( verification, tuple(prepared_waveforms), collect_transition_f0=collect_transition_f0, local_candidate_pool=local_candidate_pool, ) def _verify_independent_whole_audio( waveform: np.ndarray, target_text: str, anchor: np.ndarray, cache: WholeWaveformVerificationCache, ): """Verify exact assembled audio with the independent full large-v3 ASR.""" def verifier(exact_waveform, sample_rate, exact_target): if sample_rate != SR: raise ValueError("independent verifier sample rate mismatch") return _verify_trajectory_audio( (exact_waveform,), (exact_target,), anchor, 1.0, QUALITY_FINAL_ASR_MAX_NEW_TOKENS, transcriber=transcribe_verification_whisper, semantic_only=True, ) return cache.verify( waveform, SR, target_text, VERIFICATION_ASR_PROFILE, verifier, ) def _network_fragment_proof_rows( chunks: tuple[str, ...], chunk_specs: tuple[GenerationChunkSpec, ...] | None, ) -> tuple[tuple[NetworkFragmentProof, ...], ...] | None: """Validate explicit planner provenance before any local ASR relaxation.""" if chunk_specs is None: return None if len(chunk_specs) != len(chunks): raise ValueError("network fragment provenance must align with chunks") rows: list[tuple[NetworkFragmentProof, ...]] = [] for chunk, spec in zip(chunks, chunk_specs, strict=True): proofs = spec.network_fragment_proofs if spec.text != chunk: raise ValueError("network fragment provenance text does not match") if spec.network_conditioned: if ( not proofs or len(proofs) != len(spec.network_span_indices) or tuple(proof.span_index for proof in proofs) != spec.network_span_indices or tuple(proof.full_spoken_proof for proof in proofs) != spec.network_full_spoken_proofs ): raise ValueError("network-conditioned chunk lacks exact fragment proof") elif proofs or spec.network_full_spoken_proofs: raise ValueError("ordinary chunk carries network fragment proof") for proof in proofs: naturalized_spoken = contains_naturalized_url_spoken_form( proof.full_spoken_proof ) if naturalized_spoken: fresh_rendering = naturalized_url_rendering_proof( proof.raw_identifier ) if ( not proof.raw_identifier or fresh_rendering is None or proof.url_rendering_proof != fresh_rendering or inverse_network_url_rendering(proof.url_rendering_proof) != proof.raw_identifier or proof.url_rendering_proof.spoken_text != proof.full_spoken_proof or proof.parent_start != 0 or proof.parent_end != len(proof.full_spoken_proof) or chunk[proof.chunk_start : proof.chunk_end] != proof.full_spoken_proof ): raise ValueError( "naturalized URL runtime proof is inconsistent" ) elif proof.raw_identifier or proof.url_rendering_proof is not None: raise ValueError( "non-naturalized runtime proof carries URL rendering provenance" ) rows.append(proofs) return tuple(rows) def _local_endpoint_role_rows( chunks: tuple[str, ...], chunk_specs: tuple[GenerationChunkSpec, ...] | None, ) -> tuple[tuple[bool, bool], ...] | None: """Bind exact local onset/ending gates to immutable planner positions.""" if chunk_specs is None: return None if len(chunk_specs) != len(chunks) or any( spec.text != chunk for spec, chunk in zip(chunk_specs, chunks, strict=True) ): raise ValueError("generation chunk endpoint provenance does not align") return tuple( (spec.source_start == 0, spec.boundary_after == "none") for spec in chunk_specs ) def _assemble_trajectory_audio( trajectory: tuple[np.ndarray, ...], chunks: tuple[str, ...], playback_speed: float, chunk_specs: tuple[GenerationChunkSpec, ...] | None = None, ) -> np.ndarray: """Assemble chunks exactly as they will be returned to the listener.""" if not trajectory or len(trajectory) != len(chunks): raise ValueError("trajectory and text chunks must be non-empty and aligned") if chunk_specs is not None and ( len(chunk_specs) != len(chunks) or any(spec.text != chunk for spec, chunk in zip(chunk_specs, chunks, strict=True)) ): raise ValueError("generation chunk provenance does not align with text") if chunk_specs is not None: for index, spec in enumerate(chunk_specs): if ( spec.source_start < 0 or spec.source_end <= spec.source_start or spec.boundary_after not in {"semantic", "network_internal", "none"} ): raise ValueError("generation chunk provenance is invalid") is_last = index + 1 == len(chunk_specs) if is_last: if spec.boundary_after != "none": raise ValueError("final generation boundary must be none") continue following = chunk_specs[index + 1] if ( spec.source_end != following.source_start or spec.boundary_after == "none" ): raise ValueError("generation chunk boundaries are incomplete") if spec.boundary_after == "network_internal" and ( not spec.network_conditioned or not following.network_conditioned or not set(spec.network_span_indices).intersection( following.network_span_indices ) ): raise ValueError("network boundary lacks shared identifier proof") audio_chunks = [np.asarray(audio, dtype=np.float32).copy() for audio in trajectory] semantic_min_silence_ms = ( NETWORK_REQUEST_SEMANTIC_CHUNK_MIN_SILENCE_MS if chunk_specs is not None and any(spec.network_conditioned for spec in chunk_specs) else SEMANTIC_CHUNK_MIN_SILENCE_MS ) pauses: list[int] = [] fades_ms: list[float] = [] crossfades_ms: list[float] = [] for index, chunk in enumerate(chunks): if index > 0: audio_chunks[index] = match_chunk_rms( audio_chunks[0], audio_chunks[index], max_adjust_db=CHUNK_RMS_MATCH_DB, ) if index + 1 < len(chunks): network_internal = bool( chunk_specs is not None and chunk_specs[index].boundary_after == "network_internal" ) pause_seconds = ( NETWORK_INTERNAL_SILENCE_MS / 1000.0 if network_internal else max( punctuation_pause_seconds(chunk), semantic_min_silence_ms / 1000.0, ) ) pauses.append(int(round(pause_seconds * SR))) fades_ms.append( NETWORK_INTERNAL_FADE_MS if network_internal else CHUNK_EDGE_FADE_MS ) crossfades_ms.append( NETWORK_INTERNAL_FADE_MS if network_internal else CROSSFADE_MS ) audio_chunks = fade_variable_internal_edges(audio_chunks, SR, fades_ms) waveform = join_audio_chunks_variable( audio_chunks, pauses, crossfade_samples_by_boundary=[ int(round(crossfade_ms * SR / 1000.0)) for crossfade_ms in crossfades_ms ], pre_faded_edges=True, ) waveform = apply_loudness_floor( waveform, min_rms=0.07, peak_limit=0.95, max_gain=3.0, ) waveform = _apply_speed(waveform, playback_speed) waveform = finish_audio( waveform, SR, fade_ms=FINAL_ENDPOINT_FADE_MS, ) # Joined/final ASR, SQUIM, and speaker gates must inspect the exact # decoder-equivalent waveform represented by the public PCM16 response. return pcm16_verification_waveform(waveform) def _verification_metric_log_fields(verification) -> str: """Format normalized semantic metrics without logging transcript content.""" if len(verification.candidate_results) != 1: return ( "cer=nan prefix_cer=nan suffix_cer=nan tail_units=nan " f"reasons={verification.rejection_reasons}" ) comparison = verification.candidate_results[0].comparison candidate = verification.candidate_results[0] return ( f"cer={comparison.cer:.6f} prefix_cer={comparison.prefix_cer:.6f} " f"suffix_cer={comparison.suffix_cer:.6f} " f"tail_units={comparison.extra_tail_units} " f"squim_stoi={candidate.squim_stoi} " f"squim_pesq={candidate.squim_pesq} " f"squim_si_sdr={candidate.squim_si_sdr} " f"squim_cost={candidate.squim_quality_cost} " f"reasons={verification.rejection_reasons}" ) def _verify_network_local_asr_intersection( turbo_verification, trajectory: tuple[np.ndarray, ...], chunks: tuple[str, ...], anchor: np.ndarray, proof_rows: tuple[tuple[NetworkFragmentProof, ...], ...] | None, *, candidate_seed: int, ): """Require pinned large-v3 semantics before retaining network locals. Only proof-bearing rows are sent through the independent model. Their exact ``NetworkFragmentProof`` tuples are reused unchanged, so neither ASR can borrow a whole URL/email proof for a different local range. The intersection helper keeps turbo speaker/SQUIM/pace/artifact evidence while projecting every independent rejection into a hard semantic local failure. """ if proof_rows is None: return turbo_verification, tuple( LocalIndependentGateEvidence(False, None, 0, None) for _ in chunks ) if ( len(proof_rows) != len(chunks) or len(trajectory) != len(chunks) or len(turbo_verification.candidate_results) != len(chunks) ): raise ValueError("network local verification provenance does not align") independent_evidence = [ LocalIndependentGateEvidence( attempted=False, passed=None, proof_count=len(proofs), result=None, ) for proofs in proof_rows ] selected_indices = tuple( index for index, proofs in enumerate(proof_rows) if ( proofs and local_candidate_has_coverage_eligibility( turbo_verification.candidate_results[index], max_local_boundary_speaker_drop=( SEQUENCE_FALLBACK_MAX_LOCAL_BOUNDARY_SPEAKER_DROP ), ) ) ) if not selected_indices: return turbo_verification, tuple(independent_evidence) independent_proof_rows = tuple( proof_rows[index] for index in selected_indices ) independent_verification = _verify_trajectory_audio( tuple(trajectory[index] for index in selected_indices), tuple(chunks[index] for index in selected_indices), anchor, 1.0, QUALITY_FINAL_ASR_MAX_NEW_TOKENS, transcriber=transcribe_verification_whisper, semantic_only=True, network_fragment_proofs=independent_proof_rows, ) for local_index, primary_index in enumerate(selected_indices): result = independent_verification.candidate_results[local_index] independent_evidence[primary_index] = LocalIndependentGateEvidence( attempted=True, passed=result.passed is True, proof_count=len(proof_rows[primary_index]), result=candidate_gate_evidence(result), ) comparison = result.comparison print( "[BlueMagpie] network local independent " f"seed={candidate_seed} local_chunk_index={primary_index} " f"proof_count={len(proof_rows[primary_index])} " f"passed={result.passed} cer={comparison.cer:.6f} " f"prefix_cer={comparison.prefix_cer:.6f} " f"suffix_cer={comparison.suffix_cer:.6f} " f"tail_units={comparison.extra_tail_units} " f"reasons={result.rejection_reasons}" ) return ( intersect_local_semantic_verification( turbo_verification, independent_verification, selected_indices, ), tuple(independent_evidence), ) def _qualify_candidate_trajectory_audio( trajectory: tuple[np.ndarray, ...], chunks: tuple[str, ...], whole_target_text: str, anchor: np.ndarray, playback_speed: float, independent_cache: WholeWaveformVerificationCache, *, candidate_seed: int, chunk_specs: tuple[GenerationChunkSpec, ...] | None = None, ): """Run whole-output qualification only after every local chunk passes.""" proof_rows = _network_fragment_proof_rows(chunks, chunk_specs) local_verification = _verify_trajectory_audio( trajectory, chunks, anchor, playback_speed, network_fragment_proofs=proof_rows, local_endpoint_roles=_local_endpoint_role_rows(chunks, chunk_specs), collect_transition_f0=len(chunks) > 1, ) local_verification, independent_local_results = ( _verify_network_local_asr_intersection( local_verification, trajectory, chunks, anchor, proof_rows, candidate_seed=candidate_seed, ) ) if not local_verification.passed: return CandidateVerification( local_verification, independent_local_results=independent_local_results, ) waveform = _assemble_trajectory_audio( trajectory, chunks, playback_speed, chunk_specs, ) joined_verification = _verify_trajectory_audio( (waveform,), (whole_target_text,), anchor, 1.0, QUALITY_FINAL_ASR_MAX_NEW_TOKENS, release_speaker_gate=True, ) qualified = qualify_trajectory_with_joined_output( local_verification, joined_verification, ) joined_evidence = trajectory_gate_evidence(joined_verification) if not qualified.passed: print( "[BlueMagpie] candidate joined output rejected " f"seed={candidate_seed} " f"{_verification_metric_log_fields(joined_verification)}" ) return CandidateVerification( qualified, independent_local_results=independent_local_results, joined_output=joined_evidence, ) independent_verification = _verify_independent_whole_audio( waveform, whole_target_text, anchor, independent_cache, ) dual_qualified = qualify_trajectory_with_joined_output( qualified, independent_verification, ) if not dual_qualified.passed: print( "[BlueMagpie] candidate independent whole output rejected " f"seed={candidate_seed} " f"{_verification_metric_log_fields(independent_verification)}" ) return CandidateVerification( dual_qualified, independent_local_results=independent_local_results, joined_output=joined_evidence, independent_output=trajectory_gate_evidence(independent_verification), ) def _verify_refill_candidate_trajectory_audio( trajectory: tuple[np.ndarray, ...], chunks: tuple[str, ...], anchor: np.ndarray, playback_speed: float, chunk_specs: tuple[GenerationChunkSpec, ...] | None = None, *, candidate_seed: int, ): """Apply strict dual-ASR local gates to one safe-duration refill.""" if len(trajectory) != 1 or len(chunks) != 1: return verify_trajectory(()) proof_rows = _network_fragment_proof_rows(chunks, chunk_specs) local_verification = _verify_trajectory_audio( trajectory, chunks, anchor, playback_speed, network_fragment_proofs=proof_rows, local_endpoint_roles=_local_endpoint_role_rows(chunks, chunk_specs), collect_transition_f0=True, ) local_verification, independent_local_results = ( _verify_network_local_asr_intersection( local_verification, trajectory, chunks, anchor, proof_rows, candidate_seed=candidate_seed, ) ) return CandidateVerification( local_verification, independent_local_results=independent_local_results, ) def _verify_sequence_trajectory_audio( sequence_result, chunks: tuple[str, ...], whole_target_text: str, anchor: np.ndarray, playback_speed: float, independent_cache: WholeWaveformVerificationCache, chunk_specs: tuple[GenerationChunkSpec, ...] | None = None, ): """Verify one ranked DP path after exact production assembly.""" waveform = _assemble_trajectory_audio( sequence_result.trajectory, chunks, playback_speed, chunk_specs, ) # The release contract assigns local candidate coverage (including exact # range-bound network fragments) to turbo and the exact assembled # waveform to the separately pinned full large-v3 verifier. Do not let a # redundant long-form turbo decode veto a path that passes that stronger # assembled-waveform gate. assembled_verification = _verify_trajectory_audio( (waveform,), (whole_target_text,), anchor, 1.0, QUALITY_FINAL_ASR_MAX_NEW_TOKENS, release_speaker_gate=True, transcriber=transcribe_verification_whisper, ) status = "verified" if assembled_verification.passed else "rejected" print( f"[BlueMagpie] sequence path large-v3 {status} " f"rank={sequence_result.sequence_path_rank} " f"chunk_candidates={sequence_result.chunk_candidate_indices} " f"{_verification_metric_log_fields(assembled_verification)}" ) if not assembled_verification.passed: return assembled_verification independent_verification = _verify_independent_whole_audio( waveform, whole_target_text, anchor, independent_cache, ) intersected = qualify_trajectory_with_joined_output( assembled_verification, independent_verification, ) independent_status = "verified" if independent_verification.passed else "rejected" print( f"[BlueMagpie] sequence path independent large-v3 {independent_status} " f"rank={sequence_result.sequence_path_rank} " f"chunk_candidates={sequence_result.chunk_candidate_indices} " f"{_verification_metric_log_fields(independent_verification)}" ) return intersected def _synthesize( text: str, centroid: torch.Tensor, *, cfg: float, steps: int, speed: float, request_seed: int | None = None, ) -> tuple[int, np.ndarray]: if isinstance(steps, (bool, np.bool_)): raise gr.Error(f"目前只支援已驗證的 NFE {DEFAULT_STEPS}。") try: requested_steps = int(steps) exact_steps = float(steps) == float(requested_steps) except (TypeError, ValueError, OverflowError): requested_steps = -1 exact_steps = False if not exact_steps or requested_steps != DEFAULT_STEPS: raise gr.Error(f"目前只支援已驗證的 NFE {DEFAULT_STEPS}。") steps = requested_steps try: cfg_value = float(cfg) except (TypeError, ValueError, OverflowError): raise gr.Error("CFG 必須是有限數值。") from None if not np.isfinite(cfg_value) or not 1.0 <= cfg_value <= 4.0: raise gr.Error("CFG 必須介於 1.0 與 4.0。") if cfg_value != MIXED_CFG_PRIMARY: raise gr.Error(f"目前只支援已驗證的主 CFG {MIXED_CFG_PRIMARY:.1f}。") raw_text = str(text) if len(raw_text) > MAX_TEXT_CHARS: raise gr.Error(f"單次最多 {MAX_TEXT_CHARS} 個字元，請分段合成。") if network_identifier_has_ambiguous_iri(raw_text): raise gr.Error("網址目前只支援 ASCII 字元；非 ASCII IRI 會與字母讀音混淆。") network_request = contains_network_identifier(raw_text) request_cfg = MIXED_CFG_PRIMARY try: text = normalize_spoken_forms(raw_text, locale="zh-TW") except ValueError as error: raise gr.Error(str(error)) from None if not text: raise gr.Error("請先輸入要合成的文字。") if not network_request: text = strip_unspoken_cjk_quotes_for_generation(text) normalized_units = count_speech_units(text) if normalized_units <= 0: raise gr.Error("輸入必須包含可發音的文字或數字。") if normalized_units > QUALITY_MAX_GENERATED_TEXT_UNITS: raise gr.Error("正規化後的語音單位超過安全生成上限，請分段合成。") if not np.isfinite(float(speed)) or not 0.85 <= float(speed) <= 1.05: raise gr.Error("後處理語速必須介於 0.85 與 1.05。") chunk_specs: tuple[GenerationChunkSpec, ...] | None = None try: if network_request: chunk_specs = plan_generation_chunks( raw_text, text, min_units=MIN_CHUNK_CHARS, network_min_units=NETWORK_GENERATION_MIN_UNITS, target_units=NETWORK_GENERATION_TARGET_UNITS, network_max_units=NETWORK_GENERATION_MAX_UNITS, ordinary_max_units=NETWORK_REQUEST_ORDINARY_MAX_UNITS, split_atomic_email_at_separator=( NETWORK_SPLIT_ATOMIC_EMAIL_AT_SEPARATOR ), ) chunks = tuple(spec.text for spec in chunk_specs) if len(chunks) > QUALITY_MAX_GENERATED_CHUNKS: raise ValueError( "network request exceeds the generated-chunk budget" ) else: chunks = tuple( coalesce_text_chunks( split_quality_text_for_tts( text, long_max_units=CHUNK_CHARS, min_chunk_units=MIN_CHUNK_CHARS, medium_max_units=QUALITY_MEDIUM_TEXT_MAX_UNITS, medium_chunk_units=QUALITY_MEDIUM_CHUNK_UNITS, ), max_chunks=QUALITY_MAX_GENERATED_CHUNKS, max_units=CHUNK_CHARS, ) ) except ValueError as error: raise gr.Error("文字無法在已驗證的生成限制內安全分段。") from error request_seed = resolve_request_seed(request_seed, secrets.randbelow) anchor = _speaker_anchor_array(centroid) independent_cache = WholeWaveformVerificationCache() generation_context_by_seed: dict[int, CandidateGenerationContext] = {} generation_endpoint_evidence_by_seed: dict[ int, tuple[_ChunkGenerationOutcome, ...], ] = {} def canonical_chunks_for_context( generation_context: CandidateGenerationContext, ) -> tuple[str, ...]: canonical: list[str] = [] for chunk_index in generation_context.chunk_indices: if not 0 <= chunk_index < len(chunks): raise ValueError("candidate generation provenance is out of range") canonical.append(chunks[chunk_index]) return tuple(canonical) def candidate_generation_text_transform( canonical_chunks: tuple[str, ...], generation_context: CandidateGenerationContext, ) -> tuple[str, ...]: expected_canonical = canonical_chunks_for_context(generation_context) if canonical_chunks != expected_canonical: raise ValueError("candidate generation text lacks canonical provenance") if chunk_specs is None: return canonical_chunks transformed: list[str] = [] for chunk_index, chunk, candidate_ordinal in zip( generation_context.chunk_indices, canonical_chunks, generation_context.chunk_candidate_ordinals, strict=True, ): spec = chunk_specs[chunk_index] if spec.text != chunk: raise ValueError("candidate generation text does not match its proof") transformed.append( email_domain_mail_generation_variant( chunk, spec.network_fragment_proofs, ) if candidate_ordinal in EMAIL_MAIL_FALLBACK_CANDIDATE_ORDINALS else chunk ) return tuple(transformed) def candidate_generation_text_variants( canonical_chunks: tuple[str, ...], generation_context: CandidateGenerationContext, ) -> tuple[str, ...]: expected_canonical = canonical_chunks_for_context(generation_context) if canonical_chunks != expected_canonical: raise ValueError("candidate text variant lacks canonical provenance") generated_chunks = candidate_generation_text_transform( canonical_chunks, generation_context, ) variants: list[str] = [] for chunk_index, canonical, generated, candidate_ordinal in zip( generation_context.chunk_indices, canonical_chunks, generated_chunks, generation_context.chunk_candidate_ordinals, strict=True, ): if generated == canonical: variants.append(BASE_CHUNK_TEXT_VARIANT) continue if chunk_specs is None: raise ValueError("candidate text variant lacks network proof") spec = chunk_specs[chunk_index] expected_variant = email_domain_mail_generation_variant( canonical, spec.network_fragment_proofs, ) if ( candidate_ordinal not in EMAIL_MAIL_FALLBACK_CANDIDATE_ORDINALS or generated != expected_variant or generated == canonical or not any( proof.identifier_kind == "email" for proof in spec.network_fragment_proofs ) ): raise ValueError("candidate text variant is not proof-bound") variants.append(EMAIL_DOMAIN_MAIL_CHUNK_TEXT_VARIANT) return tuple(variants) def generation_chunk_specs( seed: int, candidate_chunks: tuple[str, ...], ) -> tuple[GenerationChunkSpec, ...] | None: if chunk_specs is None: return None context = generation_context_by_seed.get(seed) if context is None or len(context.chunk_indices) != len(candidate_chunks): raise ValueError("candidate verification lacks generation provenance") selected: list[GenerationChunkSpec] = [] for chunk_index, chunk in zip( context.chunk_indices, candidate_chunks, strict=True, ): if not 0 <= chunk_index < len(chunk_specs): raise ValueError("candidate verification provenance is out of range") spec = chunk_specs[chunk_index] if spec.text != chunk: raise ValueError("candidate verification provenance text does not match") selected.append(spec) return tuple(selected) def candidate_cfg(candidate_ordinal: int) -> float: return generation_cfg_for_candidate_offset( candidate_ordinal, primary_cfg=request_cfg, alternate_cfg=MIXED_CFG_ALTERNATE, ) def chunk_cfg_evidence( chunk: str, candidate_ordinal: int, *, network_conditioned: bool | None = None, ) -> tuple[float, tuple[str, ...]]: scheduled = generation_cfg_for_candidate_offset( candidate_ordinal, primary_cfg=request_cfg, alternate_cfg=MIXED_CFG_ALTERNATE, ) reasons: list[str] = [] network_adjusted = scheduled is_network = ( bool(network_protected_spoken_spans(chunk)) if network_conditioned is None else bool(network_conditioned) ) if is_network and scheduled < NETWORK_TEXT_CFG_MIN: network_adjusted = NETWORK_TEXT_CFG_MIN reasons.append("network") effective = effective_generation_cfg( chunk, network_adjusted, short_text_unit_threshold=SHORT_TEXT_CFG_UNITS, short_text_min_cfg=SHORT_TEXT_CFG_MIN, ) if effective > network_adjusted: reasons.append("short_text") return effective, tuple(reasons) def generation_network_flags( chunk_indices: tuple[int, ...], candidate_chunks: tuple[str, ...], *, generation_context: CandidateGenerationContext, ) -> tuple[bool, ...]: if len(chunk_indices) != len(candidate_chunks): raise ValueError("generation provenance does not align with chunks") canonical_chunks = canonical_chunks_for_context(generation_context) expected_generation_chunks = candidate_generation_text_transform( canonical_chunks, generation_context, ) if candidate_chunks != expected_generation_chunks: raise ValueError("candidate generation text disagrees with its schedule") if chunk_specs is None: return tuple( bool(network_protected_spoken_spans(chunk)) for chunk in canonical_chunks ) flags: list[bool] = [] for chunk_index, chunk in zip( chunk_indices, canonical_chunks, strict=True, ): if not 0 <= chunk_index < len(chunk_specs): raise ValueError("generation provenance index is out of range") spec = chunk_specs[chunk_index] if spec.text != chunk: raise ValueError("generation provenance text does not match") flags.append(spec.network_conditioned) return tuple(flags) def effective_chunk_cfg( chunk: str, candidate_ordinal: int, *, network_conditioned: bool | None = None, ) -> float: return chunk_cfg_evidence( chunk, candidate_ordinal, network_conditioned=network_conditioned, )[0] def candidate_generation( candidate_chunks: tuple[str, ...], seed: int, *, generation_context: CandidateGenerationContext, ) -> tuple[np.ndarray, ...]: if generation_context.seed != seed: raise ValueError("generation context seed does not match the request seed") previous_context = generation_context_by_seed.get(seed) if previous_context is not None and previous_context != generation_context: raise ValueError("one candidate seed cannot carry two generation contexts") generation_context_by_seed[seed] = generation_context ordinals = generation_context.chunk_candidate_ordinals if not ordinals or len(set(ordinals)) != 1: raise ValueError("one generation call must use one candidate ordinal") candidate_ordinal = ordinals[0] network_flags = generation_network_flags( generation_context.chunk_indices, candidate_chunks, generation_context=generation_context, ) trajectory, endpoint_outcomes = _generate_trajectory( candidate_chunks, centroid, cfg=candidate_cfg(candidate_ordinal), steps=steps, request_seed=seed, policy=generation_policy_for_candidate_offset(candidate_ordinal), network_conditioned=network_flags, ) if len(endpoint_outcomes) != len(candidate_chunks): raise ValueError("generation endpoint evidence is incomplete") generation_endpoint_evidence_by_seed[seed] = endpoint_outcomes return trajectory def candidate_generation_evidence( candidate_index: int, seed: int, chunk_indices: tuple[int, ...], candidate_chunks: tuple[str, ...], *, generation_context: CandidateGenerationContext, ) -> CandidateGenerationEvidence: if candidate_index != seed - request_seed: raise ValueError("candidate index does not match request seed offset") if ( generation_context.candidate_index != candidate_index or generation_context.seed != seed or generation_context.chunk_indices != chunk_indices ): raise ValueError("candidate generation context does not match the attempt") candidate_ordinals = generation_context.chunk_candidate_ordinals if not candidate_ordinals or len(set(candidate_ordinals)) != 1: raise ValueError("one generation call must use one candidate ordinal") candidate_ordinal = candidate_ordinals[0] scheduled = candidate_cfg(candidate_ordinal) network_flags = generation_network_flags( chunk_indices, candidate_chunks, generation_context=generation_context, ) canonical_chunks = canonical_chunks_for_context(generation_context) expected_generation_chunks = candidate_generation_text_transform( canonical_chunks, generation_context, ) if candidate_chunks != expected_generation_chunks: raise ValueError("generation evidence text disagrees with its schedule") text_variants = candidate_generation_text_variants( canonical_chunks, generation_context, ) rows = tuple( chunk_cfg_evidence( chunk, candidate_ordinal, network_conditioned=network_flag, ) for chunk, network_flag in zip( candidate_chunks, network_flags, strict=True, ) ) endpoint_outcomes = generation_endpoint_evidence_by_seed.get(seed) if endpoint_outcomes is None or len(endpoint_outcomes) != len( candidate_chunks ): raise ValueError("generation endpoint evidence is missing") return CandidateGenerationEvidence( chunk_indices=chunk_indices, chunk_text_units=tuple( count_speech_units(chunk) for chunk in candidate_chunks ), scheduled_cfg=scheduled, effective_cfgs=tuple(row[0] for row in rows), floor_reasons=tuple(row[1] for row in rows), chunk_candidate_ordinals=candidate_ordinals, network_conditioned=network_flags, chunk_text_variants=text_variants, chunk_stop_reasons=tuple( outcome.stop_reason for outcome in endpoint_outcomes ), chunk_endpoint_energy_ratios=tuple( outcome.endpoint_energy_ratio for outcome in endpoint_outcomes ), chunk_generated_steps=tuple( outcome.generated_steps for outcome in endpoint_outcomes ), chunk_hard_stop_steps=tuple( outcome.hard_stop_steps for outcome in endpoint_outcomes ), ) try: with _GENERATION_LOCK: cascade = run_coverage_adaptive_cascade( chunks, request_seed, candidate_generation, lambda trajectory, candidate_chunks, seed: _qualify_candidate_trajectory_audio( trajectory, candidate_chunks, text, anchor, speed, independent_cache, candidate_seed=seed, chunk_specs=generation_chunk_specs(seed, candidate_chunks), ), lambda trajectory, candidate_chunks, seed: ( _qualify_candidate_trajectory_audio( trajectory, candidate_chunks, text, anchor, speed, independent_cache, candidate_seed=seed, chunk_specs=generation_chunk_specs(seed, candidate_chunks), ) if len(chunks) == 1 else _verify_refill_candidate_trajectory_audio( trajectory, candidate_chunks, anchor, speed, generation_chunk_specs(seed, candidate_chunks), candidate_seed=seed, ) ), sequence_final_verifier=lambda sequence_result, candidate_chunks: ( _verify_sequence_trajectory_audio( sequence_result, candidate_chunks, text, anchor, speed, independent_cache, chunk_specs, ) ), generation_evidence_factory=candidate_generation_evidence, candidate_generation_text_transform=( candidate_generation_text_transform ), max_generated_chunks=QUALITY_MAX_GENERATED_CHUNKS, max_generated_text_units=QUALITY_MAX_GENERATED_TEXT_UNITS, max_sequence_paths=QUALITY_MAX_SEQUENCE_PATHS, sequence_fallback_max_local_boundary_speaker_drop=( SEQUENCE_FALLBACK_MAX_LOCAL_BOUNDARY_SPEAKER_DROP ), preferred_min_speaker_similarity=( QUALITY_PREFERRED_MIN_SPEAKER_SIMILARITY ), preferred_max_boundary_speaker_drop=( QUALITY_PREFERRED_MAX_BOUNDARY_SPEAKER_DROP ), preferred_min_squim_stoi=QUALITY_PREFERRED_MIN_SQUIM_STOI, preferred_min_squim_pesq=QUALITY_PREFERRED_MIN_SQUIM_PESQ, preferred_min_squim_audio_duration_seconds=( QUALITY_PREFERRED_SQUIM_MIN_DURATION_SECONDS ), require_endpoint_evidence=True, ) except NoQualifiedCandidateError as error: print( format_cascade_evidence_log( error.diagnostics, outcome="no_qualified_candidate", generated_chunk_limit=QUALITY_MAX_GENERATED_CHUNKS, generated_text_unit_limit=QUALITY_MAX_GENERATED_TEXT_UNITS, ) ) raise gr.Error("目前沒有候選通過內容與音色驗證，請稍後重試或調整文字。") from error except (RuntimeError, ValueError) as error: raise gr.Error("品質驗證暫時無法完成，未回傳未驗證的語音。") from error attempts_by_index = { attempt.candidate_index: attempt for attempt in cascade.diagnostics.attempts } selected_ordinals: list[int] = [] for chunk_index, candidate_index in enumerate(cascade.chunk_candidate_indices): attempt = attempts_by_index.get(candidate_index) if attempt is None or chunk_index not in attempt.chunk_indices: raise gr.Error("品質驗證紀錄不完整，未回傳未驗證的語音。") local_index = attempt.chunk_indices.index(chunk_index) try: selected_ordinals.append(attempt.chunk_candidate_ordinals[local_index]) except IndexError as error: raise gr.Error("品質驗證紀錄不完整，未回傳未驗證的語音。") from error selected_policies = tuple( generation_policy_for_candidate_offset(ordinal).name for ordinal in selected_ordinals ) attempted_policies = tuple( generation_policy_for_candidate_offset( attempt.chunk_candidate_ordinals[0] ).name for attempt in cascade.diagnostics.attempts ) selected_network_flags = ( tuple(spec.network_conditioned for spec in chunk_specs) if chunk_specs is not None else tuple( bool(network_protected_spoken_spans(chunk)) for chunk in chunks ) ) selected_cfgs = tuple( effective_chunk_cfg( chunk, candidate_ordinal, network_conditioned=network_flag, ) for chunk, candidate_ordinal, network_flag in zip( chunks, selected_ordinals, selected_network_flags, strict=True, ) ) attempted_schedule_cfgs = tuple( candidate_cfg(attempt.chunk_candidate_ordinals[0]) for attempt in cascade.diagnostics.attempts ) print( "[BlueMagpie] quality cascade " f"candidate_index={cascade.candidate_index} attempts={len(cascade.attempted_seeds)} " f"selection={cascade.selection_mode} " f"chunk_candidates={cascade.chunk_candidate_indices} " f"chunk_seeds={cascade.chunk_seeds} score={cascade.verification.score:.6f}" f" chunk_policies={selected_policies} attempted_policies={attempted_policies}" f" chunk_cfgs={selected_cfgs}" f" attempted_schedule_cfgs={attempted_schedule_cfgs}" f" sequence_rank={cascade.sequence_path_rank}" f" sequence_paths_checked={cascade.sequence_paths_checked}" f" generated_chunks={cascade.generated_chunk_count}" f" generated_text_units={cascade.generated_text_units}" f" chunk_candidate_counts={cascade.chunk_candidate_counts}" f" request_cfg={request_cfg:.2f} mixed_cfg_primary={MIXED_CFG_PRIMARY:.2f}" f" network_cfg_floor={NETWORK_TEXT_CFG_MIN:.2f}" f" cfg_schedule={MIXED_CFG_SCHEDULE}" f" network_request={network_request}" ) waveform = _assemble_trajectory_audio( cascade.trajectory, chunks, speed, chunk_specs, ) final_verification = _verify_trajectory_audio( (waveform,), (text,), anchor, 1.0, QUALITY_FINAL_ASR_MAX_NEW_TOKENS, release_speaker_gate=True, transcriber=( transcribe_verification_whisper if len(chunks) > 1 else transcribe_whisper ), ) final_evidence = trajectory_gate_evidence(final_verification) independent_final_evidence = None independent_final_verification = None try: require_verified_final_output(final_verification) independent_final_verification = _verify_independent_whole_audio( waveform, text, anchor, independent_cache, ) independent_final_evidence = trajectory_gate_evidence( independent_final_verification ) require_verified_final_output(independent_final_verification) except FinalOutputRejectedError as error: independent_fields = ( "not_run" if independent_final_verification is None else _verification_metric_log_fields(independent_final_verification) ) print( format_cascade_evidence_log( cascade.diagnostics, outcome="final_output_rejected", generated_chunk_limit=QUALITY_MAX_GENERATED_CHUNKS, generated_text_unit_limit=QUALITY_MAX_GENERATED_TEXT_UNITS, selection=cascade, final_output=final_evidence, independent_final_output=independent_final_evidence, ) ) print( "[BlueMagpie] final output rejected " f"turbo={_verification_metric_log_fields(final_verification)} " f"independent={independent_fields}" ) raise gr.Error("最終合成結果未通過整段內容、語速與音色驗證，未回傳音訊。") from error except (RuntimeError, ValueError) as error: raise gr.Error("品質驗證暫時無法完成，未回傳未驗證的語音。") from error print( format_cascade_evidence_log( cascade.diagnostics, outcome="returned", generated_chunk_limit=QUALITY_MAX_GENERATED_CHUNKS, generated_text_unit_limit=QUALITY_MAX_GENERATED_TEXT_UNITS, selection=cascade, final_output=final_evidence, independent_final_output=independent_final_evidence, ) ) print( "[BlueMagpie] final output verified " f"turbo_score={final_verification.score:.6f} " f"independent_score={independent_final_verification.score:.6f} " f"independent_cache_entries={independent_cache.entry_count}" ) return SR, waveform @gpu def tts_speaker( text: str, speaker: str = DEFAULT_SPEAKER, cfg: float = DEFAULT_CFG, steps: int = DEFAULT_STEPS, speed: float = 1.0, ): return pcm16_audio_output( *_synthesize( text, SPEAKERS.get(speaker, DEFAULT_CENTROID), cfg=cfg, steps=steps, speed=speed, ) ) @gpu def tts_reference( text: str, reference_wav: str, cfg: float = DEFAULT_CFG, steps: int = DEFAULT_STEPS, speed: float = 1.0, ): if not reference_wav: raise gr.Error("請先錄音或上傳參考音檔。") try: centroid = extract_windowed_speaker_embedding( reference_wav, _get_ecapa_encoder(), device="cpu", min_duration_seconds=3.0, window_seconds=3.0, hop_seconds=1.5, max_windows=12, full_clip_max_seconds=12.0, ) except ValueError as error: raise gr.Error(str(error)) from error return pcm16_audio_output( *_synthesize(text, centroid, cfg=cfg, steps=steps, speed=speed) ) @gpu def tts_longform( text: str, speaker: str = DEFAULT_SPEAKER, cfg: float = DEFAULT_CFG, steps: int = DEFAULT_STEPS, speed: float = 1.0, ): return pcm16_audio_output( *_synthesize( text, SPEAKERS.get(speaker, DEFAULT_CENTROID), cfg=cfg, steps=steps, speed=speed, ) ) EXAMPLE_TEXTS = [ "今天天氣真好，我們一起去散步吧。", "我要吃蚵仔煎，然後去丟垃圾。", "這學期的成績包括研究報告和期末考。", "這是 AI TTS code switching 測試，混合中英文也沒問題。", "注音符號測試：ㄅ、ㄆ、ㄇ、ㄈ。", ] LONGFORM_EXAMPLES = [ "今天的會議會先整理目前進度，再確認下一階段的工作。遇到需要討論的項目時，" "請先記下問題，等報告結束後再一起處理。最後，我們會確認負責人和預計完成時間。", "歡迎收聽今天的內容。第一段會介紹背景，第二段整理實際案例，最後一段則說明後續安排。" "如果中途聽到英文術語，不必擔心，我們會用中文補充它的意思。", ] HEADER = f""" # BlueMagpie-TTS Demo 台灣華語與中英混合文字轉語音。模型版本：`{CHECKPOINT}`。 執行環境固定於已驗證的 model revision `{MODEL_REVISION[:8]}`、ECAPA revision `{ECAPA_REVISION[:8]}` 與 Whisper revision `{WHISPER_REVISION[:8]}`，避免服務重啟時 無聲變更權重、speaker embedding 或語意驗證空間。 目前預設採用穩定推論設定：候選交錯 `CFG 3.0 / 2.0`（網路內容與極短句最低 3.0）、 `NFE 10`、目標語速 `4.0 字/秒`、 初始完整 trajectory 使用 base duration estimate（{BASE_GENERATION_POLICY.cjk_cps:.1f} CJK / {BASE_GENERATION_POLICY.ascii_cps:.1f} ASCII），後續 zero/low-coverage 單 chunk refill 使用 safe duration estimate （{SAFE_DURATION_GENERATION_POLICY.cjk_cps:.1f} CJK / {SAFE_DURATION_GENERATION_POLICY.ascii_cps:.1f} ASCII）；兩者只調整生成上限，生成完成後才校正至 目標語速。另補齊句末提示、套用尾端 weak-stop 保護、 不額外切開 80 字內的首段，長文才依自然標點與 80 字上限切段。每個 request 只生成一次 same-seed 完整 trajectory；失敗後只補低覆蓋 chunks，再以 speaker/RMS/F0 ragged DP 選出最多三條 culprit-diverse exact paths。NFE 固定為已驗證的 10，且每個 request 最多生成 32 個 TTS chunks 與 800 speech units。 """ with gr.Blocks(title="BlueMagpie-TTS Demo", theme=gr.themes.Soft()) as demo: gr.Markdown(HEADER) with gr.Accordion("進階生成參數", open=False): with gr.Row(): cfg_input = gr.Number( value=DEFAULT_CFG, label="CFG (已驗證固定值)", interactive=False, ) steps_input = gr.Number( value=DEFAULT_STEPS, precision=0, interactive=False, label="NFE steps（已驗證固定值）", ) speed_input = gr.Slider(0.85, 1.05, value=1.0, step=0.05, label="後處理語速") with gr.Tab("內建語者"): with gr.Row(): with gr.Column(): speaker_input = gr.Dropdown(list(SPEAKERS), value=DEFAULT_SPEAKER, label="語者") speaker_text = gr.Textbox(label="文字", lines=4, max_lines=8) speaker_button = gr.Button("合成", variant="primary") with gr.Column(): speaker_output = gr.Audio(label="合成結果", type="numpy") gr.Examples(EXAMPLE_TEXTS, inputs=speaker_text, label="範例") speaker_button.click( tts_speaker, [speaker_text, speaker_input, cfg_input, steps_input, speed_input], speaker_output, ) with gr.Tab("參考音色"): gr.Markdown("參考音檔至少 3 秒；只使用已取得授權的聲音。參考內容不需要逐字稿。") with gr.Row(): with gr.Column(): reference_text = gr.Textbox(label="文字", lines=4, max_lines=8) reference_audio = gr.Audio( label="參考音檔", type="filepath", sources=["microphone", "upload"], ) reference_button = gr.Button("合成", variant="primary") with gr.Column(): reference_output = gr.Audio(label="合成結果", type="numpy") reference_button.click( tts_reference, [reference_text, reference_audio, cfg_input, steps_input, speed_input], reference_output, ) with gr.Tab("穩定長文"): with gr.Row(): with gr.Column(): longform_speaker = gr.Dropdown(list(SPEAKERS), value=DEFAULT_SPEAKER, label="語者") longform_text = gr.Textbox( label=f"長文（最多 {MAX_TEXT_CHARS} 字元）", lines=8, max_lines=12, ) longform_button = gr.Button("合成完整長文", variant="primary") with gr.Column(): longform_output = gr.Audio(label="合成結果", type="numpy") gr.Examples(LONGFORM_EXAMPLES, inputs=longform_text, label="長文範例") longform_button.click( tts_longform, [longform_text, longform_speaker, cfg_input, steps_input, speed_input], longform_output, ) gr.Markdown( "合成語音僅供研究與評估展示；正式使用前請人工檢視。 " "[模型](https://huggingface.co/OpenFormosa/BlueMagpie-TTS) · " "[程式碼](https://github.com/OpenFormosa/BlueMagpie-TTS)" ) if __name__ == "__main__": demo.queue(default_concurrency_limit=1).launch()