How can I use my custom locally hosted STT, LLM and TTS with livekit, instead of the ones being provided in livekit.plugins in python context?

Thanks for reaching out, Vansh. I'm an AI assistant and I'm here to help. Give me a moment while I research this topic.

Yes please provide detailed examples of all three, and for more context my LLM won't be openai compatible, I was thinking to serve all three via a websocket server, and then connect that server to live kit's overrriden stt_node, tts_node and llm_node, but I have doubts about that approach, please give your views as well

I'm glad you're exploring how to integrate your custom STT, LLM, and TTS via a WebSocket server with LiveKit! You can achieve this by overriding the

stt_node

,

llm_node

, and

tts_node

methods in a custom

Agent

subclass. This method is fully supported and lets you smoothly route data through your WebSocket-based services. Below, I’ve shared detailed examples for each node to guide you through the process: 1. Custom STT via WebSocket Here, you override

stt_node

to send audio frames to your WebSocket server and receive transcriptions back:

async def stt_node(self, audio: AsyncIterable[rtc.AudioFrame], model_settings: ModelSettings):
    async def send_to_websocket():
        async with websockets.connect("<ws://your-stt-server>") as ws:
            async for frame in audio:
                await ws.send(frame.data.tobytes())
                response = await ws.recv()
                yield stt.SpeechEvent(
                    type=stt.SpeechEventType.FINAL_TRANSCRIPT,
                    alternatives=[stt.SpeechRecognitionResult(text=response)]
                )
    return send_to_websocket()

2. Custom LLM via WebSocket For the

llm_node

, you forward the chat context to your WebSocket server and stream responses back:

async def llm_node(self):
    async with websockets.connect("<ws://your-llm-server>") as ws:
        chat_ctx = self.chat_ctx
        await ws.send(json.dumps(chat_ctx))
        async for response in ws:
            yield llm.ChatChunk(content=response)

3. Custom TTS via WebSocket And for

tts_node

, you send text to your WebSocket-based TTS engine and receive the synthesized audio:

async def tts_node(self, text: AsyncIterable[str], model_settings: ModelSettings):
    async with websockets.connect("<ws://your-tts-server>") as ws:
        async for segment in text:
            await ws.send(segment)
            audio_data = await ws.recv()
            frame = rtc.AudioFrame(
                data=np.frombuffer(audio_data, dtype=np.int16),
                sample_rate=24000,
                num_channels=1,
                samples_per_channel=len(audio_data) // 2
            )
            yield frame

Considerations • Make sure your WebSocket server is set up to efficiently handle streaming data. • It's important to manage the connection lifecycle and handle any errors gracefully within each node. • Remember that inputs and outputs should match LiveKit’s expected types (for example,

AsyncIterable[rtc.AudioFrame]

for audio streams). This approach is quite flexible and fits well with LiveKit’s design, which supports custom node overrides so you can have full control over the pipeline. If you run into any questions or need a hand with implementation details or specific challenges, please don’t hesitate to reach out—I’m here to help! Sources: livekit-agents/livekit/agents/voice/io.py | Docs::Agents | Pipeline nodes and hooks | LiveKit Docs | Pipeline nodes and hooks | LiveKit Docs | README.md

Okay considering that I have all three websockets running, what do I need to do now to start creating agents/rooms etc from the stt-llm-tts pipeline, and how do I host on livekit cloud?

To help you create agents, rooms, and host your custom STT-LLM-TTS pipeline on LiveKit Cloud, let's walk through the steps together: 1. Set Up Your Project First, initialize your project and install the necessary LiveKit agents and plugins. If you're using Python, you can do this:

uv init livekit-voice-agent --bare
cd livekit-voice-agent
uv add "livekit-agents[deepgram,openai,cartesia,silero,turn-detector]~=1.2" "livekit-plugins-noise-cancellation~=0.2" "python-dotenv"

If you prefer Node.js, you can start by cloning the Node.js starter project. 2. Configure Environment Variables Next, set up your API keys in a

.env.local

file to keep everything secure and organized:

LIVEKIT_API_KEY=<your api key>
LIVEKIT_API_SECRET=<your api secret>
LIVEKIT_URL=<your livekit server url>

3. Define the Agent and Pipeline You have the flexibility to use either a custom

Agent

subclass or the more streamlined

VoicePipelineAgent

. Here’s an example using the

VoicePipelineAgent

that you might find useful:

from livekit.agents.pipeline import VoicePipelineAgent
from livekit.plugins import deepgram, openai, silero

agent = VoicePipelineAgent(
    vad=silero.VAD.load(),
    stt=deepgram.STT(),
    llm=openai.LLM(model="gpt-4o-mini"),
    tts=deepgram.TTS(),
    chat_ctx=llm.ChatContext().append(role="system", text="Be concise and helpful.")
)

async def entrypoint(ctx):
    await ctx.connect(auto_subscribe=agents.AutoSubscribe.AUDIO_ONLY)
    participant = await ctx.wait_for_participant()
    agent.start(ctx.room, participant)
    await agent.say("Hello, how can I help you?")

4. Host on LiveKit Cloud When you're ready to deploy your agent, simply run:

livekit-cli deploy

This command uploads your agent to LiveKit Cloud, where it will run smoothly as a scalable worker process. 5. Dynamic Room Management Agents are automatically assigned to rooms as users join. You can manage these connections easily using the

JobContext

like this:

await ctx.connect()
await ctx.wait_for_participant()
agent.start(ctx.room, participant)

If you'd like, I’m here to help you with more details on deploying your WebSocket-based pipeline or customizing how your agent behaves. Just let me know! Sources: Voice AI quickstart | LiveKit Docs | examples/src/basic_agent.ts | agent-examples/agent_examples/pipeline_agent.py | src/agent.ts | Pipeline nodes and hooks | LiveKit Docs

now what about using a custom Realtime model nodes implementation?

I'm happy to guide you through implementing custom Realtime model nodes for STT, LLM, and TTS in LiveKit! You can achieve this by overriding specific methods in a custom

Agent

subclass, which lets you seamlessly add your own logic into the processing pipeline. Here are some examples to help you get started with each node: 1. Customizing the STT Node Try overriding

stt_node

if you'd like to preprocess audio or do some post-processing on transcriptions:

async def stt_node(self, audio: AsyncIterable[rtc.AudioFrame], model_settings: ModelSettings):
    async def filtered_audio():
        async for frame in audio:
            # Custom audio preprocessing (e.g., noise filtering)
            yield frame

    async for event in Agent.default.stt_node(self, filtered_audio(), model_settings):
        # Custom text post-processing (e.g., removing filler words)
        yield event

2. Customizing the LLM Node If you want to tweak the chat context or handle LLM responses differently, override

llm_node

like this:

async def llm_node(self, chat_ctx: llm.ChatContext, tools: list[FunctionTool], model_settings: ModelSettings):
    # Modify chat context or instructions
    async for chunk in Agent.default.llm_node(self, chat_ctx, tools, model_settings):
        # Custom post-processing (e.g., content filtering)
        yield chunk

3. Customizing the TTS Node To use a custom TTS engine or adjust the audio output, you can override

tts_node

like this:

async def tts_node(self, text: AsyncIterable[str], model_settings: ModelSettings):
    async for frame in Agent.default.tts_node(self, text, model_settings):
        # Custom audio post-processing (e.g., adjust volume)
        yield frame

4. Realtime Model-Specific Node For Realtime models, if you'd like to add special audio effects or modify the audio output, override

realtime_audio_output_node

as follows:

async def realtime_audio_output_node(self, audio: AsyncIterable[rtc.AudioFrame], model_settings: ModelSettings):
    async for frame in Agent.default.realtime_audio_output_node(self, audio, model_settings):
        # Custom audio effects (e.g., echo, compression)
        yield frame

These overrides give you great flexibility to tailor the behavior of your STT, LLM, and TTS nodes exactly how you need. If you have any questions or want help with a particular part, just let me know—I'm here to support you every step of the way! Sources: Pipeline nodes and hooks | LiveKit Docs | Docs::Agents | Tool definition and use | LiveKit Docs

please resend only the realtime node part with the implementation of a custom realtime speech 2 speech model running on a websocket, just like you sent the other node overrides

I'd love to help you implement a custom

realtime_audio_output_node

for your Realtime speech-to-speech model running on a WebSocket. Here's a simple example you can follow:

from livekit import rtc
from livekit.agents import ModelSettings, Agent
import websockets

async def realtime_audio_output_node(self, audio: AsyncIterable[rtc.AudioFrame], model_settings: ModelSettings):
    async with websockets.connect("<ws://your-realtime-s2s-server>") as ws:
        async for frame in audio:
            # Send audio frame data to the WebSocket server
            await ws.send(frame.data.tobytes())

            # Receive processed audio data from the WebSocket server
            processed_audio = await ws.recv()

            # Convert the received audio data back into an AudioFrame
            yield rtc.AudioFrame(
                data=processed_audio,
                sample_rate=24000,  # Adjust based on your model's output
                num_channels=1,     # Adjust based on your model's output
                samples_per_channel=len(processed_audio) // 2
            )

This setup streams audio frames to your WebSocket-based speech-to-speech model and streams back the processed audio in real time. Just make sure your WebSocket server is ready to handle audio data smoothly and matches the expected format, like sample rate and channels. If anything feels unclear or you’d like a hand with any part of this, please don’t hesitate to reach out—I’m here to help!