asterisk_ai_voice_agent/tts.py
Rama main · réplica de lectura del repositorio original.
"""
TTS: Piper (local, ONNX) + ElevenLabs (cloud).
Synthesizes text to PCM, resamples to 8 kHz slin (the AudioSocket format),
and yields 320-byte chunks (one 20 ms frame each) so callers can stream them
straight into AudioSocket AUDIO frames.
Providers (persona.tts_provider):
piper - local ONNX voice from your voices dir (see download_voices.sh).
Default en_US-amy-medium (22.05 kHz). Override via
config.providers.piper.default_voice or persona.tts_voice_id
(basename of the .onnx file).
elevenlabs - cloud API (needs an ElevenLabs API key and persona.tts_voice_id
= ElevenLabs voice id). Audio is requested as pcm_16000 and
resampled to 8 kHz. Any failure falls back to the local Piper
default voice so the call never goes silent; a 401/403 disables
further ElevenLabs attempts for the rest of the call.
The voices directory is configurable (constructor arg / env AI_AGENT_VOICES_DIR)
- there are no hardcoded install paths.
Derived from ICTContact (https://www.ictcontact.com).
ICT Innovations (https://www.ictinnovations.com) - ICT Vision (https://ict.vision)
Author: Tahir Almas. MIT licensed.
"""
import asyncio
import io
import logging
import os
import wave
from typing import AsyncIterator, Optional
import numpy as np
log = logging.getLogger("ai.tts")
# Voices dir: constructor arg wins, else env, else ./voices next to the sidecar.
DEFAULT_VOICES_DIR = os.environ.get(
"AI_AGENT_VOICES_DIR",
os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "voices"),
)
DEFAULT_VOICE_NAME = "en_US-amy-medium"
TARGET_SAMPLE_RATE = 8000 # slin (16-bit, 8 kHz) / AudioSocket
FRAME_BYTES = 320 # 20 ms @ 8 kHz mono 16-bit
ELEVEN_URL_TMPL = "https://api.elevenlabs.io/v1/text-to-speech/{voice_id}/stream"
ELEVEN_DEFAULT_MODEL = "eleven_flash_v2_5"
ELEVEN_SAMPLE_RATE = 16000 # we request output_format=pcm_16000
# PiperVoice.load takes 2.5-5.5 s (ONNX model load). Loading per call meant every
# answered call started with that much dead air before the greeting. Cache
# loaded voices process-wide, keyed by path.
_VOICE_CACHE: dict = {}
_VOICE_LOCK: Optional[asyncio.Lock] = None
def resolve_voice_path(voice_id: Optional[str], voices_dir: str,
default_voice: str = DEFAULT_VOICE_NAME) -> str:
"""Pick a voice .onnx file. Priority:
1. {voices_dir}/{voice_id}.onnx if voice_id given and the file exists
2. {voices_dir}/{default_voice}.onnx if it exists
Raises if neither is present."""
voices_dir = os.path.abspath(voices_dir)
for name in (voice_id, default_voice):
if not name:
continue
candidate = os.path.join(voices_dir, f"{name}.onnx")
if os.path.isfile(candidate):
return candidate
if name == voice_id:
log.warning("voice_id=%r requested but %s not found", voice_id, candidate)
raise RuntimeError(
f"no Piper voice in {voices_dir} (tried {voice_id!r}, {default_voice!r}); "
f"run download_voices.sh {default_voice}")
def _resample_linear(pcm16: np.ndarray, src_sr: int, dst_sr: int) -> np.ndarray:
"""Cheap linear resampler. Aliasing is acceptable for 8 kHz phone audio."""
if src_sr == dst_sr:
return pcm16
ratio = dst_sr / src_sr
out_len = int(len(pcm16) * ratio)
if out_len <= 1:
return np.zeros(0, dtype=np.int16)
src_idx = np.linspace(0, len(pcm16) - 1, out_len, dtype=np.float32)
lo = src_idx.astype(np.int32)
hi = np.minimum(lo + 1, len(pcm16) - 1)
frac = src_idx - lo
out = (1.0 - frac) * pcm16[lo].astype(np.float32) + frac * pcm16[hi].astype(np.float32)
return out.astype(np.int16)
class _StreamResampler:
"""Linear resampler that can be fed in chunks without a seam at each join.
`_resample_linear` interpolates across a whole utterance. Calling it once per
network chunk would restart the interpolation at every boundary and leave a
small discontinuity there, so this keeps the unconsumed input samples and the
fractional read position between calls.
"""
def __init__(self, src_sr: int, dst_sr: int):
self.step = (src_sr / dst_sr) if dst_sr else 1.0
self.passthrough = (src_sr == dst_sr)
self._tail = np.zeros(0, dtype=np.int16)
self._t = 0.0 # next output position, in input samples
def feed(self, pcm16: np.ndarray) -> np.ndarray:
if self.passthrough:
return pcm16
buf = np.concatenate((self._tail, pcm16)) if self._tail.size else pcm16
if buf.size < 2:
self._tail = buf
return np.zeros(0, dtype=np.int16)
# Only positions whose upper neighbour is in the buffer can be produced,
# so the last usable position is buf.size - 2, not buf.size - 1.
n = int(np.floor((buf.size - 2 - self._t) / self.step)) + 1
if n <= 0:
self._tail = buf
return np.zeros(0, dtype=np.int16)
pos = self._t + self.step * np.arange(n, dtype=np.float64)
lo = pos.astype(np.int32)
frac = (pos - lo).astype(np.float32)
out = ((1.0 - frac) * buf[lo].astype(np.float32)
+ frac * buf[lo + 1].astype(np.float32)).astype(np.int16)
consumed = int(lo[-1])
self._tail = buf[consumed:]
self._t = self._t + self.step * n - consumed
return out
def flush(self) -> np.ndarray:
"""Emit any output position still inside the buffered input, then reset.
Never extrapolates. At most one output sample (0.125 ms at 8 kHz) is
dropped at the very end of a stream, which is better than inventing one:
padding the tail with a copy of the last sample used to push a whole
extra frame of near-silence onto the end of every utterance.
"""
buf, self._tail = self._tail, np.zeros(0, dtype=np.int16)
t, self._t = self._t, 0.0
if self.passthrough or buf.size < 2:
return np.zeros(0, dtype=np.int16)
n = int(np.floor((buf.size - 1 - t) / self.step)) + 1
if n <= 0:
return np.zeros(0, dtype=np.int16)
pos = t + self.step * np.arange(n, dtype=np.float64)
lo = np.minimum(pos.astype(np.int32), buf.size - 1)
hi = np.minimum(lo + 1, buf.size - 1)
frac = (pos - lo).astype(np.float32)
return ((1.0 - frac) * buf[lo].astype(np.float32)
+ frac * buf[hi].astype(np.float32)).astype(np.int16)
def _iter_frames(pcm: np.ndarray):
"""Yield 8 kHz slin PCM as 20 ms (320-byte) frames, padding the tail."""
view = pcm.tobytes()
for off in range(0, len(view), FRAME_BYTES):
chunk = view[off:off + FRAME_BYTES]
if len(chunk) < FRAME_BYTES:
chunk = chunk + b"\x00" * (FRAME_BYTES - len(chunk))
yield chunk
class StreamingTTS:
def __init__(self, provider: str, voice_id: str, model: str,
api_key: Optional[str] = None,
voices_dir: Optional[str] = None,
default_voice: str = DEFAULT_VOICE_NAME):
self.provider = (provider or "piper").strip().lower()
self.voice_id = voice_id
self.model = model
self.api_key = api_key or ""
self.voices_dir = voices_dir or DEFAULT_VOICES_DIR
self.default_voice = default_voice or DEFAULT_VOICE_NAME
self.voice_path = None
self._voice = None
self._native_sr = None
# ElevenLabs state (per call)
self._eleven = False # provider active for this call
self._eleven_disabled = False # tripped on auth failure; piper thereafter
self._client = None # httpx.AsyncClient, created in start()
async def start(self) -> None:
if self.provider == "elevenlabs":
if self.api_key and self.voice_id:
import httpx # lazy: only needed for cloud TTS
self._client = httpx.AsyncClient(timeout=httpx.Timeout(15.0, connect=5.0))
self._eleven = True
log.info("TTS ready: elevenlabs voice=%s model=%s (piper fallback armed)",
self.voice_id, self.model or ELEVEN_DEFAULT_MODEL)
return
log.warning("tts provider=elevenlabs but %s missing; using piper",
"api_key" if not self.api_key else "voice_id")
elif self.provider != "piper":
log.warning("tts provider=%s not implemented; using piper", self.provider)
await self._ensure_piper()
log.info("TTS ready: voice=%s native_sr=%d",
os.path.basename(self.voice_path), self._native_sr)
async def _ensure_piper(self) -> None:
"""Load (or fetch from cache) the local Piper voice."""
global _VOICE_LOCK
if self._voice is not None:
return
# In elevenlabs mode voice_id is a cloud id, not an .onnx basename, so
# resolution lands on the configured/default local voice.
piper_voice_id = None if self._eleven else self.voice_id
self.voice_path = resolve_voice_path(piper_voice_id, self.voices_dir, self.default_voice)
log.info("TTS voice file resolved: %s", self.voice_path)
if _VOICE_LOCK is None:
_VOICE_LOCK = asyncio.Lock()
async with _VOICE_LOCK:
voice = _VOICE_CACHE.get(self.voice_path)
if voice is None:
from piper import PiperVoice # lazy import
# Load is CPU-bound; offload so the event loop isn't blocked.
voice = await asyncio.to_thread(PiperVoice.load, self.voice_path)
_VOICE_CACHE[self.voice_path] = voice
log.info("TTS voice loaded + cached: %s", os.path.basename(self.voice_path))
self._voice = voice
self._native_sr = self._voice.config.sample_rate
async def _eleven_frames(self, text: str, should_stop=None) -> AsyncIterator[bytes]:
"""Yield 8 kHz 320-byte frames as the ElevenLabs response arrives.
The endpoint streams, so reading it incrementally puts first audio at
roughly time-to-first-byte instead of after the whole utterance has been
synthesised. Stopping early closes the HTTP stream, which is also how a
barge-in stops us paying for audio nobody will hear.
"""
url = ELEVEN_URL_TMPL.format(voice_id=self.voice_id)
model = self.model if (self.model or "").startswith("eleven") else ELEVEN_DEFAULT_MODEL
rs = _StreamResampler(ELEVEN_SAMPLE_RATE, TARGET_SAMPLE_RATE)
pending = b""
odd = b"" # a chunk can split an int16 down the middle
async with self._client.stream(
"POST", url,
params={"output_format": f"pcm_{ELEVEN_SAMPLE_RATE}"},
headers={"xi-api-key": self.api_key},
json={"text": text, "model_id": model},
) as resp:
if resp.status_code >= 400:
await resp.aread() # body is needed for a useful message
resp.raise_for_status()
async for raw in resp.aiter_bytes():
if should_stop is not None and should_stop():
return
if not raw:
continue
raw = odd + raw
usable = len(raw) // 2 * 2
odd = raw[usable:]
if not usable:
continue
out = rs.feed(np.frombuffer(raw[:usable], dtype=np.int16))
if out.size:
pending += out.tobytes()
while len(pending) >= FRAME_BYTES:
yield pending[:FRAME_BYTES]
pending = pending[FRAME_BYTES:]
if should_stop is not None and should_stop():
return
tail = rs.flush()
if tail.size:
pending += tail.tobytes()
while len(pending) >= FRAME_BYTES:
yield pending[:FRAME_BYTES]
pending = pending[FRAME_BYTES:]
if pending:
yield pending + b"\x00" * (FRAME_BYTES - len(pending))
async def _piper_pcm(self, text: str) -> np.ndarray:
"""Synthesize via local Piper; return PCM already resampled to 8 kHz."""
await self._ensure_piper()
def _synth() -> bytes:
buf = io.BytesIO()
with wave.open(buf, "wb") as wf:
self._voice.synthesize(text, wf)
buf.seek(0)
with wave.open(buf) as wf:
return wf.readframes(wf.getnframes())
# Serialize synthesis: piper phonemizes via espeak-ng, whose C API is
# not thread-safe, and the PiperVoice may be shared across calls.
async with _VOICE_LOCK:
pcm_bytes = await asyncio.to_thread(_synth)
pcm = np.frombuffer(pcm_bytes, dtype=np.int16)
if self._native_sr != TARGET_SAMPLE_RATE:
pcm = _resample_linear(pcm, self._native_sr, TARGET_SAMPLE_RATE)
return pcm
async def synthesise(self, text: str, should_stop=None) -> AsyncIterator[bytes]:
"""Yield slin PCM bytes in 20 ms (320-byte) chunks.
`should_stop` is polled around synthesis as well as between frames. The
transports already poll it per frame, but that check cannot run until the
first frame exists, so without this a barge-in during synthesis had no
effect at all and the whole sentence was rendered anyway (issue #4).
On the ElevenLabs path stopping aborts the HTTP stream. On the Piper path
an in-flight render is *not* cancelled: it runs in a worker thread under
a process-wide lock, and abandoning the await would release that lock
while espeak-ng was still using the voice, which is the thread-safety
problem the lock exists to prevent. Piper is checked either side of the
render instead, so the audio is discarded rather than played.
"""
if not text.strip():
return
if should_stop is not None and should_stop():
return
if self._eleven and not self._eleven_disabled:
produced = False
try:
async for chunk in self._eleven_frames(text, should_stop):
produced = True
yield chunk
return
except Exception as exc:
if produced:
# Some audio already reached the caller; restarting the
# sentence on piper would repeat what they just heard.
log.warning("elevenlabs stream failed mid-utterance (%s); "
"no fallback for this one", exc)
return
status = getattr(getattr(exc, "response", None), "status_code", None)
if status in (401, 403):
self._eleven_disabled = True
log.error("elevenlabs auth failed (%s); piper fallback for "
"the rest of this call", status)
else:
log.warning("elevenlabs synthesis failed (%s); piper fallback "
"for this utterance", exc)
if should_stop is not None and should_stop():
return
pcm = await self._piper_pcm(text)
if should_stop is not None and should_stop():
return # interrupted while rendering; drop it unplayed
for chunk in _iter_frames(pcm):
yield chunk
if should_stop is not None and should_stop():
return
async def close(self) -> None:
self._voice = None # cached copy stays in _VOICE_CACHE for reuse
if self._client is not None:
try:
await self._client.aclose()
except Exception:
pass
self._client = None