LiveKit Agents

import os

import openai
from livekit.agents import (
    APIConnectionError,
    APIConnectOptions,
    APIStatusError,
    APITimeoutError,
    tts,
    utils,
)
from livekit.agents.types import DEFAULT_API_CONNECT_OPTIONS

# pcm is raw 16-bit mono at 24 kHz on tts-kokoro; LiveKit decodes and resamples mp3 and wav
MIME_TYPES = {"pcm": "audio/pcm", "mp3": "audio/mpeg", "wav": "audio/wav"}


class VeniceTTS(tts.TTS):
    def __init__(
        self,
        *,
        model: str = "tts-kokoro",
        voice: str = "af_sky",
        response_format: str = "pcm",
        speed: float = 1.0,
        base_url: str = "https://api.venice.ai/api/v1",
        api_key: str | None = None,
    ) -> None:
        super().__init__(
            capabilities=tts.TTSCapabilities(streaming=False),
            sample_rate=24000,
            num_channels=1,
        )
        self._model = model
        self._voice = voice
        self._format = response_format
        self._mime_type = MIME_TYPES[response_format]
        self._speed = speed
        self._client = openai.AsyncOpenAI(
            base_url=base_url,
            api_key=api_key or os.environ["VENICE_API_KEY"],
        )

    @property
    def model(self) -> str:
        return self._model

    @property
    def provider(self) -> str:
        return "venice"

    def synthesize(
        self, text: str, *, conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS
    ) -> tts.ChunkedStream:
        return VeniceChunkedStream(tts=self, input_text=text, conn_options=conn_options)

    async def aclose(self) -> None:
        await self._client.close()


class VeniceChunkedStream(tts.ChunkedStream):
    async def _run(self, output_emitter: tts.AudioEmitter) -> None:
        venice: VeniceTTS = self._tts
        try:
            async with venice._client.audio.speech.with_streaming_response.create(
                model=venice._model,
                voice=venice._voice,
                input=self.input_text,
                response_format=venice._format,
                speed=venice._speed,
                timeout=30,
            ) as response:
                output_emitter.initialize(
                    request_id=response.request_id or utils.shortuuid(),
                    sample_rate=venice.sample_rate,
                    num_channels=venice.num_channels,
                    mime_type=venice._mime_type,
                )
                async for chunk in response.iter_bytes():
                    output_emitter.push(chunk)
            output_emitter.flush()
        except openai.APITimeoutError:
            raise APITimeoutError() from None
        except openai.APIStatusError as e:
            raise APIStatusError(
                e.message, status_code=e.status_code, request_id=e.request_id, body=e.body
            ) from None
        except Exception as e:
            raise APIConnectionError() from e

import os

from livekit import agents
from livekit.agents import Agent, AgentSession, RoomInputOptions
from livekit.plugins import openai, silero

from venice_tts import VeniceTTS

VENICE_BASE_URL = "https://api.venice.ai/api/v1"
VENICE_API_KEY = os.environ["VENICE_API_KEY"]


class Assistant(Agent):
    def __init__(self) -> None:
        super().__init__(
            instructions="You are a helpful, concise voice assistant powered by Venice.",
        )


async def entrypoint(ctx: agents.JobContext):
    session = AgentSession(
        # Speech-to-text — Venice /audio/transcriptions
        stt=openai.STT(
            model="nvidia/parakeet-tdt-0.6b-v3",
            base_url=VENICE_BASE_URL,
            api_key=VENICE_API_KEY,
        ),
        # LLM — Venice /chat/completions (streaming + tool calling supported)
        # Venice's private, uncensored model feeding the STT-LLM-TTS pipeline
        llm=openai.LLM(
            model="venice-uncensored-1-2",
            base_url=VENICE_BASE_URL,
            api_key=VENICE_API_KEY,
        ),
        # Text-to-speech — Venice /audio/speech
        tts=VeniceTTS(
            model="tts-kokoro",
            voice="af_sky",
            base_url=VENICE_BASE_URL,
            api_key=VENICE_API_KEY,
        ),
        # Local VAD handles endpointing for the batch STT
        vad=silero.VAD.load(),
    )

    await session.start(
        room=ctx.room,
        agent=Assistant(),
        room_input_options=RoomInputOptions(),
    )

    await session.generate_reply(
        instructions="Greet the user and offer your help."
    )


if __name__ == "__main__":
    agents.cli.run_app(agents.WorkerOptions(entrypoint_fnc=entrypoint))