diff --git a/backend/fluksio/ai/client.py b/backend/fluksio/ai/client.py index 59bfc5e..58efbb3 100644 --- a/backend/fluksio/ai/client.py +++ b/backend/fluksio/ai/client.py @@ -160,6 +160,26 @@ async def list_models(api_base: str, api_key: str, timeout: float = 15.0) -> lis return sorted(str(m["id"]) for m in data if isinstance(m, dict) and "id" in m) +#: One client per speech server, so its connection stays open between calls +#: (`parrot serve` keeps an idle one 600 s). A new connection over a lossy +#: link can cost a second, which a spoken reply cannot spare. +_speech: dict[str, httpx.AsyncClient] = {} + + +def _speech_client(api_base: str) -> httpx.AsyncClient: + """The client for this speech server. Used from the engine's loop only.""" + http = _speech.get(api_base) + if http is None: + limits = httpx.Limits(keepalive_expiry=300) + http = _speech[api_base] = httpx.AsyncClient(limits=limits) + return http + + +async def warm(api_base: str) -> None: + """Open the connection to a speech server ahead of the call that needs it.""" + await _speech_client(api_base).get(api_base.rstrip("/") + "/models", timeout=10) + + async def transcribe( api_base: str, audio: bytes, @@ -170,12 +190,12 @@ async def transcribe( timeout: float = 30.0, ) -> str: """``POST {api_base}/audio/transcriptions``: a whole clip, and what it says.""" - async with httpx.AsyncClient(timeout=timeout) as http: - response = await http.post( - api_base.rstrip("/") + "/audio/transcriptions", - files={"file": (filename, audio, media_type)}, - data={"model": model, "response_format": "json"}, - ) + response = await _speech_client(api_base).post( + api_base.rstrip("/") + "/audio/transcriptions", + files={"file": (filename, audio, media_type)}, + data={"model": model, "response_format": "json"}, + timeout=timeout, + ) if response.is_error: raise ProviderError(_detail(response)) return str(response.json().get("text") or "") @@ -194,16 +214,11 @@ async def speak( MP3 rather than WAV: a seventh of the bytes, which over a slow link to the server is most of the wait for a sentence. """ - async with httpx.AsyncClient(timeout=timeout) as http: - response = await http.post( - api_base.rstrip("/") + "/audio/speech", - json={ - "model": model, - "input": text, - "voice": voice, - "response_format": "mp3", - }, - ) + response = await _speech_client(api_base).post( + api_base.rstrip("/") + "/audio/speech", + json={"model": model, "input": text, "voice": voice, "response_format": "mp3"}, + timeout=timeout, + ) if response.is_error: raise ProviderError(_detail(response)) return response.content diff --git a/backend/fluksio/flow/nodes/speech.py b/backend/fluksio/flow/nodes/speech.py index c546ea2..4b76f04 100644 --- a/backend/fluksio/flow/nodes/speech.py +++ b/backend/fluksio/flow/nodes/speech.py @@ -9,6 +9,7 @@ it gets. from __future__ import annotations import asyncio +import contextlib import hashlib import logging import re @@ -128,6 +129,13 @@ class _Speech(Node): def _load(self, reference: dict[str, Any]) -> bytes: return b"".join(self._artifacts.read(str(reference["digest"]))) + def _call(self, work: Coroutine[Any, Any, Any]) -> Future[Any]: + """Run on the engine's loop, where the server's connection is kept.""" + if self._loop is None: + work.close() + raise RuntimeError(f"{self.name} only runs in a live flow") + return asyncio.run_coroutine_threadsafe(work, self._loop) + def _say(self, **values: Any) -> None: """Publish on the ports that are wired.""" wired = {spec.port for spec in self.output_ports if spec.name} @@ -231,18 +239,21 @@ class SttNode(_Speech): def _spawn(self, work: Coroutine[Any, Any, None]) -> None: """Run on the engine's loop, so this run can return at once.""" - if self._loop is None: - work.close() - raise RuntimeError("stt only hears in a live flow") - task = asyncio.run_coroutine_threadsafe(work, self._loop) + task = self._call(work) self._tasks.add(task) task.add_done_callback(self._tasks.discard) def _open(self, url: str) -> asyncio.Queue[bytes | None]: chunks: asyncio.Queue[bytes | None] = asyncio.Queue() self._spawn(self._transcribe(url, chunks)) + # The reply's first sentence then finds the connection open. + self._spawn(self._warm(url)) return chunks + async def _warm(self, url: str) -> None: + with contextlib.suppress(Exception): + await client.warm(url + "/v1") + def _feed(self, chunks: asyncio.Queue[bytes | None], chunk: bytes | None) -> None: assert self._loop is not None self._loop.call_soon_threadsafe(chunks.put_nowait, chunk) @@ -355,7 +366,7 @@ class TtsNode(_Speech): reference = spoken.get(key) if reference is not None and self._artifacts.has(reference["digest"]): return dict(reference) - data = asyncio.run(client.speak(url + "/v1", sentence, self.cfg.voice)) + data = self._call(client.speak(url + "/v1", sentence, self.cfg.voice)).result() stored: dict[str, Any] = self._artifacts.put( [data], name="speech.mp3", media_type="audio/mpeg" ) diff --git a/backend/tests/flow/test_speech.py b/backend/tests/flow/test_speech.py index 6daa787..2fa3bde 100644 --- a/backend/tests/flow/test_speech.py +++ b/backend/tests/flow/test_speech.py @@ -52,6 +52,12 @@ def test_an_utterance_is_one_session_committed_when_listening_ends(monkeypatch): yield True, " Turn on the light." monkeypatch.setattr(client, "transcribe_stream", session) + warmed: list[str] = [] + + async def warm(api_base): + warmed.append(api_base) + + monkeypatch.setattr(client, "warm", warm) async def scenario() -> None: node = SttNode( @@ -94,6 +100,8 @@ def test_an_utterance_is_one_session_committed_when_listening_ends(monkeypatch): assert pipeline.state["voice.text"] == "Turn on the light." assert pipeline.state["voice.partial"] == "Turn on the light." assert node._session is None + # Opened with the session, for the reply's first sentence. + assert warmed == [f"{URL}/v1"] # No server, no session: the node says so instead. await hear(chunks[1], True, url="") @@ -131,10 +139,16 @@ def test_a_sentence_is_spoken_once_and_the_rest_of_a_reply_queues(monkeypatch): Pipeline([node]) node._artifacts = Store() - node.execute({"voice.reply": "Kitchen light on. Heating off.", "voice.url": URL}) - # Long after both stopped playing. - node._until = -10.0 - node.execute({"voice.reply": "Kitchen light on.", "voice.url": URL}) + async def scenario() -> None: + node._loop = asyncio.get_running_loop() + reply = "Kitchen light on. Heating off." + await asyncio.to_thread(node.execute, {"voice.reply": reply, "voice.url": URL}) + # Long after both stopped playing. + node._until = -10.0 + reply = "Kitchen light on." + await asyncio.to_thread(node.execute, {"voice.reply": reply, "voice.url": URL}) + + asyncio.run(scenario()) assert spoken == ["Kitchen light on.", "Heating off."] assert ["append" in speech for speech in published] == [False, True, False] diff --git a/docs/reference/node-types.md b/docs/reference/node-types.md index 950e22b..612a58a 100644 --- a/docs/reference/node-types.md +++ b/docs/reference/node-types.md @@ -460,7 +460,8 @@ such as one that sends whole clips. | `voice` | `af_heart` | one of Kokoro's voices, such as `bm_george` | Each sentence is one call to `{url}/v1/audio/speech` and published as soon as -it is spoken. A sentence spoken before in the same voice is reused rather than +it is spoken. Calls to one server share a connection, which Speech to text opens +when an utterance starts, so a reply's first sentence does not wait for one. A sentence spoken before in the same voice is reused rather than spoken again, for the last 256 sentences. A reference carries `"append": true` while the sentences before it are still playing, which queues it on a panel's `speaker`. The first sentence of a reply carries none and cuts off whatever