"""BlueMagpie-TTS Gradio demo using a latency-first inference profile.""" from __future__ import annotations from dataclasses import dataclass, replace import hashlib import inspect import json import os from pathlib import Path import secrets import tempfile import threading from types import MappingProxyType # The public demo intentionally favors throughput and first-audio latency over # cross-process bit identity. Let cuDNN select the fastest kernels and allow # TensorFloat-32 on hardware that supports it. DETERMINISTIC_ALGORITHMS = False CUDNN_BENCHMARK = True CUDNN_DETERMINISTIC = False CUDA_MATMUL_ALLOW_TF32 = True CUDNN_ALLOW_TF32 = True os.environ["MAMBA_DETERMINISTIC"] = "0" INTERACTIVE_GPU_LEASE_SECONDS = 60 LONGFORM_GPU_LEASE_SECONDS = 120 AUTO_LONG_TEXT_CHARS = 80 def _gpu_lease_seconds(text: object) -> int: """Reserve the long-form lease only when automatic chunking needs it.""" return ( LONGFORM_GPU_LEASE_SECONDS if len(str(text)) > AUTO_LONG_TEXT_CHARS else INTERACTIVE_GPU_LEASE_SECONDS ) def _speaker_gpu_duration(text, speaker, cfg, steps, speed, seed) -> int: del speaker, cfg, steps, speed, seed return _gpu_lease_seconds(text) def _reference_gpu_duration(text, reference_wav, cfg, steps, speed, seed) -> int: del reference_wav, cfg, steps, speed, seed return _gpu_lease_seconds(text) try: import spaces speaker_gpu = spaces.GPU(duration=_speaker_gpu_duration) reference_gpu = spaces.GPU(duration=_reference_gpu_duration) USING_ZERO_GPU = True except ImportError: def speaker_gpu(function): return function reference_gpu = speaker_gpu USING_ZERO_GPU = False import gradio as gr import numpy as np import soundfile as sf import torch from huggingface_hub import snapshot_download from transformers import PreTrainedTokenizerFast 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 torch.set_float32_matmul_precision("high") from audio_wsola import wsola_time_stretch from audio_artifact_gate import ( DEFAULT_ECHO_SMEARING_GATE_CONFIG, evaluate_echo_smearing, ) from bluemagpie import BlueMagpieModel from glyph_ascii_v1 import GLYPH_ASCII_V1_SYMBOLS from glyph_assets import ( MAX_GENERATED_CHUNK_BUDGET as GLYPH_MAX_GENERATED_CHUNK_BUDGET, MAX_GENERATED_TEXT_UNIT_BUDGET as GLYPH_MAX_GENERATED_TEXT_UNIT_BUDGET, AssetSegment, CarrierSegment, GenerationAttemptRecord, GlyphAssetBundle, GlyphAssetError, SemanticPlanCommitment, assemble_hybrid_pcm16, compute_hybrid_plan_sha256, compute_ordered_attempt_ledger_sha256, load_glyph_asset_bundle, verify_hybrid_assembly, ) from glyph_hybrid_plan import ( GlyphHybridPlanError, HybridRenderPlan, build_glyph_hybrid_plan, glyph_hybrid_plan_sha256, inverse_glyph_hybrid_plan, ) from glyph_hybrid_adapter import ( GlyphAdapterRuntime, HardenedGlyphHybridAdapter, ) from glyph_hybrid_evidence import GatePolicy as GlyphEvidenceGatePolicy from latency_timing import ( latency_stage, timed_latency_request, timed_latency_stage, ) 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, 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, BREEZE25_MAX_VERIFICATION_SEGMENTS, BREEZE25_MODEL_ID, BREEZE25_REVISION, BREEZE25_WEIGHT_SHA256, GLYPH_HYBRID_MAX_VERIFICATION_SEGMENTS as QUALITY_GLYPH_HYBRID_MAX_VERIFICATION_SEGMENTS, CandidateGenerationContext, CandidateGenerationEvidence, CandidateVerification, CandidateObservation, ChunkCandidateArtifact, FinalOutputRejectedError, GenerationPolicy, LocalIndependentGateEvidence, NoQualifiedCandidateError, RELEASE_SPEAKER_TRIGGER_SECONDS, SEQUENCE_FALLBACK_MAX_LOCAL_BOUNDARY_SPEAKER_DROP, WholeWaveformTranscriptCache, 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_breeze25, trajectory_gate_evidence, verify_candidate, verify_trajectory, ) DEVICE = "cuda" if USING_ZERO_GPU or torch.cuda.is_available() else "cpu" REPO_ID = "OpenFormosa/BlueMagpie-TTS" MODEL_REVISION = "6f7cab914a1e27c56b504ec663c0144dc25cc0a3" ECAPA_REPO_ID = "speechbrain/spkrec-ecapa-voxceleb" ECAPA_REVISION = "0f99f2d0ebe89ac095bcc5903c4dd8f72b367286" DEFAULT_CFG = 3.0 REFERENCE_CFG = 2.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 = AUTO_LONG_TEXT_CHARS ONSET_CLAUSE_SEARCH_CHARS = 0 MIN_CHUNK_CHARS = 12 QUALITY_MEDIUM_TEXT_MAX_UNITS = 48 QUALITY_MEDIUM_CHUNK_UNITS = 24 GLYPH_CARRIER_MEDIUM_CHUNK_UNITS = 32 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.90 EXTREME_SHORT_MAX_UNITS = 2 EXTREME_SHORT_MAX_SPEED = 0.95 PACE_ONLY_FALLBACK_MIN_SPEED = MIN_PACE_SPEED PACE_ONLY_FALLBACK_MIN_UNITS = 30 MAX_TEXT_CHARS = 360 QUALITY_MAX_GENERATED_CHUNKS = 32 QUALITY_INTERACTIVE_MAX_GENERATED_CHUNKS = 10 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.85 QUALITY_PREFLIGHT_PACE_MARGIN_CPS = 0.05 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 # A single-chunk waveform that has passed both Breeze25 semantic projections # and every hard release gate may stop the cascade at this exact-only tier. # Local chunk proxies and multi-chunk eager paths remain on the stricter tier. QUALITY_FAST_EXACT_MIN_SPEAKER_SIMILARITY = 0.23 QUALITY_FAST_EXACT_MAX_BOUNDARY_SPEAKER_DROP = 0.08 QUALITY_MAX_SEQUENCE_PATHS = 3 SHORT_AUDIO_SPEAKER_GATE_SECONDS = 1.50 SPEAKER_GENERATION_SEED = 56_789 SPEAKER_B_LABEL = "內建語者 B" SPEAKER_B_PROJECTOR_SCALE = 1.125 SPEAKER_B_EMBEDDING_SHA256 = ( "e9556e14723c140985a104c1659d1ff8a5078d2fa28ce2fb756f04906641a8a7" ) REFERENCE_GENERATION_SEED = 1_337 REFERENCE_AUDIO_SECONDS = 6.0 GLYPH_ASSET_MANIFEST_RELATIVE_PATH = "glyph_assets_v1/manifest.json" # The asset builder replaces this unavailable sentinel with the exact frozen # manifest SHA-256 before a candidate can expose the V5 profile. GLYPH_ASSET_MANIFEST_SHA256 = "" GLYPH_SPEAKER_ANCHOR_SHA256 = ( "e33e4cb6a741d4d1237aa4ff557f1e663d0a6427dccda1f51e83bc149d4188ca" ) GLYPH_NETWORK_MARKERS = ("http://", "https://", "ftp://", "www.", "@", "://") GLYPH_FINAL_MAX_CER = 0.08 GLYPH_EVIDENCE_LOG_PREFIX = "[BlueMagpie][glyph-hybrid-evidence] " GLYPH_HYBRID_MAX_VERIFICATION_SEGMENTS = 160 if ( GLYPH_HYBRID_MAX_VERIFICATION_SEGMENTS != QUALITY_GLYPH_HYBRID_MAX_VERIFICATION_SEGMENTS ): raise RuntimeError("glyph-hybrid Breeze25 segment caps disagree") BREEZE25_SEMANTIC_VERIFICATION_PROFILE = ( f"{BREEZE25_MODEL_ID}@{BREEZE25_REVISION}:" f"weights{BREEZE25_WEIGHT_SHA256}:" 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}" ) BREEZE25_TRANSCRIPT_PROFILE = ( f"{BREEZE25_MODEL_ID}@{BREEZE25_REVISION}:" f"weights{BREEZE25_WEIGHT_SHA256}:greedy:zh:transcribe:" f"max_tokens{QUALITY_FINAL_ASR_MAX_NEW_TOKENS}:" f"segments{BREEZE25_MAX_VERIFICATION_SEGMENTS}" ) BREEZE25_RELEASE_VERIFICATION_PROFILE = ( f"{BREEZE25_TRANSCRIPT_PROFILE}:release-speaker-squim-echo-wsola-v2:" f"pace{QUALITY_MAX_PACE_CPS}" ) BREEZE25_SNAPSHOT_ALLOW_PATTERNS = ( "added_tokens.json", "config.json", "generation_config.json", "merges.txt", "model.safetensors", "normalizer.json", "preprocessor_config.json", "special_tokens_map.json", "tokenizer.json", "tokenizer_config.json", "vocab.json", ) print(f"[BlueMagpie] downloading model from {REPO_ID}@{MODEL_REVISION} ...") MODEL_DIR = snapshot_download(REPO_ID, revision=MODEL_REVISION) # Optional reference-speaker conditioning is loaded on first use. The former # Breeze/SQUIM verification stack is deliberately absent from the hot path. ECAPA_DIR: str | None = None BREEZE25_ASR_DIR: str | None = None 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) centroids = table["centroids"].detach().float().cpu().clone() embedding_path = os.path.join( MODEL_DIR, "checkpoints", "speaker_b_embedding.pt", ) if not os.path.exists(embedding_path): raise RuntimeError("speaker_b_embedding.pt is missing from the model release") with open(embedding_path, "rb") as handle: embedding_sha256 = hashlib.sha256(handle.read()).hexdigest() if embedding_sha256 != SPEAKER_B_EMBEDDING_SHA256: raise RuntimeError("speaker_b_embedding.pt has an unexpected SHA-256") embedding_payload = torch.load( embedding_path, map_location="cpu", weights_only=True, ) embedding = embedding_payload.get("embedding") if ( embedding_payload.get("speaker_id") != "female_voice" or not isinstance(embedding, torch.Tensor) or centroids.ndim != 2 or centroids.shape[0] < 2 or embedding.shape != centroids[1].shape or not torch.isfinite(embedding).all() or float(torch.linalg.vector_norm(embedding.float())) <= 0.0 or embedding_payload.get("generation_seed") != SPEAKER_GENERATION_SEED or embedding_payload.get("speaker_projector_scale") != SPEAKER_B_PROJECTOR_SCALE ): raise RuntimeError("speaker_b_embedding.pt has an invalid contract") centroids[1] = torch.nn.functional.normalize(embedding.float(), dim=0) labels = {f"內建語者 {chr(65 + index)}": centroid for index, centroid in enumerate(centroids)} default_index = 0 return labels, f"內建語者 {chr(65 + default_index)}" SPEAKERS, DEFAULT_SPEAKER = _load_speakers() DEFAULT_CENTROID = SPEAKERS[DEFAULT_SPEAKER] @dataclass(frozen=True) class _GlyphProfileState: bundle: GlyphAssetBundle | None asset_entry_sha256_by_asset_id: object unavailable_reason: str | None @property def available(self) -> bool: return ( self.bundle is not None and self.unavailable_reason is None and isinstance( self.asset_entry_sha256_by_asset_id, MappingProxyType, ) ) def _canonical_json_sha256(value: object) -> str: encoded = json.dumps( value, ensure_ascii=True, separators=(",", ":"), sort_keys=True, ).encode("ascii") return hashlib.sha256(encoded).hexdigest() def _glyph_asset_entry_sha256(asset) -> str: """Bind the exact schema-v7 trusted-asset fields used by the evaluator.""" return _canonical_json_sha256( { "asset_id": asset.asset_id, "raw_text": asset.raw_byte, "canonical_text": asset.canonical_token, "pcm_sha256": hashlib.sha256(asset.pcm16_le).hexdigest(), "pcm_sample_count": len(asset.pcm16_le) // 2, } ) def _load_glyph_profile_state() -> _GlyphProfileState: def unavailable(reason: str) -> _GlyphProfileState: return _GlyphProfileState( bundle=None, asset_entry_sha256_by_asset_id=MappingProxyType({}), unavailable_reason=reason, ) expected_manifest = GLYPH_ASSET_MANIFEST_SHA256 if ( type(expected_manifest) is not str or len(expected_manifest) != 64 or any(character not in "0123456789abcdef" for character in expected_manifest) ): return unavailable("manifest_sha256_not_frozen") manifest_path = ( Path(__file__).resolve().parent / GLYPH_ASSET_MANIFEST_RELATIVE_PATH ) try: bundle = load_glyph_asset_bundle( manifest_path, expected_manifest_sha256=expected_manifest, ) centroid = ( torch.as_tensor(DEFAULT_CENTROID) .detach() .cpu() .to(dtype=torch.float32) .contiguous() .numpy() ) centroid_sha256 = hashlib.sha256( np.asarray(centroid, dtype=" None: """Move deterministic CPU verifier loads and first forwards out of requests.""" print("[BlueMagpie] preloading release CPU verifiers ...") encoder = _get_ecapa_encoder() if encoder is None: raise RuntimeError("release speaker verifier preload failed") sample_count = int(round(6.0 * SR)) timeline = np.arange(sample_count, dtype=np.float64) / float(SR) warmup_waveform = np.asarray( 0.08 * np.sin(2.0 * np.pi * 180.0 * timeline) + 0.04 * np.sin(2.0 * np.pi * 360.0 * timeline) + 0.02 * np.sin(2.0 * np.pi * 540.0 * timeline), dtype=np.float32, ) anchor = _speaker_anchor_array(DEFAULT_CENTROID) warmup_evidence = ( speaker_evidence_from_audio( warmup_waveform, SR, encoder, anchor, device="cpu", ), release_speaker_evidence_from_audio( warmup_waveform, SR, encoder, anchor, device="cpu", ), ) if any( not np.isfinite(evidence.similarity) or not np.isfinite(evidence.speaker_embedding).all() for evidence in warmup_evidence ): raise RuntimeError("release speaker verifier warmup failed") squim_evidence = squim_objective_evidence_from_audio( warmup_waveform, SR, ) if not all( np.isfinite(value) for value in ( squim_evidence.stoi, squim_evidence.pesq, squim_evidence.si_sdr, ) ): raise RuntimeError("release SQUIM verifier warmup failed") median_f0_hz = active_audio_median_f0_hz( warmup_waveform, SR, ) if ( median_f0_hz is None or not np.isfinite(median_f0_hz) or median_f0_hz < 1.0 ): raise RuntimeError("release F0 transition runtime warmup failed") print("[BlueMagpie] release CPU verifiers ready") def _apply_speed( audio: np.ndarray, speed: float, *, network_conditioned: bool = False, ) -> np.ndarray: del network_conditioned try: speed = float(speed) except (TypeError, ValueError, OverflowError) as error: raise ValueError("speed must be finite and positive") from error if not np.isfinite(speed) or speed <= 0.0: raise ValueError("speed must be finite and positive") if speed < MIN_PACE_SPEED: raise ValueError( f"speed is below the production WSOLA floor {MIN_PACE_SPEED:.2f}" ) if abs(speed - 1.0) < 1.0e-3: return np.asarray(audio, dtype=np.float32) return wsola_time_stretch( audio, rate=speed, sample_rate=SR, ) def _apply_echo_smearing_gate( verification, waveforms: tuple[np.ndarray | None, ...], ): """Reject analyzable candidates with fixed echo or smearing evidence. The no-reference metric needs at least one second of waveform evidence. Shorter utterances remain governed by the existing semantic, endpoint and SQUIM contracts instead of being rejected by an inapplicable estimator. Every analyzable waveform is fail-closed: missing/non-finite/insufficient active-window evidence becomes an acoustic rejection. """ if len(waveforms) != len(verification.candidate_results): raise ValueError("audio artifact evidence does not align") updated_results = [] for index, (waveform, result) in enumerate( zip(waveforms, verification.candidate_results, strict=True) ): if waveform is None or not result.passed: updated_results.append(result) continue duration_seconds = waveform.size / float(SR) if ( duration_seconds < DEFAULT_ECHO_SMEARING_GATE_CONFIG.min_duration_seconds ): print( "[BlueMagpie] audio artifact gate " f"chunk={index} applicable=false " f"duration={duration_seconds:.6f}" ) updated_results.append(result) continue diagnostics = evaluate_echo_smearing(waveform, SR) print( "[BlueMagpie] audio artifact gate " f"chunk={index} applicable=true passed={diagnostics.passed} " f"delayed_score={diagnostics.delayed_copy_score:.6f} " f"smearing_score={diagnostics.smearing_score:.6f} " f"delay_ms={diagnostics.dominant_delay_ms:.3f} " f"peak_ratio={diagnostics.cepstral_peak_ratio:.6f} " f"windows={diagnostics.analysis_window_count} " f"reasons={diagnostics.rejection_reasons}" ) if diagnostics.passed: updated_results.append(result) continue artifact_reasons = tuple( f"audio_artifact_{reason}" for reason in diagnostics.rejection_reasons ) or ("audio_artifact_unavailable",) updated_results.append( replace( result, passed=False, score=float("inf"), rejection_reasons=tuple( dict.fromkeys( (*result.rejection_reasons, *artifact_reasons) ) ), ) ) candidate_results = tuple(updated_results) passed = bool(candidate_results) and all( result.passed for result in candidate_results ) score = ( sum(result.score for result in candidate_results) if passed else float("inf") ) rejection_reasons = tuple( f"chunk_{index}:{reason}" for index, result in enumerate(candidate_results) for reason in result.rejection_reasons ) return replace( verification, passed=passed, candidate_results=candidate_results, score=score, rejection_reasons=rejection_reasons, ) @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, ) with latency_stage("generation"): try: audio = model.generate(**kwargs) finally: if _STOP_CONTROLLER is not None: _STOP_CONTROLLER.end() # Copying to CPU closes the asynchronous CUDA work inside this stage, # so the reported wall time is not shifted into later verification. audio = audio.detach().float().cpu().numpy().reshape(-1) 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}" ) 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 WSOLA 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) # Verifiers are intentionally not preloaded in latency-first mode. 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 _enrich_transition_artifact_f0( waveform: np.ndarray, artifact: ChunkCandidateArtifact, ) -> ChunkCandidateArtifact: """Attach the unchanged pYIN statistic only when DP must compare paths.""" if not isinstance(artifact, ChunkCandidateArtifact): raise TypeError("transition artifact is invalid") if artifact.median_f0_hz is not None: return artifact try: median_f0_hz = active_audio_median_f0_hz(waveform, SR) except (RuntimeError, TypeError, ValueError): median_f0_hz = None return replace(artifact, median_f0_hz=median_f0_hz) 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_breeze25, semantic_only: bool = False, network_fragment_proofs: ( tuple[tuple[NetworkFragmentProof, ...], ...] | None ) = None, local_endpoint_roles: tuple[tuple[bool, bool], ...] | None = None, ): 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") 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 ) semantic_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, "max_pace_cps": None if semantic_only else QUALITY_MAX_PACE_CPS, } encoder = None observations: list[CandidateObservation] = [] artifacts: list[ChunkCandidateArtifact] = [] prepared_waveforms: list[np.ndarray | None] = [] proof_rows = network_fragment_proofs or ((),) * len(chunks) for index, (chunk, audio, fragment_proofs) in enumerate( zip( chunks, trajectory, proof_rows, strict=True, ) ): try: waveform = np.asarray(audio, dtype=np.float32) if ( waveform.ndim != 1 or waveform.size == 0 or not np.isfinite(waveform).all() ): raise ValueError("candidate waveform is invalid") duration = active_voiced_duration_seconds(waveform, SR) except (TypeError, ValueError, OverflowError): observations.append( CandidateObservation( target_text=chunk, transcript_text="", audio_duration_seconds=0.0, pace_cps=None, ) ) artifacts.append(ChunkCandidateArtifact()) prepared_waveforms.append(None) continue speech_units = count_speech_units(chunk) pace_cps = ( speech_units / duration * float(playback_speed) if duration > 0.0 else None ) if ( not semantic_only and ( pace_cps is None or pace_cps > QUALITY_MAX_PACE_CPS + QUALITY_PREFLIGHT_PACE_MARGIN_CPS ) ): # A candidate that cannot pass the immutable pace gate must never # be returned. Reject it before Breeze25, ECAPA, SQUIM and echo # work; later cascade candidates still face every unchanged gate. observations.append( CandidateObservation( target_text=chunk, transcript_text="", audio_duration_seconds=duration, pace_cps=pace_cps, ) ) artifacts.append(ChunkCandidateArtifact()) prepared_waveforms.append(None) continue prepared = prepare_candidate_audio( waveform, 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 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 "" ) semantic_observation = CandidateObservation( target_text=chunk, transcript_text=transcript, audio_duration_seconds=duration, pace_cps=pace_cps, ) if not semantic_only: row_gate_kwargs = ( indexed_gate_kwargs[index] if indexed_gate_kwargs is not None else {} ) semantic_preflight = verify_candidate( semantic_observation, **{ **semantic_gate_kwargs, **row_gate_kwargs, "max_pace_cps": ( QUALITY_MAX_PACE_CPS + QUALITY_PREFLIGHT_PACE_MARGIN_CPS ), }, speaker_gate_enabled=False, squim_gate_enabled=False, ) if not semantic_preflight.passed: # Preserve the exact semantic/pace evidence while avoiding # ECAPA, SQUIM, echo and F0 work for an already rejected row. observations.append(semantic_observation) artifacts.append(ChunkCandidateArtifact()) prepared_waveforms.append(None) continue prepared_waveforms.append(waveform) 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=( speech_units / 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, ) ) verification = verify_trajectory( observations, chunk_artifacts=artifacts, candidate_gate_kwargs_by_index=indexed_gate_kwargs, **semantic_gate_kwargs, 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 ), squim_gate_enabled=not semantic_only, min_squim_stoi=QUALITY_MIN_SQUIM_STOI, min_squim_pesq=QUALITY_MIN_SQUIM_PESQ, skip_acoustic_on_hard_failure=True, ) if not semantic_only: verification = _apply_echo_smearing_gate( verification, tuple(prepared_waveforms), ) return verification def _verify_independent_whole_audio( waveform: np.ndarray, target_text: str, anchor: np.ndarray, cache: WholeWaveformVerificationCache, *, transcriber=transcribe_breeze25, ): """Apply the semantic-only projection using the cached Breeze25 transcript.""" 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=transcriber, semantic_only=True, ) return cache.verify( waveform, SR, target_text, BREEZE25_SEMANTIC_VERIFICATION_PROFILE, verifier, ) def _cached_request_transcriber( cache: WholeWaveformTranscriptCache, decoder_profile: str, transcriber, ): """Share one pinned ASR decode for an exact waveform within a request.""" def cached_transcriber( waveform: np.ndarray, sample_rate: int, *, max_new_tokens: int = QUALITY_FINAL_ASR_MAX_NEW_TOKENS, ) -> str: if ( type(max_new_tokens) is not int or max_new_tokens != QUALITY_FINAL_ASR_MAX_NEW_TOKENS ): raise ValueError("cached whole-waveform ASR token cap mismatch") return cache.transcribe( waveform, sample_rate, decoder_profile, lambda exact_waveform, exact_rate: transcriber( exact_waveform, exact_rate, max_new_tokens=QUALITY_FINAL_ASR_MAX_NEW_TOKENS, ), ) return cached_transcriber def _verify_release_whole_audio( waveform: np.ndarray, target_text: str, anchor: np.ndarray, cache: WholeWaveformVerificationCache, *, transcriber, verifier_profile: str, ): """Cache the full semantic, speaker, SQUIM and echo release gate.""" def verifier(exact_waveform, sample_rate, exact_target): if sample_rate != SR: raise ValueError("release verifier sample rate mismatch") return _verify_trajectory_audio( (exact_waveform,), (exact_target,), anchor, 1.0, QUALITY_FINAL_ASR_MAX_NEW_TOKENS, release_speaker_gate=True, transcriber=transcriber, ) return cache.verify( waveform, SR, target_text, verifier_profile, verifier, ) def _transcribe_glyph_hybrid_breeze_primary( audio: np.ndarray, sample_rate: int, *, language: str = "zh", task: str = "transcribe", max_new_tokens: int = 128, max_verification_segments: int = GLYPH_HYBRID_MAX_VERIFICATION_SEGMENTS, ) -> str: """Run the fixed-zh Breeze25 profile with the V5 segment cap.""" return transcribe_breeze25( audio, sample_rate, language=language, task=task, max_new_tokens=max_new_tokens, max_verification_segments=max_verification_segments, ) def _transcribe_glyph_hybrid_breeze_confirmation( audio: np.ndarray, sample_rate: int, *, language: str | None = None, task: str = "transcribe", max_new_tokens: int = 128, max_verification_segments: int = GLYPH_HYBRID_MAX_VERIFICATION_SEGMENTS, ) -> str: """Run the auto-language Breeze25 confirmation profile for V5.""" return transcribe_breeze25( audio, sample_rate, language=language, task=task, max_new_tokens=max_new_tokens, max_verification_segments=max_verification_segments, ) _GLYPH_HYBRID_EVIDENCE_ADAPTER = HardenedGlyphHybridAdapter( GlyphAdapterRuntime( sample_rate=SR, inference_steps=DEFAULT_STEPS, model_repo_id=REPO_ID, model_revision=MODEL_REVISION, ecapa_repo_id=ECAPA_REPO_ID, ecapa_revision=ECAPA_REVISION, speaker_anchor_sha256=GLYPH_SPEAKER_ANCHOR_SHA256, speaker_centroid=DEFAULT_CENTROID, generation_lock=_GENERATION_LOCK, generate_chunk=_generate_chunk, generation_policy_resolver=( generation_policy_for_candidate_offset ), generation_cfg_resolver=lambda ordinal: ( generation_cfg_for_candidate_offset( ordinal, primary_cfg=MIXED_CFG_PRIMARY, alternate_cfg=MIXED_CFG_ALTERNATE, ) ), speaker_encoder_factory=_get_ecapa_encoder, speaker_anchor_factory=lambda: _speaker_anchor_array( DEFAULT_CENTROID ), gate_policy=GlyphEvidenceGatePolicy( sample_rate=SR, max_cer=GLYPH_FINAL_MAX_CER, max_prefix_cer=0.0, max_suffix_cer=0.0, max_prefix_deletions=0, max_suffix_deletions=0, max_extra_tail_units=0, max_pace_cps=QUALITY_MAX_PACE_CPS, min_speaker_similarity=( QUALITY_RELEASE_MIN_SPEAKER_SIMILARITY ), max_boundary_speaker_drop=( QUALITY_RELEASE_MAX_BOUNDARY_SPEAKER_DROP ), min_squim_stoi=QUALITY_PREFERRED_MIN_SQUIM_STOI, min_squim_pesq=QUALITY_PREFERRED_MIN_SQUIM_PESQ, min_squim_si_sdr=-20.0, max_endpoint_tail_energy_ratio=0.02, ), asr_max_new_tokens=QUALITY_FINAL_ASR_MAX_NEW_TOKENS, asr_max_verification_segments=( GLYPH_HYBRID_MAX_VERIFICATION_SEGMENTS ), breeze_primary_transcriber=( _transcribe_glyph_hybrid_breeze_primary ), breeze_confirmation_transcriber=( _transcribe_glyph_hybrid_breeze_confirmation ), ) ) 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 ) @timed_latency_stage("assemble") 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( primary_verification, trajectory: tuple[np.ndarray, ...], chunks: tuple[str, ...], anchor: np.ndarray, proof_rows: tuple[tuple[NetworkFragmentProof, ...], ...] | None, *, candidate_seed: int, transcriber=transcribe_breeze25, ): """Project cached Breeze25 semantics before retaining network locals. Only proof-bearing rows receive the duplicate semantic projection. Their exact ``NetworkFragmentProof`` tuples are reused unchanged, so neither semantic projection can borrow a whole URL/email proof for another local range. The intersection helper keeps primary speaker/SQUIM/pace/artifact evidence while projecting every semantic rejection into a hard local failure. The request-local transcriber cache prevents a second decode of exact PCM. """ if proof_rows is None: return primary_verification, tuple( LocalIndependentGateEvidence(False, None, 0, None) for _ in chunks ) if ( len(proof_rows) != len(chunks) or len(trajectory) != len(chunks) or len(primary_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( primary_verification.candidate_results[index], max_local_boundary_speaker_drop=( SEQUENCE_FALLBACK_MAX_LOCAL_BOUNDARY_SPEAKER_DROP ), ) ) ) if not selected_indices: return primary_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=transcriber, 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 Breeze25 projection " 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( primary_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, release_transcriber=transcribe_breeze25, independent_transcriber=transcribe_breeze25, ): """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, QUALITY_FINAL_ASR_MAX_NEW_TOKENS, transcriber=release_transcriber, network_fragment_proofs=proof_rows, local_endpoint_roles=_local_endpoint_role_rows(chunks, chunk_specs), ) local_verification, independent_local_results = ( _verify_network_local_asr_intersection( local_verification, trajectory, chunks, anchor, proof_rows, candidate_seed=candidate_seed, transcriber=independent_transcriber, ) ) 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_release_whole_audio( waveform, whole_target_text, anchor, independent_cache, transcriber=release_transcriber, verifier_profile=BREEZE25_RELEASE_VERIFICATION_PROFILE, ) 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, transcriber=independent_transcriber, ) semantic_qualified = qualify_trajectory_with_joined_output( qualified, independent_verification, ) if not semantic_qualified.passed: print( "[BlueMagpie] candidate Breeze25 semantic projection rejected " f"seed={candidate_seed} " f"{_verification_metric_log_fields(independent_verification)}" ) return CandidateVerification( semantic_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, transcriber=transcribe_breeze25, ): """Apply strict cached-Breeze25 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, QUALITY_FINAL_ASR_MAX_NEW_TOKENS, transcriber=transcriber, network_fragment_proofs=proof_rows, local_endpoint_roles=_local_endpoint_role_rows(chunks, chunk_specs), ) local_verification, independent_local_results = ( _verify_network_local_asr_intersection( local_verification, trajectory, chunks, anchor, proof_rows, candidate_seed=candidate_seed, transcriber=transcriber, ) ) 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, *, release_transcriber=transcribe_breeze25, independent_transcriber=transcribe_breeze25, ): """Verify one ranked DP path after exact production assembly.""" waveform = _assemble_trajectory_audio( sequence_result.trajectory, chunks, playback_speed, chunk_specs, ) # Local and assembled checks share one request-local Breeze25 transcript # cache. The semantic-only projection remains a separate hard gate over # the same exact transcript, never a second decoder invocation. assembled_verification = _verify_release_whole_audio( waveform, whole_target_text, anchor, independent_cache, transcriber=release_transcriber, verifier_profile=BREEZE25_RELEASE_VERIFICATION_PROFILE, ) status = "verified" if assembled_verification.passed else "rejected" print( f"[BlueMagpie] sequence path Breeze25 {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, transcriber=independent_transcriber, ) intersected = qualify_trajectory_with_joined_output( assembled_verification, independent_verification, ) independent_status = "verified" if independent_verification.passed else "rejected" print( f"[BlueMagpie] sequence path Breeze25 projection {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 @dataclass(frozen=True) class _GlyphAttemptReservation: carrier_segment_id: str candidate_index: int candidate_ordinal: int network_conditioned: bool attempt_id: str seed: int text: str @dataclass(frozen=True) class _GlyphCompletedAttempt: reservation: _GlyphAttemptReservation endpoint_outcome: _ChunkGenerationOutcome record: GenerationAttemptRecord class _GlyphRequestAttemptLedger: """Transient exact accounting for every model invocation in one request.""" def __init__(self) -> None: self._records: list[GenerationAttemptRecord] = [] self._record_by_carrier_candidate: dict[ tuple[str, int], GenerationAttemptRecord, ] = {} self._selected_by_carrier: dict[str, GenerationAttemptRecord] = {} self._completed: list[_GlyphCompletedAttempt] = [] self._selected_waveform_by_carrier: dict[str, np.ndarray] = {} @property def records(self) -> tuple[GenerationAttemptRecord, ...]: return tuple(self._records) @property def completed_attempts(self) -> tuple[_GlyphCompletedAttempt, ...]: return tuple(self._completed) def begin( self, carrier_segment_id: str, candidate_chunks: tuple[str, ...], seed: int, generation_context: CandidateGenerationContext, candidate_ordinal: int, network_conditioned: bool, ) -> _GlyphAttemptReservation: if ( type(carrier_segment_id) is not str or not carrier_segment_id or type(candidate_chunks) is not tuple or len(candidate_chunks) != 1 or len(generation_context.chunk_indices) != 1 or generation_context.chunk_candidate_ordinals != (candidate_ordinal,) or type(network_conditioned) is not bool ): raise ValueError( "one hybrid carrier candidate must be one exact model invocation" ) text = candidate_chunks[0] units = count_speech_units(text) generated_chunks = sum( record.generated_chunk_count for record in self._records ) generated_units = sum( record.generated_text_units for record in self._records ) if ( units <= 0 or generated_chunks + 1 > GLYPH_MAX_GENERATED_CHUNK_BUDGET or generated_units + units > GLYPH_MAX_GENERATED_TEXT_UNIT_BUDGET ): raise ValueError("hybrid request exceeds its frozen generation budget") key = (carrier_segment_id, generation_context.candidate_index) if key in self._record_by_carrier_candidate: raise ValueError("hybrid carrier candidate invocation is duplicated") return _GlyphAttemptReservation( carrier_segment_id=carrier_segment_id, candidate_index=generation_context.candidate_index, candidate_ordinal=candidate_ordinal, network_conditioned=network_conditioned, attempt_id=( f"glyph-carrier:{carrier_segment_id}:attempt:{len(self._records)}" ), seed=int(seed), text=text, ) def complete( self, reservation: _GlyphAttemptReservation, endpoint_outcomes: tuple[_ChunkGenerationOutcome, ...], ) -> None: if ( type(reservation) is not _GlyphAttemptReservation or type(endpoint_outcomes) is not tuple or len(endpoint_outcomes) != 1 ): raise ValueError("hybrid generation endpoint evidence is incomplete") outcome = endpoint_outcomes[0] endpoint_sha256 = _canonical_json_sha256( { "schema": "bluemagpie.glyph-carrier.endpoint.v1", "stop_reason": outcome.stop_reason, "tail_energy_ratio": float(outcome.endpoint_energy_ratio), "generated_steps": int(outcome.generated_steps), "hard_cap_steps": int(outcome.hard_stop_steps), } ) record = GenerationAttemptRecord( ordinal=len(self._records), attempt_id=reservation.attempt_id, seed=reservation.seed, text_sha256=hashlib.sha256( reservation.text.encode("utf-8") ).hexdigest(), endpoint_evidence_sha256=endpoint_sha256, generated_chunk_count=1, generated_text_units=count_speech_units(reservation.text), ) key = ( reservation.carrier_segment_id, reservation.candidate_index, ) if key in self._record_by_carrier_candidate: raise ValueError("hybrid generation completion is duplicated") self._records.append(record) self._record_by_carrier_candidate[key] = record self._completed.append( _GlyphCompletedAttempt( reservation=reservation, endpoint_outcome=outcome, record=record, ) ) def select( self, carrier_segment_id: str, cascade, waveform: np.ndarray, ) -> None: if ( carrier_segment_id in self._selected_by_carrier or len(cascade.chunk_candidate_indices) != 1 ): raise ValueError("hybrid carrier selection is inconsistent") key = (carrier_segment_id, cascade.chunk_candidate_indices[0]) record = self._record_by_carrier_candidate.get(key) if record is None: raise ValueError("hybrid selected carrier lacks generation provenance") self._selected_by_carrier[carrier_segment_id] = record selected_waveform = np.asarray(waveform, dtype=np.float32).reshape(-1).copy() if selected_waveform.size <= 0 or not np.isfinite(selected_waveform).all(): raise ValueError("hybrid selected carrier waveform is invalid") self._selected_waveform_by_carrier[carrier_segment_id] = selected_waveform def selected(self, carrier_segment_id: str) -> GenerationAttemptRecord: record = self._selected_by_carrier.get(carrier_segment_id) if record is None: raise ValueError("hybrid selected carrier provenance is unavailable") return record def selected_waveform(self, carrier_segment_id: str) -> np.ndarray: waveform = self._selected_waveform_by_carrier.get(carrier_segment_id) if waveform is None: raise ValueError("hybrid selected carrier waveform is unavailable") return waveform.copy() class _GlyphCarrierGenerationAudit: def __init__( self, request_ledger: _GlyphRequestAttemptLedger, carrier_segment_id: str, ) -> None: self._request_ledger = request_ledger self._carrier_segment_id = carrier_segment_id def begin( self, candidate_chunks: tuple[str, ...], seed: int, generation_context: CandidateGenerationContext, candidate_ordinal: int, network_conditioned: bool, ) -> _GlyphAttemptReservation: return self._request_ledger.begin( self._carrier_segment_id, candidate_chunks, seed, generation_context, candidate_ordinal, network_conditioned, ) def complete( self, reservation: _GlyphAttemptReservation, endpoint_outcomes: tuple[_ChunkGenerationOutcome, ...], ) -> None: self._request_ledger.complete(reservation, endpoint_outcomes) def select(self, cascade, waveform: np.ndarray) -> None: self._request_ledger.select( self._carrier_segment_id, cascade, waveform, ) def selected(self) -> GenerationAttemptRecord: return self._request_ledger.selected(self._carrier_segment_id) def selected_waveform(self) -> np.ndarray: return self._request_ledger.selected_waveform(self._carrier_segment_id) def _quality_request_generated_chunk_limit( normalized_units: int, *, network_request: bool, carrier_request: bool, ) -> int: """Bound interactive retries without changing network/carrier contracts.""" if ( not network_request and not carrier_request and normalized_units <= QUALITY_MEDIUM_TEXT_MAX_UNITS ): return QUALITY_INTERACTIVE_MAX_GENERATED_CHUNKS return QUALITY_MAX_GENERATED_CHUNKS def _fast_text_chunks(text: str) -> tuple[str, ...]: """Split text cheaply without invoking any release-gate planner.""" text = str(text).strip() if not text: return () chunks: list[str] = [] break_characters = frozenset("。!?!?;;,,、::\n") while len(text) > CHUNK_CHARS: cut = CHUNK_CHARS for index in range(CHUNK_CHARS - 1, CHUNK_CHARS // 2 - 1, -1): if text[index] in break_characters: cut = index + 1 break chunk = text[:cut].strip() if chunk: chunks.append(chunk) text = text[cut:].strip() if text: chunks.append(text) return tuple(chunks) def _prepare_reference_audio(reference_wav: str) -> tuple[str, str | None]: """Deterministically center-crop long references to the trained 6s cap.""" info = sf.info(reference_wav) max_frames = int(round(REFERENCE_AUDIO_SECONDS * int(info.samplerate))) if int(info.frames) <= max_frames: return reference_wav, None start = max(0, (int(info.frames) - max_frames) // 2) waveform, sample_rate = sf.read( reference_wav, start=start, frames=max_frames, dtype="float32", always_2d=True, ) with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as handle: temporary_path = handle.name try: sf.write( temporary_path, waveform, sample_rate, subtype="PCM_16", ) except Exception: os.unlink(temporary_path) raise return temporary_path, temporary_path def _generate_fast_chunk( text: str, centroid: torch.Tensor | None, *, cfg: float, steps: int, request_seed: int, reference_wav_path: str = "", short_text_min_cfg: float = SHORT_TEXT_CFG_MIN, ) -> np.ndarray: """Run one model call and return it without candidate qualification.""" generation_cps = select_generation_cps( text, cjk_cps=BASE_GENERATION_POLICY.cjk_cps, ascii_cps=BASE_GENERATION_POLICY.ascii_cps, ) model_text, expected_steps, hard_stop_steps = endpoint_generation_plan( text, generation_cps=generation_cps, step_seconds=STEP_SECONDS, margin_steps=BASE_GENERATION_POLICY.hard_stop_margin_steps, add_terminal_punctuation=count_speech_units(text) >= MIN_ENDPOINT_CUE_UNITS, ) generation_cfg = effective_generation_cfg( text, float(cfg), short_text_unit_threshold=SHORT_TEXT_CFG_UNITS, short_text_min_cfg=float(short_text_min_cfg), ) set_generation_seed(request_seed) kwargs = { "target_text": model_text, "speaker_centroid": centroid, "cfg_value": generation_cfg, "inference_timesteps": int(steps), "min_len": 2, "max_len": hard_stop_steps, "retry_badcase": False, } if reference_wav_path: kwargs["reference_wav_path"] = reference_wav_path 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( 2, expected_steps=expected_steps, hard_stop_steps=hard_stop_steps, ) try: with latency_stage("generation"): audio = model.generate(**kwargs) return audio.detach().float().cpu().numpy().reshape(-1) finally: if _STOP_CONTROLLER is not None: _STOP_CONTROLLER.end() def _assemble_fast_audio(trajectory: tuple[np.ndarray, ...]) -> np.ndarray: """Concatenate first-pass chunks with a short seam and no analysis.""" if not trajectory: return np.zeros(1, dtype=np.float32) if len(trajectory) == 1: waveform = np.asarray(trajectory[0], dtype=np.float32).reshape(-1) else: seam = np.zeros(int(round(0.12 * SR)), dtype=np.float32) parts: list[np.ndarray] = [] for index, audio in enumerate(trajectory): if index: parts.append(seam) parts.append(np.asarray(audio, dtype=np.float32).reshape(-1)) waveform = np.concatenate(parts) return finish_audio(waveform, SR, fade_ms=FINAL_ENDPOINT_FADE_MS) def _synthesize( text: str, centroid: torch.Tensor | None, *, cfg: float, steps: int, speed: float, generation_audit: _GlyphCarrierGenerationAudit | None = None, request_seed: int | None = None, speaker_projector_scale: float = 1.0, reference_wav_path: str = "", short_text_min_cfg: float = SHORT_TEXT_CFG_MIN, ) -> tuple[int, np.ndarray]: """Latency-first single-pass synthesis with no output qualification.""" del speed, generation_audit raw_text = str(text) try: normalized = normalize_spoken_forms(raw_text, locale="zh-TW") except (TypeError, ValueError): normalized = raw_text normalized = strip_unspoken_cjk_quotes_for_generation(normalized) chunks = _fast_text_chunks(normalized) if not chunks: return SR, np.zeros(1, dtype=np.float32) seed = ( secrets.randbelow(2_147_483_648) if request_seed is None else int(request_seed) ) with _GENERATION_LOCK: projector_hook = None if speaker_projector_scale != 1.0: projector_hook = model.speaker_projector.register_forward_hook( lambda _module, _inputs, output: output * speaker_projector_scale ) try: trajectory = tuple( _generate_fast_chunk( chunk, centroid, cfg=float(cfg), steps=int(steps), request_seed=seed, reference_wav_path=reference_wav_path, short_text_min_cfg=short_text_min_cfg, ) for chunk in chunks ) finally: if projector_hook is not None: projector_hook.remove() waveform = _assemble_fast_audio(trajectory) print( "[BlueMagpie] latency-first synthesis " f"chunks={len(chunks)} model_calls={len(chunks)} verification=disabled" ) return SR, waveform def _synthesize_verified( text: str, centroid: torch.Tensor, *, cfg: float, steps: int, speed: float, generation_audit: _GlyphCarrierGenerationAudit | None = None, request_seed: int | None = None, ) -> tuple[int, np.ndarray]: def emit_public_cascade_evidence(line: str) -> None: # Carrier-local cascades are implementation details of one hybrid # request. Publishing schema 6 here would make an external evaluator # mistake a carrier for the returned request. Ordinary requests keep # their byte-for-byte existing schema-6 emission. if generation_audit is None: print(line) 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 float(speed) != 1.0: raise gr.Error("後處理語速固定為 1.0;減速已由品質管線控制。") 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: medium_chunk_units = ( GLYPH_CARRIER_MEDIUM_CHUNK_UNITS if generation_audit is not None else QUALITY_MEDIUM_CHUNK_UNITS ) 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=medium_chunk_units, ), max_chunks=QUALITY_MAX_GENERATED_CHUNKS, max_units=CHUNK_CHARS, ) ) except ValueError as error: raise gr.Error("文字無法在已驗證的生成限制內安全分段。") from error request_generated_chunk_limit = _quality_request_generated_chunk_limit( normalized_units, network_request=network_request, carrier_request=generation_audit is not None, ) request_seed = resolve_request_seed(request_seed, secrets.randbelow) anchor = _speaker_anchor_array(centroid) independent_cache = WholeWaveformVerificationCache() transcript_cache = WholeWaveformTranscriptCache() cached_breeze25_transcriber = _cached_request_transcriber( transcript_cache, BREEZE25_TRANSCRIPT_PROFILE, transcribe_breeze25, ) 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, ) audit_reservation = ( generation_audit.begin( candidate_chunks, seed, generation_context, candidate_ordinal, network_flags[0], ) if generation_audit is not None else None ) 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") if generation_audit is not None: generation_audit.complete( audit_reservation, endpoint_outcomes, ) 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), release_transcriber=cached_breeze25_transcriber, independent_transcriber=cached_breeze25_transcriber, ), 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), release_transcriber=cached_breeze25_transcriber, independent_transcriber=cached_breeze25_transcriber, ) if len(chunks) == 1 else _verify_refill_candidate_trajectory_audio( trajectory, candidate_chunks, anchor, speed, generation_chunk_specs(seed, candidate_chunks), candidate_seed=seed, transcriber=cached_breeze25_transcriber, ) ), sequence_final_verifier=lambda sequence_result, candidate_chunks: ( _verify_sequence_trajectory_audio( sequence_result, candidate_chunks, text, anchor, speed, independent_cache, chunk_specs, release_transcriber=cached_breeze25_transcriber, independent_transcriber=cached_breeze25_transcriber, ) ), generation_evidence_factory=candidate_generation_evidence, candidate_generation_text_transform=( candidate_generation_text_transform ), transition_artifact_enricher=( _enrich_transition_artifact_f0 ), max_generated_chunks=request_generated_chunk_limit, 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 ), single_exact_min_speaker_similarity=( QUALITY_FAST_EXACT_MIN_SPEAKER_SIMILARITY ), single_exact_max_boundary_speaker_drop=( QUALITY_FAST_EXACT_MAX_BOUNDARY_SPEAKER_DROP ), require_endpoint_evidence=True, ) except NoQualifiedCandidateError as error: emit_public_cascade_evidence( format_cascade_evidence_log( error.diagnostics, outcome="no_qualified_candidate", generated_chunk_limit=request_generated_chunk_limit, 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_release_whole_audio( waveform, text, anchor, independent_cache, transcriber=cached_breeze25_transcriber, verifier_profile=BREEZE25_RELEASE_VERIFICATION_PROFILE, ) 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, transcriber=cached_breeze25_transcriber, ) 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) ) emit_public_cascade_evidence( format_cascade_evidence_log( cascade.diagnostics, outcome="final_output_rejected", generated_chunk_limit=request_generated_chunk_limit, 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"breeze25={_verification_metric_log_fields(final_verification)} " f"semantic_projection={independent_fields}" ) raise gr.Error("最終合成結果未通過整段內容、語速與音色驗證,未回傳音訊。") from error except (RuntimeError, ValueError) as error: raise gr.Error("品質驗證暫時無法完成,未回傳未驗證的語音。") from error emit_public_cascade_evidence( format_cascade_evidence_log( cascade.diagnostics, outcome="returned", generated_chunk_limit=request_generated_chunk_limit, 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"breeze25_score={final_verification.score:.6f} " f"semantic_projection_score={independent_final_verification.score:.6f} " f"independent_cache_entries={independent_cache.entry_count} " f"transcript_cache_entries={transcript_cache.entry_count}" ) if generation_audit is not None: generation_audit.select(cascade, waveform) return SR, waveform def _glyph_request_network_looking(text: object) -> bool: folded = str(text).lower() return any(marker in folded for marker in GLYPH_NETWORK_MARKERS) def _glyph_profile_selected(speaker: object) -> bool: return ( type(speaker) is str and speaker == DEFAULT_SPEAKER and DEFAULT_SPEAKER == "內建語者 B" and SPEAKERS.get(speaker) is DEFAULT_CENTROID ) def _validate_glyph_runtime_controls( *, cfg: object, steps: object, speed: object, ) -> None: if isinstance(steps, (bool, np.bool_)): raise gr.Error(f"目前只支援已驗證的 NFE {DEFAULT_STEPS}。") try: exact_steps = float(steps) == float(DEFAULT_STEPS) cfg_value = float(cfg) speed_value = float(speed) except (TypeError, ValueError, OverflowError): raise gr.Error("V5 字元語音只支援固定的已驗證參數。") from None if ( not exact_steps or not np.isfinite(cfg_value) or cfg_value != MIXED_CFG_PRIMARY or not np.isfinite(speed_value) or speed_value != 1.0 ): raise gr.Error("V5 字元語音只支援 CFG 3.0、NFE 10 與原始語速 1.0。") def _validate_glyph_carrier_text(carrier_text: str) -> None: try: normalized = normalize_spoken_forms(carrier_text, locale="zh-TW") except ValueError as error: raise GlyphHybridPlanError("hybrid carrier normalization failed") from error generated_text = strip_unspoken_cjk_quotes_for_generation(normalized) if normalized != carrier_text or generated_text != carrier_text: raise GlyphHybridPlanError( "hybrid carrier text is not stable under the legacy frontend" ) chunks = tuple( coalesce_text_chunks( split_quality_text_for_tts( generated_text, long_max_units=CHUNK_CHARS, min_chunk_units=MIN_CHUNK_CHARS, medium_max_units=QUALITY_MEDIUM_TEXT_MAX_UNITS, medium_chunk_units=GLYPH_CARRIER_MEDIUM_CHUNK_UNITS, ), max_chunks=QUALITY_MAX_GENERATED_CHUNKS, max_units=CHUNK_CHARS, ) ) if chunks != (carrier_text,): raise GlyphHybridPlanError( "one renderer carrier must remain one exact generation chunk" ) def _require_glyph_final_verification(verification) -> None: require_verified_final_output(verification) if len(verification.candidate_results) != 1: raise FinalOutputRejectedError( "glyph final verifier did not return one whole-waveform result" ) result = verification.candidate_results[0] comparison = result.comparison if ( not np.isfinite(float(comparison.cer)) or float(comparison.cer) > GLYPH_FINAL_MAX_CER or comparison.prefix_deletions != 0 or comparison.suffix_deletions != 0 or comparison.extra_tail_units != 0 ): raise FinalOutputRejectedError( "glyph final semantic gate exceeded the frozen release limits" ) def _glyph_inverse_proof_sha256( plan: HybridRenderPlan, reconstructed_raw: str, ) -> str: return _canonical_json_sha256( { "schema": "bluemagpie.glyph-hybrid.inverse.v1", "plan_sha256": plan.plan_sha256, "raw_input_sha256": hashlib.sha256( reconstructed_raw.encode("utf-8") ).hexdigest(), "canonical_target_sha256": plan.canonical_target_sha256, } ) def _emit_glyph_hybrid_evidence( *, raw_text: str, plan: HybridRenderPlan, profile_state: _GlyphProfileState, segments: tuple[AssetSegment | CarrierSegment, ...], assembly, semantic_commitment: SemanticPlanCommitment, request_ledger: _GlyphRequestAttemptLedger, final_verification, independent_final_verification, ) -> None: adapter = _GLYPH_HYBRID_EVIDENCE_ADAPTER if not callable(adapter): raise RuntimeError("schema-7 hybrid evidence adapter is unavailable") serialized = adapter( raw_text=raw_text, plan=plan, profile_state=profile_state, segments=segments, assembly=assembly, semantic_commitment=semantic_commitment, request_ledger=request_ledger, final_verification=final_verification, independent_final_verification=independent_final_verification, ) if type(serialized) is not bytes or not serialized or b"\n" in serialized: raise RuntimeError("schema-7 hybrid evidence is not one canonical record") try: payload = json.loads(serialized.decode("ascii")) except (UnicodeDecodeError, json.JSONDecodeError) as error: raise RuntimeError("schema-7 hybrid evidence is invalid") from error try: canonical_serialized = json.dumps( payload, ensure_ascii=True, allow_nan=False, separators=(",", ":"), sort_keys=True, ).encode("ascii") except (TypeError, ValueError) as error: raise RuntimeError("schema-7 hybrid evidence is not finite JSON") from error if canonical_serialized != serialized: raise RuntimeError("schema-7 hybrid evidence is not canonical JSON") forbidden_content_keys = { "raw_input", "raw_text", "canonical_target", "canonical_text", "target_text", "transcript_text", "text", } def contains_content(value: object) -> bool: if isinstance(value, dict): return bool(forbidden_content_keys.intersection(value)) or any( contains_content(item) for item in value.values() ) if isinstance(value, list): return any(contains_content(item) for item in value) return False if not isinstance(payload, dict): raise RuntimeError("schema-7 hybrid evidence root is invalid") request_proof = payload.get("request_proof") strict_gates = ( request_proof.get("additional_strict_gates") if isinstance(request_proof, dict) else None ) final_gate = ( request_proof.get("final_gate") if isinstance(request_proof, dict) else None ) independent_gate = ( request_proof.get("independent_gate") if isinstance(request_proof, dict) else None ) exact_pcm_sha256 = hashlib.sha256(assembly.pcm16_le).hexdigest() budget_rows = ( request_proof.get("budgets") if isinstance(request_proof, dict) else None ) expected_budget_names = ( "model_invocations", "generated_chunks", "generated_speech_units", "asset_count", "asset_atoms", "asset_canonical_units", "asset_audio_samples", "asr_calls", "ecapa_calls", "squim_calls", ) valid_budget_rows = ( isinstance(budget_rows, list) and len(budget_rows) == len(expected_budget_names) and all( isinstance(row, dict) and set(row) == {"name", "actual", "cap"} and type(row["name"]) is str and type(row["actual"]) is int and type(row["cap"]) is int for row in budget_rows ) and tuple(row["name"] for row in budget_rows) == expected_budget_names ) budgets = ( { row["name"]: (row["actual"], row["cap"]) for row in budget_rows } if valid_budget_rows else {} ) records = request_ledger.records renderer_contract = payload.get("renderer_contract") renderer_contract_sha256 = payload.get("renderer_contract_sha256") digest_payload = dict(payload) evidence_sha256 = digest_payload.pop("evidence_sha256", None) recomputed_evidence_sha256 = _canonical_json_sha256(digest_payload) expected_strict_gate_names = { "strict_inverse_reconstruction", "asset_reconstruction", "public_pcm_equality", } def has_exact_internal_bindings() -> bool: if not isinstance(request_proof, dict): return False segment_rows = request_proof.get("segments") call_rows = payload.get("runtime_call_ledger") selection_rows = payload.get("carrier_selections") if not all( isinstance(rows, list) for rows in (segment_rows, call_rows, selection_rows) ): return False if not all( isinstance(row, dict) for rows in (segment_rows, call_rows, selection_rows) for row in rows ): return False segment_by_id = { row.get("segment_id"): row for row in segment_rows if type(row.get("segment_id")) is str } call_by_id = { row.get("call_id"): row for row in call_rows if type(row.get("call_id")) is str } if ( len(segment_by_id) != len(segment_rows) or len(call_by_id) != len(call_rows) ): return False canonical_commitment = request_proof.get( "canonical_target_sha256" ) for gate in (final_gate, independent_gate): if not isinstance(gate, dict): return False asr_call = call_by_id.get(gate.get("asr_call_id")) if ( not isinstance(asr_call, dict) or asr_call.get("kind") != "asr" or "subject_model_call_id" not in asr_call or asr_call.get("subject_model_call_id") is not None or asr_call.get("scope") != gate.get("scope") or asr_call.get("target_sha256") != canonical_commitment ): return False asset_segments = { segment_id: row for segment_id, row in segment_by_id.items() if row.get("kind") == "asset" } asset_loads: dict[str, list[dict]] = {} for row in call_rows: if row.get("kind") == "asset_load": asset_loads.setdefault(row.get("segment_id"), []).append(row) if set(asset_loads) != set(asset_segments): return False for segment_id, segment in asset_segments.items(): loads = asset_loads[segment_id] if len(loads) != 1: return False loaded = loads[0] if ( loaded.get("asset_id") != segment.get("asset_id") or loaded.get("manifest_entry_sha256") != segment.get("asset_manifest_entry_sha256") or loaded.get("loaded_pcm_sha256") != segment.get("realized_pcm_sha256") or loaded.get("loaded_sample_count") != segment.get("output_sample_count") ): return False carrier_segments = { segment_id: row for segment_id, row in segment_by_id.items() if row.get("kind") == "carrier" } selections_by_segment: dict[str, list[dict]] = {} for row in selection_rows: selections_by_segment.setdefault( row.get("segment_id"), [], ).append(row) if set(selections_by_segment) != set(carrier_segments): return False for segment_id, segment in carrier_segments.items(): selections = selections_by_segment[segment_id] if len(selections) != 1: return False selected = call_by_id.get(selections[0].get("selected_call_id")) if ( not isinstance(selected, dict) or selected.get("kind") != "model_generation" or selected.get("segment_id") != segment_id or selected.get("text_sha256") != segment.get("carrier_text_sha256") or selected.get("output_pcm_sha256") != segment.get("realized_pcm_sha256") or selected.get("output_sample_count") != segment.get("output_sample_count") ): return False return True if ( payload.get("schema") != "bluemagpie.glyph-hybrid.evidence.v7" or payload.get("version") != 7 or contains_content(payload) or evidence_sha256 != recomputed_evidence_sha256 or not isinstance(renderer_contract, dict) or renderer_contract_sha256 != _canonical_json_sha256(renderer_contract) or renderer_contract.get("speaker_anchor_sha256") != GLYPH_SPEAKER_ANCHOR_SHA256 or not isinstance(request_proof, dict) or request_proof.get("content_commitment_scheme") != "hmac-sha256-v1" or not has_exact_internal_bindings() or request_proof.get("assembled_pcm_sha256") != exact_pcm_sha256 or request_proof.get("public_pcm_sha256") != exact_pcm_sha256 or budgets.get("model_invocations") != (len(records), GLYPH_MAX_GENERATED_CHUNK_BUDGET) or budgets.get("generated_chunks") != ( sum(record.generated_chunk_count for record in records), GLYPH_MAX_GENERATED_CHUNK_BUDGET, ) or budgets.get("generated_speech_units") != ( sum(record.generated_text_units for record in records), GLYPH_MAX_GENERATED_TEXT_UNIT_BUDGET, ) or not valid_budget_rows or not isinstance(strict_gates, dict) or set(strict_gates) != expected_strict_gate_names or not all(value is True for value in strict_gates.values()) or not isinstance(final_gate, dict) or final_gate.get("passed") is not True or not isinstance(independent_gate, dict) or independent_gate.get("passed") is not True ): raise RuntimeError("schema-7 hybrid evidence is not bound to this output") print(GLYPH_EVIDENCE_LOG_PREFIX + serialized.decode("ascii")) def _synthesize_glyph_hybrid( raw_text: str, plan: HybridRenderPlan, profile_state: _GlyphProfileState, *, cfg: float, steps: int, speed: float, ) -> tuple[int, np.ndarray]: if ( not profile_state.available or profile_state.bundle is None or glyph_hybrid_plan_sha256(plan) != plan.plan_sha256 ): raise GlyphHybridPlanError("hybrid profile or renderer plan is unavailable") manifest_entries = profile_state.asset_entry_sha256_by_asset_id reconstructed_raw = inverse_glyph_hybrid_plan( plan, asset_entry_sha256_by_asset_id=manifest_entries, expected_raw_request=raw_text, ) if reconstructed_raw != raw_text: raise GlyphHybridPlanError("hybrid inverse reconstruction failed") inverse_proof_sha256 = _glyph_inverse_proof_sha256( plan, reconstructed_raw, ) request_seed = resolve_request_seed(None, secrets.randbelow) request_ledger = _GlyphRequestAttemptLedger() assembled_segments: list[AssetSegment | CarrierSegment] = [] carrier_count = 0 for segment in plan.segments: segment_id = f"glyph-plan-segment-{segment.ordinal}" if segment.kind == "asset_atom": if segment.asset_id is None: raise GlyphHybridPlanError("asset segment has no immutable asset id") assembled_segments.append( AssetSegment( segment_id=segment_id, asset_id=segment.asset_id, speed=1.0, ) ) elif segment.kind == "carrier": _validate_glyph_carrier_text(segment.canonical_text) generation_audit = _GlyphCarrierGenerationAudit( request_ledger, segment_id, ) carrier_rate, carrier_waveform = _synthesize( segment.canonical_text, DEFAULT_CENTROID, cfg=cfg, steps=steps, speed=1.0, generation_audit=generation_audit, request_seed=request_seed, ) if ( carrier_rate != SR or not np.array_equal( carrier_waveform, generation_audit.selected_waveform(), ) ): raise RuntimeError("hybrid carrier finalization is inconsistent") selected = generation_audit.selected() assembled_segments.append( CarrierSegment( segment_id=segment_id, audio=carrier_waveform, selected_attempt_id=selected.attempt_id, selected_seed=selected.seed, selected_text_sha256=selected.text_sha256, selected_endpoint_evidence_sha256=( selected.endpoint_evidence_sha256 ), sample_rate=SR, speed=1.0, ) ) carrier_count += 1 elif segment.kind not in {"control", "separator"}: raise GlyphHybridPlanError("renderer plan has an unknown segment kind") immutable_segments = tuple(assembled_segments) if not immutable_segments: raise GlyphHybridPlanError("hybrid renderer produced no audio segments") records = request_ledger.records if (carrier_count == 0) != (len(records) == 0): raise RuntimeError("hybrid request model accounting is inconsistent") commitment = SemanticPlanCommitment( external_semantic_plan_sha256=plan.plan_sha256, inverse_proof_sha256=inverse_proof_sha256, ordered_attempt_ledger_sha256=( compute_ordered_attempt_ledger_sha256(records) ), attempt_record_count=len(records), assembled_carrier_count=carrier_count, request_model_generator_invocations=len(records), request_generated_chunk_count=sum( record.generated_chunk_count for record in records ), request_generated_text_units=sum( record.generated_text_units for record in records ), request_generated_chunk_budget=GLYPH_MAX_GENERATED_CHUNK_BUDGET, request_generated_text_unit_budget=( GLYPH_MAX_GENERATED_TEXT_UNIT_BUDGET ), ordered_attempt_records=records, ) assembly_plan_sha256 = compute_hybrid_plan_sha256( profile_state.bundle, immutable_segments, semantic_commitment=commitment, ) with latency_stage("assemble"): assembly = assemble_hybrid_pcm16( profile_state.bundle, immutable_segments, semantic_commitment=commitment, expected_plan_sha256=assembly_plan_sha256, speed=1.0, post_gain=1.0, ) verify_hybrid_assembly( assembly, trusted_bundle=profile_state.bundle, original_segments=immutable_segments, expected_semantic_commitment=commitment, ) if ( inverse_glyph_hybrid_plan( plan, asset_entry_sha256_by_asset_id=manifest_entries, expected_raw_request=raw_text, ) != raw_text ): raise GlyphHybridPlanError("post-assembly inverse reconstruction failed") public_pcm = assembly.public_pcm16() waveform = public_pcm.astype(np.float32) / np.float32(32768.0) if ( assembly.public_pcm16_le is not assembly.verification_pcm16_le or not np.array_equal( pcm16_audio_output(assembly.sample_rate, waveform)[1], public_pcm, ) ): raise RuntimeError("hybrid public and verification PCM differ") anchor = _speaker_anchor_array(DEFAULT_CENTROID) independent_cache = WholeWaveformVerificationCache() final_verification = _verify_trajectory_audio( (waveform,), (plan.canonical_target,), anchor, 1.0, QUALITY_FINAL_ASR_MAX_NEW_TOKENS, release_speaker_gate=True, transcriber=_transcribe_glyph_hybrid_breeze_primary, ) independent_final_verification = None try: _require_glyph_final_verification(final_verification) independent_final_verification = _verify_independent_whole_audio( waveform, plan.canonical_target, anchor, independent_cache, transcriber=_transcribe_glyph_hybrid_breeze_confirmation, ) _require_glyph_final_verification(independent_final_verification) except FinalOutputRejectedError as error: raise gr.Error( "V5 字元語音未通過整段內容、音色、音質與語速驗證,未回傳音訊。" ) from error _emit_glyph_hybrid_evidence( raw_text=raw_text, plan=plan, profile_state=profile_state, segments=immutable_segments, assembly=assembly, semantic_commitment=commitment, request_ledger=request_ledger, final_verification=final_verification, independent_final_verification=independent_final_verification, ) return assembly.sample_rate, waveform def _maybe_synthesize_glyph_speaker_request( text: object, speaker: object, *, cfg: object, steps: object, speed: object, ) -> tuple[int, np.ndarray] | None: if not _glyph_profile_selected(speaker): return None raw_text = str(text) if not _glyph_request_network_looking(raw_text): return None _validate_glyph_runtime_controls(cfg=cfg, steps=steps, speed=speed) profile_state = _GLYPH_PROFILE_STATE if not profile_state.available: raise gr.Error( "V5 字元語音資產目前不可用或完整性驗證失敗,未改用未驗證路徑。" ) try: plan = build_glyph_hybrid_plan( raw_text, asset_entry_sha256_by_asset_id=( profile_state.asset_entry_sha256_by_asset_id ), ) except (GlyphHybridPlanError, GlyphAssetError) as error: raise gr.Error("V5 字元語音規劃失敗,未改用未驗證路徑。") from error if plan is None: raise gr.Error( "此網路字串不符合 V5 嚴格小寫 ASCII 規格,未進行部分合成或舊路徑 fallback。" ) try: return _synthesize_glyph_hybrid( raw_text, plan, profile_state, cfg=float(cfg), steps=int(steps), speed=float(speed), ) except gr.Error: raise except (GlyphHybridPlanError, GlyphAssetError, RuntimeError, ValueError) as error: raise gr.Error( "V5 字元語音完整性或品質驗證失敗,未回傳未驗證音訊。" ) from error @speaker_gpu @timed_latency_request def tts_speaker( text: str, speaker: str = DEFAULT_SPEAKER, cfg: float = DEFAULT_CFG, steps: int = DEFAULT_STEPS, speed: float = 1.0, seed: int = SPEAKER_GENERATION_SEED, ): return pcm16_audio_output( *_synthesize( text, SPEAKERS.get(speaker, DEFAULT_CENTROID), cfg=cfg, steps=steps, speed=speed, request_seed=int(seed), speaker_projector_scale=( SPEAKER_B_PROJECTOR_SCALE if speaker == SPEAKER_B_LABEL else 1.0 ), ) ) @reference_gpu @timed_latency_request def tts_reference( text: str, reference_wav: str, cfg: float = REFERENCE_CFG, steps: int = DEFAULT_STEPS, speed: float = 1.0, seed: int = REFERENCE_GENERATION_SEED, ): if not reference_wav: raise gr.Error("請先錄音或上傳參考音檔。") prepared_reference, temporary_reference = _prepare_reference_audio(reference_wav) try: return pcm16_audio_output( *_synthesize( text, None, cfg=cfg, steps=steps, speed=speed, request_seed=int(seed), reference_wav_path=prepared_reference, short_text_min_cfg=REFERENCE_CFG, ) ) finally: if temporary_reference is not None: os.unlink(temporary_reference) EXAMPLE_TEXTS = [ "今天天氣真好,我們一起去散步吧。", "我要吃蚵仔煎,然後去丟垃圾。", "這學期的成績包括研究報告和期末考。", "這是 AI TTS code switching 測試,混合中英文也沒問題。", "注音符號測試:ㄅ、ㄆ、ㄇ、ㄈ。", "今天的會議會先整理目前進度,再確認下一階段的工作。遇到需要討論的項目時," "請先記下問題,等報告結束後再一起處理。最後,我們會確認負責人和預計完成時間。", ] HEADER = """ # BlueMagpie-TTS Demo 台灣華語與中英混合文字轉語音。模型版本:`BlueMagpie-TTS`。 """ 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=True, ) reference_cfg_input = gr.Number( value=REFERENCE_CFG, label="參考音色 CFG", interactive=True, ) steps_input = gr.Number( value=DEFAULT_STEPS, precision=0, interactive=True, label="NFE steps", ) speed_input = gr.Number( value=1.0, interactive=False, label="後處理語速(效率模式停用)", ) with gr.Row(): speaker_seed_input = gr.Number( value=SPEAKER_GENERATION_SEED, minimum=0, maximum=2_147_483_647, precision=0, interactive=True, label="內建語者 Seed", ) reference_seed_input = gr.Number( value=REFERENCE_GENERATION_SEED, minimum=0, maximum=2_147_483_647, precision=0, interactive=True, label="參考音色 Seed", ) 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_seed_input, ], speaker_output, ) with gr.Tab("參考音色"): gr.Markdown( "建議錄製至少 6 秒、單一語者且背景乾淨的授權參考音檔;" "超過 6 秒會固定取中央 6 秒。錄音會直接作為模型的 " "reference audio,不轉成 ECAPA 音色。" ) 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, reference_cfg_input, steps_input, speed_input, reference_seed_input, ], reference_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(ssr_mode=False)