fix(nvidia_riva): send the api key exactly once by calling riva over grpc directly (#50140)

* fix(nvidia_riva): send the api key once per call by talking to Riva over grpc directly

Drops nvidia-riva-client from the stt-nvidia-riva extra. Depending on its version it sent
the bearer token zero times on plaintext or twice on TLS, so token-checking Riva deployments
rejected every request (#50059). The provider now declares the few Riva messages it uses,
opens its own grpc channel and puts authorization and function-id on each call exactly once

* test(nvidia_riva): run the provider against a real grpc server built from nvidia's own definitions

* fix(nvidia_riva): validate endpointing settings with pydantic before building the riva message

* test(nvidia_riva): check the key is sent once over tls, and drop a comment that restated the code
This commit is contained in:
ryan-crabbe-berri 2026-11-04 18:07:31 -08:00 • committed by GitHub
parent 5e91bc8e86
commit 42b4cfc5bf
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
9 changed files with 985 additions and 991 deletions

View file

@ -13,24 +13,26 @@ The handler is intentionally a thin orchestration layer:
2. Build ``RecognitionConfig`` / ``StreamingRecognitionConfig`` protobufs
from the structured dict produced by
:class:`NvidiaRivaAudioTranscriptionConfig`.
3. Construct ``riva.client.Auth`` honoring NVCF (function-id metadata + TLS)
vs self-hosted (any host:port, optional TLS) modes.
4. Stream the audio through Riva's ``streaming_response_generator`` and
aggregate ``is_final`` results into a single transcript, or, when
``riva_offline`` is set, send the whole buffer through the unary
``offline_recognize`` RPC for NIMs that only ship offline engines.
3. Open a gRPC channel honoring NVCF (TLS by default) vs self-hosted (any
host:port, optional TLS) modes, and attach the ``authorization`` and NVCF
``function-id`` metadata to each call exactly once.
4. Stream the audio through Riva's ``StreamingRecognize`` RPC and aggregate
``is_final`` results into a single transcript, or, when ``riva_offline``
is set, send the whole buffer through the unary ``Recognize`` RPC for NIMs
that only ship offline engines.
5. Return a normalized ``TranscriptionResponse`` with ``duration`` exposed
on ``_hidden_params`` so cost calculation works.
``riva-client`` is imported lazily so ``litellm`` core remains usable
without the optional STT extras installed.
``grpcio`` and ``protobuf`` are imported lazily so ``litellm`` core remains
usable without the optional STT extras installed.
"""
from __future__ import annotations
import asyncio
import inspect
from collections.abc import Callable, Iterable, Mapping, Sequence
from types import MappingProxyType
from typing import TYPE_CHECKING, Any, Final, Protocol
from collections.abc import Iterable, Sequence
from contextlib import closing
from typing import TYPE_CHECKING, Final
import httpx
@ -64,89 +66,19 @@ if TYPE_CHECKING:
Logging as LiteLLMLoggingObj,
)
# Stream audio to Riva in ~50 ms slices (1600 samples at 16 kHz). Matches
# NVIDIA's recommended chunk size for streaming ASR — small enough for
# responsive endpointing, large enough to keep per-RPC overhead low.
_DEFAULT_CHUNK_SAMPLES: Final = 1600
_DEFAULT_CHUNK_BYTES: Final = _DEFAULT_CHUNK_SAMPLES * 2 # int16 = 2 bytes/sample
_RIVA_INSTALL_HINT = "NVIDIA Riva client is not installed. Install with `pip install 'litellm[stt-nvidia-riva]'`."
class _RivaAuth(Protocol):
"""Opaque ``riva.client.Auth`` handle."""
class _RecognizeFuture(Protocol):
def result(self, timeout: float | None = None) -> object: ...
def done(self) -> bool: ...
def cancel(self) -> bool: ...
class _AsrService(Protocol):
@property
def streaming_response_generator(self) -> Callable[..., Iterable[object]]: ...
def offline_recognize(self, audio_bytes: bytes, config: object, future: bool) -> _RecognizeFuture: ...
class _EndpointingConfig(Protocol):
"""Opaque ``EndpointingConfig`` protobuf message."""
class _EndpointingConfigField(Protocol):
CopyFrom: Callable[[_EndpointingConfig], None]
class _RecognitionConfig(Protocol):
@property
def endpointing_config(self) -> _EndpointingConfigField: ...
class _StreamingRecognitionConfig(Protocol):
"""Opaque ``StreamingRecognitionConfig`` protobuf message."""
class _AudioEncoding(Protocol):
@property
def LINEAR_PCM(self) -> object: ...
class _RivaClientModule(Protocol):
"""The ``riva.client`` entry points this handler calls."""
@property
def Auth(self) -> Callable[..., _RivaAuth]: ...
@property
def ASRService(self) -> Callable[[_RivaAuth], _AsrService]: ...
class _RivaAsrModule(Protocol):
"""The protobuf constructors this handler calls, from whichever module exposes them."""
@property
def AudioEncoding(self) -> _AudioEncoding: ...
@property
def RecognitionConfig(self) -> Callable[..., _RecognitionConfig]: ...
@property
def StreamingRecognitionConfig(self) -> Callable[..., _StreamingRecognitionConfig]: ...
@property
def EndpointingConfig(self) -> Callable[..., _EndpointingConfig]: ...
def _auth_factory(riva_module: _RivaClientModule) -> Callable[..., _RivaAuth]:
return riva_module.Auth
def _audio_encoding(riva_asr_module: _RivaAsrModule) -> _AudioEncoding:
return riva_asr_module.AudioEncoding
def _use_ssl(optional_params: dict) -> bool:
"""
NVCF endpoints are TLS-only, so ``nvcf_function_id`` defaults ``use_ssl``
to True. Self-hosted Riva defaults to plaintext but honors an explicit
``use_ssl``, since TLS-terminating ingresses in front of it are common.
"""
use_ssl_override: Final = optional_params.get("use_ssl")
if use_ssl_override is not None:
return bool(use_ssl_override)
return bool(optional_params.get("nvcf_function_id"))
def _parse_riva_offline(value: object, drop_params: bool) -> bool:
@ -192,7 +124,7 @@ class NvidiaRivaAudioTranscription:
litellm_params: dict,
model_response: TranscriptionResponse,
timeout: float | str | httpx.Timeout | None,
logging_obj: "LiteLLMLoggingObj",
logging_obj: LiteLLMLoggingObj,
api_key: str | None,
api_base: str | None,
atranscription: bool = False,
@ -237,13 +169,11 @@ class NvidiaRivaAudioTranscription:
litellm_params: dict,
model_response: TranscriptionResponse,
timeout: float | str | httpx.Timeout | None,
logging_obj: "LiteLLMLoggingObj",
logging_obj: LiteLLMLoggingObj,
api_key: str | None,
api_base: str | None,
provider_config: NvidiaRivaAudioTranscriptionConfig | None = None,
) -> TranscriptionResponse:
# ``riva-client`` exposes a sync streaming generator, so we offload
# the blocking call to a worker thread to keep the event loop free.
return await asyncio.to_thread(
self._run_sync,
model=model,
@ -267,7 +197,7 @@ class NvidiaRivaAudioTranscription:
litellm_params: dict,
model_response: TranscriptionResponse,
timeout: float | str | httpx.Timeout | None,
logging_obj: "LiteLLMLoggingObj",
logging_obj: LiteLLMLoggingObj,
api_key: str | None,
api_base: str | None,
provider_config: NvidiaRivaAudioTranscriptionConfig,
@ -317,18 +247,12 @@ class NvidiaRivaAudioTranscription:
) or litellm.drop_params is True
riva_offline: Final = _parse_riva_offline(optional_params.get("riva_offline"), drop_params=drop_params)
riva_module, riva_asr_module = _import_riva()
auth_obj: Final = self._construct_auth(
riva_module=riva_module,
api_base=api_base,
api_key=api_key,
optional_params=optional_params,
)
try:
from litellm.llms.nvidia_riva.audio_transcription import riva_grpc
except ImportError as e:
raise NvidiaRivaException(status_code=500, message=_RIVA_INSTALL_HINT) from e
recognition_config: Final = self._build_recognition_config_proto(
riva_asr_module=riva_asr_module,
recognition_config_dict=recognition_config_dict,
)
recognition_config: Final = riva_grpc.recognition_config(recognition_config_dict)
logging_obj.pre_call(
input=None,
api_key=api_key,
@ -344,26 +268,18 @@ class NvidiaRivaAudioTranscription:
},
)
metadata: Final = riva_grpc.call_metadata(api_key, optional_params.get("nvcf_function_id"))
try:
asr_service: Final[_AsrService] = riva_module.ASRService(auth_obj)
responses: Final = (
self._offline_responses(
asr_service=asr_service,
pcm_bytes=resampled.pcm_bytes,
recognition_config=recognition_config,
timeout=None if timeout is None else _deadline_seconds(timeout, drop_params),
deadline: Final = None if timeout is None else _deadline_seconds(timeout, drop_params)
with closing(riva_grpc.open_channel(api_base, _use_ssl(optional_params))) as channel:
responses: Final = (
riva_grpc.recognize(channel, recognition_config, resampled.pcm_bytes, metadata, deadline)
if riva_offline
else riva_grpc.streaming_recognize(
channel, recognition_config, resampled.pcm_bytes, metadata, deadline
)
)
if riva_offline
else self._streaming_responses(
asr_service=asr_service,
riva_asr_module=riva_asr_module,
pcm_bytes=resampled.pcm_bytes,
recognition_config=recognition_config,
timeout=timeout,
drop_params=drop_params,
)
)
final_results: Final = self._collect_final_results(responses, require_final=not riva_offline)
final_results: Final = self._collect_final_results(responses, require_final=not riva_offline)
except NvidiaRivaException:
raise
except Exception as e:
@ -400,142 +316,6 @@ class NvidiaRivaAudioTranscription:
return final_response
def _construct_auth(
self,
riva_module: _RivaClientModule,
api_base: str,
api_key: str | None,
optional_params: dict,
) -> _RivaAuth:
"""
Build a ``riva.client.Auth`` object.
- When ``nvcf_function_id`` is provided we attach the NVCF
``function-id`` and bearer ``authorization`` metadata, and default
``use_ssl`` to True (NVCF endpoints are TLS-only).
- Otherwise (self-hosted) we default ``use_ssl`` to False but still
honor an explicit override — self-hosted Riva behind an ingress
with TLS termination is a real deployment topology.
"""
nvcf_function_id: Final = optional_params.get("nvcf_function_id")
use_ssl_override: Final = optional_params.get("use_ssl")
use_ssl: Final = bool(use_ssl_override) if use_ssl_override is not None else bool(nvcf_function_id)
metadata: Final[list[tuple[str, str]]] = []
if nvcf_function_id:
metadata.append(("function-id", str(nvcf_function_id)))
if api_key:
metadata.append(("authorization", f"Bearer {api_key}"))
try:
return _auth_factory(riva_module)(uri=api_base, use_ssl=use_ssl, metadata_args=metadata)
except TypeError:
# Older riva-client signatures used positional-only args.
return _auth_factory(riva_module)(None, use_ssl, api_base, metadata)
def _build_recognition_config_proto(
self, riva_asr_module: _RivaAsrModule, recognition_config_dict: dict[str, Any]
) -> _RecognitionConfig:
encoding_name: Final = (recognition_config_dict.get("encoding") or "LINEAR_PCM").upper()
encoding_enum: Final[object] = getattr(
_audio_encoding(riva_asr_module),
encoding_name,
_audio_encoding(riva_asr_module).LINEAR_PCM,
)
config: Final[_RecognitionConfig] = riva_asr_module.RecognitionConfig(
encoding=encoding_enum,
sample_rate_hertz=int(recognition_config_dict["sample_rate_hertz"]),
language_code=recognition_config_dict["language_code"],
audio_channel_count=int(recognition_config_dict["audio_channel_count"]),
enable_automatic_punctuation=bool(recognition_config_dict.get("enable_automatic_punctuation", True)),
enable_word_time_offsets=bool(recognition_config_dict.get("enable_word_time_offsets", False)),
max_alternatives=int(recognition_config_dict.get("max_alternatives", 1)),
model=recognition_config_dict.get("model", "") or "",
verbatim_transcripts=bool(recognition_config_dict.get("verbatim_transcripts", False)),
profanity_filter=bool(recognition_config_dict.get("profanity_filter", False)),
)
endpointing: Final = recognition_config_dict.get("endpointing_config")
if isinstance(endpointing, dict) and endpointing:
try:
ep: Final[_EndpointingConfig] = riva_asr_module.EndpointingConfig(**endpointing)
config.endpointing_config.CopyFrom(ep)
except Exception:
# If the user supplied an unknown EndpointingConfig field
# (older Riva server), fall back to Riva's defaults rather
# than failing the whole request.
pass
return config
def _streaming_responses(
self,
asr_service: _AsrService,
riva_asr_module: _RivaAsrModule,
pcm_bytes: bytes,
recognition_config: _RecognitionConfig,
timeout: float | str | httpx.Timeout | None,
drop_params: bool,
) -> Iterable[object]:
streaming_config: Final[_StreamingRecognitionConfig] = riva_asr_module.StreamingRecognitionConfig(
config=recognition_config, interim_results=False
)
return asr_service.streaming_response_generator(
audio_chunks=self._iter_audio_chunks(pcm_bytes),
streaming_config=streaming_config,
**self._timeout_kwargs(asr_service.streaming_response_generator, timeout, drop_params),
)
def _offline_responses(
self,
asr_service: _AsrService,
pcm_bytes: bytes,
recognition_config: _RecognitionConfig,
timeout: float | None,
) -> Iterable[object]:
pending: Final = asr_service.offline_recognize(audio_bytes=pcm_bytes, config=recognition_config, future=True)
try:
return (pending.result(timeout=timeout),)
except Exception as e:
if pending.done():
raise
pending.cancel()
raise NvidiaRivaException(
status_code=408,
message=f"NVIDIA Riva offline recognition did not finish within {timeout} seconds",
) from e
@classmethod
def _timeout_kwargs(
cls, rpc: Callable[..., object], timeout: float | str | httpx.Timeout | None, drop_params: bool
) -> Mapping[str, float]:
# Forward the deadline so the call cannot block forever if the
# server stalls. Older riva-client versions do not accept a
# ``timeout`` kwarg, so pass it only when supported.
if timeout is None or not cls._supports_timeout_kwarg(rpc):
return MappingProxyType({})
return MappingProxyType({"timeout": _deadline_seconds(timeout, drop_params)})
@staticmethod
def _supports_timeout_kwarg(callable_obj: Callable[..., object]) -> bool:
try:
sig: Final = inspect.signature(callable_obj)
except (TypeError, ValueError):
return False
params: Final = sig.parameters
if "timeout" in params:
return True
return any(p.kind == inspect.Parameter.VAR_KEYWORD for p in params.values())
@staticmethod
def _iter_audio_chunks(pcm_bytes: bytes):
for offset in range(0, len(pcm_bytes), _DEFAULT_CHUNK_BYTES):
chunk = pcm_bytes[offset : offset + _DEFAULT_CHUNK_BYTES]
if not chunk:
continue
yield chunk
@staticmethod
def _collect_final_results(stream: Iterable[object], require_final: bool = True) -> Sequence[RivaFinalResult]:
"""
@ -570,27 +350,3 @@ class NvidiaRivaAudioTranscription:
}
)
return final_results
def _import_riva() -> tuple[_RivaClientModule, _RivaAsrModule]:
"""
Lazy import of ``riva.client`` and ``riva.client.proto.riva_asr_pb2``.
We try the SDK first (preferred) and fall back to importing the proto
module separately when the SDK packaging changes between versions.
"""
try:
import riva.client as riva_client
except ImportError as e:
raise NvidiaRivaException(status_code=500, message=_RIVA_INSTALL_HINT) from e
riva_asr_module = riva_client
if not hasattr(riva_asr_module, "RecognitionConfig"):
try:
from riva.client.proto import riva_asr_pb2
riva_asr_module = riva_asr_pb2
except ImportError as e:
raise NvidiaRivaException(status_code=500, message=_RIVA_INSTALL_HINT) from e
return riva_client, riva_asr_module

View file

@ -0,0 +1,199 @@
"""
The part of NVIDIA Riva's speech recognition gRPC schema that LiteLLM sends and reads.
Names and numbers match ``riva/proto/riva_asr.proto`` and ``riva/proto/riva_audio.proto`` in
https://github.com/nvidia-riva/common. Fields LiteLLM never sets or reads are left out, and the
wire format skips them. The messages live in their own descriptor pool so they never collide
with nvidia-riva-client's copies when both are imported in one process.
"""
from dataclasses import dataclass
from enum import Enum, IntEnum
from typing import Final
from google.protobuf import descriptor_pb2, descriptor_pool, message_factory
from google.protobuf.message import Message
_PACKAGE: Final = "nvidia.riva.asr"
_SERVICE: Final = f"/{_PACKAGE}.RivaSpeechRecognition"
RECOGNIZE_METHOD: Final = f"{_SERVICE}/Recognize"
STREAMING_RECOGNIZE_METHOD: Final = f"{_SERVICE}/StreamingRecognize"
class _Kind(Enum):
FLOAT = "TYPE_FLOAT"
INT32 = "TYPE_INT32"
BOOL = "TYPE_BOOL"
STRING = "TYPE_STRING"
MESSAGE = "TYPE_MESSAGE"
BYTES = "TYPE_BYTES"
ENUM = "TYPE_ENUM"
class AudioEncoding(IntEnum):
ENCODING_UNSPECIFIED = 0
LINEAR_PCM = 1
LINEAR_PCM: Final = AudioEncoding.LINEAR_PCM.value
@dataclass(frozen=True, slots=True)
class _Field:
name: str
number: int
kind: _Kind
type_name: str = ""
repeated: bool = False
optional: bool = False
oneof: str = ""
_MESSAGES: Final[tuple[tuple[str, tuple[_Field, ...]], ...]] = (
(
"RecognitionConfig",
(
_Field("encoding", 1, _Kind.ENUM, "AudioEncoding"),
_Field("sample_rate_hertz", 2, _Kind.INT32),
_Field("language_code", 3, _Kind.STRING),
_Field("max_alternatives", 4, _Kind.INT32),
_Field("profanity_filter", 5, _Kind.BOOL),
_Field("audio_channel_count", 7, _Kind.INT32),
_Field("enable_word_time_offsets", 8, _Kind.BOOL),
_Field("enable_automatic_punctuation", 11, _Kind.BOOL),
_Field("model", 13, _Kind.STRING),
_Field("verbatim_transcripts", 14, _Kind.BOOL),
_Field("endpointing_config", 25, _Kind.MESSAGE, "EndpointingConfig", optional=True),
),
),
(
"EndpointingConfig",
(
_Field("start_history", 1, _Kind.INT32, optional=True),
_Field("start_threshold", 2, _Kind.FLOAT, optional=True),
_Field("stop_history", 3, _Kind.INT32, optional=True),
_Field("stop_threshold", 4, _Kind.FLOAT, optional=True),
_Field("stop_history_eou", 5, _Kind.INT32, optional=True),
_Field("stop_threshold_eou", 6, _Kind.FLOAT, optional=True),
),
),
(
"StreamingRecognitionConfig",
(
_Field("config", 1, _Kind.MESSAGE, "RecognitionConfig"),
_Field("interim_results", 2, _Kind.BOOL),
),
),
(
"StreamingRecognizeRequest",
(
_Field("streaming_config", 1, _Kind.MESSAGE, "StreamingRecognitionConfig", oneof="streaming_request"),
_Field("audio_content", 2, _Kind.BYTES, oneof="streaming_request"),
),
),
(
"WordInfo",
(
_Field("start_time", 1, _Kind.INT32),
_Field("end_time", 2, _Kind.INT32),
_Field("word", 3, _Kind.STRING),
),
),
(
"SpeechRecognitionAlternative",
(
_Field("transcript", 1, _Kind.STRING),
_Field("words", 3, _Kind.MESSAGE, "WordInfo", repeated=True),
),
),
(
"StreamingRecognitionResult",
(
_Field("alternatives", 1, _Kind.MESSAGE, "SpeechRecognitionAlternative", repeated=True),
_Field("is_final", 2, _Kind.BOOL),
),
),
(
"StreamingRecognizeResponse",
(_Field("results", 1, _Kind.MESSAGE, "StreamingRecognitionResult", repeated=True),),
),
(
"RecognizeRequest",
(
_Field("config", 1, _Kind.MESSAGE, "RecognitionConfig"),
_Field("audio", 2, _Kind.BYTES),
),
),
(
"SpeechRecognitionResult",
(_Field("alternatives", 1, _Kind.MESSAGE, "SpeechRecognitionAlternative", repeated=True),),
),
(
"RecognizeResponse",
(_Field("results", 1, _Kind.MESSAGE, "SpeechRecognitionResult", repeated=True),),
),
)
def _oneof_names(fields: tuple[_Field, ...]) -> tuple[str, ...]:
declared: Final = tuple(dict.fromkeys(field.oneof for field in fields if field.oneof))
synthetic: Final = tuple(f"_{field.name}" for field in fields if field.optional)
return declared + synthetic
def _field_proto(field: _Field, oneofs: tuple[str, ...]) -> descriptor_pb2.FieldDescriptorProto:
oneof: Final = f"_{field.name}" if field.optional else field.oneof
return descriptor_pb2.FieldDescriptorProto(
name=field.name,
number=field.number,
type=descriptor_pb2.FieldDescriptorProto.Type.Value(field.kind.value),
label=(
descriptor_pb2.FieldDescriptorProto.LABEL_REPEATED
if field.repeated
else descriptor_pb2.FieldDescriptorProto.LABEL_OPTIONAL
),
type_name=f".{_PACKAGE}.{field.type_name}" if field.type_name else None,
oneof_index=oneofs.index(oneof) if oneof else None,
proto3_optional=field.optional or None,
)
def _message_proto(name: str, fields: tuple[_Field, ...]) -> descriptor_pb2.DescriptorProto:
oneofs: Final = _oneof_names(fields)
return descriptor_pb2.DescriptorProto(
name=name,
field=[_field_proto(field, oneofs) for field in fields],
oneof_decl=[descriptor_pb2.OneofDescriptorProto(name=oneof) for oneof in oneofs],
)
_FILE: Final = descriptor_pb2.FileDescriptorProto(
name="litellm/nvidia_riva/riva_asr.proto",
package=_PACKAGE,
syntax="proto3",
enum_type=[
descriptor_pb2.EnumDescriptorProto(
name="AudioEncoding",
value=[
descriptor_pb2.EnumValueDescriptorProto(name=encoding.name, number=encoding.value)
for encoding in AudioEncoding
],
)
],
message_type=[_message_proto(name, fields) for name, fields in _MESSAGES],
)
_POOL: Final = descriptor_pool.DescriptorPool()
_POOL.AddSerializedFile(_FILE.SerializeToString())
def _message_class(name: str) -> type[Message]:
return message_factory.GetMessageClass(_POOL.FindMessageTypeByName(f"{_PACKAGE}.{name}"))
RecognitionConfig: Final = _message_class("RecognitionConfig")
EndpointingConfig: Final = _message_class("EndpointingConfig")
StreamingRecognitionConfig: Final = _message_class("StreamingRecognitionConfig")
StreamingRecognizeRequest: Final = _message_class("StreamingRecognizeRequest")
StreamingRecognizeResponse: Final = _message_class("StreamingRecognizeResponse")
RecognizeRequest: Final = _message_class("RecognizeRequest")
RecognizeResponse: Final = _message_class("RecognizeResponse")

View file

@ -0,0 +1,113 @@
"""
gRPC calls to NVIDIA Riva's speech recognition service.
The ``authorization`` and NVCF ``function-id`` metadata ride on each call, never on the channel, so every
request carries them exactly once over both plaintext and TLS.
"""
from collections.abc import Iterator, Mapping, Sequence
from typing import Any, Final, TypeAlias
import grpc
from google.protobuf.message import Message
from pydantic import TypeAdapter
from litellm.llms.nvidia_riva.audio_transcription import riva_asr_schema as schema
from litellm.llms.nvidia_riva.common_utils import NvidiaRivaException
# Stream audio to Riva in ~50 ms slices (1600 samples at 16 kHz). Matches
# NVIDIA's recommended chunk size for streaming ASR, small enough for
# responsive endpointing, large enough to keep per-RPC overhead low.
_CHUNK_SAMPLES: Final = 1600
_CHUNK_BYTES: Final = _CHUNK_SAMPLES * 2
CallMetadata: TypeAlias = tuple[tuple[str, str], ...]
_ENDPOINTING_FIELDS: Final = TypeAdapter(Mapping[str, object])
def call_metadata(api_key: str | None, nvcf_function_id: object) -> CallMetadata:
function_id: Final = (("function-id", str(nvcf_function_id)),) if nvcf_function_id else ()
authorization: Final = (("authorization", f"Bearer {api_key}"),) if api_key else ()
return function_id + authorization
def open_channel(api_base: str, use_ssl: bool) -> grpc.Channel:
if use_ssl:
return grpc.secure_channel(api_base, grpc.ssl_channel_credentials())
return grpc.insecure_channel(api_base)
def recognition_config(config: Mapping[str, Any]) -> Message:
endpointing: Final = _endpointing_config(config.get("endpointing_config"))
return schema.RecognitionConfig(
encoding=schema.LINEAR_PCM,
sample_rate_hertz=int(config["sample_rate_hertz"]),
language_code=config["language_code"],
audio_channel_count=int(config["audio_channel_count"]),
enable_automatic_punctuation=bool(config.get("enable_automatic_punctuation", True)),
enable_word_time_offsets=bool(config.get("enable_word_time_offsets", False)),
max_alternatives=int(config.get("max_alternatives", 1)),
model=config.get("model", "") or "",
verbatim_transcripts=bool(config.get("verbatim_transcripts", False)),
profanity_filter=bool(config.get("profanity_filter", False)),
**({} if endpointing is None else {"endpointing_config": endpointing}),
)
def _endpointing_config(endpointing: object) -> Message | None:
try:
fields: Final = _ENDPOINTING_FIELDS.validate_python(endpointing)
return schema.EndpointingConfig(**fields) if fields else None
except (ValueError, TypeError):
# A missing, unknown or mistyped field falls back to Riva's default
# endpointing rather than failing the whole request.
return None
def _streaming_requests(config: Message, pcm_bytes: bytes) -> Iterator[Message]:
yield schema.StreamingRecognizeRequest(
streaming_config=schema.StreamingRecognitionConfig(config=config, interim_results=False)
)
for offset in range(0, len(pcm_bytes), _CHUNK_BYTES):
yield schema.StreamingRecognizeRequest(audio_content=pcm_bytes[offset : offset + _CHUNK_BYTES])
def streaming_recognize(
channel: grpc.Channel,
config: Message,
pcm_bytes: bytes,
metadata: CallMetadata,
timeout: float | None,
) -> Iterator[Message]:
rpc: Final = channel.stream_stream(
schema.STREAMING_RECOGNIZE_METHOD,
request_serializer=schema.StreamingRecognizeRequest.SerializeToString,
response_deserializer=schema.StreamingRecognizeResponse.FromString,
)
return rpc(_streaming_requests(config, pcm_bytes), metadata=metadata, timeout=timeout)
def recognize(
channel: grpc.Channel,
config: Message,
pcm_bytes: bytes,
metadata: CallMetadata,
timeout: float | None,
) -> Sequence[Message]:
rpc: Final = channel.unary_unary(
schema.RECOGNIZE_METHOD,
request_serializer=schema.RecognizeRequest.SerializeToString,
response_deserializer=schema.RecognizeResponse.FromString,
)
pending: Final = rpc.future(schema.RecognizeRequest(config=config, audio=pcm_bytes), metadata=metadata)
try:
return (pending.result(timeout=timeout),)
except Exception as e:
if pending.done():
raise
pending.cancel()
raise NvidiaRivaException(
status_code=408,
message=f"NVIDIA Riva offline recognition did not finish within {timeout} seconds",
) from e

View file

@ -155,7 +155,8 @@ stt-vertex-chirp = [
stt-nvidia-riva = [
# NVIDIA Riva STT provider (gRPC). These are imported lazily inside the
# provider handler so litellm core remains usable without them.
"nvidia-riva-client>=2.15.0",
"grpcio>=1.67.1,<2.0",
"protobuf>=5.26.1,<7.0",
"soundfile>=0.12.1",
"audioread>=3.0.1",
"numpy>=1.26.0",
@ -241,6 +242,7 @@ dev = [
"pytest-timeout==2.4.0",
"vcrpy==8.2.1",
"pytest-recording==0.13.4",
"nvidia-riva-client==2.16.0",
]
e2e-dev = [
"playwright==1.61.0",
@ -275,6 +277,9 @@ ci = [
# protobuf, Pillow is a compiled C extension).
"tenacity==8.5.0",
"google-generativeai==0.8.6",
# The Riva tests serve NVIDIA's own message definitions, which checks
# LiteLLM's hand-written copy of them on every run.
"nvidia-riva-client==2.16.0",
"Pillow==12.3.0; python_version == '3.10'",
"Pillow==13.0.0; python_version >= '3.11'",
# Azure batch E2E tests still import psycopg2 directly.

View file

@ -39,6 +39,7 @@ class RivaCall:
rpc: Rpc
audio: bytes
cancelled: bool
authorization: tuple[str, ...] = ()
@dataclass(frozen=True, slots=True)
@ -68,6 +69,10 @@ def pcm_bytes(seconds: float) -> int:
return int(seconds * SAMPLE_RATE_HZ) * 2
def _authorization(context: grpc.ServicerContext) -> tuple[str, ...]:
return tuple(str(value) for key, value in context.invocation_metadata() if key == "authorization")
def carries_clip(audio: bytes, seconds: float) -> bool:
sent: Final = _clip_samples(seconds)
if len(audio) != len(sent) * 2:
@ -100,9 +105,9 @@ class _Asr(riva_asr_pb2_grpc.RivaSpeechRecognitionServicer):
released: Final = threading.Event()
context.add_callback(released.set)
released.wait(timeout=_STALL_LIMIT_SECONDS)
self._calls.put(RivaCall(Rpc.RECOGNIZE, request.audio, cancelled=not context.is_active()))
self._calls.put(RivaCall(Rpc.RECOGNIZE, request.audio, not context.is_active(), _authorization(context)))
return riva_asr_pb2.RecognizeResponse()
self._calls.put(RivaCall(Rpc.RECOGNIZE, request.audio, cancelled=False))
self._calls.put(RivaCall(Rpc.RECOGNIZE, request.audio, False, _authorization(context)))
return riva_asr_pb2.RecognizeResponse(
results=[riva_asr_pb2.SpeechRecognitionResult(alternatives=[self._alternative(request.config)])]
)
@ -111,13 +116,12 @@ class _Asr(riva_asr_pb2_grpc.RivaSpeechRecognitionServicer):
self, request_iterator: Iterable[riva_asr_pb2.StreamingRecognizeRequest], context: grpc.ServicerContext
) -> Iterator[riva_asr_pb2.StreamingRecognizeResponse]:
if self._offline_only:
self._calls.put(RivaCall(Rpc.STREAMING, b"", cancelled=False))
self._calls.put(RivaCall(Rpc.STREAMING, b"", False, _authorization(context)))
context.abort(grpc.StatusCode.INVALID_ARGUMENT, OFFLINE_ONLY_REFUSAL)
requests: Final = tuple(request_iterator)
config: Final = requests[0].streaming_config.config if requests else riva_asr_pb2.RecognitionConfig()
self._calls.put(
RivaCall(Rpc.STREAMING, b"".join(request.audio_content for request in requests), cancelled=False)
)
audio: Final = b"".join(request.audio_content for request in requests)
self._calls.put(RivaCall(Rpc.STREAMING, audio, False, _authorization(context)))
yield riva_asr_pb2.StreamingRecognizeResponse(
results=[riva_asr_pb2.StreamingRecognitionResult(alternatives=[self._alternative(config)], is_final=True)]
)

View file

@ -135,3 +135,12 @@ def test_a_stalled_offline_recognize_answers_408_at_the_deployment_timeout(gatew
assert timed_out.status_code == 408, timed_out.text
calls: Final = eventually(server.drain, lambda observed: len(observed) > 0, seconds=30)
assert all(call.rpc is Rpc.RECOGNIZE and call.cancelled for call in calls), calls
@pytest.mark.parametrize("riva_offline", (False, True), ids=("streaming", "offline"))
def test_the_deployment_api_key_reaches_riva_as_exactly_one_bearer_token(gateway: Gateway, riva_offline: bool) -> None:
with riva_server(_WORDS) as server, gateway.scenario() as scenario:
model: Final = _riva_model(scenario, server, api_key="riva-deployment-key", riva_offline=riva_offline)
served: Final = _transcribe(gateway, scenario.key(), model)
assert served.status_code == 200, served.text
assert [call.authorization for call in server.drain()] == [("Bearer riva-deployment-key",)]

File diff suppressed because it is too large Load diff

View file

@ -0,0 +1,41 @@
from collections.abc import Iterator
from typing import Final
from google.protobuf.descriptor import FieldDescriptor
from riva.client.proto import riva_asr_pb2, riva_audio_pb2
from litellm.llms.nvidia_riva.audio_transcription import riva_asr_schema
def _declared_fields() -> Iterator[tuple[str, FieldDescriptor]]:
for message_name, _ in riva_asr_schema._MESSAGES:
message: Final = riva_asr_schema._POOL.FindMessageTypeByName(f"{riva_asr_schema._PACKAGE}.{message_name}")
yield from ((message_name, field) for field in message.fields)
def _wire_shape(field: FieldDescriptor | None) -> tuple[object, ...] | None:
if field is None:
return None
return (
field.number,
field.type,
field.label,
field.has_presence,
field.message_type.full_name if field.message_type else None,
field.enum_type.name if field.enum_type else None,
field.containing_oneof.name if field.containing_oneof else None,
)
def _nvidia_field(message_name: str, field_name: str) -> FieldDescriptor | None:
return riva_asr_pb2.DESCRIPTOR.message_types_by_name[message_name].fields_by_name.get(field_name)
def test_every_declared_field_matches_nvidias_riva_schema() -> None:
declared: Final = tuple(_declared_fields())
assert len(declared) == sum(len(fields) for _, fields in riva_asr_schema._MESSAGES)
assert {f"{message}.{field.name}": _wire_shape(field) for message, field in declared} == {
f"{message}.{field.name}": _wire_shape(_nvidia_field(message, field.name)) for message, field in declared
}
assert riva_asr_schema.LINEAR_PCM == riva_audio_pb2.AudioEncoding.LINEAR_PCM

10
uv.lock generated
View file

@ -5134,9 +5134,10 @@ semantic-router = [
]
stt-nvidia-riva = [
{ name = "audioread" },
{ name = "grpcio" },
{ name = "numpy", version = "1.26.4", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.12'" },
{ name = "numpy", version = "2.4.4", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.12'" },
{ name = "nvidia-riva-client" },
{ name = "protobuf" },
{ name = "soundfile" },
]
stt-vertex-chirp = [
@ -5176,6 +5177,7 @@ ci = [
{ name = "logfire" },
{ name = "lunary", version = "1.4.36", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.11'" },
{ name = "lunary", version = "1.4.37", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.11'" },
{ name = "nvidia-riva-client" },
{ name = "pillow", version = "12.3.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.11'" },
{ name = "pillow", version = "13.0.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.11'" },
{ name = "psutil" },
@ -5200,6 +5202,7 @@ dev = [
{ name = "mypy" },
{ name = "numpy", version = "1.26.4", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.12'" },
{ name = "numpy", version = "2.4.4", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.12'" },
{ name = "nvidia-riva-client" },
{ name = "openapi-core" },
{ name = "opentelemetry-api" },
{ name = "opentelemetry-exporter-otlp" },
@ -5300,6 +5303,7 @@ requires-dist = [
{ name = "granian", marker = "python_full_version < '3.15' and extra == 'proxy'", specifier = ">=2.7.4,<3.0" },
{ name = "grpcio", marker = "extra == 'grpc'", specifier = "==1.84.1" },
{ name = "grpcio", marker = "extra == 'proxy-runtime'", specifier = "==1.84.1" },
{ name = "grpcio", marker = "extra == 'stt-nvidia-riva'", specifier = ">=1.67.1,<2.0" },
{ name = "gunicorn", marker = "extra == 'proxy'", specifier = ">=23.0.0,<24.0" },
{ name = "hiredis", marker = "python_full_version >= '3.15' and extra == 'proxy'", specifier = ">=3.5.0,<4.0" },
{ name = "hiredis", marker = "python_full_version < '3.15' and extra == 'proxy'", specifier = ">=3.0.0,<4.0" },
@ -5323,7 +5327,6 @@ requires-dist = [
{ name = "mlflow", marker = "extra == 'mlflow'", specifier = ">=3.17.2,<4.0" },
{ name = "numpy", marker = "extra == 'stt-nvidia-riva'", specifier = ">=1.26.0" },
{ name = "numpydoc", marker = "extra == 'utils'", specifier = ">=1.8.0,<2.0" },
{ name = "nvidia-riva-client", marker = "extra == 'stt-nvidia-riva'", specifier = ">=2.15.0" },
{ name = "openai", specifier = ">=2.20.0,<4.0.0" },
{ name = "opentelemetry-api", marker = "extra == 'proxy-runtime'", specifier = "==1.45.1" },
{ name = "opentelemetry-exporter-otlp", marker = "extra == 'proxy-runtime'", specifier = "==1.45.1" },
@ -5335,6 +5338,7 @@ requires-dist = [
{ name = "polars", marker = "extra == 'proxy'", specifier = ">=1.38.1,<2.0" },
{ name = "prisma", marker = "extra == 'extra-proxy'", specifier = ">=0.11.0,<1.0" },
{ name = "prometheus-client", marker = "extra == 'proxy-runtime'", specifier = ">=0.20.0,<1.0" },
{ name = "protobuf", marker = "extra == 'stt-nvidia-riva'", specifier = ">=5.26.1,<7.0" },
{ name = "psycopg", marker = "extra == 'extra-proxy'", specifier = ">=3.2,<4.0" },
{ name = "psycopg-binary", marker = "extra == 'extra-proxy'", specifier = ">=3.2,<4.0" },
{ name = "pydantic", marker = "python_full_version < '3.14'", specifier = ">=2.11.0,<3.0.0" },
@ -5406,6 +5410,7 @@ ci = [
{ name = "logfire", specifier = "==5.2.0" },
{ name = "lunary", marker = "python_full_version == '3.10.*'", specifier = "==1.4.36" },
{ name = "lunary", marker = "python_full_version >= '3.11'", specifier = "==1.4.37" },
{ name = "nvidia-riva-client", specifier = "==2.16.0" },
{ name = "pillow", marker = "python_full_version == '3.10.*'", specifier = "==12.3.0" },
{ name = "pillow", marker = "python_full_version >= '3.11'", specifier = "==13.0.0" },
{ name = "psutil", specifier = "==7.2.2" },
@ -5429,6 +5434,7 @@ dev = [
{ name = "langfuse", specifier = ">=4.7,<5.0" },
{ name = "mypy", specifier = "==1.20.1" },
{ name = "numpy", specifier = ">=1.26.0,<3.0" },
{ name = "nvidia-riva-client", specifier = "==2.16.0" },
{ name = "openapi-core", specifier = "==0.22.0" },
{ name = "opentelemetry-api", specifier = "==1.45.1" },
{ name = "opentelemetry-exporter-otlp", specifier = "==1.45.1" },