feat(voice): add Yandex streaming ASR

2026-04-17 01:06:45 +05:00
parent 5d40cfafd8
commit f66b289371
9 changed files with 516 additions and 2 deletions
@@ -78,6 +78,7 @@ AI_VOICE_AUDIOSOCKET_PORT=9019
 AI_VOICE_ASR_MODEL=gpt-4o-mini-transcribe
 AI_VOICE_ASR_YANDEX_API_BASE=https://stt.api.ml.yandexcloud.kz
 AI_VOICE_ASR_YANDEX_API_VERSION=auto
+AI_VOICE_ASR_YANDEX_GRPC_TARGET=stt.api.ml.yandexcloud.kz:443
 AI_VOICE_ASR_YANDEX_API_KEY=
 AI_VOICE_ASR_YANDEX_IAM_TOKEN=
 AI_VOICE_ASR_YANDEX_FOLDER_ID=
@@ -86,6 +87,8 @@ AI_VOICE_ASR_YANDEX_TOPIC=general
 AI_VOICE_ASR_YANDEX_SAMPLE_RATE_HZ=8000
 AI_VOICE_ASR_YANDEX_TIMEOUT_SECONDS=8
 AI_VOICE_ASR_YANDEX_POLL_INTERVAL_SECONDS=0.25
+AI_VOICE_ASR_YANDEX_EOU_MAX_PAUSE_MS=450
+AI_VOICE_ASR_YANDEX_STREAM_QUEUE_MAX_CHUNKS=300
 AI_VOICE_TTS_MODEL=gpt-4o-mini-tts
 AI_VOICE_TTS_VOICE=alloy
 AI_VOICE_TTS_YANDEX_API_BASE=https://tts.api.ml.yandexcloud.kz
@@ -27,6 +27,7 @@ x-app-env: &app_env
  AI_VOICE_V2_STREAMING_ASR_BACKEND: yandex_speechkit
  AI_VOICE_V2_STREAMING_ASR_BASE_URL: http://streaming-asr-sidecar:8021
  AI_VOICE_V2_STREAMING_ASR_TIMEOUT_SECONDS: "4"
+  AI_VOICE_V2_STREAMING_ASR_PARTIAL_POLL_ENABLED: "1"

 x-service-defaults: &service_defaults
  build:
@@ -27,6 +27,7 @@ x-app-env: &app_env
  AI_VOICE_V2_STREAMING_ASR_BACKEND: yandex_speechkit
  AI_VOICE_V2_STREAMING_ASR_BASE_URL: http://streaming-asr-sidecar:8021
  AI_VOICE_V2_STREAMING_ASR_TIMEOUT_SECONDS: "4"
+  AI_VOICE_V2_STREAMING_ASR_PARTIAL_POLL_ENABLED: "1"

 x-service-defaults: &service_defaults
  image: ${APP_IMAGE:?Set APP_IMAGE in deployment/.env.images or shell env}
@@ -27,6 +27,7 @@ x-app-env: &app_env
  AI_VOICE_V2_STREAMING_ASR_BACKEND: yandex_speechkit
  AI_VOICE_V2_STREAMING_ASR_BASE_URL: http://streaming-asr-sidecar:8021
  AI_VOICE_V2_STREAMING_ASR_TIMEOUT_SECONDS: "4"
+  AI_VOICE_V2_STREAMING_ASR_PARTIAL_POLL_ENABLED: "1"

 x-service-defaults: &service_defaults
  build:
@@ -38,6 +38,7 @@ x-app-env: &app_env
  AI_VOICE_V2_STREAMING_ASR_BACKEND: yandex_speechkit
  AI_VOICE_V2_STREAMING_ASR_BASE_URL: http://streaming-asr-sidecar:8021
  AI_VOICE_V2_STREAMING_ASR_TIMEOUT_SECONDS: "4"
+  AI_VOICE_V2_STREAMING_ASR_PARTIAL_POLL_ENABLED: "1"
  AI_VOICE_V2_PREBAKED_ACK_ENABLED: "1"
  AI_VOICE_V2_PREBAKED_ACK_DIR: /app/.data/voice_v2_ack_bank
  AI_VOICE_V2_STREAMING_TTS: "1"
@@ -9,4 +9,8 @@ pika==1.3.2
 paramiko==3.5.0
 python-multipart==0.0.20
 faster-whisper==1.1.1
+grpcio==1.80.0
+grpcio-tools==1.80.0
+protobuf==6.33.6
 requests==2.32.5
+yandexcloud==0.386.0
@@ -319,6 +319,7 @@ def spawn_service(spec: dict[str, Any], runtime_dir: Path, data_dir: Path, base_
    env["AI_VOICE_V2_STREAMING_ASR_BACKEND"] = env.get("AI_VOICE_V2_STREAMING_ASR_BACKEND", "yandex_speechkit")
    env["AI_VOICE_V2_STREAMING_ASR_BASE_URL"] = env.get("AI_VOICE_V2_STREAMING_ASR_BASE_URL", "http://127.0.0.1:8021")
    env["AI_VOICE_V2_STREAMING_ASR_TIMEOUT_SECONDS"] = env.get("AI_VOICE_V2_STREAMING_ASR_TIMEOUT_SECONDS", "4")
+    env["AI_VOICE_V2_STREAMING_ASR_PARTIAL_POLL_ENABLED"] = env.get("AI_VOICE_V2_STREAMING_ASR_PARTIAL_POLL_ENABLED", "1")
    env["AI_VOICE_V2_STREAMING_ASR_MODEL"] = env.get("AI_VOICE_V2_STREAMING_ASR_MODEL", "base")
    env["AI_VOICE_V2_STREAMING_ASR_COMPUTE_TYPE"] = env.get("AI_VOICE_V2_STREAMING_ASR_COMPUTE_TYPE", "int8")
    env["AI_VOICE_V2_STREAMING_ASR_DEVICE"] = env.get("AI_VOICE_V2_STREAMING_ASR_DEVICE", "cpu")
@@ -3,11 +3,12 @@ from __future__ import annotations
 import base64
 import io
 import os
+import queue
 import threading
 import time
 import uuid
 import wave
-from dataclasses import dataclass
+from dataclasses import dataclass, field

 import httpx

@@ -112,6 +113,33 @@ def _yandex_asr_poll_interval_seconds() -> float:
    return max(min(value, 2.0), 0.05)


+def _yandex_asr_grpc_target(api_base: str) -> str:
+    explicit = os.getenv("AI_VOICE_ASR_YANDEX_GRPC_TARGET", "").strip()
+    if explicit:
+        return explicit
+    if "yandexcloud.kz" in str(api_base or "").lower():
+        return "stt.api.ml.yandexcloud.kz:443"
+    return "stt.api.cloud.yandex.net:443"
+
+
+def _yandex_asr_stream_queue_max_chunks() -> int:
+    raw = os.getenv("AI_VOICE_ASR_YANDEX_STREAM_QUEUE_MAX_CHUNKS", "300").strip()
+    try:
+        value = int(raw)
+    except ValueError:
+        value = 300
+    return max(min(value, 2000), 10)
+
+
+def _yandex_asr_eou_max_pause_ms() -> int:
+    raw = os.getenv("AI_VOICE_ASR_YANDEX_EOU_MAX_PAUSE_MS", "450").strip()
+    try:
+        value = int(raw)
+    except ValueError:
+        value = 450
+    return max(min(value, 3000), 100)
+
+
 def _yandex_asr_default_language() -> str:
    return os.getenv("AI_VOICE_ASR_YANDEX_LANGUAGE", "ru-RU").strip() or "ru-RU"

@@ -551,6 +579,353 @@ class LocalSidecarStreamingASRProvider(StreamingASRProvider):
            return


+@dataclass
+class _YandexGrpcStreamState:
+    stream_id: str
+    session_id: str
+    language: str
+    input_queue: queue.Queue[bytes | None]
+    updates_queue: queue.Queue[StreamingASRPartial]
+    thread: threading.Thread | None = None
+    latest_partial: StreamingASRPartial | None = None
+    final_transcription: ASRTranscription | None = None
+    error: BaseException | None = None
+    finish_requested: bool = False
+    lock: threading.Lock = field(default_factory=threading.Lock)
+
+
+class YandexSpeechKitGrpcStreamingASRProvider(StreamingASRProvider):
+    name = "yandex-speechkit-grpc"
+    supports_streaming = True
+
+    def __init__(
+        self,
+        *,
+        api_base: str | None = None,
+        api_key: str | None = None,
+        iam_token: str | None = None,
+        folder_id: str | None = None,
+        timeout_seconds: float | None = None,
+        sample_rate_hz: int | None = None,
+        topic: str | None = None,
+        grpc_target: str | None = None,
+        queue_max_chunks: int | None = None,
+        eou_max_pause_ms: int | None = None,
+        grpc_module: object | None = None,
+        stt_pb2_module: object | None = None,
+        stt_service_pb2_grpc_module: object | None = None,
+    ) -> None:
+        self._api_base = str(api_base or _yandex_asr_api_base()).strip().rstrip("/")
+        self._api_key = str(api_key if api_key is not None else _yandex_asr_api_key()).strip()
+        self._iam_token = str(iam_token if iam_token is not None else _yandex_asr_iam_token()).strip()
+        self._folder_id = str(folder_id if folder_id is not None else _yandex_asr_folder_id()).strip()
+        self._timeout_seconds = max(
+            float(timeout_seconds if timeout_seconds is not None else _yandex_asr_timeout_seconds()),
+            3.0,
+        )
+        self._sample_rate_hz = int(sample_rate_hz if sample_rate_hz is not None else _yandex_asr_sample_rate_hz())
+        self._topic = str(topic if topic is not None else _yandex_asr_topic()).strip() or "general"
+        self._grpc_target = str(grpc_target or _yandex_asr_grpc_target(self._api_base)).strip()
+        self._queue_max_chunks = int(queue_max_chunks or _yandex_asr_stream_queue_max_chunks())
+        self._eou_max_pause_ms = int(eou_max_pause_ms or _yandex_asr_eou_max_pause_ms())
+        self._grpc_module = grpc_module
+        self._stt_pb2_module = stt_pb2_module
+        self._stt_service_pb2_grpc_module = stt_service_pb2_grpc_module
+        self._streams: dict[str, _YandexGrpcStreamState] = {}
+        self._lock = threading.Lock()
+
+    def _load_grpc_modules(self) -> tuple[object, object, object]:
+        if (
+            self._grpc_module is not None
+            and self._stt_pb2_module is not None
+            and self._stt_service_pb2_grpc_module is not None
+        ):
+            return self._grpc_module, self._stt_pb2_module, self._stt_service_pb2_grpc_module
+        try:
+            import grpc  # type: ignore[import-not-found]
+            from yandex.cloud.ai.stt.v3 import stt_pb2  # type: ignore[import-not-found]
+            from yandex.cloud.ai.stt.v3 import stt_service_pb2_grpc  # type: ignore[import-not-found]
+        except ImportError as exc:
+            raise StreamingASRUnavailable("Yandex SpeechKit gRPC dependencies are not installed") from exc
+        return grpc, stt_pb2, stt_service_pb2_grpc
+
+    def _metadata(self) -> tuple[tuple[str, str], ...]:
+        if not self._api_key and not self._iam_token:
+            raise StreamingASRUnavailable(
+                "AI_VOICE_ASR_YANDEX_API_KEY or AI_VOICE_ASR_YANDEX_IAM_TOKEN is required for Yandex streaming ASR"
+            )
+        metadata: list[tuple[str, str]] = []
+        if self._api_key:
+            metadata.append(("authorization", f"Api-Key {self._api_key}"))
+        else:
+            metadata.append(("authorization", f"Bearer {self._iam_token}"))
+        if self._folder_id:
+            metadata.append(("x-folder-id", self._folder_id))
+        return tuple(metadata)
+
+    @staticmethod
+    def _enum_value(owner: object, name: str, default: int) -> int:
+        return int(getattr(owner, name, default))
+
+    def _build_session_options(self, stt_pb2: object, *, language: str) -> object:
+        raw_audio = stt_pb2.RawAudio(  # type: ignore[attr-defined]
+            audio_encoding=self._enum_value(stt_pb2.RawAudio, "LINEAR16_PCM", 1),  # type: ignore[attr-defined]
+            sample_rate_hertz=self._sample_rate_hz,
+            audio_channel_count=1,
+        )
+        recognition_model = stt_pb2.RecognitionModelOptions(  # type: ignore[attr-defined]
+            model=self._topic,
+            audio_format=stt_pb2.AudioFormatOptions(raw_audio=raw_audio),  # type: ignore[attr-defined]
+            text_normalization=stt_pb2.TextNormalizationOptions(  # type: ignore[attr-defined]
+                text_normalization=self._enum_value(
+                    stt_pb2.TextNormalizationOptions,  # type: ignore[attr-defined]
+                    "TEXT_NORMALIZATION_ENABLED",
+                    1,
+                ),
+                profanity_filter=False,
+                literature_text=False,
+                phone_formatting_mode=self._enum_value(
+                    stt_pb2.TextNormalizationOptions,  # type: ignore[attr-defined]
+                    "PHONE_FORMATTING_MODE_DISABLED",
+                    1,
+                ),
+            ),
+            language_restriction=stt_pb2.LanguageRestrictionOptions(  # type: ignore[attr-defined]
+                restriction_type=self._enum_value(stt_pb2.LanguageRestrictionOptions, "WHITELIST", 1),  # type: ignore[attr-defined]
+                language_code=[language],
+            ),
+            audio_processing_type=self._enum_value(stt_pb2.RecognitionModelOptions, "REAL_TIME", 1),  # type: ignore[attr-defined]
+        )
+        eou_classifier = stt_pb2.EouClassifierOptions(  # type: ignore[attr-defined]
+            default_classifier=stt_pb2.DefaultEouClassifier(  # type: ignore[attr-defined]
+                type=self._enum_value(stt_pb2.DefaultEouClassifier, "HIGH", 2),  # type: ignore[attr-defined]
+                max_pause_between_words_hint_ms=self._eou_max_pause_ms,
+            )
+        )
+        return stt_pb2.StreamingOptions(  # type: ignore[attr-defined]
+            recognition_model=recognition_model,
+            eou_classifier=eou_classifier,
+        )
+
+    def _request_iterator(self, state: _YandexGrpcStreamState, stt_pb2: object):
+        yield stt_pb2.StreamingRequest(  # type: ignore[attr-defined]
+            session_options=self._build_session_options(stt_pb2, language=state.language)
+        )
+        while True:
+            item = state.input_queue.get()
+            try:
+                if item is None:
+                    yield stt_pb2.StreamingRequest(eou=stt_pb2.Eou())  # type: ignore[attr-defined]
+                    return
+                if item:
+                    yield stt_pb2.StreamingRequest(chunk=stt_pb2.AudioChunk(data=item))  # type: ignore[attr-defined]
+            finally:
+                state.input_queue.task_done()
+
+    @staticmethod
+    def _has_field(message: object, field_name: str) -> bool:
+        try:
+            return bool(message.HasField(field_name))  # type: ignore[attr-defined]
+        except (AttributeError, ValueError):
+            return getattr(message, field_name, None) is not None
+
+    @classmethod
+    def _partial_from_response(cls, response: object) -> StreamingASRPartial | None:
+        for field_name, is_final in (
+            ("final_refinement", True),
+            ("final", True),
+            ("partial", False),
+        ):
+            if not cls._has_field(response, field_name):
+                continue
+            payload = getattr(response, field_name, None)
+            if field_name == "final_refinement" and payload is not None:
+                payload = getattr(payload, "normalized_text", None)
+            text, confidence, language = cls._extract_alternatives(payload)
+            if text:
+                return StreamingASRPartial(
+                    text=text,
+                    language=language,
+                    confidence=confidence,
+                    is_final=is_final,
+                    is_stable=is_final,
+                )
+        return None
+
+    @staticmethod
+    def _extract_alternatives(payload: object) -> tuple[str, float | None, str | None]:
+        alternatives = getattr(payload, "alternatives", None)
+        if not alternatives:
+            return "", None, None
+        texts: list[str] = []
+        confidence: float | None = None
+        language: str | None = None
+        for alternative in alternatives:
+            text = str(getattr(alternative, "text", "") or "").strip()
+            if text:
+                texts.append(text)
+            if confidence is None:
+                raw_confidence = getattr(alternative, "confidence", None)
+                if isinstance(raw_confidence, (int, float)):
+                    confidence = float(raw_confidence)
+            if language is None:
+                languages = getattr(alternative, "languages", None)
+                if languages:
+                    first_language = languages[0]
+                    language_code = str(getattr(first_language, "language_code", "") or "").strip()
+                    if language_code:
+                        language = language_code
+        return " ".join(texts).strip(), confidence, language
+
+    def _drain_updates(self, state: _YandexGrpcStreamState) -> None:
+        while True:
+            try:
+                update = state.updates_queue.get_nowait()
+            except queue.Empty:
+                return
+            with state.lock:
+                state.latest_partial = update
+                if update.is_final:
+                    state.final_transcription = ASRTranscription(
+                        text=update.text,
+                        language=update.language or state.language,
+                        confidence=update.confidence,
+                    )
+
+    def _run_stream(self, state: _YandexGrpcStreamState) -> None:
+        channel = None
+        try:
+            grpc, stt_pb2, stt_service_pb2_grpc = self._load_grpc_modules()
+            channel = grpc.secure_channel(  # type: ignore[attr-defined]
+                self._grpc_target,
+                grpc.ssl_channel_credentials(),  # type: ignore[attr-defined]
+            )
+            stub = stt_service_pb2_grpc.RecognizerStub(channel)  # type: ignore[attr-defined]
+            responses = stub.RecognizeStreaming(
+                self._request_iterator(state, stt_pb2),
+                metadata=self._metadata(),
+                timeout=self._timeout_seconds,
+            )
+            for response in responses:
+                partial = self._partial_from_response(response)
+                if partial is not None:
+                    state.updates_queue.put(partial)
+        except BaseException as exc:
+            with state.lock:
+                state.error = exc
+        finally:
+            if channel is not None:
+                close = getattr(channel, "close", None)
+                if callable(close):
+                    close()
+
+    def _state(self, stream_id: str) -> _YandexGrpcStreamState:
+        with self._lock:
+            state = self._streams.get(stream_id)
+        if state is None:
+            raise StreamingASRUnavailable("Yandex SpeechKit gRPC stream is not active")
+        return state
+
+    def _signal_finish(self, state: _YandexGrpcStreamState) -> None:
+        with state.lock:
+            if state.finish_requested:
+                return
+            state.finish_requested = True
+        try:
+            state.input_queue.put_nowait(None)
+        except queue.Full:
+            try:
+                state.input_queue.get_nowait()
+                state.input_queue.task_done()
+            except queue.Empty:
+                pass
+            state.input_queue.put_nowait(None)
+
+    def open_stream(self, session_id: str, *, language_hint: str | None = None) -> str:
+        self._metadata()
+        self._load_grpc_modules()
+        stream_id = f"yasr_grpc_{uuid.uuid4().hex}"
+        language = _normalize_yandex_asr_language(language_hint)
+        state = _YandexGrpcStreamState(
+            stream_id=stream_id,
+            session_id=str(session_id or "").strip(),
+            language=language,
+            input_queue=queue.Queue(maxsize=self._queue_max_chunks),
+            updates_queue=queue.Queue(),
+        )
+        thread = threading.Thread(
+            target=self._run_stream,
+            args=(state,),
+            name=f"yandex-speechkit-asr-{stream_id[:16]}",
+            daemon=True,
+        )
+        state.thread = thread
+        with self._lock:
+            self._streams[stream_id] = state
+        thread.start()
+        return stream_id
+
+    def push_pcm(self, stream_id: str, pcm_8k_chunk: bytes) -> None:
+        if not pcm_8k_chunk:
+            return
+        state = self._state(stream_id)
+        with state.lock:
+            if state.error is not None:
+                raise StreamingASRUnavailable(str(state.error)[:500])
+            if state.finish_requested:
+                raise StreamingASRUnavailable("Yandex SpeechKit gRPC stream is already finalizing")
+        try:
+            state.input_queue.put(pcm_8k_chunk, timeout=0.1)
+        except queue.Full as exc:
+            raise StreamingASRUnavailable("Yandex SpeechKit gRPC input queue is full") from exc
+
+    def poll_partial(self, stream_id: str) -> StreamingASRPartial | None:
+        state = self._state(stream_id)
+        self._drain_updates(state)
+        with state.lock:
+            if state.error is not None and state.latest_partial is None:
+                raise StreamingASRUnavailable(str(state.error)[:500])
+            return state.latest_partial
+
+    def finalize(self, stream_id: str) -> ASRTranscription:
+        state = self._state(stream_id)
+        self._signal_finish(state)
+        deadline = time.monotonic() + self._timeout_seconds
+        while True:
+            self._drain_updates(state)
+            thread = state.thread
+            if thread is None or not thread.is_alive():
+                break
+            if time.monotonic() >= deadline:
+                raise StreamingASRUnavailable("Yandex SpeechKit gRPC stream finalize timed out")
+            thread.join(timeout=0.05)
+        self._drain_updates(state)
+        with state.lock:
+            if state.final_transcription is not None:
+                return state.final_transcription
+            if state.latest_partial is not None:
+                return ASRTranscription(
+                    text=state.latest_partial.text,
+                    language=state.latest_partial.language or state.language,
+                    confidence=state.latest_partial.confidence,
+                )
+            if state.error is not None:
+                raise StreamingASRUnavailable(str(state.error)[:500])
+        return ASRTranscription(text="", language=state.language, confidence=None)
+
+    def close_stream(self, stream_id: str) -> None:
+        try:
+            state = self._state(stream_id)
+        except StreamingASRUnavailable:
+            return
+        self._signal_finish(state)
+        thread = state.thread
+        if thread is not None and thread.is_alive():
+            thread.join(timeout=0.5)
+        with self._lock:
+            self._streams.pop(stream_id, None)
+
+
 class YandexSpeechKitBufferedStreamingASRProvider(StreamingASRProvider):
    name = "yandex-speechkit-buffered"
    supports_streaming = True
@@ -612,6 +987,8 @@ def build_streaming_asr_provider(name: str) -> StreamingASRProvider:
    normalized = str(name or "disabled").strip().lower()
    if normalized in {"local_sidecar", "local-sidecar", "sidecar"}:
        return LocalSidecarStreamingASRProvider()
-    if normalized in {"yandex", "yandex_speechkit", "yandex-speechkit", "speechkit"}:
+    if normalized in {"yandex", "yandex_speechkit", "yandex-speechkit", "speechkit", "yandex_grpc"}:
+        return YandexSpeechKitGrpcStreamingASRProvider()
+    if normalized in {"yandex_buffered", "yandex_speechkit_buffered", "yandex-speechkit-buffered"}:
        return YandexSpeechKitBufferedStreamingASRProvider()
    return StreamingASRProvider()
@@ -1,5 +1,7 @@
 from __future__ import annotations

+import time
+
 from services.ai_voice_runtime_service.audiosocket import pcm16le_to_wav_bytes
 from services.ai_voice_runtime_service.providers import asr as asr_module

@@ -198,10 +200,133 @@ def test_yandex_buffered_streaming_provider_buffers_pcm_until_finalize():
    assert calls == [{"audio_bytes": b"\x10\x00\x20\x00", "language_hint": "ru"}]


+def test_yandex_grpc_streaming_provider_returns_partial_and_final():
+    calls: list[dict] = []
+
+    class _Message:
+        def __init__(self, **kwargs) -> None:
+            self.__dict__.update(kwargs)
+
+        def HasField(self, name: str) -> bool:
+            return getattr(self, name, None) is not None
+
+    class _RawAudio(_Message):
+        LINEAR16_PCM = 1
+
+    class _TextNormalizationOptions(_Message):
+        TEXT_NORMALIZATION_ENABLED = 1
+        PHONE_FORMATTING_MODE_DISABLED = 1
+
+    class _LanguageRestrictionOptions(_Message):
+        WHITELIST = 1
+
+    class _RecognitionModelOptions(_Message):
+        REAL_TIME = 1
+
+    class _DefaultEouClassifier(_Message):
+        HIGH = 2
+
+    class _FakeSttPb2:
+        RawAudio = _RawAudio
+        AudioFormatOptions = _Message
+        TextNormalizationOptions = _TextNormalizationOptions
+        LanguageRestrictionOptions = _LanguageRestrictionOptions
+        RecognitionModelOptions = _RecognitionModelOptions
+        DefaultEouClassifier = _DefaultEouClassifier
+        EouClassifierOptions = _Message
+        StreamingOptions = _Message
+        StreamingRequest = _Message
+        AudioChunk = _Message
+        Eou = _Message
+
+    class _FakeChannel:
+        def close(self) -> None:
+            calls.append({"method": "close"})
+
+    class _FakeGrpc:
+        @staticmethod
+        def ssl_channel_credentials():
+            return "ssl"
+
+        @staticmethod
+        def secure_channel(target: str, credentials):
+            calls.append({"method": "secure_channel", "target": target, "credentials": credentials})
+            return _FakeChannel()
+
+    class _RecognizerStub:
+        def __init__(self, channel) -> None:
+            calls.append({"method": "stub_init", "channel": channel})
+
+        def RecognizeStreaming(self, request_iterator, *, metadata, timeout):
+            calls.append({"method": "recognize", "metadata": metadata, "timeout": timeout})
+            for request in request_iterator:
+                calls.append({"method": "request", "request": request})
+                if getattr(request, "chunk", None) is not None:
+                    yield _Message(
+                        partial=_Message(
+                            alternatives=[
+                                _Message(text="need schedule", confidence=0.7, languages=[]),
+                            ]
+                        )
+                    )
+                    continue
+                if getattr(request, "eou", None) is not None:
+                    yield _Message(
+                        final_refinement=_Message(
+                            normalized_text=_Message(
+                                alternatives=[
+                                    _Message(text="need schedule in almaty", confidence=0.91, languages=[]),
+                                ]
+                            )
+                        )
+                    )
+                    return
+
+    class _FakeGrpcPb2:
+        RecognizerStub = _RecognizerStub
+
+    provider = asr_module.YandexSpeechKitGrpcStreamingASRProvider(
+        api_key="asr-key",
+        grpc_target="stt.test:443",
+        timeout_seconds=3,
+        grpc_module=_FakeGrpc,
+        stt_pb2_module=_FakeSttPb2,
+        stt_service_pb2_grpc_module=_FakeGrpcPb2,
+    )
+    stream_id = provider.open_stream("session-1", language_hint="ru")
+    provider.push_pcm(stream_id, b"\x10\x00\x20\x00")
+
+    partial = None
+    for _ in range(20):
+        partial = provider.poll_partial(stream_id)
+        if partial is not None:
+            break
+        time.sleep(0.02)
+
+    assert partial is not None
+    assert partial.text == "need schedule"
+    assert partial.is_final is False
+
+    final = provider.finalize(stream_id)
+    provider.close_stream(stream_id)
+
+    assert final.text == "need schedule in almaty"
+    recognize_call = next(call for call in calls if call.get("method") == "recognize")
+    assert ("authorization", "Api-Key asr-key") in recognize_call["metadata"]
+    requests = [call["request"] for call in calls if call.get("method") == "request"]
+    assert getattr(requests[0], "session_options", None) is not None
+    assert getattr(requests[1], "chunk", None).data == b"\x10\x00\x20\x00"
+    assert getattr(requests[-1], "eou", None) is not None
+
+
 def test_yandex_asr_builders():
    assert isinstance(asr_module.build_asr_provider("yandex"), asr_module.YandexSpeechKitASRProvider)
    assert isinstance(asr_module.build_asr_provider("speechkit"), asr_module.YandexSpeechKitASRProvider)
    assert isinstance(
        asr_module.build_streaming_asr_provider("yandex_speechkit"),
+        asr_module.YandexSpeechKitGrpcStreamingASRProvider,
+    )
+    assert isinstance(
+        asr_module.build_streaming_asr_provider("yandex_speechkit_buffered"),
        asr_module.YandexSpeechKitBufferedStreamingASRProvider,
    )