minke.yu/sglang
1
0
Fork 0
Code Issues Pull Requests Actions 12 Packages Projects Releases Wiki Activity

fix: fix transcription & audio-understanding for ASR/audio/speech models (#32611)

Browse Source 
Co-authored-by: Singh <rohitsi2@iil-login.iind.intel.com>
This commit is contained in:
Rohit Kumar Singh
2026-08-19 18:44:23 +08:00
committed by GitHub
co-authored by Singh
parent f446e853e7
commit 3e5ce26c2d
10 changed files with 608 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files
python/sglang/srt/entrypoints/openai/transcription_adapters/__init__.py
+12
View File
@@ -7,9 +7,18 @@ from sglang.srt.entrypoints.openai.transcription_adapters.base import ( # noqa:
)
# Import built-in adapters so they self-register via @register_transcription_adapter.
from sglang.srt.entrypoints.openai.transcription_adapters.glmasr import ( # noqa: F401
GlmAsrAdapter,
)
from sglang.srt.entrypoints.openai.transcription_adapters.granite_speech import ( # noqa: F401
GraniteSpeechAdapter,
)
from sglang.srt.entrypoints.openai.transcription_adapters.mimo_v2_asr import ( # noqa: F401
MiMoV2ASRAdapter,
)
from sglang.srt.entrypoints.openai.transcription_adapters.qwen2_audio import ( # noqa: F401
Qwen2AudioAdapter,
)
from sglang.srt.entrypoints.openai.transcription_adapters.qwen3_asr import ( # noqa: F401
Qwen3ASRAdapter,
)
@@ -24,4 +33,7 @@ __all__ = [
"WhisperAdapter",
"Qwen3ASRAdapter",
"MiMoV2ASRAdapter",
"GraniteSpeechAdapter",
"Qwen2AudioAdapter",
"GlmAsrAdapter",
]
python/sglang/srt/entrypoints/openai/transcription_adapters/glmasr.py
+91
View File
@@ -0,0 +1,91 @@
from __future__ import annotations
from sglang.srt.entrypoints.openai.protocol import (
TranscriptionRequest,
TranscriptionUsage,
TranscriptionVerboseResponse,
)
from sglang.srt.entrypoints.openai.transcription_adapters.base import (
TranscriptionAdapter,
register_transcription_adapter,
)
# Per-second output-token budget for clips long enough to exceed the floor.
# Speech runs ~2-3 words/s and tokenizers emit >1 token/word, so 15 tok/s
# leaves generous headroom over the natural transcript length while still
# bounding runaway generation.
_MAX_NEW_TOKENS_PER_SECOND = 15
# Floor for the generation-length cap; sized so a ~30s clip always fits.
_DEFAULT_MAX_NEW_TOKENS = 448
@register_transcription_adapter("GlmAsr")
class GlmAsrAdapter(TranscriptionAdapter):
"""Adapter for GLM-ASR.
GLM-ASR is a decoder-only speech LM: the multimodal processor
(``GlmAsrProcessor``) inserts the ``<|begin_of_audio|>...<|end_of_audio|>``
placeholder and the model free-form generates the transcript. There is no
Whisper-style forced language/task prefix, so language auto-detection is
disabled and the default (no-op) prompt/response handling applies.
"""
# Assistant framing GLM-ASR was trained to emit around the transcript.
# Mirrors HF ``GlmAsrProcessor.decode(strip_prefix=True)`` so the raw
# transcript is returned
_ASSISTANT_PREFIXES = (
"The spoken content of the audio is",
"The transcription of the audio is",
"The content of the input audio is",
)
def build_sampling_params(self, request: TranscriptionRequest) -> dict:
# ``/v1/audio/transcriptions`` has no request-side length field, so the
# adapter is the only control. Short clips use the floor; longer clips
# scale with duration so the transcript is not silently truncated.
duration_s = request.audio_duration_s or 0.0
return {
"temperature": request.temperature,
"max_new_tokens": max(
_DEFAULT_MAX_NEW_TOKENS,
int(duration_s * _MAX_NEW_TOKENS_PER_SECOND),
),
}
def postprocess_text(self, text: str) -> str:
# Strip the assistant prefix and surrounding quotes GLM-ASR wraps the
# transcript in (mirrors HF's ``strip_prefix=True``).
stripped = text.strip()
for prefix in self._ASSISTANT_PREFIXES:
if stripped.startswith(prefix):
stripped = stripped[len(prefix) :].strip()
break
if stripped.endswith("."):
stripped = stripped[:-1].strip()
if (
len(stripped) >= 2
and stripped[0] == stripped[-1]
and stripped[0]
in {
"'",
'"',
}
):
stripped = stripped[1:-1].strip()
return stripped
def build_verbose_response(
self,
request: TranscriptionRequest,
text: str,
ret: dict,
tokenizer,
usage: TranscriptionUsage,
) -> TranscriptionVerboseResponse:
return TranscriptionVerboseResponse(
language=None,
duration=round(request.audio_duration_s, 2),
text=text,
segments=[],
usage=usage,
)
python/sglang/srt/entrypoints/openai/transcription_adapters/granite_speech.py
+63
View File
@@ -0,0 +1,63 @@
from __future__ import annotations
from sglang.srt.entrypoints.openai.protocol import (
TranscriptionRequest,
TranscriptionUsage,
TranscriptionVerboseResponse,
)
from sglang.srt.entrypoints.openai.transcription_adapters.base import (
TranscriptionAdapter,
register_transcription_adapter,
)
# Per-second output-token budget for clips long enough to exceed the floor.
# Speech runs ~2-3 words/s and tokenizers emit >1 token/word, so 15 tok/s
# leaves generous headroom over the natural transcript length while still
# bounding runaway generation.
_MAX_NEW_TOKENS_PER_SECOND = 15
# Floor for the generation-length cap; sized so a ~30s clip always fits.
_DEFAULT_MAX_NEW_TOKENS = 448
@register_transcription_adapter("GraniteSpeech")
class GraniteSpeechAdapter(TranscriptionAdapter):
"""Transcription adapter for IBM Granite Speech.
Granite Speech is a decoder-only speech LM: the audio encoder features are
merged into the ``<|audio|>`` placeholder positions and the model free-form
generates the transcript. It has no Whisper-style forced language/task
prefix, so language auto-detection and fused prefix parsing are disabled and
the default (no-op) prompt/response handling applies. Granite is
English-only, so the verbose response reports ``"en"`` when the request
omits a language.
"""
def build_sampling_params(self, request: TranscriptionRequest) -> dict:
# ``/v1/audio/transcriptions`` has no request-side length field, so the
# adapter is the only control. Short clips use the floor; longer clips
# scale with duration so the transcript is not silently truncated.
duration_s = request.audio_duration_s or 0.0
return {
"temperature": request.temperature,
"max_new_tokens": max(
_DEFAULT_MAX_NEW_TOKENS,
int(duration_s * _MAX_NEW_TOKENS_PER_SECOND),
),
}
def build_verbose_response(
self,
request: TranscriptionRequest,
text: str,
ret: dict,
tokenizer,
usage: TranscriptionUsage,
) -> TranscriptionVerboseResponse:
# ``language`` left unset. It is multilingual and infers the language from the audio.
return TranscriptionVerboseResponse(
language=None,
duration=round(request.audio_duration_s, 2),
text=text,
segments=[],
usage=usage,
)
python/sglang/srt/entrypoints/openai/transcription_adapters/qwen2_audio.py
+61
View File
@@ -0,0 +1,61 @@
from __future__ import annotations
from sglang.srt.entrypoints.openai.protocol import (
TranscriptionRequest,
TranscriptionUsage,
TranscriptionVerboseResponse,
)
from sglang.srt.entrypoints.openai.transcription_adapters.base import (
TranscriptionAdapter,
register_transcription_adapter,
)
# Per-second output-token budget for clips long enough to exceed the floor.
# Speech runs ~2-3 words/s and tokenizers emit >1 token/word, so 15 tok/s
# leaves generous headroom over the natural transcript length while still
# bounding runaway generation.
_MAX_NEW_TOKENS_PER_SECOND = 15
# Floor for the generation-length cap; sized so a ~30s clip always fits.
_DEFAULT_MAX_NEW_TOKENS = 448
@register_transcription_adapter("Qwen2Audio")
class Qwen2AudioAdapter(TranscriptionAdapter):
"""Adapter for Qwen2-Audio.
Qwen2-Audio is a decoder-only audio LM: the multimodal processor
(``Qwen2AudioMultimodalProcessor``) inserts the audio placeholder and the
model free-form generates the transcript. There is no Whisper-style forced
language/task prefix, so language auto-detection is disabled and the
default (no-op) prompt/response handling applies.
"""
def build_sampling_params(self, request: TranscriptionRequest) -> dict:
# ``/v1/audio/transcriptions`` has no request-side length field, so the
# adapter is the only control. Short clips use the floor; longer clips
# scale with duration so the transcript is not silently truncated.
duration_s = request.audio_duration_s or 0.0
return {
"temperature": request.temperature,
"max_new_tokens": max(
_DEFAULT_MAX_NEW_TOKENS,
int(duration_s * _MAX_NEW_TOKENS_PER_SECOND),
),
}
def build_verbose_response(
self,
request: TranscriptionRequest,
text: str,
ret: dict,
tokenizer,
usage: TranscriptionUsage,
) -> TranscriptionVerboseResponse:
# ``language`` left unset. It auto-detects the spoken language internally.
return TranscriptionVerboseResponse(
language=None,
duration=round(request.audio_duration_s, 2),
text=text,
segments=[],
usage=usage,
)
python/sglang/srt/models/glmasr.py
+51 -10
View File
@@ -92,18 +92,59 @@ class GlmAsrForConditionalGeneration(nn.Module):
return self.pattern.pad_input_tokens(input_ids, mm_inputs)
def get_audio_feature(self, items: List[MultimodalDataItem]) -> torch.Tensor:
# Extract audio features from input items
input_features = torch.cat([item.feature for item in items], dim=0).type(
self.audio_tower.dtype
)
# The processor emits mel features as a batch of fixed-size 30s windows
# (``(num_chunks, num_mel_bins, 3000)``) plus an ``input_features_mask``
# marking the valid frames per window; long clips span several windows.
# The projector stacks ``merge_factor`` encoder frames into one
# embedding, so we reshape per-window to
# ``(num_chunks, -1, intermediate_size)`` and then keep only the valid
# embeddings per window (derived from the mask via the encoder's
# conv-subsampling + merge downsampling). This keeps the emitted count
# aligned with the audio placeholder tokens the processor inserted,
# mirroring the HF ``CohereAsr``/``GlmAsr`` ``get_audio_features`` path.
audio_config = self.config.audio_config
merge_factor = audio_config.intermediate_size // audio_config.hidden_size
audio_embeds = self.audio_tower(input_features).last_hidden_state
audio_embeds = audio_embeds.reshape(
-1, self.config.audio_config.intermediate_size
)
audio_embeds = self.multi_modal_projector(audio_embeds)
audio_embeds_list = []
for item in items:
input_features = item.feature.type(self.audio_tower.dtype)
input_features_mask = item.input_features_mask
if input_features_mask.dim() == 1:
input_features_mask = input_features_mask.unsqueeze(0)
input_features_mask = input_features_mask.to(input_features.device)
return audio_embeds
hidden_states = self.audio_tower(input_features).last_hidden_state
# The frame axis is not guaranteed to be a multiple of
# ``merge_factor`` (e.g. a short unpadded clip yields an odd frame
# count), so right-pad it before stacking ``merge_factor`` frames
# per embedding. Padded rows are trimmed back per-window below via
# ``post_lengths``, so the extra frames never reach the LM.
num_frames = hidden_states.shape[1]
pad = (-num_frames) % merge_factor
if pad:
hidden_states = torch.nn.functional.pad(hidden_states, (0, 0, 0, pad))
hidden_states = hidden_states.reshape(
input_features.shape[0], -1, audio_config.intermediate_size
)
embeds = self.multi_modal_projector(hidden_states)
# Valid embeddings per window: mel frames -> conv subsampling
# (stride 1 then stride 2) -> merge_factor downsampling.
audio_lengths = input_features_mask.sum(-1)
for padding, kernel_size, stride in [(1, 3, 1), (1, 3, 2)]:
audio_lengths = (
audio_lengths + 2 * padding - (kernel_size - 1) - 1
) // stride + 1
post_lengths = (audio_lengths - merge_factor) // merge_factor + 1
valid = (
torch.arange(embeds.shape[1], device=post_lengths.device)[None, :]
< post_lengths[:, None]
)
audio_embeds_list.append(embeds[valid.to(embeds.device)])
return torch.cat(audio_embeds_list, dim=0)
def forward(
self,
python/sglang/srt/multimodal/processors/base_processor.py
+1
View File
@@ -676,6 +676,7 @@ class BaseMultimodalProcessor(ABC):
"Gemma4Processor",
"Gemma4UnifiedProcessor",
"GlmAsrProcessor",
"GraniteSpeechProcessor",
"Qwen2AudioProcessor",
"Qwen3ASRProcessor",
"Qwen3OmniMoeProcessor",
python/sglang/srt/multimodal/processors/glmasr.py
+34 -1
View File
@@ -29,14 +29,47 @@ class GlmAsrProcessor(BaseMultimodalProcessor):
audio_token_id=self.audio_token_id,
).build(_processor)
# GLM-ASR's chat template keys on ``{"type": "audio"}`` (or any mapping
# containing an ``audio`` key) to emit its audio placeholder span.
_TRANSCRIPTION_CONVERSATION = [
{
"role": "user",
"content": [
{"type": "audio", "audio": ""},
{"type": "text", "text": "Transcribe the audio."},
],
}
]
def _build_transcription_prompt(self, input_text) -> str:
"""Fall back to a default ASR prompt for audio-only requests.
The ``/v1/audio/transcriptions`` endpoint sends empty text (and hence
empty ``input_ids``), which carries no audio placeholder. Render the
GLM-ASR chat prompt with one audio span so the encoder features have a
slot to fill; otherwise the caller-supplied text is used as-is.
"""
if isinstance(input_text, list):
input_text = (
self._processor.tokenizer.decode(input_text) if input_text else ""
)
if input_text and input_text.strip():
return input_text
return self._processor.apply_chat_template(
self._TRANSCRIPTION_CONVERSATION,
add_generation_prompt=True,
tokenize=False,
)
async def process_mm_data_async(
self,
audio_data,
input_text,
**kwargs,
):
prompt = self._build_transcription_prompt(input_text)
base_output = await self.load_mm_data(
prompt=input_text,
prompt=prompt,
audio_data=audio_data,
multimodal_tokens=self.mm_tokens,
)
python/sglang/srt/multimodal/processors/qwen_audio.py
+73 -1
View File
@@ -1,5 +1,8 @@
import logging
import re
import numpy as np
from sglang.srt.managers.schedule_batch import (
Modality,
MultimodalDataItem,
@@ -11,6 +14,8 @@ from sglang.srt.multimodal.processors.base_processor import (
MultimodalSpecialTokens,
)
logger = logging.getLogger(__name__)
class Qwen2AudioMultimodalProcessor(BaseMultimodalProcessor):
models = [Qwen2AudioForConditionalGeneration]
@@ -35,6 +40,70 @@ class Qwen2AudioMultimodalProcessor(BaseMultimodalProcessor):
self.ATTR_NAME_TO_MODALITY.update({"feature_attention_mask": Modality.AUDIO})
# Qwen2-Audio's audio tower requires exactly 3000 mel frames (a fixed
# 30s window); its HF feature extractor truncates to that by default.
# BaseMultimodalProcessor otherwise passes ``truncation=False`` to audio
# processors (needed by chunking encoders), which would feed >3000-frame
# mel for clips longer than 30s and make the tower raise. Force
# truncation so long clips are capped to the model's window.
self.audio_config = {**self.audio_config, "truncation": True}
# Qwen2-Audio's chat template matches a bare ``audio`` key; the strict instruction
# keeps the model from emitting a "The content of this audio is:" preamble
# that would otherwise inflate WER.
_TRANSCRIPTION_CONVERSATION = [
{
"role": "user",
"content": [
{"type": "audio", "audio": ""},
{
"type": "text",
"text": (
"Transcribe the audio. Output only the exact transcription, "
"with no preamble, prefix, commentary, or quotation marks."
),
},
],
}
]
def _build_transcription_prompt(self, input_text) -> str:
"""Fall back to a default ASR prompt for audio-only requests.
The ``/v1/audio/transcriptions`` endpoint sends empty text (and hence
empty ``input_ids``), which carries no audio placeholder. Render the
Qwen2-Audio chat prompt with one audio span so the encoder features have
a slot to fill; otherwise the caller-supplied text is used as-is.
"""
if isinstance(input_text, list):
input_text = (
self._processor.tokenizer.decode(input_text) if input_text else ""
)
if input_text and input_text.strip():
return input_text
return self._processor.apply_chat_template(
self._TRANSCRIPTION_CONVERSATION,
add_generation_prompt=True,
tokenize=False,
)
def _warn_if_audio_exceeds_window(self, audios) -> None:
# Qwen2-Audio's encoder is a single fixed 30s window, so
# warn user if audio is truncated.
feature_extractor = self._processor.feature_extractor
max_samples = int(
feature_extractor.sampling_rate * feature_extractor.chunk_length
)
for audio in audios:
if isinstance(audio, np.ndarray) and audio.shape[-1] > max_samples:
logger.warning(
"Qwen2-Audio input is %.1fs but the encoder window is %ds; "
"only the first %ds will be transcribed (audio truncated).",
audio.shape[-1] / feature_extractor.sampling_rate,
feature_extractor.chunk_length,
feature_extractor.chunk_length,
)
def get_mm_data(self, prompt, embeddings, **kwargs):
audio_feature_lens = kwargs.get("audio_feature_lens", None)
@@ -87,14 +156,17 @@ class Qwen2AudioMultimodalProcessor(BaseMultimodalProcessor):
input_text,
**kwargs,
):
prompt = self._build_transcription_prompt(input_text)
base_output = await self.load_mm_data(
prompt=input_text,
prompt=prompt,
audio_data=audio_data,
multimodal_tokens=self.mm_tokens,
)
if base_output is None:
return None
self._warn_if_audio_exceeds_window(base_output.audios)
mm_items, input_ids, ret = self.process_and_combine_mm_data(
base_output, self.mm_tokens
)
python/sglang/srt/multimodal/processors/transformers_auto.py
+59 -2
View File
@@ -59,7 +59,7 @@ class TransformersAutoMultimodalProcessor(BaseMultimodalProcessor):
),
audio_token_id=_first_attr(
hf_config,
("audio_token_id",),
("audio_token_id", "audio_token_index"),
),
).build(_processor)
@@ -141,6 +141,40 @@ class TransformersAutoMultimodalProcessor(BaseMultimodalProcessor):
return items
def _build_audio_prompt(self, input_text) -> str:
"""Ensure the prompt carries the model's audio placeholder token.
Audio-only entrypoints (``/v1/audio/transcriptions``) send empty text,
which would otherwise make the HF processor crash (empty ``text`` list)
and leave no ``<|audio|>`` placeholder for the encoder features to fill.
Build a minimal chat prompt with one audio token per clip.
"""
if isinstance(input_text, list):
input_text = (
self._processor.tokenizer.decode(input_text) if input_text else ""
)
if (
input_text
and input_text.strip()
and self.mm_tokens.audio_token in input_text
):
return input_text
audio_token = self.mm_tokens.audio_token
content = (
f"{audio_token}Transcribe the speech into written text. "
f"Output only the exact transcription, with no preamble, commentary, "
f"quotation marks, or explanation."
)
tokenizer = self._processor.tokenizer
if hasattr(tokenizer, "apply_chat_template"):
return tokenizer.apply_chat_template(
[{"role": "user", "content": content}],
tokenize=False,
add_generation_prompt=True,
)
return content
async def process_mm_data_async(
self,
image_data,
@@ -153,9 +187,32 @@ class TransformersAutoMultimodalProcessor(BaseMultimodalProcessor):
if video_data is not None and not isinstance(video_data, list):
video_data = [video_data]
# Audio path: load clips and run the shared token-expansion machinery,
# which passes ``audio=`` to the HF processor and packs audio features
# into MultimodalDataItem objects with the right token offsets.
if audio_data:
prompt = self._build_audio_prompt(input_text)
base_output = await self.load_mm_data(
prompt=prompt,
audio_data=audio_data,
multimodal_tokens=self.mm_tokens,
)
if base_output is None:
return None
mm_items, input_ids, _ = self.process_and_combine_mm_data(
base_output, self.mm_tokens
)
ret = MultimodalProcessorOutput(
input_ids=input_ids.tolist(),
mm_items=mm_items,
)
if self.mm_tokens.audio_token_id is not None:
ret.audio_token_id = self.mm_tokens.audio_token_id
return ret
# Load raw media
images = self._load_images(image_data)
# TODO: video / audio loading when needed
# TODO: video loading when needed
# Apply HF processor — handles token expansion internally
processor_output = self._apply_hf_processor(
test/registered/unit/entrypoints/openai/test_transcription_adapters.py
+163
View File
@@ -0,0 +1,163 @@
"""Unit tests for the transcription-adapter registry and the ASR/audio adapters.
Covers the pieces that are easy to regress and cheap to check on CPU:
* ``resolve_adapter`` maps each real HF architecture string to the intended
adapter class (guards the substring-matching resolver against collisions when
new keys are added), and an unknown arch falls back to Whisper.
* The decoder-only speech-LM adapters (GLM-ASR, Qwen2-Audio, Granite Speech)
keep language auto-detection off and each implement ``build_sampling_params``
(duration-scaled ``max_new_tokens`` above a per-model floor) and
``build_verbose_response`` (no segments, per-model default language).
"""
import unittest
from sglang.srt.entrypoints.openai.protocol import TranscriptionRequest
from sglang.srt.entrypoints.openai.transcription_adapters import (
GlmAsrAdapter,
GraniteSpeechAdapter,
Qwen2AudioAdapter,
WhisperAdapter,
resolve_adapter,
)
from sglang.test.ci.ci_register import register_cpu_ci
from sglang.test.test_utils import CustomTestCase
register_cpu_ci(est_time=4, suite="base-a-test-cpu")
# Per-second scaling rate every speech-LM adapter uses for max_new_tokens.
_TOKENS_PER_SECOND = 15
# (arch string, adapter class, expected floor max_new_tokens, expected response language)
# Response language is always None: these adapters neither honor a request-side
# hint nor detect+report a language, so they must not claim one.
_SPEECH_LM_CASES = [
("GlmAsrForConditionalGeneration", GlmAsrAdapter, 448, None),
("Qwen2AudioForConditionalGeneration", Qwen2AudioAdapter, 448, None),
("GraniteSpeechForConditionalGeneration", GraniteSpeechAdapter, 448, None),
]
class TestTranscriptionAdapterResolution(CustomTestCase):
def test_resolves_expected_adapter_per_architecture(self):
for arch, cls, _, _ in _SPEECH_LM_CASES:
with self.subTest(arch=arch):
self.assertIsInstance(resolve_adapter([arch]), cls)
def test_unknown_architecture_falls_back_to_whisper(self):
# Whisper is the registered default; a non-ASR arch must not match one
# of the speech-LM keys via substring.
self.assertIsInstance(
resolve_adapter(["SomeUnknownForConditionalGeneration"]), WhisperAdapter
)
self.assertIsInstance(
resolve_adapter(["WhisperForConditionalGeneration"]), WhisperAdapter
)
self.assertIsInstance(resolve_adapter([]), WhisperAdapter)
class TestSpeechLMAdapterContract(CustomTestCase):
def _request(self, temperature=0.0, duration=3.0, language=None):
return TranscriptionRequest(
model="m",
temperature=temperature,
audio_duration_s=duration,
language=language,
)
def test_language_detection_disabled(self):
for _, cls, _, _ in _SPEECH_LM_CASES:
with self.subTest(adapter=cls.__name__):
self.assertFalse(cls().supports_language_detection)
def test_sampling_params_short_clip_uses_floor(self):
# A 3s clip scales to well under any adapter's floor, so max_new_tokens
# must stay pinned at the per-model floor (guards the max(...) floor).
req = self._request(temperature=0.2, duration=3.0)
for _, cls, max_new_tokens, _ in _SPEECH_LM_CASES:
with self.subTest(adapter=cls.__name__):
self.assertEqual(
cls().build_sampling_params(req),
{"temperature": 0.2, "max_new_tokens": max_new_tokens},
)
def test_sampling_params_long_clip_scales_with_duration(self):
# A long clip must lift max_new_tokens above the floor so the transcript
# isn't silently truncated. Each adapter scales at _TOKENS_PER_SECOND, so
# 600s -> 9000 tokens, well above every per-model floor.
req = self._request(temperature=0.0, duration=600.0)
expected = int(600.0 * _TOKENS_PER_SECOND)
for _, cls, floor, _ in _SPEECH_LM_CASES:
with self.subTest(adapter=cls.__name__):
params = cls().build_sampling_params(req)
self.assertEqual(params["max_new_tokens"], expected)
self.assertGreater(params["max_new_tokens"], floor)
def test_sampling_params_at_floor_scale_boundary(self):
# Pin the floor<->scale crossover (max_new_tokens = max(floor,
# int(duration * rate))). The knee is at floor/rate seconds; bracket it
# by +/-1s. Just below must stay at the floor; just above must switch to
# the scaled value. A max()->min() swap or a wrong comparison at the
# knee would flip one of these and is invisible to the deep-floor (3s)
# and deep-scaled (600s) cases.
for _, cls, floor, _ in _SPEECH_LM_CASES:
knee_s = floor / _TOKENS_PER_SECOND
below_s = knee_s - 1.0
above_s = knee_s + 1.0
with self.subTest(adapter=cls.__name__):
below = cls().build_sampling_params(self._request(duration=below_s))
self.assertEqual(below["max_new_tokens"], floor)
above = cls().build_sampling_params(self._request(duration=above_s))
self.assertEqual(
above["max_new_tokens"], int(above_s * _TOKENS_PER_SECOND)
)
self.assertGreater(above["max_new_tokens"], floor)
def test_verbose_response_language_unset_and_has_no_segments(self):
for _, cls, _, expected_language in _SPEECH_LM_CASES:
with self.subTest(adapter=cls.__name__):
resp = cls().build_verbose_response(
self._request(duration=3.456), "hello", {}, None, None
)
self.assertEqual(resp.language, expected_language)
self.assertEqual(resp.duration, 3.46)
self.assertEqual(resp.text, "hello")
self.assertEqual(resp.segments, [])
def test_verbose_response_does_not_claim_request_language(self):
# Even when the client sends a language hint, these adapters neither
# honor nor detect it, so the response must not claim it (guards against
# re-introducing a misleading echo).
for _, cls, _, _ in _SPEECH_LM_CASES:
with self.subTest(adapter=cls.__name__):
resp = cls().build_verbose_response(
self._request(language="fr"), "bonjour", {}, None, None
)
self.assertIsNone(resp.language)
def test_glm_postprocess_strips_assistant_prefix(self):
# GLM-ASR wraps the transcript in an assistant preamble + quotes;
# postprocess_text must strip it (mirrors HF strip_prefix=True) so it
# doesn't leak into the transcript and inflate WER.
adapter = GlmAsrAdapter()
self.assertEqual(
adapter.postprocess_text(
'The spoken content of the audio is "hello world".'
),
"hello world",
)
self.assertEqual(
adapter.postprocess_text("The transcription of the audio is 'bonjour'."),
"bonjour",
)
# A raw transcript with no preamble must pass through unchanged.
self.assertEqual(
adapter.postprocess_text("just the transcript"), "just the transcript"
)
if __name__ == "__main__":
unittest.main()