Fluksio/app
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Speech calls keep one connection per server, opened when listening starts

Browse Source 
transcribe and speak share an httpx client per speech server with a 300 s
keep-alive, and run on the engine's loop. stt warms it with GET /v1/models
when an utterance opens, so a reply's first sentence finds the connection
open: a new one over a lossy link cost up to a second (openpanel V4).
This commit is contained in:
Melvin Strobl
2026-10-19 22:06:08 +02:00
parent f50d8b9570
commit 4df7bdb057
4 changed files with 67 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/fluksio/ai/client.py
+31 -16
View File
@@ -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
backend/fluksio/flow/nodes/speech.py
+16 -5
View File
@@ -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"
)
backend/tests/flow/test_speech.py
+18 -4
View File
@@ -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]
docs/reference/node-types.md
+2 -1
View File
@@ -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