API reference¶
Generated from the apogee-ai-voice source with mkdocstrings. Every symbol below is exported from apogee_ai_voice, so it is part of the supported public surface.
Application · DTOs¶
Bases: BaseModel
Bases: BaseModel
Bases: BaseModel
class-attribute
instance-attribute
¶
Python
languages: list[LanguageCode] = Field(default_factory=list)
Bases: BaseModel
Bases: BaseModel
Bases: BaseModel
Application · Use cases¶
Python
GenerateImageUseCase(backend: IImageGen)
Source code in apogee_ai_voice/application/use_cases/generate_image_use_case.py
async
¶
Python
execute(request: ImageRequest) -> ImageResult
Python
GenerateVideoUseCase(backend: IVideoGen)
Source code in apogee_ai_voice/application/use_cases/generate_video_use_case.py
async
¶
Python
execute(request: VideoRequest) -> VideoResult
Python
ListVoicesUseCase(tts: ITTS)
Source code in apogee_ai_voice/application/use_cases/list_voices_use_case.py
Python
RunVoicePipelineUseCase(runtime: PipecatStyleRuntime)
Source code in apogee_ai_voice/application/use_cases/run_voice_pipeline_use_case.py
async
¶
Python
execute(chunks: AsyncIterator[AudioChunk] | None = None) -> AsyncIterator[RealtimeEvent]
Python
SynthesizeUseCase(tts: ITTS)
Source code in apogee_ai_voice/application/use_cases/synthesize_use_case.py
async
¶
Python
execute(request: SynthesisRequest) -> SynthesisResult
Python
TranscribeUseCase(stt: ISTT)
Source code in apogee_ai_voice/application/use_cases/transcribe_use_case.py
async
¶
Python
execute(audio: bytes, *, language: LanguageCode = AUTO) -> Transcript
Domain¶
dataclass
¶
Python
AudioChunk(data: bytes, format: AudioFormatSpec = AudioFormatSpec(), sequence: int = 0, timestamp: datetime = (lambda: now(utc))(), is_final: bool = False)
A slice of audio bytes — either captured from mic or synthesised.
class-attribute
instance-attribute
¶
Python
format: AudioFormatSpec = field(default_factory=AudioFormatSpec)
class-attribute
instance-attribute
¶
Bases: str, Enum
dataclass
¶
Python
AudioFormatSpec(format: AudioFormat = PCM16, sample_rate_hz: int = 24000, channels: int = 1, bits_per_sample: int = 16)
Codec + sample rate + channels.
dataclass
¶
Python
ImageAsset(data: bytes = b'', mime_type: str = 'image/png', url: str | None = None, width: int = 0, height: int = 0, seed: int | None = None)
dataclass
¶
Python
ImageRequest(prompt: str, style: ImageStyle = NATURAL, width: int = 1024, height: int = 1024, n: int = 1, negative_prompt: str | None = None, seed: int | None = None, metadata: dict[str, str] = dict())
class-attribute
instance-attribute
¶
dataclass
¶
Python
ImageResult(backend: str, images: tuple[ImageAsset, ...] = tuple(), latency_ms: float = 0.0, cost_usd: float = 0.0, prompt: str = '')
class-attribute
instance-attribute
¶
Python
images: tuple[ImageAsset, ...] = field(default_factory=tuple)
Bases: str, Enum
Bases: str, Enum
dataclass
¶
Python
LatencyBudget(ttft_ms: int = 500, full_response_ms: int = 2000, vad_ms: int = 50, stt_partial_ms: int = 200, tts_first_audio_ms: int = 250)
End-to-end budget for a voice turn (in milliseconds).
Pipecat-style runtime breaks the budget into VAD + STT + LLM + TTS; each adapter may report observed latency back so we can detect drift.
dataclass
¶
Python
RealtimeEvent(kind: RealtimeEventKind, payload: dict[str, Any] = dict(), timestamp: datetime = (lambda: now(utc))())
dataclass
¶
Python
RealtimeSession(session_id: str = (lambda: token_hex(8))(), backend: str = 'echo', started_at: datetime = (lambda: now(utc))(), ended_at: datetime | None = None, metadata: dict[str, str] = dict(), events: list[RealtimeEvent] = list())
class-attribute
instance-attribute
¶
class-attribute
instance-attribute
¶
class-attribute
instance-attribute
¶
class-attribute
instance-attribute
¶
Python
events: list[RealtimeEvent] = field(default_factory=list)
Python
append(event: RealtimeEvent) -> None
dataclass
¶
Python
SynthesisRequest(text: str, voice_id: str = 'default', profile: VoiceProfile | None = None, language: LanguageCode = AUTO, output_format: AudioFormatSpec = AudioFormatSpec(), streaming: bool = False)
class-attribute
instance-attribute
¶
Python
output_format: AudioFormatSpec = field(default_factory=AudioFormatSpec)
class-attribute
instance-attribute
¶
If True, expect chunked output via stream().
dataclass
¶
Python
SynthesisResult(audio: bytes, format: AudioFormatSpec, backend: str = 'echo', voice_id: str = 'default', latency_ms: float = 0.0, text: str = '')
dataclass
¶
Python
Transcript(text: str, language: LanguageCode = AUTO, segments: tuple[TranscriptSegment, ...] = tuple(), is_final: bool = True, backend: str = 'echo', latency_ms: float = 0.0)
class-attribute
instance-attribute
¶
Python
segments: tuple[TranscriptSegment, ...] = field(default_factory=tuple)
dataclass
¶
Python
TranscriptSegment(text: str, start_seconds: float, end_seconds: float, speaker: str | None = None, confidence: float = 1.0)
dataclass
¶
Python
VadEvent(decision: VadDecision, energy: float = 0.0, timestamp: datetime = (lambda: now(utc))())
class-attribute
instance-attribute
¶
dataclass
¶
Python
VideoRequest(prompt: str, model: VideoModel = RUNWAY, duration_seconds: float = 5.0, width: int = 1280, height: int = 720, fps: int = 24, seed_image: bytes | None = None, metadata: dict[str, str] = dict())
class-attribute
instance-attribute
¶
dataclass
¶
Python
VideoResult(backend: str, data: bytes = b'', url: str | None = None, duration_seconds: float = 0.0, latency_ms: float = 0.0, cost_usd: float = 0.0)
dataclass
¶
Python
Voice(voice_id: str, name: str, backend: str, language: LanguageCode = AUTO, gender: str | None = None, description: str | None = None, sample_url: str | None = None, tags: tuple[str, ...] = tuple())
Voice metadata returned by ITTS.list_voices.
class-attribute
instance-attribute
¶
dataclass
¶
Python
VoiceProfile(voice_id: str, name: str = '', language: LanguageCode = AUTO, speed: float = 1.0, pitch: float = 0.0, style: str | None = None)
Domain · Enums¶
Domain · Exceptions¶
Bases: VoiceError
Source code in apogee_ai_voice/domain/exceptions/voice_exceptions.py
Bases: VoiceError
Bases: VoiceError
Bases: Exception
Base for apogee-ai-voice errors.
Domain · Protocols (ports)¶
Bases: Protocol
async
¶
Python
generate(request: ImageRequest) -> ImageResult
Bases: Protocol
Bidirectional duplex voice channel.
async
¶
Python
open(*, system_prompt: str = '', voice_id: str = 'default') -> RealtimeSession
async
¶
Python
send_audio(session: RealtimeSession, chunk: AudioChunk) -> None
async
¶
Python
send_text(session: RealtimeSession, text: str) -> None
async
¶
Python
events(session: RealtimeSession) -> AsyncIterator[RealtimeEvent]
async
¶
Python
interrupt(session: RealtimeSession) -> None
async
¶
Python
close(session: RealtimeSession) -> None
Bases: Protocol
async
¶
Python
transcribe(audio: bytes, *, language: LanguageCode = AUTO) -> Transcript
async
¶
Python
stream(chunks: AsyncIterator[AudioChunk], *, language: LanguageCode = AUTO) -> AsyncIterator[Transcript]
async
¶
Bases: Protocol
async
¶
Python
synthesize(request: SynthesisRequest) -> SynthesisResult
async
¶
Python
stream(request: SynthesisRequest) -> AsyncIterator[AudioChunk]
async
¶
Bases: Protocol
Audio transport — local mic/speaker, WebRTC, LiveKit, etc.
async
¶
async
¶
Python
stream_input() -> AsyncIterator[AudioChunk]
async
¶
Python
play(chunk: AudioChunk) -> None
async
¶
Bases: Protocol
async
¶
Python
generate(request: VideoRequest) -> VideoResult
Bases: Protocol
Python
evaluate(chunk: AudioChunk) -> VadEvent
Infrastructure¶
Adapter for AssemblyAI.
Lazy import: install via pip install 'apogee-ai-voice[assemblyai]'.
Source code in apogee_ai_voice/infrastructure/stt/assemblyai_stt.py
async
¶
Python
transcribe(audio: bytes, *, language: LanguageCode = AUTO) -> Transcript
Source code in apogee_ai_voice/infrastructure/stt/assemblyai_stt.py
Python
async def transcribe(
self,
audio: bytes,
*,
language: LanguageCode = LanguageCode.AUTO,
) -> Transcript:
try:
import assemblyai as aai # type: ignore
except ImportError as exc: # pragma: no cover
raise STTUnavailableException(self.name, str(exc)) from exc
if self._api_key:
aai.settings.api_key = self._api_key
def _run() -> aai.Transcript: # type: ignore
transcriber = aai.Transcriber()
return transcriber.transcribe(audio)
start = time.perf_counter()
try:
transcript_obj = await asyncio.to_thread(_run)
except Exception as exc: # noqa: BLE001
raise STTUnavailableException(self.name, str(exc)) from exc
latency = (time.perf_counter() - start) * 1000.0
text = getattr(transcript_obj, "text", "") or ""
return Transcript(
text=text,
language=language,
segments=(
TranscriptSegment(text=text, start_seconds=0.0, end_seconds=0.0),
) if text else (),
backend=self.name,
latency_ms=latency,
)
async
¶
Python
stream(chunks: AsyncIterator[AudioChunk], *, language: LanguageCode = AUTO) -> AsyncIterator[Transcript]
Source code in apogee_ai_voice/infrastructure/stt/assemblyai_stt.py
Python
async def stream(
self,
chunks: AsyncIterator[AudioChunk],
*,
language: LanguageCode = LanguageCode.AUTO,
) -> AsyncIterator[Transcript]:
buffer = bytearray()
async for chunk in chunks:
buffer.extend(chunk.data)
if chunk.is_final:
break
async def gen() -> AsyncIterator[Transcript]:
yield await self.transcribe(bytes(buffer), language=language)
return gen()
async
¶
Adapter for Cartesia (Sonic).
Lazy import: install via pip install 'apogee-ai-voice[cartesia]'.
Source code in apogee_ai_voice/infrastructure/tts/cartesia_tts.py
Python
def __init__(self, *, api_key: str | None = None, model_id: str = "sonic-english") -> None:
try:
import cartesia # type: ignore # noqa: F401
except ImportError as exc:
raise ImportError(
"CartesiaTTS requires `cartesia`. "
"Install with: pip install 'apogee-ai-voice[cartesia]'"
) from exc
self._api_key = api_key
self._model_id = model_id
async
¶
Python
synthesize(request: SynthesisRequest) -> SynthesisResult
Source code in apogee_ai_voice/infrastructure/tts/cartesia_tts.py
Python
async def synthesize(self, request: SynthesisRequest) -> SynthesisResult:
try:
from cartesia import AsyncCartesia # type: ignore
except ImportError as exc: # pragma: no cover
raise TTSUnavailableException(self.name, str(exc)) from exc
client = AsyncCartesia(api_key=self._api_key)
start = time.perf_counter()
try:
output = await client.tts.bytes(
model_id=self._model_id,
transcript=request.text,
voice={"id": request.voice_id} if request.voice_id else {"mode": "id", "id": "default"},
output_format={
"container": "raw",
"encoding": "pcm_s16le",
"sample_rate": request.output_format.sample_rate_hz,
},
)
except Exception as exc: # noqa: BLE001
raise TTSUnavailableException(self.name, str(exc)) from exc
latency = (time.perf_counter() - start) * 1000.0
return SynthesisResult(
audio=output,
format=request.output_format,
backend=self.name,
voice_id=request.voice_id,
latency_ms=latency,
text=request.text,
)
async
¶
Python
stream(request: SynthesisRequest) -> AsyncIterator[AudioChunk]
Source code in apogee_ai_voice/infrastructure/tts/cartesia_tts.py
async
¶
Adapter for Deepgram Nova.
Lazy import: install via pip install 'apogee-ai-voice[deepgram]'.
Source code in apogee_ai_voice/infrastructure/stt/deepgram_stt.py
Python
def __init__(
self,
*,
api_key: str | None = None,
model: str = "nova-2",
) -> None:
try:
import deepgram # type: ignore # noqa: F401
except ImportError as exc:
raise ImportError(
"DeepgramSTT requires `deepgram-sdk`. "
"Install with: pip install 'apogee-ai-voice[deepgram]'"
) from exc
self._api_key = api_key
self._model = model
async
¶
Python
transcribe(audio: bytes, *, language: LanguageCode = AUTO) -> Transcript
Source code in apogee_ai_voice/infrastructure/stt/deepgram_stt.py
Python
async def transcribe(
self,
audio: bytes,
*,
language: LanguageCode = LanguageCode.AUTO,
) -> Transcript:
try:
from deepgram import DeepgramClient, PrerecordedOptions # type: ignore
except ImportError as exc: # pragma: no cover
raise STTUnavailableException(self.name, str(exc)) from exc
client = DeepgramClient(self._api_key)
start = time.perf_counter()
try:
options = PrerecordedOptions(
model=self._model,
smart_format=True,
language=None if language == LanguageCode.AUTO else language.value,
)
response = client.listen.rest.v("1").transcribe_file(
{"buffer": audio}, options
)
except Exception as exc: # noqa: BLE001
raise STTUnavailableException(self.name, str(exc)) from exc
latency = (time.perf_counter() - start) * 1000.0
results = (
response.get("results", {}) if isinstance(response, dict)
else getattr(response, "results", {}) or {}
)
channels = (
results.get("channels", []) if isinstance(results, dict)
else getattr(results, "channels", []) or []
)
if not channels:
return Transcript(text="", language=language, backend=self.name, latency_ms=latency)
alternatives = channels[0].get("alternatives") if isinstance(channels[0], dict) else None
alt = (alternatives or [{}])[0]
text = alt.get("transcript", "") if isinstance(alt, dict) else ""
return Transcript(
text=text,
language=language,
segments=(
TranscriptSegment(
text=text,
start_seconds=0.0,
end_seconds=0.0,
confidence=alt.get("confidence", 1.0) if isinstance(alt, dict) else 1.0,
),
) if text else (),
backend=self.name,
latency_ms=latency,
)
async
¶
Python
stream(chunks: AsyncIterator[AudioChunk], *, language: LanguageCode = AUTO) -> AsyncIterator[Transcript]
Source code in apogee_ai_voice/infrastructure/stt/deepgram_stt.py
Python
async def stream(
self,
chunks: AsyncIterator[AudioChunk],
*,
language: LanguageCode = LanguageCode.AUTO,
) -> AsyncIterator[Transcript]:
# Real streaming would use Deepgram WebSocket; we collapse to
# batched transcription for simplicity. Production code should
# plug deepgram.listen.live API here.
buffer = bytearray()
async for chunk in chunks:
buffer.extend(chunk.data)
if chunk.is_final:
break
async def gen() -> AsyncIterator[Transcript]:
yield await self.transcribe(bytes(buffer), language=language)
return gen()
async
¶
Adapter for Deepgram Aura.
Lazy import: install via pip install 'apogee-ai-voice[deepgram]'.
Source code in apogee_ai_voice/infrastructure/tts/deepgram_tts.py
Python
def __init__(self, *, api_key: str | None = None, voice_model: str = "aura-asteria-en") -> None:
try:
import deepgram # type: ignore # noqa: F401
except ImportError as exc:
raise ImportError(
"DeepgramTTS requires `deepgram-sdk`. "
"Install with: pip install 'apogee-ai-voice[deepgram]'"
) from exc
self._api_key = api_key
self._voice_model = voice_model
async
¶
Python
synthesize(request: SynthesisRequest) -> SynthesisResult
Source code in apogee_ai_voice/infrastructure/tts/deepgram_tts.py
Python
async def synthesize(self, request: SynthesisRequest) -> SynthesisResult:
try:
from deepgram import DeepgramClient, SpeakOptions # type: ignore
except ImportError as exc: # pragma: no cover
raise TTSUnavailableException(self.name, str(exc)) from exc
client = DeepgramClient(self._api_key)
start = time.perf_counter()
try:
options = SpeakOptions(model=self._voice_model)
response = client.speak.v("1").stream({"text": request.text}, options)
audio = response.stream.read() if hasattr(response, "stream") else response
except Exception as exc: # noqa: BLE001
raise TTSUnavailableException(self.name, str(exc)) from exc
latency = (time.perf_counter() - start) * 1000.0
return SynthesisResult(
audio=audio if isinstance(audio, bytes) else b"",
format=request.output_format,
backend=self.name,
voice_id=request.voice_id,
latency_ms=latency,
text=request.text,
)
async
¶
Python
stream(request: SynthesisRequest) -> AsyncIterator[AudioChunk]
Source code in apogee_ai_voice/infrastructure/tts/deepgram_tts.py
async
¶
Python
list_voices() -> list[Voice]
Source code in apogee_ai_voice/infrastructure/tts/deepgram_tts.py
Python
async def list_voices(self) -> list[Voice]:
names = [
"aura-asteria-en", "aura-luna-en", "aura-stella-en",
"aura-orion-en", "aura-arcas-en", "aura-perseus-en",
"aura-angus-en", "aura-orpheus-en", "aura-helios-en",
"aura-zeus-en",
]
return [Voice(voice_id=n, name=n, backend=self.name) for n in names]
async
¶
Returns deterministic 1x1 PNG bytes — for tests/dev only.
async
¶
Python
generate(request: ImageRequest) -> ImageResult
Source code in apogee_ai_voice/infrastructure/image_gen/echo_image_gen.py
Python
async def generate(self, request: ImageRequest) -> ImageResult:
start = time.perf_counter()
assets = tuple(
ImageAsset(
data=self._ONE_PIXEL_PNG,
mime_type="image/png",
width=request.width,
height=request.height,
seed=request.seed,
)
for _ in range(request.n)
)
latency = (time.perf_counter() - start) * 1000.0
return ImageResult(
backend=self.name,
images=assets,
latency_ms=latency,
prompt=request.prompt,
)
Deterministic STT for tests: returns a fixed phrase scaled to audio length so callers can validate plumbing without paid providers.
Source code in apogee_ai_voice/infrastructure/stt/echo_stt.py
async
¶
Python
transcribe(audio: bytes, *, language: LanguageCode = AUTO) -> Transcript
Source code in apogee_ai_voice/infrastructure/stt/echo_stt.py
Python
async def transcribe(
self,
audio: bytes,
*,
language: LanguageCode = LanguageCode.AUTO,
) -> Transcript:
start = time.perf_counter()
# Use byte length as a proxy for duration (24kHz mono PCM16 → /48000)
duration = max(0.1, len(audio) / 48000.0)
text = self._phrase
latency = (time.perf_counter() - start) * 1000.0
return Transcript(
text=text,
language=language,
segments=(
TranscriptSegment(
text=text,
start_seconds=0.0,
end_seconds=duration,
confidence=1.0,
),
),
backend=self.name,
latency_ms=latency,
)
async
¶
Python
stream(chunks: AsyncIterator[AudioChunk], *, language: LanguageCode = AUTO) -> AsyncIterator[Transcript]
Source code in apogee_ai_voice/infrastructure/stt/echo_stt.py
Python
async def stream(
self,
chunks: AsyncIterator[AudioChunk],
*,
language: LanguageCode = LanguageCode.AUTO,
) -> AsyncIterator[Transcript]:
async def gen() -> AsyncIterator[Transcript]:
buffer = bytearray()
async for chunk in chunks:
buffer.extend(chunk.data)
yield Transcript(
text=self._phrase,
language=language,
is_final=chunk.is_final,
backend=self.name,
)
if chunk.is_final:
break
del buffer
return gen()
async
¶
Synthesises a deterministic 440Hz sine wave mod by text length.
No external API. Useful in CI to verify the contract end-to-end without hitting paid providers. Output is PCM16, 24kHz mono.
async
¶
Python
synthesize(request: SynthesisRequest) -> SynthesisResult
Source code in apogee_ai_voice/infrastructure/tts/echo_tts.py
Python
async def synthesize(self, request: SynthesisRequest) -> SynthesisResult:
start = time.perf_counter()
spec = request.output_format
# Duration roughly proportional to text length (50ms per char)
duration_seconds = max(0.1, len(request.text) * 0.05)
sample_count = int(spec.sample_rate_hz * duration_seconds)
amplitude = 8000
pcm = bytearray()
for i in range(sample_count):
value = int(amplitude * math.sin(2 * math.pi * 440 * i / spec.sample_rate_hz))
pcm.extend(struct.pack("<h", value))
latency = (time.perf_counter() - start) * 1000.0
return SynthesisResult(
audio=bytes(pcm),
format=spec,
backend=self.name,
voice_id=request.voice_id,
latency_ms=latency,
text=request.text,
)
async
¶
Python
stream(request: SynthesisRequest) -> AsyncIterator[AudioChunk]
Source code in apogee_ai_voice/infrastructure/tts/echo_tts.py
Python
async def stream(self, request: SynthesisRequest) -> AsyncIterator[AudioChunk]:
result = await self.synthesize(request)
chunk_bytes = max(1, request.output_format.bytes_per_second // 10)
async def gen() -> AsyncIterator[AudioChunk]:
seq = 0
data = result.audio
for offset in range(0, len(data), chunk_bytes):
yield AudioChunk(
data=data[offset : offset + chunk_bytes],
format=result.format,
sequence=seq,
is_final=offset + chunk_bytes >= len(data),
)
seq += 1
await asyncio.sleep(0)
return gen()
async
¶
Returns deterministic empty MP4 stub — for tests.
async
¶
Python
generate(request: VideoRequest) -> VideoResult
Source code in apogee_ai_voice/infrastructure/video_gen/echo_video_gen.py
Adapter for ElevenLabs.
Lazy import: install via pip install 'apogee-ai-voice[elevenlabs]'.
Source code in apogee_ai_voice/infrastructure/tts/elevenlabs_tts.py
Python
def __init__(self, *, api_key: str | None = None, model: str = "eleven_multilingual_v2") -> None:
try:
import elevenlabs # type: ignore # noqa: F401
except ImportError as exc:
raise ImportError(
"ElevenLabsTTS requires `elevenlabs`. "
"Install with: pip install 'apogee-ai-voice[elevenlabs]'"
) from exc
self._api_key = api_key
self._model = model
async
¶
Python
synthesize(request: SynthesisRequest) -> SynthesisResult
Source code in apogee_ai_voice/infrastructure/tts/elevenlabs_tts.py
Python
async def synthesize(self, request: SynthesisRequest) -> SynthesisResult:
try:
from elevenlabs.client import AsyncElevenLabs # type: ignore
except ImportError as exc: # pragma: no cover
raise TTSUnavailableException(self.name, str(exc)) from exc
client = AsyncElevenLabs(api_key=self._api_key)
start = time.perf_counter()
try:
stream = client.text_to_speech.convert(
voice_id=request.voice_id,
model_id=self._model,
text=request.text,
output_format="mp3_44100_128",
)
audio = b""
async for piece in stream:
audio += piece
except Exception as exc: # noqa: BLE001
raise TTSUnavailableException(self.name, str(exc)) from exc
latency = (time.perf_counter() - start) * 1000.0
return SynthesisResult(
audio=audio,
format=request.output_format,
backend=self.name,
voice_id=request.voice_id,
latency_ms=latency,
text=request.text,
)
async
¶
Python
stream(request: SynthesisRequest) -> AsyncIterator[AudioChunk]
Source code in apogee_ai_voice/infrastructure/tts/elevenlabs_tts.py
Python
async def stream(self, request: SynthesisRequest) -> AsyncIterator[AudioChunk]:
try:
from elevenlabs.client import AsyncElevenLabs # type: ignore
except ImportError as exc: # pragma: no cover
raise TTSUnavailableException(self.name, str(exc)) from exc
client = AsyncElevenLabs(api_key=self._api_key)
async def gen() -> AsyncIterator[AudioChunk]:
try:
stream = client.text_to_speech.convert(
voice_id=request.voice_id,
model_id=self._model,
text=request.text,
output_format="pcm_24000",
)
except Exception as exc: # noqa: BLE001
raise TTSUnavailableException(self.name, str(exc)) from exc
seq = 0
async for piece in stream:
yield AudioChunk(
data=piece,
format=request.output_format,
sequence=seq,
)
seq += 1
await asyncio.sleep(0)
return gen()
async
¶
Python
list_voices() -> list[Voice]
Source code in apogee_ai_voice/infrastructure/tts/elevenlabs_tts.py
Python
async def list_voices(self) -> list[Voice]:
try:
from elevenlabs.client import AsyncElevenLabs # type: ignore
except ImportError as exc: # pragma: no cover
raise TTSUnavailableException(self.name, str(exc)) from exc
client = AsyncElevenLabs(api_key=self._api_key)
try:
response = await client.voices.get_all()
except Exception as exc: # noqa: BLE001
raise TTSUnavailableException(self.name, str(exc)) from exc
voices = getattr(response, "voices", None) or []
return [
Voice(
voice_id=getattr(v, "voice_id", ""),
name=getattr(v, "name", ""),
backend=self.name,
description=getattr(v, "description", None),
)
for v in voices
if getattr(v, "voice_id", None)
]
async
¶
RMS-energy VAD — no deps, OK for tests/dev.
Real production should use Silero or WebRTC VAD adapters.
Source code in apogee_ai_voice/infrastructure/vad/energy_vad.py
Python
evaluate(chunk: AudioChunk) -> VadEvent
Source code in apogee_ai_voice/infrastructure/vad/energy_vad.py
Python
def evaluate(self, chunk: AudioChunk) -> VadEvent:
if not chunk.data or chunk.format.bits_per_sample != 16:
return VadEvent(decision=VadDecision.UNCERTAIN, energy=0.0)
# Treat as little-endian PCM16
sample_count = len(chunk.data) // 2
if sample_count == 0:
return VadEvent(decision=VadDecision.SILENCE, energy=0.0)
samples = struct.unpack(f"<{sample_count}h", chunk.data[: sample_count * 2])
energy = (sum(s * s for s in samples) / sample_count) ** 0.5
decision = VadDecision.SPEECH if energy >= self._threshold else VadDecision.SILENCE
return VadEvent(decision=decision, energy=float(energy))
Bases: _HttpImageGen
Adapter for Replicate-style Flux endpoints.
Source code in apogee_ai_voice/infrastructure/image_gen/_http_image_gen.py
Bases: _HttpImageGen
Adapter for Google Imagen / Vertex AI image generation.
Source code in apogee_ai_voice/infrastructure/image_gen/_http_image_gen.py
Thin LiveKit Agents wrapper.
Lazy import: install via pip install 'apogee-ai-voice[livekit]'.
Production deployments should plug LiveKit Agents' room participant
APIs here. This adapter exposes the framework-agnostic ITransport
surface so the rest of the stack can stay decoupled.
Source code in apogee_ai_voice/infrastructure/transports/livekit_transport.py
Python
def __init__(
self,
*,
url: str,
token: str,
sample_rate_hz: int = 16000,
channels: int = 1,
) -> None:
try:
import livekit # type: ignore # noqa: F401
except ImportError as exc:
raise ImportError(
"LiveKitTransport requires `livekit`. "
"Install with: pip install 'apogee-ai-voice[livekit]'"
) from exc
self._url = url
self._token = token
self._format = AudioFormatSpec(
sample_rate_hz=sample_rate_hz, channels=channels
)
self._room = None
self._inbound: asyncio.Queue[AudioChunk | None] = asyncio.Queue()
self._opened = False
async
¶
Source code in apogee_ai_voice/infrastructure/transports/livekit_transport.py
Python
async def open(self) -> None:
try:
from livekit import rtc # type: ignore
except ImportError as exc: # pragma: no cover
raise TransportError(str(exc)) from exc
try:
self._room = rtc.Room()
await self._room.connect(self._url, self._token)
self._opened = True
except Exception as exc: # noqa: BLE001
raise TransportError(str(exc)) from exc
async
¶
Python
stream_input() -> AsyncIterator[AudioChunk]
async
¶
Python
play(chunk: AudioChunk) -> None
async
¶
In-process queue-based transport for tests and unit demos.
Implements ITransport: callers can inject(chunk) to feed audio
in, and played accumulates the bytes that the runtime "plays".
Source code in apogee_ai_voice/infrastructure/transports/local_audio_transport.py
async
¶
async
¶
Python
stream_input() -> AsyncIterator[AudioChunk]
async
¶
Python
play(chunk: AudioChunk) -> None
async
¶
async
¶
Python
inject(chunk: AudioChunk) -> None
async
¶
Adapter for OpenAI images.generate (DALL-E 3 / gpt-image-1).
Lazy import: install via pip install 'apogee-ai-voice[openai]'.
Source code in apogee_ai_voice/infrastructure/image_gen/openai_image_gen.py
Python
def __init__(self, *, api_key: str | None = None, model: str = "gpt-image-1") -> None:
try:
import openai # type: ignore # noqa: F401
except ImportError as exc:
raise ImportError(
"OpenAIImageGen requires `openai`. "
"Install with: pip install 'apogee-ai-voice[openai]'"
) from exc
self._api_key = api_key
self._model = model
async
¶
Python
generate(request: ImageRequest) -> ImageResult
Source code in apogee_ai_voice/infrastructure/image_gen/openai_image_gen.py
Python
async def generate(self, request: ImageRequest) -> ImageResult:
try:
from openai import AsyncOpenAI # type: ignore
except ImportError as exc: # pragma: no cover
raise ImageGenUnavailable(self.name, str(exc)) from exc
client = AsyncOpenAI(api_key=self._api_key)
start = time.perf_counter()
try:
response = await client.images.generate(
model=self._model,
prompt=request.prompt,
n=request.n,
size=f"{request.width}x{request.height}",
response_format="b64_json",
)
except Exception as exc: # noqa: BLE001
raise ImageGenUnavailable(self.name, str(exc)) from exc
latency = (time.perf_counter() - start) * 1000.0
assets: list[ImageAsset] = []
for item in response.data:
data = b""
url = None
if getattr(item, "b64_json", None):
data = base64.b64decode(item.b64_json)
elif getattr(item, "url", None):
url = item.url
assets.append(
ImageAsset(
data=data,
url=url,
width=request.width,
height=request.height,
seed=request.seed,
)
)
return ImageResult(
backend=self.name,
images=tuple(assets),
latency_ms=latency,
prompt=request.prompt,
)
Adapter for OpenAI's Realtime API (gpt-4o-realtime).
Lazy import: install via pip install 'apogee-ai-voice[openai]'.
Source code in apogee_ai_voice/infrastructure/realtime/openai_realtime.py
Python
def __init__(
self,
*,
api_key: str | None = None,
model: str = "gpt-4o-realtime-preview",
) -> None:
try:
import openai # type: ignore # noqa: F401
except ImportError as exc:
raise ImportError(
"OpenAIRealtimeAdapter requires `openai`. "
"Install with: pip install 'apogee-ai-voice[openai]'"
) from exc
self._api_key = api_key
self._model = model
self._connections: dict[str, object] = {}
async
¶
Python
open(*, system_prompt: str = '', voice_id: str = 'alloy') -> RealtimeSession
Source code in apogee_ai_voice/infrastructure/realtime/openai_realtime.py
Python
async def open(self, *, system_prompt: str = "", voice_id: str = "alloy") -> RealtimeSession:
try:
from openai import AsyncOpenAI # type: ignore
except ImportError as exc: # pragma: no cover
raise RealtimeError(str(exc)) from exc
client = AsyncOpenAI(api_key=self._api_key)
session = RealtimeSession(backend=self.name, metadata={"voice_id": voice_id})
try:
connection_ctx = client.beta.realtime.connect(model=self._model)
connection = await connection_ctx.__aenter__()
await connection.session.update(
session={"voice": voice_id, "instructions": system_prompt}
)
self._connections[session.session_id] = connection
except Exception as exc: # noqa: BLE001
raise RealtimeError(str(exc)) from exc
return session
async
¶
Python
send_audio(session: RealtimeSession, chunk: AudioChunk) -> None
Source code in apogee_ai_voice/infrastructure/realtime/openai_realtime.py
Python
async def send_audio(self, session: RealtimeSession, chunk: AudioChunk) -> None:
connection = self._connections.get(session.session_id)
if connection is None:
raise RealtimeError(f"session {session.session_id!r} is not open")
try:
await connection.input_audio_buffer.append(audio=chunk.data)
except Exception as exc: # noqa: BLE001
raise RealtimeError(str(exc)) from exc
async
¶
Python
send_text(session: RealtimeSession, text: str) -> None
Source code in apogee_ai_voice/infrastructure/realtime/openai_realtime.py
Python
async def send_text(self, session: RealtimeSession, text: str) -> None:
connection = self._connections.get(session.session_id)
if connection is None:
raise RealtimeError(f"session {session.session_id!r} is not open")
try:
await connection.conversation.item.create(
item={"type": "message", "role": "user", "content": [{"type": "input_text", "text": text}]}
)
await connection.response.create()
except Exception as exc: # noqa: BLE001
raise RealtimeError(str(exc)) from exc
async
¶
Python
events(session: RealtimeSession) -> AsyncIterator[RealtimeEvent]
Source code in apogee_ai_voice/infrastructure/realtime/openai_realtime.py
Python
async def events(self, session: RealtimeSession) -> AsyncIterator[RealtimeEvent]:
connection = self._connections.get(session.session_id)
if connection is None:
raise RealtimeError(f"session {session.session_id!r} is not open")
async def gen() -> AsyncIterator[RealtimeEvent]:
try:
async for event in connection:
payload = (
event.model_dump() if hasattr(event, "model_dump")
else (event if isinstance(event, dict) else {})
)
kind = _map_event(payload.get("type", ""))
yield RealtimeEvent(kind=kind, payload=payload)
except Exception as exc: # noqa: BLE001
raise RealtimeError(str(exc)) from exc
return gen()
async
¶
Python
interrupt(session: RealtimeSession) -> None
Source code in apogee_ai_voice/infrastructure/realtime/openai_realtime.py
async
¶
Python
close(session: RealtimeSession) -> None
Source code in apogee_ai_voice/infrastructure/realtime/openai_realtime.py
Adapter for OpenAI audio.speech (tts-1 / tts-1-hd / gpt-4o-mini-tts).
Lazy import: install via pip install 'apogee-ai-voice[openai]'.
Source code in apogee_ai_voice/infrastructure/tts/openai_tts.py
Python
def __init__(self, *, api_key: str | None = None, model: str = "tts-1") -> None:
try:
import openai # type: ignore # noqa: F401
except ImportError as exc:
raise ImportError(
"OpenAITTS requires `openai`. "
"Install with: pip install 'apogee-ai-voice[openai]'"
) from exc
self._api_key = api_key
self._model = model
async
¶
Python
synthesize(request: SynthesisRequest) -> SynthesisResult
Source code in apogee_ai_voice/infrastructure/tts/openai_tts.py
Python
async def synthesize(self, request: SynthesisRequest) -> SynthesisResult:
try:
from openai import AsyncOpenAI # type: ignore
except ImportError as exc: # pragma: no cover
raise TTSUnavailableException(self.name, str(exc)) from exc
client = AsyncOpenAI(api_key=self._api_key)
start = time.perf_counter()
try:
response = await client.audio.speech.create(
model=self._model,
voice=request.voice_id or "alloy",
input=request.text,
response_format=_format_to_response(request.output_format.format),
speed=request.profile.speed if request.profile else 1.0,
)
except Exception as exc: # noqa: BLE001
raise TTSUnavailableException(self.name, str(exc)) from exc
audio = await response.aread() if hasattr(response, "aread") else response.read()
latency = (time.perf_counter() - start) * 1000.0
return SynthesisResult(
audio=audio,
format=request.output_format,
backend=self.name,
voice_id=request.voice_id,
latency_ms=latency,
text=request.text,
)
async
¶
Python
stream(request: SynthesisRequest) -> AsyncIterator[AudioChunk]
Source code in apogee_ai_voice/infrastructure/tts/openai_tts.py
async
¶
dataclass
¶
Python
PipecatRunStats(vad_ms: float = 0.0, stt_ms: float = 0.0, llm_ms: float = 0.0, tts_first_audio_ms: float = 0.0, full_response_ms: float = 0.0)
Latency stats accumulated during a single user turn.
Python
PipecatStyleRuntime(*, vad: IVoiceActivityDetector, stt: ISTT, tts: ITTS, responder: LLMResponder, transport: ITransport | None = None, format: AudioFormatSpec | None = None, budget: LatencyBudget | None = None, language: LanguageCode = AUTO, silence_chunks_to_finalise: int = 3)
VAD → STT → LLM → TTS pipeline.
Mirrors Pipecat's flow: stream chunks from the transport into the VAD,
accumulate during SPEECH, transcribe on SILENCE boundary, ask
the LLM, synthesise and play. Supports interruption: a new SPEECH
chunk while the agent is talking emits an INTERRUPTED event and
the in-flight TTS is cancelled.
Source code in apogee_ai_voice/infrastructure/realtime/pipecat_runtime.py
Python
def __init__(
self,
*,
vad: IVoiceActivityDetector,
stt: ISTT,
tts: ITTS,
responder: LLMResponder,
transport: ITransport | None = None,
format: AudioFormatSpec | None = None,
budget: LatencyBudget | None = None,
language: LanguageCode = LanguageCode.AUTO,
silence_chunks_to_finalise: int = 3,
) -> None:
self._vad = vad
self._stt = stt
self._tts = tts
self._responder = responder
self._transport = transport
self._format = format or AudioFormatSpec()
self._budget = budget or LatencyBudget()
self._language = language
self._silence_threshold = max(1, silence_chunks_to_finalise)
async
¶
Python
run(chunks: AsyncIterator[AudioChunk] | None = None) -> AsyncIterator[RealtimeEvent]
Drive a single conversation. Yields :class:RealtimeEvent items.
Source code in apogee_ai_voice/infrastructure/realtime/pipecat_runtime.py
Python
async def run(
self,
chunks: AsyncIterator[AudioChunk] | None = None,
) -> AsyncIterator[RealtimeEvent]:
"""Drive a single conversation. Yields :class:`RealtimeEvent` items."""
async def gen() -> AsyncIterator[RealtimeEvent]:
session = RealtimeSession(backend=self.name)
yield RealtimeEvent(
kind=RealtimeEventKind.SESSION_STARTED,
payload={"session_id": session.session_id},
)
input_stream = chunks
if input_stream is None:
if self._transport is None:
raise ValueError("Either chunks or transport must be provided")
await self._transport.open()
input_stream = await self._transport.stream_input()
buffer = bytearray()
silence_run = 0
speaking = False
stats = PipecatRunStats()
agent_busy = False
async for chunk in input_stream:
vad_started = time.perf_counter()
event = self._vad.evaluate(chunk)
stats.vad_ms += (time.perf_counter() - vad_started) * 1000.0
if event.decision == VadDecision.SPEECH:
if agent_busy:
yield RealtimeEvent(kind=RealtimeEventKind.INTERRUPTED)
agent_busy = False
if not speaking:
speaking = True
yield RealtimeEvent(
kind=RealtimeEventKind.USER_SPEECH_START,
payload={"energy": event.energy},
)
buffer.extend(chunk.data)
silence_run = 0
elif event.decision == VadDecision.SILENCE and speaking:
silence_run += 1
buffer.extend(chunk.data)
if silence_run >= self._silence_threshold:
speaking = False
yield RealtimeEvent(kind=RealtimeEventKind.USER_SPEECH_END)
async for evt in self._handle_turn(buffer, stats):
yield evt
if evt.kind == RealtimeEventKind.AGENT_AUDIO_DONE:
agent_busy = False
elif evt.kind == RealtimeEventKind.AGENT_AUDIO_CHUNK:
agent_busy = True
buffer.clear()
silence_run = 0
if speaking and buffer:
async for evt in self._handle_turn(buffer, stats):
yield evt
if self._transport is not None:
await self._transport.close()
session.ended_at = None
yield RealtimeEvent(
kind=RealtimeEventKind.SESSION_ENDED,
payload={
"session_id": session.session_id,
"vad_ms": stats.vad_ms,
"stt_ms": stats.stt_ms,
"llm_ms": stats.llm_ms,
"tts_first_audio_ms": stats.tts_first_audio_ms,
"full_response_ms": stats.full_response_ms,
},
)
return gen()
Adapter for Runway ML Gen-3 / Gen-4.
Lazy import: install via pip install 'apogee-ai-voice[runway]'.
Source code in apogee_ai_voice/infrastructure/video_gen/runway_video_gen.py
Python
def __init__(self, *, api_key: str | None = None, model: str = "gen3a_turbo") -> None:
try:
import runwayml # type: ignore # noqa: F401
except ImportError as exc:
raise ImportError(
"RunwayVideoGen requires `runwayml`. "
"Install with: pip install 'apogee-ai-voice[runway]'"
) from exc
self._api_key = api_key
self._model = model
async
¶
Python
generate(request: VideoRequest) -> VideoResult
Source code in apogee_ai_voice/infrastructure/video_gen/runway_video_gen.py
Python
async def generate(self, request: VideoRequest) -> VideoResult:
try:
from runwayml import AsyncRunwayML # type: ignore
except ImportError as exc: # pragma: no cover
raise VideoGenUnavailable(self.name, str(exc)) from exc
client = AsyncRunwayML(api_key=self._api_key)
start = time.perf_counter()
try:
task = await client.text_to_video.create(
model=self._model,
prompt_text=request.prompt,
duration=int(request.duration_seconds),
ratio=f"{request.width}:{request.height}",
)
# Poll until done
for _ in range(60):
refreshed = await client.tasks.retrieve(task.id)
if getattr(refreshed, "status", "") in {"SUCCEEDED", "FAILED"}:
task = refreshed
break
await asyncio.sleep(2)
except Exception as exc: # noqa: BLE001
raise VideoGenUnavailable(self.name, str(exc)) from exc
latency = (time.perf_counter() - start) * 1000.0
urls = getattr(task, "output", []) or []
url = urls[0] if urls else None
return VideoResult(
backend=self.name,
url=url,
duration_seconds=request.duration_seconds,
latency_ms=latency,
)
Adapter for OpenAI Sora video generation.
Lazy import via pip install 'apogee-ai-voice[openai]'. The actual
Sora endpoint surface evolves quickly; this adapter wraps the
high-level client.videos call when present and degrades gracefully
when it isn't.
Source code in apogee_ai_voice/infrastructure/video_gen/openai_sora_gen.py
Python
def __init__(self, *, api_key: str | None = None, model: str = "sora-1") -> None:
try:
import openai # type: ignore # noqa: F401
except ImportError as exc:
raise ImportError(
"SoraVideoGen requires `openai`. "
"Install with: pip install 'apogee-ai-voice[openai]'"
) from exc
self._api_key = api_key
self._model = model
async
¶
Python
generate(request: VideoRequest) -> VideoResult
Source code in apogee_ai_voice/infrastructure/video_gen/openai_sora_gen.py
Python
async def generate(self, request: VideoRequest) -> VideoResult:
try:
from openai import AsyncOpenAI # type: ignore
except ImportError as exc: # pragma: no cover
raise VideoGenUnavailable(self.name, str(exc)) from exc
client = AsyncOpenAI(api_key=self._api_key)
videos_api = getattr(client, "videos", None)
if videos_api is None:
raise VideoGenUnavailable(
self.name,
"openai>=… with videos API is required",
)
start = time.perf_counter()
try:
response = await videos_api.generate(
model=self._model,
prompt=request.prompt,
duration=int(request.duration_seconds),
size=f"{request.width}x{request.height}",
)
except Exception as exc: # noqa: BLE001
raise VideoGenUnavailable(self.name, str(exc)) from exc
latency = (time.perf_counter() - start) * 1000.0
url = getattr(response, "url", None)
return VideoResult(
backend=self.name,
url=url,
duration_seconds=request.duration_seconds,
latency_ms=latency,
)
Bases: _HttpImageGen
Adapter for Hugging Face Inference API stable-diffusion endpoints.
Source code in apogee_ai_voice/infrastructure/image_gen/_http_image_gen.py
Adapter for Google Veo via the Generative Language API.
Source code in apogee_ai_voice/infrastructure/video_gen/google_veo_gen.py
async
¶
Python
generate(request: VideoRequest) -> VideoResult
Source code in apogee_ai_voice/infrastructure/video_gen/google_veo_gen.py
Python
async def generate(self, request: VideoRequest) -> VideoResult:
payload = {
"prompt": request.prompt,
"duration": int(request.duration_seconds),
"aspectRatio": _aspect(request.width, request.height),
}
headers = {
"x-goog-api-key": self._api_key,
"Content-Type": "application/json",
}
start = time.perf_counter()
try:
async with httpx.AsyncClient(timeout=180.0) as client:
response = await client.post(self._ENDPOINT, json=payload, headers=headers)
response.raise_for_status()
data = response.json()
except Exception as exc: # noqa: BLE001
raise VideoGenUnavailable(self.name, str(exc)) from exc
latency = (time.perf_counter() - start) * 1000.0
url = (data.get("video") or {}).get("uri") if isinstance(data, dict) else None
return VideoResult(
backend=self.name,
url=url,
duration_seconds=request.duration_seconds,
latency_ms=latency,
)
Python
VoiceRegistry(*, tts: Mapping[str, ITTS] | None = None, stt: Mapping[str, ISTT] | None = None, image: Mapping[str, IImageGen] | None = None, video: Mapping[str, IVideoGen] | None = None)
Holds backends keyed by category + name.
Source code in apogee_ai_voice/infrastructure/registry/voice_registry.py
Python
def __init__(
self,
*,
tts: Mapping[str, ITTS] | None = None,
stt: Mapping[str, ISTT] | None = None,
image: Mapping[str, IImageGen] | None = None,
video: Mapping[str, IVideoGen] | None = None,
) -> None:
self._tts: dict[str, ITTS] = dict(tts or {})
self._stt: dict[str, ISTT] = dict(stt or {})
self._image: dict[str, IImageGen] = dict(image or {})
self._video: dict[str, IVideoGen] = dict(video or {})
Adapter for OpenAI Whisper via audio.transcriptions.
Lazy import: install via pip install 'apogee-ai-voice[openai]'.
Source code in apogee_ai_voice/infrastructure/stt/whisper_stt.py
Python
def __init__(self, *, api_key: str | None = None, model: str = "whisper-1") -> None:
try:
import openai # type: ignore # noqa: F401
except ImportError as exc:
raise ImportError(
"WhisperSTT requires `openai`. "
"Install with: pip install 'apogee-ai-voice[openai]'"
) from exc
self._api_key = api_key
self._model = model
async
¶
Python
transcribe(audio: bytes, *, language: LanguageCode = AUTO) -> Transcript
Source code in apogee_ai_voice/infrastructure/stt/whisper_stt.py
Python
async def transcribe(
self,
audio: bytes,
*,
language: LanguageCode = LanguageCode.AUTO,
) -> Transcript:
try:
from openai import AsyncOpenAI # type: ignore
except ImportError as exc: # pragma: no cover
raise STTUnavailableException(self.name, str(exc)) from exc
client = AsyncOpenAI(api_key=self._api_key)
start = time.perf_counter()
try:
buffer = io.BytesIO(audio)
buffer.name = "audio.wav"
response = await client.audio.transcriptions.create(
model=self._model,
file=buffer,
language=None if language == LanguageCode.AUTO else language.value.split("-")[0],
response_format="verbose_json",
)
except Exception as exc: # noqa: BLE001
raise STTUnavailableException(self.name, str(exc)) from exc
latency = (time.perf_counter() - start) * 1000.0
text = getattr(response, "text", "") or ""
segments_raw = getattr(response, "segments", []) or []
segments = tuple(
TranscriptSegment(
text=str(s.get("text", "")),
start_seconds=float(s.get("start", 0.0)),
end_seconds=float(s.get("end", 0.0)),
confidence=float(s.get("no_speech_prob", 0.0)) if isinstance(s, dict) else 1.0,
)
for s in segments_raw
if isinstance(s, dict)
)
return Transcript(
text=text,
language=language,
segments=segments,
backend=self.name,
latency_ms=latency,
)
async
¶
Python
stream(chunks: AsyncIterator[AudioChunk], *, language: LanguageCode = AUTO) -> AsyncIterator[Transcript]
Source code in apogee_ai_voice/infrastructure/stt/whisper_stt.py
Python
async def stream(
self,
chunks: AsyncIterator[AudioChunk],
*,
language: LanguageCode = LanguageCode.AUTO,
) -> AsyncIterator[Transcript]:
# Whisper REST API doesn't stream; collect and call once
buffer = bytearray()
last_format = None
async for chunk in chunks:
buffer.extend(chunk.data)
last_format = chunk.format
if chunk.is_final:
break
del last_format
async def gen() -> AsyncIterator[Transcript]:
yield await self.transcribe(bytes(buffer), language=language)
return gen()