fix(whisperx): reject unconfigured diarization

WhisperX silently returned a plain transcript when diarization lacked
the Hugging Face token required to load pyannote. Reject that request
clearly so callers do not mistake missing speaker labels for a
successful diarization.

Convert WhisperX seconds to the nanosecond duration unit used by the
transcription API.

Assisted-by: Codex:gpt-5
Signed-off-by: Ettore Di Giacinto <mudler@localai.io>
This commit is contained in:
Ettore Di Giacinto committed 2026-09-05 22:09:16 +00:00
1 parent 9ca516d6ef
commit 8744de44d4
4 files changed
+47 -3

No files matched your search

+8 -2
View File
@@ -16,6 +16,7 @@ import grpc
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', 'common'))
sys.path.insert(0, os.path.join(os.path.dirname(__file__), 'common'))
from grpc_auth import get_auth_interceptors
from transcript_utils import require_diarization_token, seconds_to_nanoseconds
@@ -81,6 +82,11 @@ class BackendServicer(backend_pb2_grpc.BackendServicer):
import whisperx
from whisperx.diarize import DiarizationPipeline
try:
require_diarization_token(request.diarize, self.hf_token)
except ValueError as err:
context.abort(grpc.StatusCode.FAILED_PRECONDITION, str(err))
resultSegments = []
text = ""
try:
@@ -117,8 +123,8 @@ class BackendServicer(backend_pb2_grpc.BackendServicer):
# Build result segments
for idx, seg in enumerate(transcript["segments"]):
seg_text = seg.get("text", "")
start = int(seg.get("start", 0))
end = int(seg.get("end", 0))
start = seconds_to_nanoseconds(seg.get("start", 0))
end = seconds_to_nanoseconds(seg.get("end", 0))
speaker = seg.get("speaker", "")
resultSegments.append(backend_pb2.TranscriptSegment(
@@ -0,0 +1,25 @@
import unittest
import transcript_utils
class TestTranscriptUtils(unittest.TestCase):
def test_diarization_requires_hugging_face_token(self):
with self.assertRaisesRegex(
ValueError,
"HF_TOKEN is required for WhisperX diarization",
):
transcript_utils.require_diarization_token(True, None)
def test_diarization_does_not_require_token_when_disabled(self):
transcript_utils.require_diarization_token(False, None)
def test_seconds_are_serialized_as_nanoseconds(self):
self.assertEqual(
transcript_utils.seconds_to_nanoseconds(3.25),
3_250_000_000,
)
if __name__ == "__main__":
unittest.main()
@@ -0,0 +1,12 @@
"""Helpers for WhisperX transcript responses."""
def require_diarization_token(diarize, token):
"""Reject diarization when WhisperX cannot load its gated pipeline."""
if diarize and not token:
raise ValueError("HF_TOKEN is required for WhisperX diarization")
def seconds_to_nanoseconds(seconds):
"""Convert WhisperX timestamps to the duration unit used by LocalAI."""
return int(seconds * 1_000_000_000)