feat: migrate realtime voice service to OpenAI provider pipeline

providers/__init__.py

@@ -1,7 +1,9 @@
 from realtime_voice_service.providers.base import BaseLLM, BaseSTT, BaseTTS, MockLLM, MockSTT, MockTTS
 from realtime_voice_service.providers.llm import OpenAILLM
 from realtime_voice_service.providers.stt import ElevenLabsSTT
+from realtime_voice_service.providers.stt_openai import OpenAISTT
 from realtime_voice_service.providers.tts import ElevenLabsTTS
+from realtime_voice_service.providers.factory import create_stt_provider, create_tts_provider
 
 __all__ = [
     "BaseLLM",
@@ -13,4 +15,7 @@ __all__ = [
     "MockSTT",
     "MockTTS",
     "OpenAILLM",
+    "OpenAISTT",
+    "create_stt_provider",
+    "create_tts_provider",
 ]

providers/base.py

@@ -6,6 +6,27 @@ import re
 import struct
 from abc import ABC, abstractmethod
 from collections.abc import AsyncGenerator
+from collections.abc import AsyncIterable
+from collections.abc import Awaitable
+from collections.abc import Callable
+from dataclasses import dataclass
+
+
+PartialTranscriptCallback = Callable[[str], Awaitable[None] | None]
+
+
+class BaseSTTStream(ABC):
+    @abstractmethod
+    async def push_audio(self, audio_chunk: bytes) -> None:
+        raise NotImplementedError
+
+    @abstractmethod
+    async def finish(self) -> str:
+        raise NotImplementedError
+
+    @abstractmethod
+    async def cancel(self) -> None:
+        raise NotImplementedError
 
 
 class BaseSTT(ABC):
@@ -13,25 +34,41 @@ class BaseSTT(ABC):
     async def transcribe(self, audio_bytes: bytes) -> str:
         raise NotImplementedError
 
+    async def start_stream(
+        self,
+        *,
+        partial_callback: PartialTranscriptCallback | None = None,
+    ) -> BaseSTTStream | None:
+        del partial_callback
+        return None
+
 
 class BaseLLM(ABC):
     @abstractmethod
-    async def generate_stream(self, text: str, context: list) -> AsyncGenerator[str, None]:
+    async def generate_stream(self, text: str, context: list) -> AsyncGenerator[LLMStreamEvent, None]:
         raise NotImplementedError
 
 
 class BaseTTS(ABC):
     @abstractmethod
-    async def synthesize_stream(self, text: str) -> AsyncGenerator[bytes, None]:
+    async def synthesize_stream(self, text_stream: AsyncIterable[str]) -> AsyncGenerator[bytes, None]:
         raise NotImplementedError
 
 
+@dataclass(slots=True)
+class LLMStreamEvent:
+    type: str
+    content: str | None = None
+    name: str | None = None
+    tool_call_id: str | None = None
+
+
 class MockSTT(BaseSTT):
     def __init__(
         self,
         *,
         latency_ms: int = 40,
-        sample_rate_hz: int = 8000,
+        sample_rate_hz: int = 16000,
         scripted_transcripts: list[str] | None = None,
     ) -> None:
         self._latency_ms = max(latency_ms, 0)
@@ -47,6 +84,13 @@ class MockSTT(BaseSTT):
         duration_ms = int(((len(audio_bytes) // 2) / float(self._sample_rate_hz)) * 1000.0)
         return f"mock user utterance {self._call_count} ({duration_ms} ms)"
 
+    async def start_stream(
+        self,
+        *,
+        partial_callback: PartialTranscriptCallback | None = None,
+    ) -> BaseSTTStream | None:
+        return _MockSTTStream(parent=self, partial_callback=partial_callback)
+
 
 class MockLLM(BaseLLM):
     def __init__(
@@ -58,7 +102,7 @@ class MockLLM(BaseLLM):
         self._token_delay_ms = max(token_delay_ms, 0)
         self._scripted_responses = list(scripted_responses or [])
 
-    async def generate_stream(self, text: str, context: list) -> AsyncGenerator[str, None]:
+    async def generate_stream(self, text: str, context: list) -> AsyncGenerator[LLMStreamEvent, None]:
         del context
         response_text = (
             self._scripted_responses.pop(0)
@@ -68,14 +112,14 @@ class MockLLM(BaseLLM):
         chunks = re.findall(r"\S+\s*", response_text) or [response_text]
         for chunk in chunks:
             await asyncio.sleep(self._token_delay_ms / 1000.0)
-            yield chunk
+            yield LLMStreamEvent(type="text", content=chunk)
 
 
 class MockTTS(BaseTTS):
     def __init__(
         self,
         *,
-        sample_rate_hz: int = 8000,
+        sample_rate_hz: int = 16000,
         chunk_duration_ms: int = 40,
         chunk_delay_ms: int = 15,
         tone_hz: float = 440.0,
@@ -89,18 +133,60 @@ class MockTTS(BaseTTS):
         self._amplitude = amplitude
         self._milliseconds_per_word = max(milliseconds_per_word, 40)
 
-    async def synthesize_stream(self, text: str) -> AsyncGenerator[bytes, None]:
-        word_count = max(len(text.split()), 1)
-        total_ms = min(word_count * self._milliseconds_per_word, 2400)
-        total_samples = max(int(self._sample_rate_hz * (total_ms / 1000.0)), 1)
-        chunk_samples = max(int(self._sample_rate_hz * (self._chunk_duration_ms / 1000.0)), 1)
+    async def synthesize_stream(self, text_stream: AsyncIterable[str]) -> AsyncGenerator[bytes, None]:
+        async for text in text_stream:
+            normalized = text.strip()
+            if not normalized:
+                continue
+            word_count = max(len(normalized.split()), 1)
+            total_ms = min(word_count * self._milliseconds_per_word, 2400)
+            total_samples = max(int(self._sample_rate_hz * (total_ms / 1000.0)), 1)
+            chunk_samples = max(int(self._sample_rate_hz * (self._chunk_duration_ms / 1000.0)), 1)
 
-        for start in range(0, total_samples, chunk_samples):
-            end = min(start + chunk_samples, total_samples)
-            pcm = bytearray()
-            for index in range(start, end):
-                angle = 2.0 * math.pi * self._tone_hz * (index / float(self._sample_rate_hz))
-                sample = int(self._amplitude * math.sin(angle))
-                pcm.extend(struct.pack("<h", sample))
-            await asyncio.sleep(self._chunk_delay_ms / 1000.0)
-            yield bytes(pcm)
+            for start in range(0, total_samples, chunk_samples):
+                end = min(start + chunk_samples, total_samples)
+                pcm = bytearray()
+                for index in range(start, end):
+                    angle = 2.0 * math.pi * self._tone_hz * (index / float(self._sample_rate_hz))
+                    sample = int(self._amplitude * math.sin(angle))
+                    pcm.extend(struct.pack("<h", sample))
+                await asyncio.sleep(self._chunk_delay_ms / 1000.0)
+                yield bytes(pcm)
+
+
+class _MockSTTStream(BaseSTTStream):
+    def __init__(
+        self,
+        *,
+        parent: MockSTT,
+        partial_callback: PartialTranscriptCallback | None = None,
+    ) -> None:
+        self._parent = parent
+        self._partial_callback = partial_callback
+        self._audio_buffer = bytearray()
+        self._cancelled = False
+        self._partial_emitted = False
+
+    async def push_audio(self, audio_chunk: bytes) -> None:
+        if self._cancelled or not audio_chunk:
+            return
+        self._audio_buffer.extend(audio_chunk)
+        if self._partial_callback is None or self._partial_emitted:
+            return
+        duration_ms = int(((len(self._audio_buffer) // 2) / float(self._parent._sample_rate_hz)) * 1000.0)
+        if duration_ms < 600:
+            return
+        self._partial_emitted = True
+        partial = f"mock partial utterance {self._parent._call_count + 1}"
+        maybe_awaitable = self._partial_callback(partial)
+        if maybe_awaitable is not None:
+            await maybe_awaitable
+
+    async def finish(self) -> str:
+        if self._cancelled:
+            return ""
+        return await self._parent.transcribe(bytes(self._audio_buffer))
+
+    async def cancel(self) -> None:
+        self._cancelled = True
+        self._audio_buffer.clear()

providers/factory.py

@@ -0,0 +1,123 @@
+from __future__ import annotations
+
+import logging
+import os
+
+from realtime_voice_service.providers.base import BaseLLM, BaseSTT, BaseTTS
+from realtime_voice_service.providers.llm import OllamaLLM, OpenAILLM
+from realtime_voice_service.providers.stt import ElevenLabsSTT, FallbackSTT, YandexSpeechKitSTT
+from realtime_voice_service.providers.stt_openai import OpenAISTT
+from realtime_voice_service.providers.tts import ElevenLabsTTS
+
+
+LOGGER = logging.getLogger("uvicorn.error")
+
+
+def _first_env(names: tuple[str, ...], default: str = "") -> str:
+    for name in names:
+        value = os.getenv(name)
+        if value is not None and value.strip():
+            return value.strip()
+    return default
+
+
+def _stt_provider_name() -> str:
+    return _first_env(
+        (
+            "STT_PROVIDER",
+            "REALTIME_VOICE_STT_PROVIDER",
+            "AI_VOICE_ASR_PROVIDER",
+        ),
+        "elevenlabs",
+    ).lower()
+
+
+def _stt_fallback_provider_name() -> str:
+    return _first_env(
+        (
+            "STT_FALLBACK_PROVIDER",
+            "REALTIME_VOICE_STT_FALLBACK_PROVIDER",
+        )
+    ).lower()
+
+
+def _llm_provider_name() -> str:
+    return _first_env(("LLM_PROVIDER", "REALTIME_VOICE_LLM_PROVIDER"), "openai").lower()
+
+
+def _tts_provider_name() -> str:
+    return _first_env(("TTS_PROVIDER", "REALTIME_VOICE_TTS_PROVIDER"), "elevenlabs").lower()
+
+
+def create_llm_provider() -> BaseLLM:
+    provider = _llm_provider_name()
+    if provider in {"openai", "openai_chat", "gpt"}:
+        LOGGER.info("LLM provider selected: provider=%s", provider)
+        return OpenAILLM()
+    if provider in {"ollama", "local", "qwen"}:
+        LOGGER.info("LLM provider selected: provider=%s", provider)
+        return OllamaLLM()
+    raise RuntimeError(f"Unsupported LLM provider: {provider}")
+
+
+def _build_single_stt_provider(
+    provider: str,
+    *,
+    input_sample_rate_hz: int,
+    target_sample_rate_hz: int,
+) -> BaseSTT:
+    normalized = provider.lower().strip()
+    if normalized in {"openai", "whisper", "openai_whisper"}:
+        # Whisper accepts an 8 kHz WAV container; keep the exact AudioSocket PCM, no resampling.
+        return OpenAISTT(input_sample_rate_hz=input_sample_rate_hz)
+    if normalized in {"elevenlabs", "eleven_labs", "scribe"}:
+        return ElevenLabsSTT(
+            input_sample_rate_hz=input_sample_rate_hz,
+            target_sample_rate_hz=target_sample_rate_hz,
+        )
+    if normalized in {"yandex", "yandex_speechkit", "speechkit"}:
+        return YandexSpeechKitSTT(
+            input_sample_rate_hz=input_sample_rate_hz,
+            target_sample_rate_hz=target_sample_rate_hz,
+        )
+    raise RuntimeError(f"Unsupported STT provider: {provider}")
+
+
+def create_stt_provider(
+    *,
+    input_sample_rate_hz: int,
+    target_sample_rate_hz: int,
+) -> BaseSTT:
+    provider = _stt_provider_name()
+    primary = _build_single_stt_provider(
+        provider,
+        input_sample_rate_hz=input_sample_rate_hz,
+        target_sample_rate_hz=target_sample_rate_hz,
+    )
+    fallback_provider = _stt_fallback_provider_name()
+    if not fallback_provider or fallback_provider == provider:
+        LOGGER.info("STT provider selected: provider=%s", provider)
+        return primary
+
+    fallback = _build_single_stt_provider(
+        fallback_provider,
+        input_sample_rate_hz=input_sample_rate_hz,
+        target_sample_rate_hz=target_sample_rate_hz,
+    )
+    LOGGER.info("STT provider selected: provider=%s fallback=%s", provider, fallback_provider)
+    return FallbackSTT(primary=primary, fallback=fallback)
+
+
+def create_tts_provider(
+    *,
+    target_sample_rate_hz: int,
+    output_format: str | None = None,
+) -> BaseTTS:
+    provider = _tts_provider_name()
+    if provider in {"elevenlabs", "eleven_labs"}:
+        LOGGER.info("TTS provider selected: provider=%s", provider)
+        return ElevenLabsTTS(
+            output_format=output_format or f"pcm_{target_sample_rate_hz}",
+            target_sample_rate_hz=target_sample_rate_hz,
+        )
+    raise RuntimeError(f"Unsupported TTS provider: {provider}")

providers/llm.py

@@ -1,10 +1,26 @@
 from __future__ import annotations
 
+import asyncio
+import inspect
+import json
+import logging
 import os
+import time
 from collections.abc import AsyncGenerator
 from typing import Any
 
 from realtime_voice_service.providers.base import BaseLLM
+from realtime_voice_service.providers.base import LLMStreamEvent
+
+
+LOGGER = logging.getLogger("uvicorn.error")
+
+
+def _preview_text(text: str, *, limit: int = 160) -> str:
+    normalized = " ".join(str(text or "").split())
+    if len(normalized) <= limit:
+        return normalized
+    return f"{normalized[:limit]}..."
 
 
 def _timeout_seconds() -> float:
@@ -17,6 +33,33 @@ def _timeout_seconds() -> float:
         return 30.0
 
 
+def _max_context_messages() -> int:
+    raw = os.getenv("OPENAI_LLM_MAX_CONTEXT_MESSAGES")
+    if raw is None:
+        return 8
+    try:
+        return max(int(raw.strip()), 0)
+    except ValueError:
+        return 8
+
+
+def _read_bool_env(name: str, default: bool) -> bool:
+    raw = os.getenv(name)
+    if raw is None:
+        return default
+    return str(raw).strip().lower() in {"1", "true", "yes", "on"}
+
+
+def _read_int_env(name: str, default: int) -> int:
+    raw = os.getenv(name)
+    if raw is None:
+        return default
+    try:
+        return int(str(raw).strip())
+    except ValueError:
+        return default
+
+
 class OpenAILLM(BaseLLM):
     def __init__(
         self,
@@ -28,6 +71,12 @@ class OpenAILLM(BaseLLM):
         timeout_seconds: float | None = None,
         temperature: float = 0.3,
         max_retries: int = 2,
+        max_context_messages: int | None = None,
+        reasoning_effort: str | None = None,
+        enable_tools: bool | None = None,
+        max_tool_roundtrips: int | None = None,
+        serper_api_key: str | None = None,
+        serper_api_base: str | None = None,
     ) -> None:
         self._api_key = str(api_key if api_key is not None else os.getenv("OPENAI_API_KEY", "")).strip()
         self._model = str(model or os.getenv("OPENAI_LLM_MODEL", "gpt-4o-mini")).strip() or "gpt-4o-mini"
@@ -42,33 +91,225 @@ class OpenAILLM(BaseLLM):
         ).strip()
         self._timeout_seconds = max(float(timeout_seconds if timeout_seconds is not None else _timeout_seconds()), 1.0)
         self._temperature = max(min(float(temperature), 2.0), 0.0)
+        self._reasoning_effort = str(
+            reasoning_effort if reasoning_effort is not None else os.getenv("OPENAI_LLM_REASONING_EFFORT", "")
+        ).strip().lower()
         self._max_retries = max(int(max_retries), 0)
+        self._max_context_messages = max(
+            int(max_context_messages if max_context_messages is not None else _max_context_messages()),
+            0,
+        )
+        self._enable_tools = (
+            _read_bool_env("OPENAI_LLM_ENABLE_TOOLS", True)
+            if enable_tools is None
+            else bool(enable_tools)
+        )
+        self._max_tool_roundtrips = max(
+            int(max_tool_roundtrips if max_tool_roundtrips is not None else _read_int_env("OPENAI_LLM_MAX_TOOL_ROUNDTRIPS", 2)),
+            0,
+        )
+        self._serper_api_key = str(
+            serper_api_key if serper_api_key is not None else os.getenv("SERPER_API_KEY", "")
+        ).strip()
+        self._serper_api_base = (
+            str(serper_api_base or os.getenv("SERPER_API_BASE", "https://google.serper.dev")).strip().rstrip("/")
+            or "https://google.serper.dev"
+        )
         self._client: Any | None = None
         self._openai_module: Any | None = None
+        self._serper_session = None
+        self._serper_session_lock = asyncio.Lock()
+        LOGGER.info(
+            "OpenAI LLM config: model=%s base_url=%s timeout=%s max_context_messages=%s "
+            "reasoning_effort=%s tools_enabled=%s serper_configured=%s max_tool_roundtrips=%s",
+            self._model,
+            self._base_url or "default",
+            self._timeout_seconds,
+            self._max_context_messages,
+            self._reasoning_effort or "default",
+            self._enable_tools,
+            bool(self._serper_api_key),
+            self._max_tool_roundtrips,
+        )
 
-    async def generate_stream(self, text: str, context: list) -> AsyncGenerator[str, None]:
+    async def generate_stream(self, text: str, context: list) -> AsyncGenerator[LLMStreamEvent, None]:
         if not self._api_key:
             raise RuntimeError("OPENAI_API_KEY is required for OpenAI LLM")
 
-        client = self._get_client()
         messages = self._build_messages(text, context)
-        try:
-            stream = await client.chat.completions.create(
-                model=self._model,
+        turn_started_monotonic = time.perf_counter()
+        LOGGER.info(
+            "OpenAI LLM turn start: model=%s input_chars=%s context_entries=%s messages=%s "
+            "tools_available=%s input_preview=%r",
+            self._model,
+            len(text),
+            len(context),
+            len(messages),
+            bool(self._build_tools()),
+            _preview_text(text),
+        )
+        for round_index in range(self._max_tool_roundtrips + 1):
+            tool_buffers: dict[int, dict[str, str]] = {}
+            announced_tool_indexes: set[int] = set()
+            text_event_count = 0
+            text_char_count = 0
+            round_started_monotonic = time.perf_counter()
+            async for event in self._stream_completion(
                 messages=messages,
-                temperature=self._temperature,
-                stream=True,
+                tool_buffers=tool_buffers,
+                announced_tool_indexes=announced_tool_indexes,
+            ):
+                if event.type == "text":
+                    content = str(event.content or "")
+                    text_event_count += 1
+                    text_char_count += len(content)
+                    if text_event_count == 1 or text_event_count % 20 == 0:
+                        LOGGER.info(
+                            "OpenAI LLM text stream: round=%s events=%s chars=%s latest=%r",
+                            round_index,
+                            text_event_count,
+                            text_char_count,
+                            _preview_text(content, limit=80),
+                        )
+                elif event.type == "tool_call_start":
+                    LOGGER.info(
+                        "OpenAI LLM tool_call_start: round=%s name=%s tool_call_id=%s",
+                        round_index,
+                        event.name,
+                        event.tool_call_id,
+                    )
+                yield event
+            LOGGER.info(
+                "OpenAI LLM stream round completed: round=%s text_events=%s text_chars=%s "
+                "tool_calls=%s latency_ms=%s",
+                round_index,
+                text_event_count,
+                text_char_count,
+                len(tool_buffers),
+                int((time.perf_counter() - round_started_monotonic) * 1000.0),
             )
+            if not tool_buffers:
+                LOGGER.info(
+                    "OpenAI LLM turn completed: total_latency_ms=%s",
+                    int((time.perf_counter() - turn_started_monotonic) * 1000.0),
+                )
+                return
+
+            assistant_tool_calls = self._finalize_tool_calls(tool_buffers)
+            if not assistant_tool_calls:
+                LOGGER.warning("OpenAI LLM produced tool buffer without finalized tool calls")
+                return
+            messages.append(
+                {
+                    "role": "assistant",
+                    "content": None,
+                    "tool_calls": assistant_tool_calls,
+                }
+            )
+            tool_messages = await self._execute_tool_calls(assistant_tool_calls)
+            messages.extend(tool_messages)
+        LOGGER.warning(
+            "OpenAI LLM max tool roundtrips reached: max_tool_roundtrips=%s total_latency_ms=%s",
+            self._max_tool_roundtrips,
+            int((time.perf_counter() - turn_started_monotonic) * 1000.0),
+        )
+        final_text_event_count = 0
+        final_text_char_count = 0
+        final_round_started_monotonic = time.perf_counter()
+        async for event in self._stream_completion(
+            messages=messages,
+            tool_buffers={},
+            announced_tool_indexes=set(),
+            enable_tools=False,
+        ):
+            if event.type == "text":
+                content = str(event.content or "")
+                final_text_event_count += 1
+                final_text_char_count += len(content)
+                if final_text_event_count == 1 or final_text_event_count % 20 == 0:
+                    LOGGER.info(
+                        "OpenAI LLM final no-tool stream: events=%s chars=%s latest=%r",
+                        final_text_event_count,
+                        final_text_char_count,
+                        _preview_text(content, limit=80),
+                    )
+                yield event
+        LOGGER.info(
+            "OpenAI LLM final no-tool round completed: text_events=%s text_chars=%s latency_ms=%s total_latency_ms=%s",
+            final_text_event_count,
+            final_text_char_count,
+            int((time.perf_counter() - final_round_started_monotonic) * 1000.0),
+            int((time.perf_counter() - turn_started_monotonic) * 1000.0),
+        )
+
+    async def close(self) -> None:
+        client = self._client
+        self._client = None
+        if client is not None and hasattr(client, "close"):
+            result = client.close()
+            if inspect.isawaitable(result):
+                await result
+        session = self._serper_session
+        self._serper_session = None
+        if session is not None and not session.closed:
+            await session.close()
+
+    async def _stream_completion(
+        self,
+        *,
+        messages: list[dict[str, Any]],
+        tool_buffers: dict[int, dict[str, str]],
+        announced_tool_indexes: set[int],
+        enable_tools: bool = True,
+    ) -> AsyncGenerator[LLMStreamEvent, None]:
+        client = self._get_client()
+        request_kwargs: dict[str, Any] = {
+            "model": self._model,
+            "messages": messages,
+            "stream": True,
+        }
+        if self._supports_custom_temperature():
+            request_kwargs["temperature"] = self._temperature
+        tools = self._build_tools() if enable_tools else None
+        reasoning_effort = self._reasoning_effort
+        if tools and self._model.lower().startswith("gpt-5.5"):
+            reasoning_effort = ""
+        if reasoning_effort:
+            request_kwargs["reasoning_effort"] = reasoning_effort
+        if tools:
+            request_kwargs["tools"] = tools
+            request_kwargs["tool_choice"] = "auto"
+        LOGGER.info(
+            "OpenAI LLM stream request: model=%s messages=%s tools=%s temperature=%s reasoning_effort=%s enable_tools=%s",
+            self._model,
+            len(messages),
+            len(tools or []),
+            self._temperature if self._supports_custom_temperature() else "default",
+            reasoning_effort or "default",
+            enable_tools,
+        )
+
+        try:
+            stream = await client.chat.completions.create(**request_kwargs)
             async for chunk in stream:
                 choices = getattr(chunk, "choices", None) or []
                 if not choices:
                     continue
-                delta = getattr(choices[0], "delta", None)
+                choice = choices[0]
+                delta = getattr(choice, "delta", None)
                 if delta is None:
                     continue
                 content = getattr(delta, "content", None)
                 if content:
-                    yield str(content)
+                    yield LLMStreamEvent(type="text", content=str(content))
+                tool_calls = getattr(delta, "tool_calls", None) or []
+                for tool_delta in tool_calls:
+                    for event in self._consume_tool_delta(
+                        tool_delta=tool_delta,
+                        tool_buffers=tool_buffers,
+                        announced_tool_indexes=announced_tool_indexes,
+                    ):
+                        yield event
         except Exception as exc:  # noqa: BLE001
             openai_module = self._openai_module
             if openai_module is not None and isinstance(exc, getattr(openai_module, "APITimeoutError", ())):
@@ -82,6 +323,198 @@ class OpenAILLM(BaseLLM):
                 raise RuntimeError("OpenAI LLM connection failed") from exc
             raise RuntimeError("OpenAI LLM streaming failed") from exc
 
+    def _consume_tool_delta(
+        self,
+        *,
+        tool_delta: Any,
+        tool_buffers: dict[int, dict[str, str]],
+        announced_tool_indexes: set[int],
+    ) -> list[LLMStreamEvent]:
+        index = int(getattr(tool_delta, "index", 0) or 0)
+        state = tool_buffers.setdefault(index, {"id": "", "name": "", "arguments": ""})
+        tool_id = getattr(tool_delta, "id", None)
+        if tool_id:
+            state["id"] = str(tool_id)
+        function = getattr(tool_delta, "function", None)
+        if function is not None:
+            function_name = getattr(function, "name", None)
+            if function_name:
+                state["name"] = str(function_name)
+            function_arguments = getattr(function, "arguments", None)
+            if function_arguments:
+                state["arguments"] += str(function_arguments)
+        if state["name"] and index not in announced_tool_indexes:
+            announced_tool_indexes.add(index)
+            return [
+                LLMStreamEvent(
+                    type="tool_call_start",
+                    name=state["name"],
+                    tool_call_id=state["id"] or f"tool-call-{index}",
+                )
+            ]
+        return []
+
+    async def _execute_tool_calls(self, assistant_tool_calls: list[dict[str, Any]]) -> list[dict[str, str]]:
+        LOGGER.info("OpenAI LLM executing tool calls: count=%s", len(assistant_tool_calls))
+        results = await asyncio.gather(
+            *(self._execute_tool_call(tool_call) for tool_call in assistant_tool_calls),
+            return_exceptions=False,
+        )
+        tool_messages: list[dict[str, str]] = []
+        for tool_call, tool_result in zip(assistant_tool_calls, results, strict=False):
+            tool_messages.append(
+                {
+                    "role": "tool",
+                    "tool_call_id": str(tool_call["id"]),
+                    "content": tool_result,
+                }
+            )
+        return tool_messages
+
+    async def _execute_tool_call(self, tool_call: dict[str, Any]) -> str:
+        function = tool_call.get("function") or {}
+        name = str(function.get("name") or "").strip().lower()
+        raw_arguments = str(function.get("arguments") or "{}")
+        try:
+            arguments = json.loads(raw_arguments)
+        except json.JSONDecodeError:
+            arguments = {}
+
+        started_monotonic = time.perf_counter()
+        LOGGER.info(
+            "OpenAI LLM tool execution start: name=%s tool_call_id=%s args=%s",
+            name,
+            tool_call.get("id"),
+            raw_arguments[:500],
+        )
+        if name == "serper":
+            result = await self._run_serper_tool(arguments)
+        else:
+            result = f"Tool `{name}` is not supported by this runtime."
+        LOGGER.info(
+            "OpenAI LLM tool execution done: name=%s tool_call_id=%s latency_ms=%s result_chars=%s",
+            name,
+            tool_call.get("id"),
+            int((time.perf_counter() - started_monotonic) * 1000.0),
+            len(result),
+        )
+        return result
+
+    async def _run_serper_tool(self, arguments: dict[str, Any]) -> str:
+        if not self._serper_api_key:
+            return "Serper API is unavailable: SERPER_API_KEY is not configured."
+        query = str(arguments.get("query") or arguments.get("q") or "").strip()
+        if not query:
+            return "Serper API error: empty search query."
+
+        try:
+            import aiohttp
+        except Exception as exc:  # noqa: BLE001
+            raise RuntimeError("The `aiohttp` package is required for Serper tool execution") from exc
+
+        session = await self._get_serper_session()
+        request_payload = {
+            "q": query,
+            "gl": str(arguments.get("gl") or os.getenv("SERPER_SEARCH_GL", "kz")).strip(),
+            "hl": str(arguments.get("hl") or os.getenv("SERPER_SEARCH_HL", "ru")).strip(),
+            "num": max(int(arguments.get("num") or os.getenv("SERPER_SEARCH_NUM", 5)), 1),
+        }
+        started_monotonic = time.perf_counter()
+        LOGGER.info(
+            "Serper request start: query=%r gl=%s hl=%s num=%s",
+            _preview_text(query),
+            request_payload["gl"],
+            request_payload["hl"],
+            request_payload["num"],
+        )
+        try:
+            async with session.post(
+                f"{self._serper_api_base}/search",
+                headers={
+                    "X-API-KEY": self._serper_api_key,
+                    "Content-Type": "application/json",
+                },
+                json=request_payload,
+            ) as response:
+                payload_text = await response.text()
+                LOGGER.info(
+                    "Serper response: status=%s latency_ms=%s response_bytes=%s",
+                    response.status,
+                    int((time.perf_counter() - started_monotonic) * 1000.0),
+                    len(payload_text),
+                )
+                if response.status >= 400:
+                    return f"Serper API returned HTTP {response.status}: {payload_text[:300]}"
+        except (TimeoutError, asyncio.TimeoutError):
+            return "Serper API timed out while searching."
+        except aiohttp.ClientError as exc:
+            return f"Serper API request failed: {exc}"
+
+        try:
+            payload = json.loads(payload_text)
+        except json.JSONDecodeError:
+            return "Serper API returned invalid JSON."
+        summary = self._summarize_serper_payload(query=query, payload=payload)
+        LOGGER.info("Serper summary built: chars=%s preview=%r", len(summary), _preview_text(summary))
+        return summary
+
+    async def _get_serper_session(self):
+        try:
+            import aiohttp
+        except Exception as exc:  # noqa: BLE001
+            raise RuntimeError("The `aiohttp` package is required for Serper tool execution") from exc
+
+        if self._serper_session is not None and not self._serper_session.closed:
+            return self._serper_session
+        async with self._serper_session_lock:
+            if self._serper_session is not None and not self._serper_session.closed:
+                return self._serper_session
+            timeout = aiohttp.ClientTimeout(total=self._timeout_seconds)
+            connector = aiohttp.TCPConnector(limit=16, ttl_dns_cache=300)
+            self._serper_session = aiohttp.ClientSession(timeout=timeout, connector=connector)
+            return self._serper_session
+
+    def _build_tools(self) -> list[dict[str, Any]] | None:
+        if not self._enable_tools or not self._serper_api_key:
+            return None
+        return [
+            {
+                "type": "function",
+                "function": {
+                    "name": "serper",
+                    "description": (
+                        "Search the public web for recent or external information when the user asks "
+                        "about current facts, websites, company data, schedules, or anything requiring live search."
+                    ),
+                    "parameters": {
+                        "type": "object",
+                        "properties": {
+                            "query": {
+                                "type": "string",
+                                "description": "Precise search query to send to Serper.",
+                            },
+                            "num": {
+                                "type": "integer",
+                                "description": "How many results to fetch, usually 3 to 5.",
+                                "minimum": 1,
+                                "maximum": 10,
+                            },
+                            "hl": {
+                                "type": "string",
+                                "description": "UI language code, for example ru or en.",
+                            },
+                            "gl": {
+                                "type": "string",
+                                "description": "Country code for result localization, for example kz or us.",
+                            },
+                        },
+                        "required": ["query"],
+                        "additionalProperties": False,
+                    },
+                },
+            }
+        ]
+
     def _get_client(self):
         if self._client is not None:
             return self._client
@@ -107,11 +540,15 @@ class OpenAILLM(BaseLLM):
         )
         return self._client
 
-    def _build_messages(self, text: str, context: list) -> list[dict[str, str]]:
-        messages: list[dict[str, str]] = []
+    def _supports_custom_temperature(self) -> bool:
+        return not self._model.lower().startswith("gpt-5")
+
+    def _build_messages(self, text: str, context: list) -> list[dict[str, Any]]:
+        messages: list[dict[str, Any]] = []
         if self._system_prompt:
             messages.append({"role": "system", "content": self._system_prompt})
 
+        context_messages: list[dict[str, Any]] = []
         for entry in context:
             role: str | None = None
             content: str | None = None
@@ -123,9 +560,280 @@ class OpenAILLM(BaseLLM):
                 role = "assistant" if speaker == "assistant" else "user"
                 content = str(entry[1]).strip() or None
             if role and content:
-                messages.append({"role": role, "content": content})
+                context_messages.append({"role": role, "content": content})
 
+        original_context_count = len(context_messages)
+        if self._max_context_messages > 0 and len(context_messages) > self._max_context_messages:
+            context_messages = context_messages[-self._max_context_messages :]
+            LOGGER.info(
+                "OpenAI LLM context trimmed: original=%s retained=%s max_context_messages=%s",
+                original_context_count,
+                len(context_messages),
+                self._max_context_messages,
+            )
+
+        messages.extend(context_messages)
         if text.strip():
             if not messages or messages[-1].get("role") != "user" or messages[-1].get("content") != text:
                 messages.append({"role": "user", "content": text})
         return messages
+
+    @staticmethod
+    def _finalize_tool_calls(tool_buffers: dict[int, dict[str, str]]) -> list[dict[str, Any]]:
+        tool_calls: list[dict[str, Any]] = []
+        for index in sorted(tool_buffers):
+            state = tool_buffers[index]
+            name = str(state.get("name") or "").strip()
+            arguments = str(state.get("arguments") or "{}").strip() or "{}"
+            if not name:
+                continue
+            tool_calls.append(
+                {
+                    "id": str(state.get("id") or f"tool-call-{index}"),
+                    "type": "function",
+                    "function": {
+                        "name": name,
+                        "arguments": arguments,
+                    },
+                }
+            )
+        return tool_calls
+
+    @staticmethod
+    def _summarize_serper_payload(*, query: str, payload: dict[str, Any]) -> str:
+        lines = [f"Search query: {query}"]
+        answer_box = payload.get("answerBox")
+        if isinstance(answer_box, dict):
+            answer_text = str(answer_box.get("answer") or answer_box.get("snippet") or "").strip()
+            if answer_text:
+                lines.append(f"Answer box: {answer_text}")
+
+        knowledge_graph = payload.get("knowledgeGraph")
+        if isinstance(knowledge_graph, dict):
+            title = str(knowledge_graph.get("title") or "").strip()
+            description = str(knowledge_graph.get("description") or "").strip()
+            if title or description:
+                lines.append(f"Knowledge graph: {title} {description}".strip())
+
+        organic = payload.get("organic")
+        if isinstance(organic, list):
+            for index, item in enumerate(organic[:5], start=1):
+                if not isinstance(item, dict):
+                    continue
+                title = str(item.get("title") or "").strip()
+                snippet = str(item.get("snippet") or "").strip()
+                link = str(item.get("link") or "").strip()
+                if title or snippet or link:
+                    lines.append(f"{index}. {title} | {snippet} | {link}".strip())
+        if len(lines) == 1:
+            lines.append("No useful search results were returned.")
+        return "\n".join(lines)
+
+
+
+class OllamaLLM(BaseLLM):
+    def __init__(
+        self,
+        *,
+        model: str | None = None,
+        base_url: str | None = None,
+        system_prompt: str | None = None,
+        timeout_seconds: float | None = None,
+        temperature: float | None = None,
+        max_context_messages: int | None = None,
+    ) -> None:
+        self._model = str(model or os.getenv("OLLAMA_LLM_MODEL", "qwen2.5:1.5b")).strip() or "qwen2.5:1.5b"
+        self._base_url = (
+            str(base_url or os.getenv("OLLAMA_BASE_URL", "http://host.docker.internal:11434")).strip().rstrip("/")
+            or "http://host.docker.internal:11434"
+        )
+        self._system_prompt = str(
+            system_prompt
+            if system_prompt is not None
+            else os.getenv(
+                "OLLAMA_LLM_SYSTEM_PROMPT",
+                os.getenv(
+                    "OPENAI_LLM_SYSTEM_PROMPT",
+                    "You are a concise voice assistant for a telecom call center. Answer clearly and briefly.",
+                ),
+            )
+        ).strip()
+        self._timeout_seconds = max(
+            float(timeout_seconds if timeout_seconds is not None else self._read_float_env("OLLAMA_TIMEOUT_SECONDS", _timeout_seconds())),
+            1.0,
+        )
+        self._temperature = max(
+            min(float(temperature if temperature is not None else self._read_float_env("OLLAMA_LLM_TEMPERATURE", 0.3)), 2.0),
+            0.0,
+        )
+        self._max_context_messages = max(
+            int(
+                max_context_messages
+                if max_context_messages is not None
+                else self._read_int_env("OLLAMA_LLM_MAX_CONTEXT_MESSAGES", _max_context_messages())
+            ),
+            0,
+        )
+        self._num_predict = max(self._read_int_env("OLLAMA_LLM_NUM_PREDICT", 64), 0)
+        self._num_ctx = max(self._read_int_env("OLLAMA_LLM_NUM_CTX", 1024), 0)
+        self._client: Any | None = None
+        LOGGER.info(
+            "Ollama LLM config: model=%s base_url=%s timeout=%s max_context_messages=%s "
+            "temperature=%s num_predict=%s num_ctx=%s",
+            self._model,
+            self._base_url,
+            self._timeout_seconds,
+            self._max_context_messages,
+            self._temperature,
+            self._num_predict or "default",
+            self._num_ctx or "default",
+        )
+
+    async def generate_stream(self, text: str, context: list) -> AsyncGenerator[LLMStreamEvent, None]:
+        messages = self._build_messages(text, context)
+        options: dict[str, Any] = {"temperature": self._temperature}
+        if self._num_predict > 0:
+            options["num_predict"] = self._num_predict
+        if self._num_ctx > 0:
+            options["num_ctx"] = self._num_ctx
+        payload: dict[str, Any] = {
+            "model": self._model,
+            "messages": messages,
+            "stream": True,
+            "options": options,
+        }
+        started_monotonic = time.perf_counter()
+        text_event_count = 0
+        text_char_count = 0
+        LOGGER.info(
+            "Ollama LLM turn start: model=%s input_chars=%s context_entries=%s messages=%s input_preview=%r",
+            self._model,
+            len(text),
+            len(context),
+            len(messages),
+            _preview_text(text),
+        )
+        try:
+            client = self._get_client()
+            async with client.stream("POST", f"{self._base_url}/api/chat", json=payload) as response:
+                if response.status_code >= 400:
+                    body = (await response.aread()).decode("utf-8", "replace")
+                    raise RuntimeError(f"Ollama LLM returned HTTP {response.status_code}: {body[:300]}")
+                async for line in response.aiter_lines():
+                    if not line:
+                        continue
+                    try:
+                        chunk = json.loads(line)
+                    except json.JSONDecodeError:
+                        LOGGER.warning("Ollama LLM ignored invalid stream line: %r", line[:200])
+                        continue
+                    error = str(chunk.get("error") or "").strip()
+                    if error:
+                        raise RuntimeError(f"Ollama LLM error: {error}")
+                    message = chunk.get("message")
+                    content = ""
+                    if isinstance(message, dict):
+                        content = str(message.get("content") or "")
+                    if content:
+                        text_event_count += 1
+                        text_char_count += len(content)
+                        if text_event_count == 1 or text_event_count % 20 == 0:
+                            LOGGER.info(
+                                "Ollama LLM text stream: events=%s chars=%s latest=%r",
+                                text_event_count,
+                                text_char_count,
+                                _preview_text(content, limit=80),
+                            )
+                        yield LLMStreamEvent(type="text", content=content)
+                    if bool(chunk.get("done")):
+                        break
+        except (TimeoutError, asyncio.TimeoutError) as exc:
+            raise RuntimeError("Ollama LLM request timed out") from exc
+        except RuntimeError:
+            raise
+        except Exception as exc:  # noqa: BLE001
+            raise RuntimeError("Ollama LLM streaming failed") from exc
+        finally:
+            LOGGER.info(
+                "Ollama LLM turn completed: text_events=%s text_chars=%s total_latency_ms=%s",
+                text_event_count,
+                text_char_count,
+                int((time.perf_counter() - started_monotonic) * 1000.0),
+            )
+
+    async def close(self) -> None:
+        client = self._client
+        self._client = None
+        if client is not None:
+            await client.aclose()
+
+    def _get_client(self):
+        if self._client is not None:
+            return self._client
+        try:
+            import httpx
+        except Exception as exc:  # noqa: BLE001
+            raise RuntimeError("The `httpx` package is required for Ollama LLM") from exc
+        timeout = httpx.Timeout(
+            self._timeout_seconds,
+            connect=min(self._timeout_seconds, 3.0),
+            write=min(self._timeout_seconds, 10.0),
+            read=self._timeout_seconds,
+        )
+        self._client = httpx.AsyncClient(timeout=timeout)
+        return self._client
+
+    def _build_messages(self, text: str, context: list) -> list[dict[str, Any]]:
+        messages: list[dict[str, Any]] = []
+        if self._system_prompt:
+            messages.append({"role": "system", "content": self._system_prompt})
+
+        context_messages: list[dict[str, Any]] = []
+        for entry in context:
+            role: str | None = None
+            content: str | None = None
+            if isinstance(entry, dict):
+                role = str(entry.get("role") or "").strip().lower() or None
+                content = str(entry.get("content") or "").strip() or None
+            elif isinstance(entry, (tuple, list)) and len(entry) >= 2:
+                speaker = str(entry[0]).strip().lower()
+                role = "assistant" if speaker == "assistant" else "user"
+                content = str(entry[1]).strip() or None
+            if role and content:
+                context_messages.append({"role": role, "content": content})
+
+        original_context_count = len(context_messages)
+        if self._max_context_messages > 0 and len(context_messages) > self._max_context_messages:
+            context_messages = context_messages[-self._max_context_messages :]
+            LOGGER.info(
+                "Ollama LLM context trimmed: original=%s retained=%s max_context_messages=%s",
+                original_context_count,
+                len(context_messages),
+                self._max_context_messages,
+            )
+
+        messages.extend(context_messages)
+        if text.strip():
+            if not messages or messages[-1].get("role") != "user" or messages[-1].get("content") != text:
+                messages.append({"role": "user", "content": text})
+        return messages
+
+    @staticmethod
+    def _read_float_env(name: str, default: float) -> float:
+        raw = os.getenv(name)
+        if raw is None:
+            return default
+        try:
+            return float(str(raw).strip())
+        except ValueError:
+            return default
+
+    @staticmethod
+    def _read_int_env(name: str, default: int) -> int:
+        raw = os.getenv(name)
+        if raw is None:
+            return default
+        try:
+            return int(str(raw).strip())
+        except ValueError:
+            return default

providers/stt_openai.py

@@ -0,0 +1,188 @@
+from __future__ import annotations
+
+import asyncio
+import io
+import logging
+import os
+import time
+import wave
+from typing import Any
+
+from realtime_voice_service.providers.base import BaseSTT
+
+
+LOGGER = logging.getLogger("uvicorn.error")
+
+
+def _timeout_seconds() -> float:
+    raw = os.getenv("OPENAI_STT_TIMEOUT_SECONDS") or os.getenv("OPENAI_TIMEOUT_SECONDS")
+    if raw is None:
+        return 30.0
+    try:
+        return max(float(raw.strip()), 1.0)
+    except ValueError:
+        return 30.0
+
+
+def _max_retries() -> int:
+    raw = os.getenv("OPENAI_STT_MAX_RETRIES")
+    if raw is None:
+        return 2
+    try:
+        return max(int(raw.strip()), 0)
+    except ValueError:
+        return 2
+
+
+def _preview_text(text: str, *, limit: int = 120) -> str:
+    normalized = " ".join(str(text or "").split())
+    if len(normalized) <= limit:
+        return normalized
+    return f"{normalized[:limit]}..."
+
+
+def _pcm16le_to_wav_bytes(
+    pcm_bytes: bytes,
+    *,
+    sample_rate_hz: int,
+    channels: int = 1,
+    sample_width_bytes: int = 2,
+) -> bytes:
+    handle = io.BytesIO()
+    with wave.open(handle, "wb") as wav_file:
+        wav_file.setnchannels(channels)
+        wav_file.setsampwidth(sample_width_bytes)
+        wav_file.setframerate(sample_rate_hz)
+        wav_file.writeframes(pcm_bytes)
+    return handle.getvalue()
+
+
+class OpenAISTT(BaseSTT):
+    name = "openai"
+
+    def __init__(
+        self,
+        *,
+        api_key: str | None = None,
+        model: str | None = None,
+        base_url: str | None = None,
+        input_sample_rate_hz: int = 8000,
+        prompt: str | None = None,
+        language: str | None = None,
+        timeout_seconds: float | None = None,
+        max_retries: int | None = None,
+    ) -> None:
+        self._api_key = str(api_key if api_key is not None else os.getenv("OPENAI_API_KEY", "")).strip()
+        self._model = str(model or os.getenv("OPENAI_STT_MODEL", "whisper-1")).strip() or "whisper-1"
+        self._base_url = str(
+            base_url if base_url is not None else (os.getenv("OPENAI_STT_BASE_URL") or os.getenv("OPENAI_BASE_URL") or "")
+        ).strip() or None
+        self._input_sample_rate_hz = max(int(input_sample_rate_hz), 1)
+        self._prompt = str(prompt if prompt is not None else os.getenv("STT_PROMPT", "")).strip()
+        self._language = str(language if language is not None else os.getenv("OPENAI_STT_LANGUAGE", "ru")).strip() or None
+        self._timeout_seconds = max(float(timeout_seconds if timeout_seconds is not None else _timeout_seconds()), 1.0)
+        self._max_retries = max(int(max_retries if max_retries is not None else _max_retries()), 0)
+        self._client: Any | None = None
+        self._openai_module: Any | None = None
+        LOGGER.info(
+            "OpenAI STT config: model=%s input_sample_rate=%s language=%s prompt_configured=%s base_url=%s timeout=%s",
+            self._model,
+            self._input_sample_rate_hz,
+            self._language,
+            bool(self._prompt),
+            self._base_url or "default",
+            self._timeout_seconds,
+        )
+
+    async def transcribe(self, audio_bytes: bytes) -> str:
+        if not audio_bytes:
+            return ""
+        if not self._api_key:
+            raise RuntimeError("OPENAI_API_KEY is required for OpenAI STT")
+
+        wav_bytes = _pcm16le_to_wav_bytes(audio_bytes, sample_rate_hz=self._input_sample_rate_hz)
+        audio_file = io.BytesIO(wav_bytes)
+        audio_file.name = "utterance.wav"
+
+        started_monotonic = time.perf_counter()
+        LOGGER.info(
+            "OpenAI STT request start: model=%s pcm_bytes=%s wav_bytes=%s sample_rate=%s language=%s prompt=%r",
+            self._model,
+            len(audio_bytes),
+            len(wav_bytes),
+            self._input_sample_rate_hz,
+            self._language,
+            _preview_text(self._prompt),
+        )
+
+        request: dict[str, object] = {
+            "file": audio_file,
+            "model": self._model,
+            "response_format": "json",
+        }
+        if self._prompt:
+            request["prompt"] = self._prompt
+        if self._language:
+            request["language"] = self._language
+
+        try:
+            response = await self._get_client().audio.transcriptions.create(**request)
+        except (TimeoutError, asyncio.TimeoutError) as exc:
+            raise RuntimeError("OpenAI STT request timed out") from exc
+        except Exception as exc:  # noqa: BLE001
+            openai_module = self._openai_module
+            if openai_module is not None and isinstance(exc, getattr(openai_module, "APITimeoutError", ())):
+                raise RuntimeError("OpenAI STT request timed out") from exc
+            if openai_module is not None and isinstance(exc, getattr(openai_module, "RateLimitError", ())):
+                raise RuntimeError("OpenAI STT rate limit exceeded") from exc
+            if openai_module is not None and isinstance(exc, getattr(openai_module, "APIStatusError", ())):
+                status_code = getattr(exc, "status_code", "unknown")
+                raise RuntimeError(f"OpenAI STT returned HTTP {status_code}") from exc
+            if openai_module is not None and isinstance(exc, getattr(openai_module, "APIConnectionError", ())):
+                raise RuntimeError("OpenAI STT connection failed") from exc
+            raise RuntimeError("OpenAI STT request failed") from exc
+
+        transcript_value = getattr(response, "text", None)
+        if transcript_value is None and isinstance(response, dict):
+            transcript_value = response.get("text")
+        transcript = str(transcript_value or "").strip()
+        LOGGER.info(
+            "OpenAI STT transcript result: latency_ms=%s chars=%s transcript=%r",
+            int((time.perf_counter() - started_monotonic) * 1000.0),
+            len(transcript),
+            _preview_text(transcript),
+        )
+        return transcript
+
+    def _get_client(self):
+        if self._client is not None:
+            return self._client
+        try:
+            import httpx
+            import openai
+            from openai import AsyncOpenAI
+        except Exception as exc:  # noqa: BLE001
+            raise RuntimeError("The `openai` package is required for OpenAI STT") from exc
+
+        timeout = httpx.Timeout(
+            self._timeout_seconds,
+            connect=min(self._timeout_seconds, 5.0),
+            write=min(self._timeout_seconds, 15.0),
+            read=self._timeout_seconds,
+        )
+        client_kwargs: dict[str, object] = {
+            "api_key": self._api_key,
+            "timeout": timeout,
+            "max_retries": self._max_retries,
+        }
+        if self._base_url:
+            client_kwargs["base_url"] = self._base_url
+        self._openai_module = openai
+        self._client = AsyncOpenAI(**client_kwargs)
+        return self._client
+
+    async def close(self) -> None:
+        client = self._client
+        self._client = None
+        if client is not None:
+            await client.close()

providers/tts.py

@@ -2,16 +2,35 @@ from __future__ import annotations
 
 import asyncio
 import audioop
+import base64
+import json
+import logging
 import os
+import time
 from collections.abc import AsyncGenerator
+from collections.abc import AsyncIterable
+from urllib.parse import quote
+from urllib.parse import urlencode
 
 from realtime_voice_service.providers.base import BaseTTS
 
 
+LOGGER = logging.getLogger("uvicorn.error")
+
+
 def _api_base() -> str:
     return (os.getenv("ELEVENLABS_API_BASE", "https://api.elevenlabs.io").strip() or "https://api.elevenlabs.io").rstrip("/")
 
 
+def _ws_base(http_base: str) -> str:
+    normalized = http_base.rstrip("/")
+    if normalized.startswith("https://"):
+        return f"wss://{normalized[len('https://') :]}"
+    if normalized.startswith("http://"):
+        return f"ws://{normalized[len('http://') :]}"
+    return normalized
+
+
 def _timeout_seconds() -> float:
     raw = os.getenv("ELEVENLABS_TIMEOUT_SECONDS")
     if raw is None:
@@ -22,78 +41,33 @@ def _timeout_seconds() -> float:
         return 30.0
 
 
-def _parse_output_format(output_format: str) -> tuple[str, int]:
+def _parse_chunk_schedule(raw: str | None) -> list[int]:
+    if not raw:
+        return [80, 120, 160, 220]
+    values: list[int] = []
+    for part in raw.split(","):
+        try:
+            values.append(max(int(part.strip()), 20))
+        except ValueError:
+            continue
+    return values or [80, 120, 160, 220]
+
+
+def _sample_rate_from_pcm_format(output_format: str) -> int | None:
     normalized = str(output_format or "").strip().lower()
-    if normalized == "ulaw_8000":
-        return "ulaw", 8000
-    if normalized == "alaw_8000":
-        return "alaw", 8000
     if not normalized.startswith("pcm_"):
-        raise ValueError("ElevenLabsTTS expects `pcm_*`, `ulaw_8000`, or `alaw_8000` output formats")
-    suffix = normalized.split("_", 1)[1]
+        return None
     try:
-        return "pcm16le", int(suffix)
-    except ValueError as exc:
-        raise ValueError(f"Unsupported ElevenLabs output format: {output_format}") from exc
+        return int(normalized.split("_", 1)[1])
+    except (IndexError, ValueError):
+        return None
 
 
-class _PCM16StreamAdapter:
-    def __init__(self, *, input_codec: str, input_rate_hz: int, output_rate_hz: int) -> None:
-        self._input_codec = input_codec
-        self._input_rate_hz = input_rate_hz
-        self._output_rate_hz = output_rate_hz
-        self._carry = b""
-        self._state = None
-
-    def process(self, chunk: bytes) -> bytes:
-        if not chunk:
-            return b""
-        data = self._carry + chunk
-        sample_width_bytes = 2 if self._input_codec == "pcm16le" else 1
-        usable_length = len(data) - (len(data) % sample_width_bytes)
-        self._carry = data[usable_length:]
-        if usable_length <= 0:
-            return b""
-        pcm16 = self._decode_to_pcm16(data[:usable_length])
-        if self._input_rate_hz == self._output_rate_hz:
-            return pcm16
-        converted, self._state = audioop.ratecv(
-            pcm16,
-            2,
-            1,
-            self._input_rate_hz,
-            self._output_rate_hz,
-            self._state,
-        )
-        return converted
-
-    def flush(self) -> bytes:
-        if not self._carry:
-            return b""
-        sample_width_bytes = 2 if self._input_codec == "pcm16le" else 1
-        padded = self._carry + (b"\x00" * ((sample_width_bytes - len(self._carry)) % sample_width_bytes))
-        self._carry = b""
-        pcm16 = self._decode_to_pcm16(padded)
-        if self._input_rate_hz == self._output_rate_hz:
-            return pcm16
-        converted, self._state = audioop.ratecv(
-            pcm16,
-            2,
-            1,
-            self._input_rate_hz,
-            self._output_rate_hz,
-            self._state,
-        )
-        return converted
-
-    def _decode_to_pcm16(self, chunk: bytes) -> bytes:
-        if self._input_codec == "pcm16le":
-            return chunk
-        if self._input_codec == "ulaw":
-            return audioop.ulaw2lin(chunk, 2)
-        if self._input_codec == "alaw":
-            return audioop.alaw2lin(chunk, 2)
-        raise ValueError(f"Unsupported input codec: {self._input_codec}")
+def _preview_text(text: str, *, limit: int = 120) -> str:
+    normalized = " ".join(str(text or "").split())
+    if len(normalized) <= limit:
+        return normalized
+    return f"{normalized[:limit]}..."
 
 
 class ElevenLabsTTS(BaseTTS):
@@ -106,78 +80,357 @@ class ElevenLabsTTS(BaseTTS):
         model_id: str | None = None,
         language_code: str | None = None,
         output_format: str | None = None,
-        target_sample_rate_hz: int = 8000,
+        target_sample_rate_hz: int | None = None,
+        inactivity_timeout_seconds: int = 20,
         timeout_seconds: float | None = None,
-        stream_chunk_bytes: int = 4096,
+        auto_mode: bool | None = None,
+        chunk_length_schedule: list[int] | None = None,
     ) -> None:
         self._api_key = str(api_key if api_key is not None else os.getenv("ELEVENLABS_API_KEY", "")).strip()
         self._api_base = str(api_base or _api_base()).strip().rstrip("/")
+        self._ws_base = _ws_base(self._api_base)
         self._voice_id = str(voice_id or os.getenv("ELEVENLABS_TTS_VOICE_ID", "")).strip()
-        self._model_id = (
-            str(model_id or os.getenv("ELEVENLABS_TTS_MODEL_ID", "eleven_flash_v2_5")).strip()
-            or "eleven_flash_v2_5"
+        self._requested_model_id = (
+            str(model_id or os.getenv("ELEVENLABS_TTS_MODEL_ID", "eleven_turbo_v2_5")).strip()
+            or "eleven_turbo_v2_5"
+        )
+        self._websocket_fallback_model_id = (
+            str(os.getenv("ELEVENLABS_TTS_WS_FALLBACK_MODEL_ID", "eleven_turbo_v2_5")).strip()
+            or "eleven_turbo_v2_5"
+        )
+        self._model_id = self._resolve_websocket_model_id(
+            requested_model_id=self._requested_model_id,
+            fallback_model_id=self._websocket_fallback_model_id,
         )
         self._language_code = str(language_code or os.getenv("ELEVENLABS_TTS_LANGUAGE_CODE", "ru")).strip() or None
-        self._output_format = str(output_format or os.getenv("ELEVENLABS_TTS_OUTPUT_FORMAT", "pcm_16000")).strip().lower() or "pcm_16000"
-        self._source_codec, self._source_sample_rate_hz = _parse_output_format(self._output_format)
-        self._target_sample_rate_hz = max(int(target_sample_rate_hz), 1)
+        requested_output_format = (
+            str(output_format or os.getenv("ELEVENLABS_TTS_OUTPUT_FORMAT", "pcm_16000")).strip().lower()
+            or "pcm_16000"
+        )
+        self._target_sample_rate_hz = int(
+            target_sample_rate_hz
+            if target_sample_rate_hz is not None
+            else (_sample_rate_from_pcm_format(requested_output_format) or 16000)
+        )
+        self._output_format = self._resolve_provider_output_format(
+            requested_output_format=requested_output_format,
+            target_sample_rate_hz=self._target_sample_rate_hz,
+        )
+        self._provider_sample_rate_hz = _sample_rate_from_pcm_format(self._output_format) or self._target_sample_rate_hz
         self._timeout_seconds = max(float(timeout_seconds if timeout_seconds is not None else _timeout_seconds()), 1.0)
-        self._stream_chunk_bytes = max(int(stream_chunk_bytes), 256)
+        self._inactivity_timeout_seconds = max(int(inactivity_timeout_seconds), 5)
+        self._auto_mode = (
+            str(os.getenv("ELEVENLABS_TTS_AUTO_MODE", "1")).strip().lower() in {"1", "true", "yes", "on"}
+            if auto_mode is None
+            else bool(auto_mode)
+        )
+        self._chunk_length_schedule = list(
+            chunk_length_schedule
+            or _parse_chunk_schedule(os.getenv("ELEVENLABS_TTS_CHUNK_LENGTH_SCHEDULE"))
+        )
+        self._voice_settings = {
+            "stability": self._read_float_env("ELEVENLABS_TTS_STABILITY", 0.35),
+            "similarity_boost": self._read_float_env("ELEVENLABS_TTS_SIMILARITY_BOOST", 0.75),
+            "speed": self._read_float_env("ELEVENLABS_TTS_SPEED", 1.0),
+            "use_speaker_boost": self._read_bool_env("ELEVENLABS_TTS_USE_SPEAKER_BOOST", False),
+        }
+        LOGGER.info(
+            "ElevenLabs TTS config: voice_id=%s requested_model=%s websocket_model=%s "
+            "provider_output_format=%s provider_sample_rate=%s target_sample_rate=%s "
+            "language=%s auto_mode=%s chunk_schedule=%s voice_settings=%s",
+            self._voice_id,
+            self._requested_model_id,
+            self._model_id,
+            self._output_format,
+            self._provider_sample_rate_hz,
+            self._target_sample_rate_hz,
+            self._language_code,
+            self._auto_mode,
+            self._chunk_length_schedule,
+            self._voice_settings,
+        )
 
-    async def synthesize_stream(self, text: str) -> AsyncGenerator[bytes, None]:
-        if not text.strip():
-            return
+    async def synthesize_stream(self, text_stream: AsyncIterable[str]) -> AsyncGenerator[bytes, None]:
         if not self._api_key:
             raise RuntimeError("ELEVENLABS_API_KEY is required for ElevenLabs TTS")
         if not self._voice_id:
             raise RuntimeError("ELEVENLABS_TTS_VOICE_ID is required for ElevenLabs TTS")
 
         try:
-            import aiohttp
+            from websockets.exceptions import WebSocketException
+            from websockets.legacy.client import connect as websocket_connect
         except Exception as exc:  # noqa: BLE001
-            raise RuntimeError("The `aiohttp` package is required for ElevenLabs TTS") from exc
+            raise RuntimeError("The `websockets` package is required for ElevenLabs TTS WebSocket streaming") from exc
 
-        payload = {
-            "text": text,
+        websocket_url = self._build_websocket_url()
+        started_monotonic = time.perf_counter()
+        audio_chunk_count = 0
+        audio_byte_count = 0
+        yielded_byte_count = 0
+        resample_state = None
+        pcm_remainder = b""
+        LOGGER.info(
+            "ElevenLabs TTS WebSocket connecting: voice_id=%s model=%s output_format=%s "
+            "provider_sample_rate=%s target_sample_rate=%s language=%s auto_mode=%s",
+            self._voice_id,
+            self._model_id,
+            self._output_format,
+            self._provider_sample_rate_hz,
+            self._target_sample_rate_hz,
+            self._language_code,
+            self._auto_mode,
+        )
+        try:
+            async with websocket_connect(
+                websocket_url,
+                extra_headers=[("xi-api-key", self._api_key)],
+                open_timeout=min(self._timeout_seconds, 10.0),
+                close_timeout=1.0,
+                ping_interval=20.0,
+                ping_timeout=20.0,
+                max_size=None,
+            ) as websocket:
+                LOGGER.info(
+                    "ElevenLabs TTS WebSocket connected: voice_id=%s model=%s connect_ms=%s",
+                    self._voice_id,
+                    self._model_id,
+                    int((time.perf_counter() - started_monotonic) * 1000.0),
+                )
+                await websocket.send(json.dumps(self._initial_payload()))
+                sender_task = asyncio.create_task(
+                    self._send_text_chunks(websocket, text_stream),
+                    name="elevenlabs-tts-sender",
+                )
+                try:
+                    while True:
+                        raw_message = await asyncio.wait_for(websocket.recv(), timeout=self._timeout_seconds)
+                        if isinstance(raw_message, bytes):
+                            raw_message = raw_message.decode("utf-8")
+                        payload = json.loads(raw_message)
+                        if not isinstance(payload, dict):
+                            continue
+                        if payload.get("audio"):
+                            audio_chunk = base64.b64decode(str(payload["audio"]))
+                            audio_chunk_count += 1
+                            audio_byte_count += len(audio_chunk)
+                            audio_chunk, resample_state, pcm_remainder = self._normalize_pcm_chunk(
+                                audio_chunk,
+                                resample_state=resample_state,
+                                pcm_remainder=pcm_remainder,
+                            )
+                            yielded_byte_count += len(audio_chunk)
+                            if audio_chunk_count == 1:
+                                LOGGER.info(
+                                    "ElevenLabs TTS first audio: voice_id=%s model=%s ttfa_ms=%s "
+                                    "provider_bytes=%s yielded_bytes=%s",
+                                    self._voice_id,
+                                    self._model_id,
+                                    int((time.perf_counter() - started_monotonic) * 1000.0),
+                                    audio_byte_count,
+                                    len(audio_chunk),
+                                )
+                            elif audio_chunk_count % 10 == 0:
+                                LOGGER.info(
+                                    "ElevenLabs TTS audio summary: voice_id=%s chunks=%s provider_bytes=%s yielded_bytes=%s",
+                                    self._voice_id,
+                                    audio_chunk_count,
+                                    audio_byte_count,
+                                    yielded_byte_count,
+                                )
+                            if audio_chunk:
+                                yield audio_chunk
+                        if self._is_error_payload(payload):
+                            LOGGER.error(
+                                "ElevenLabs TTS error payload: voice_id=%s model=%s payload=%s",
+                                self._voice_id,
+                                self._model_id,
+                                self._format_error_payload(payload),
+                            )
+                            raise RuntimeError(self._format_error_payload(payload))
+                        if bool(payload.get("isFinal")) or bool(payload.get("is_final")):
+                            LOGGER.info(
+                                "ElevenLabs TTS final payload: voice_id=%s model=%s chunks=%s "
+                                "provider_bytes=%s yielded_bytes=%s total_ms=%s",
+                                self._voice_id,
+                                self._model_id,
+                                audio_chunk_count,
+                                audio_byte_count,
+                                yielded_byte_count,
+                                int((time.perf_counter() - started_monotonic) * 1000.0),
+                            )
+                            if sender_task.done():
+                                break
+                finally:
+                    if not sender_task.done():
+                        sender_task.cancel()
+                        try:
+                            await sender_task
+                        except asyncio.CancelledError:
+                            pass
+        except asyncio.TimeoutError as exc:
+            raise RuntimeError("ElevenLabs TTS WebSocket request timed out") from exc
+        except WebSocketException as exc:
+            raise RuntimeError("ElevenLabs TTS WebSocket stream failed") from exc
+        except OSError as exc:
+            raise RuntimeError("ElevenLabs TTS WebSocket connection failed") from exc
+
+    def _build_websocket_url(self) -> str:
+        query = {
             "model_id": self._model_id,
+            "output_format": self._output_format,
+            "inactivity_timeout": self._inactivity_timeout_seconds,
+            "auto_mode": str(self._auto_mode).lower(),
+            "sync_alignment": "false",
             "apply_text_normalization": "auto",
         }
         if self._language_code:
-            payload["language_code"] = self._language_code
+            query["language_code"] = self._language_code
+        encoded_voice_id = quote(self._voice_id, safe="")
+        return f"{self._ws_base}/v1/text-to-speech/{encoded_voice_id}/stream-input?{urlencode(query)}"
 
-        adapter = _PCM16StreamAdapter(
-            input_codec=self._source_codec,
-            input_rate_hz=self._source_sample_rate_hz,
-            output_rate_hz=self._target_sample_rate_hz,
+    def _initial_payload(self) -> dict[str, object]:
+        payload: dict[str, object] = {
+            "text": " ",
+            "xi_api_key": self._api_key,
+            "voice_settings": self._voice_settings,
+        }
+        if not self._auto_mode:
+            payload["generation_config"] = {
+                "chunk_length_schedule": self._chunk_length_schedule,
+            }
+        return payload
+
+    async def _send_text_chunks(self, websocket, text_stream: AsyncIterable[str]) -> None:
+        pending_chunk: str | None = None
+        sent_count = 0
+        sent_chars = 0
+        async for raw_chunk in text_stream:
+            normalized = str(raw_chunk).strip()
+            if not normalized:
+                continue
+            normalized_chunk = self._normalize_stream_text(normalized)
+            if pending_chunk is not None:
+                await websocket.send(json.dumps({
+                    "text": pending_chunk,
+                    "try_trigger_generation": True,
+                }))
+                sent_count += 1
+                sent_chars += len(pending_chunk)
+                LOGGER.info(
+                    "ElevenLabs TTS text chunk sent: index=%s chars=%s total_chars=%s preview=%r",
+                    sent_count,
+                    len(pending_chunk),
+                    sent_chars,
+                    _preview_text(pending_chunk),
+                )
+            pending_chunk = normalized_chunk
+
+        if pending_chunk is not None:
+            await websocket.send(json.dumps({
+                "text": pending_chunk,
+                "try_trigger_generation": True,
+                "flush": True,
+            }))
+            sent_count += 1
+            sent_chars += len(pending_chunk)
+            LOGGER.info(
+                "ElevenLabs TTS final text chunk sent: index=%s chars=%s total_chars=%s preview=%r",
+                sent_count,
+                len(pending_chunk),
+                sent_chars,
+                _preview_text(pending_chunk),
+            )
+
+        await websocket.send(json.dumps({"text": ""}))
+        LOGGER.info("ElevenLabs TTS text stream closed: chunks=%s chars=%s", sent_count, sent_chars)
+
+    @staticmethod
+    def _normalize_stream_text(text: str) -> str:
+        if text.endswith((" ", "\n", "\t")):
+            return text
+        return f"{text} "
+
+    @staticmethod
+    def _resolve_websocket_model_id(*, requested_model_id: str, fallback_model_id: str) -> str:
+        normalized_requested = requested_model_id.strip() or "eleven_turbo_v2_5"
+        if normalized_requested not in {"eleven_v3", "eleven_ttv_v3"}:
+            return normalized_requested
+
+        normalized_fallback = fallback_model_id.strip() or "eleven_turbo_v2_5"
+        LOGGER.warning(
+            "ElevenLabs WebSocket TTS does not support model_id=%s; falling back to model_id=%s",
+            normalized_requested,
+            normalized_fallback,
         )
-        timeout = aiohttp.ClientTimeout(total=self._timeout_seconds)
-        try:
-            async with aiohttp.ClientSession(timeout=timeout) as session:
-                async with session.post(
-                    f"{self._api_base}/v1/text-to-speech/{self._voice_id}/stream",
-                    params={"output_format": self._output_format},
-                    headers={
-                        "xi-api-key": self._api_key,
-                        "Content-Type": "application/json",
-                    },
-                    json=payload,
-                ) as response:
-                    if response.status >= 400:
-                        error_text = await response.text()
-                        raise RuntimeError(
-                            f"ElevenLabs TTS returned HTTP {response.status}: {error_text[:300]}"
-                        )
-                    async for chunk in response.content.iter_chunked(self._stream_chunk_bytes):
-                        if not chunk:
-                            continue
-                        converted = adapter.process(chunk)
-                        if converted:
-                            yield converted
-        except (TimeoutError, asyncio.TimeoutError) as exc:
-            raise RuntimeError("ElevenLabs TTS request timed out") from exc
-        except aiohttp.ClientError as exc:
-            raise RuntimeError("ElevenLabs TTS request failed") from exc
+        return normalized_fallback
 
-        tail = adapter.flush()
-        if tail:
-            yield tail
+    @staticmethod
+    def _resolve_provider_output_format(*, requested_output_format: str, target_sample_rate_hz: int) -> str:
+        normalized = requested_output_format.strip().lower() or "pcm_16000"
+        if normalized == "pcm_8000":
+            LOGGER.warning(
+                "ElevenLabs TTS does not provide reliable PCM16 8kHz streaming; requesting pcm_16000 "
+                "and resampling to %sHz locally",
+                target_sample_rate_hz,
+            )
+            return "pcm_16000"
+        return normalized
+
+    def _normalize_pcm_chunk(
+        self,
+        audio_chunk: bytes,
+        *,
+        resample_state,
+        pcm_remainder: bytes,
+    ) -> tuple[bytes, object, bytes]:
+        if not audio_chunk:
+            return b"", resample_state, pcm_remainder
+        if self._provider_sample_rate_hz == self._target_sample_rate_hz:
+            return audio_chunk, resample_state, pcm_remainder
+
+        payload = pcm_remainder + audio_chunk
+        if len(payload) % 2:
+            pcm_remainder = payload[-1:]
+            payload = payload[:-1]
+        else:
+            pcm_remainder = b""
+        if not payload:
+            return b"", resample_state, pcm_remainder
+        converted, resample_state = audioop.ratecv(
+            payload,
+            2,
+            1,
+            self._provider_sample_rate_hz,
+            self._target_sample_rate_hz,
+            resample_state,
+        )
+        return converted, resample_state, pcm_remainder
+
+    @staticmethod
+    def _is_error_payload(payload: dict[str, object]) -> bool:
+        message_type = str(payload.get("message_type") or payload.get("type") or "").strip().lower()
+        return message_type.endswith("error") or "error" in payload
+
+    @staticmethod
+    def _format_error_payload(payload: dict[str, object]) -> str:
+        detail = str(payload.get("message") or payload.get("detail") or payload.get("error") or "").strip()
+        if detail:
+            return f"ElevenLabs TTS WebSocket error: {detail}"
+        return "ElevenLabs TTS WebSocket error"
+
+    @staticmethod
+    def _read_bool_env(name: str, default: bool) -> bool:
+        raw = os.getenv(name)
+        if raw is None:
+            return default
+        return str(raw).strip().lower() in {"1", "true", "yes", "on"}
+
+    @staticmethod
+    def _read_float_env(name: str, default: float) -> float:
+        raw = os.getenv(name)
+        if raw is None:
+            return default
+        try:
+            return float(str(raw).strip())
+        except ValueError:
+            return default