is it possible to stream the response from tts to ...
# ask-aif
future-shampoo-29056
09/23/2025, 7:41 AMis it possible to stream the response from tts to the room the make the agent respond faster
t
tall-airline-16756
09/23/2025, 7:41 AMThanks for reaching out, Huy. I'm an AI assistant and I'm here to help. Give me a moment while I research this topic.
tall-airline-16756
09/23/2025, 7:42 AMAbsolutely, you can stream TTS (Text-to-Speech) responses directly to a LiveKit room to help agents respond more quickly. It’s a great way to make interactions feel more natural and immediate! This works by using the
stream()AudioSourcestream()f
future-shampoo-29056
09/23/2025, 7:44 AMcan you write a custom tts class that inherit the base one but for an api calling tts that return as streaming
t
tall-airline-16756
09/23/2025, 7:44 AMTo help you create a custom TTS class that supports streaming responses from an API, you can extend the base
tts.TTStts.TTSstream()SynthesizeStreamSynthesizeStreamAudioFrame Copy code
class CustomTTS(tts.TTS):
def __init__(self, api_key, sample_rate=24000):
super().__init__(sample_rate, 1, streaming=True)
self.api_key = api_key
def stream(self):
return CustomSynthesizeStream(self)
class CustomSynthesizeStream(tts.SynthesizeStream):
def __init__(self, tts):
super().__init__(tts)
async def run(self):
# Connect to API and handle streaming
passf
future-shampoo-29056
09/23/2025, 7:51 AMif im trying to make the flow faster :stt,llm,tts i already did streaming what else can i do (for example pre warm something)
t
tall-airline-16756
09/23/2025, 7:52 AMI'd love to help you optimize your STT, LLM, and TTS flow for better and faster performance! To get a clearer picture, could you please share a bit more about your current setup? For instance:
• Are you working with any specific APIs or frameworks for STT, LLM, and TTS?
• What kind of latency are you noticing right now?
• Are there particular parts of the flow—like STT processing, LLM response generation, or TTS synthesis—that seem slower than you'd expect?
Once I have these details, I’ll be able to suggest some targeted improvements, like pre-warming techniques or other optimizations, to help speed things up for you.
f
future-shampoo-29056
09/23/2025, 7:52 AMi create my own api for all 3 then write the custom stt,lmm,tts for each of them
t
tall-airline-16756
09/23/2025, 7:53 AMTo help you get the most out of your custom STT, LLM, and TTS APIs and make them run faster, here are some friendly tips you might find useful:
• Reduce Latency:
◦ Try deploying your services close to each other—that way, you can cut down on network delays.
◦ Consider using providers known for low-latency inference, like Groq for LLMs or Google Cloud for STT/TTS, to speed things up.
◦ Processing audio in small chunks (around 20ms) can be a great way to balance speed and accuracy.
• Increase Throughput:
◦ Enabling parallel STT and TTS operations can really help when you’re handling multiple streams or quick interactions.
◦ Streaming LLM tokens incrementally and feeding them into the TTS pipeline as they come can save you from waiting for the full response.
• Pre-Warming Techniques:
◦ Keeping your STT, LLM, and TTS models loaded and ready means you won’t have to wait for initialization each time.
◦ Caching commonly used phrases or responses can also speed things along nicely.
• Monitor Performance:
◦ Keeping an eye on metrics like Time To First Token (TTFT), Time To First Byte (TTFB), and overall latency can help you spot where things might slow down.
I hope these ideas help you achieve faster and smoother performance! If you’d like, I’m here to guide you through any of these steps in more detail—just let me know!
Sources: Live Conversations with AI using ChatGPT and WebRTC | Speech & Audio Processing | livekit/livekit_composite | DeepWiki | README.md | An open source stack for real-time multimodal AI
f
future-shampoo-29056
09/25/2025, 9:21 AMis there any format when it come to streaming tts
t
tall-airline-16756
09/25/2025, 9:22 AMWhen it comes to streaming TTS, the format and implementation can vary depending on the provider and the specific needs of your application. Let’s walk through some key points together:
• Streaming Protocol: Most TTS providers use WebSocket for real-time streaming. For example, Cartesia, Neuphonic, and Resemble AI utilize WebSocket connections to send text bit by bit and receive audio frames live.
• Audio Format: The audio output is usually in raw PCM format (like
pcm_s16leStreamAdapterf
future-shampoo-29056
09/25/2025, 9:23 AMhow about openai with streaming =True
t
tall-airline-16756
09/25/2025, 9:23 AMI understand you’re looking to create a streaming experience with OpenAI TTS. While it doesn’t natively support
streaming=TrueStreamAdapterStreamAdapter Copy code
from livekit.agents.tts import StreamAdapter
from livekit.agents.tokenize import SentenceTokenizer
streaming_tts = StreamAdapter(tts=openai_tts, sentence_tokenizer=SentenceTokenizer())
stream = streaming_tts.stream()
stream.push_text("Hello world.")
stream.flush()f
future-shampoo-29056
09/25/2025, 9:30 AMwhat do i need to change if i want tts to be streaming? from kittentts import KittenTTS
import numpy as np
from livekit.agents import tts
from livekit.agents.types import DEFAULT_API_CONNECT_OPTIONS
class KittenTTSPlugin(tts.TTS):
def __init__(self, model_name="KittenML/kitten-tts-nano-0.1", voice="expr-voice-2-f", sample_rate=24000):
# Initialize parent with required parameters (matching OpenAI TTS pattern)
super().__init__(
capabilities=tts.TTSCapabilities(streaming=False),
sample_rate=sample_rate,
num_channels=1, # Mono audio
)
self.model = KittenTTS(model_name)
self.voice = voice
# Verify voice availability
if voice not in self.model.available_voices:
raise ValueError(f"Voice {voice} not available. Choose from {self.model.available_voices}")
def synthesize(self, text: str, *, conn_options=DEFAULT_API_CONNECT_OPTIONS) -> tts.ChunkedStream:
"""
Synthesize text to speech using KittenTTS.
Args:
text: The text to synthesize
conn_options: Connection options (ignored for KittenTTS)
Returns:
ChunkedStream: Stream of synthesized audio chunks
"""
return KittenTTSChunkedStream(tts=self, input_text=text, conn_options=conn_options)
@property
def name(self):
return f"KittenTTS-{self.voice}"
class KittenTTSChunkedStream(tts.ChunkedStream):
def __init__(self, *, tts: KittenTTSPlugin, input_text: str, conn_options):
super().__init__(tts=tts, input_text=input_text, conn_options=conn_options)
self._tts: KittenTTSPlugin = tts
self._audio_generated = False
async def _run(self, output_emitter: tts.AudioEmitter):
"""
Generate audio using KittenTTS and emit it through the output emitter.
This is the core method that must be implemented.
"""
if self._audio_generated:
return
try:
# Add padding to avoid cut-offs
padded_text = self.input_text.strip() + " ... "
# Generate audio waveform (numpy float32 array)
audio = self._tts.model.generate(padded_text, voice=self._tts.voice)
# Convert float32 to int16 (LiveKit requirement)
# Scale from [-1, 1] to [-32768, 32767]
audio_int16 = (audio * 32767).astype(np.int16)
# Convert to bytes
audio_bytes = audio_int16.tobytes()
# Initialize the output emitter with PCM format
output_emitter.initialize(
request_id=str(id(self)),
sample_rate=self._tts.sample_rate,
num_channels=1, # Mono
mime_type="audio/pcm", # Use PCM format for raw audio
)
# Push the audio data through the emitter
output_emitter.push(audio_bytes)
# Flush the emitter to indicate completion
output_emitter.flush()
self._audio_generated = True
except Exception as e:
# Handle any errors during synthesis
raise RuntimeError(f"KittenTTS synthesis failed: {str(e)}")
finally:
# Ensure the emitter is properly closed
try:
output_emitter.end_input()
except:
pass
t
tall-airline-16756
09/25/2025, 9:31 AMTo help you modify your KittenTTS plugin to support streaming, here are some friendly steps you can follow:
• Enable Streaming Capability: In the constructor, update the
capabilitiesstreaming=True Copy code
super().__init__(
capabilities=tts.TTSCapabilities(streaming=True),
sample_rate=sample_rate,
num_channels=1,
)stream()stream()SynthesizeStream Copy code
def stream(self, *, conn_options=DEFAULT_API_CONNECT_OPTIONS):
return KittenTTSSynthesizeStream(tts=self, conn_options=conn_options)KittenTTSSynthesizeStreamtts.SynthesizeStreamAudioEmitter_run() Copy code
async def _run(self, output_emitter):
output_emitter.initialize(
request_id=str(id(self)),
sample_rate=self._tts.sample_rate,
num_channels=1,
mime_type="audio/pcm",
stream=True,
)
for text_chunk in self.input_text.split(" "):
audio = self._tts.model.generate(text_chunk, voice=self._tts.voice)
audio_bytes = (audio * 32767).astype(np.int16).tobytes()
output_emitter.push(audio_bytes)
output_emitter.end_input()f
future-shampoo-29056
09/25/2025, 9:34 AMwhat if the output of my tts is not streaming what would happens then
t
tall-airline-16756
09/25/2025, 9:34 AMI understand that your TTS output might not be streaming, and that's completely okay! LiveKit takes care of this by using the
synthesize()StreamAdapteruse_tts_aligned_transcript=Truebefore_tts_cbf
future-shampoo-29056
09/25/2025, 9:41 AMhow about how to check time used for each process (stt , llm. tts)
t
tall-airline-16756
09/25/2025, 9:42 AMI'm glad you're interested in measuring the time used for each process (STT, LLM, TTS) in LiveKit! You can easily do this using the built-in metrics system. Here’s a friendly guide to help you track timing for each stage:
• STT (Speech-to-Text):
◦ `duration`: This tells you the time taken to create the transcript (for non-streaming STT).
◦ `audio_duration`: This shows the duration of the audio input that was processed.
• LLM (Large Language Model):
◦
ttftttfbmetrics_collectedf
future-shampoo-29056
09/25/2025, 9:47 AMhow to subsrcibe to the metrics coleected event
t
tall-airline-16756
09/25/2025, 9:47 AMI'm happy to guide you through subscribing to the
metrics_collected Copy code
from livekit.agents import metrics, MetricsCollectedEvent
@session.on("metrics_collected")
def on_metrics_collected(ev: MetricsCollectedEvent):
metrics.log_metrics(ev.metrics) Copy code
from livekit.agents import metrics
@agent.on("metrics_collected")
def _on_metrics_collected(mtrcs: metrics.AgentMetrics):
metrics.log_metrics(mtrcs)UsageCollector Copy code
usage_collector = metrics.UsageCollector()
@agent.on("metrics_collected")
def _on_metrics_collected(mtrcs: metrics.AgentMetrics):
usage_collector.collect(mtrcs)
summary = usage_collector.get_summary()
<http://logger.info|logger.info>(f"Usage: {summary}")f
future-shampoo-29056
09/25/2025, 9:49 AMso when i do
Copy code
from livekit.agents import metrics, MetricsCollectedEvent
@session.on("metrics_collected")
def on_metrics_collected(ev: MetricsCollectedEvent):
metrics.log_metrics(ev.metrics)t
tall-airline-16756
09/25/2025, 9:50 AMAbsolutely! When you use
@session.on("metrics_collected")sessionf
future-shampoo-29056
09/25/2025, 9:51 AMif i write a custom llm or custom tts do i need to add antything in order for the event listener to listen
t
tall-airline-16756
09/25/2025, 9:52 AMI’m glad you’re working on a custom LLM or TTS! You actually don’t need to add anything extra for the
metrics_collectedttftdurationttfbtts.TTSllm.LLMmetrics_collected Copy code
@session.on("metrics_collected")
def on_metrics_collected(ev):
metrics.log_metrics(ev.metrics)synthesize()stream()f
future-shampoo-29056
09/25/2025, 10:03 AMit doesnt have metrics
from future import annotations
import aiohttp
import asyncio
import json
import uuid
from dataclasses import dataclass
from typing import Any
from livekit.agents import (
APIConnectionError,
APIStatusError,
APITimeoutError,
llm,
)
from livekit.agents.types import DEFAULT_API_CONNECT_OPTIONS, APIConnectOptions, NOT_GIVEN, NotGivenOr
from livekit.agents.llm.chat_context import ChatContext, ChatRole
from livekit.agents.llm.tool_context import FunctionTool, RawFunctionTool, ToolChoice
@dataclass
class _LLMOptions:
api_base_url: str
api_key: str
model: str
temperature: float
max_tokens: int
class CustomLLM(llm.LLM):
def __init__(
self,
*,
_api_base_url_: str = "https://api.openai.com/v1",
_api_key_: str | None = None,
_model_: str = "gpt-4o-mini",
_temperature_: float = 0.7,
_max_tokens_: int = 1000,
) -> None:
"""
Create a new instance of Custom LLM that makes HTTP calls to GPT-4o mini.
Args:
api_base_url: Base URL of the OpenAI API
api_key: OpenAI API key
model: Model name to use
temperature: Sampling temperature
max_tokens: Maximum tokens in response
"""
super().__init__()
self._opts = _LLMOptions(
_api_base_url_=_api_base_url_.rstrip('/'),
_api_key_=_api_key_ or "",
model=model,
temperature=temperature,
_max_tokens_=_max_tokens_,
)
# Create HTTP client with proper timeout settings
self._client_session: aiohttp.ClientSession | None = None
async def _get_client_session(self) -> aiohttp.ClientSession:
"""Get or create HTTP client session"""
if self._client_session is None or self._client_session.closed:
timeout = aiohttp.ClientTimeout(
total=60.0,
connect=15.0,
_sock_read_=30.0,
_sock_connect_=5.0
)
self._client_session = aiohttp.ClientSession(
timeout=timeout,
headers={
"Authorization": f"Bearer {self._opts.api_key}",
"Content-Type": "application/json",
},
connector=aiohttp.TCPConnector(
limit=50,
_limit_per_host_=50,
_keepalive_timeout_=120,
_enable_cleanup_closed_=True
)
)
return self._client_session
@property
def model(self) -> str:
return self._opts.model
def chat(
self,
*,
_chat_ctx_: ChatContext,
_tools_: list[FunctionTool | RawFunctionTool] | None = None,
_conn_options_: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS,
_parallel_tool_calls_: NotGivenOr[bool] = NOT_GIVEN,
_tool_choice_: NotGivenOr[ToolChoice] = NOT_GIVEN,
_extra_kwargs_: NotGivenOr[dict[str, Any]] = NOT_GIVEN,
) -> llm.LLMStream:
"""
Create a chat completion stream.
"""
return CustomLLMStream(
llm=self,
_chat_ctx_=_chat_ctx_,
tools=tools or [],
_conn_options_=_conn_options_,
_parallel_tool_calls_=_parallel_tool_calls_,
_tool_choice_=_tool_choice_,
_extra_kwargs_=_extra_kwargs_,
)
async def aclose(self) -> None:
"""Close the LLM instance and cleanup resources"""
if self._client_session and not self._client_session.closed:
await self._client_session.close()
self._client_session = None
class CustomLLMStream(llm.LLMStream):
def __init__(
self,
_llm_: CustomLLM,
*,
_chat_ctx_: ChatContext,
_tools_: list[FunctionTool | RawFunctionTool],
_conn_options_: APIConnectOptions,
_parallel_tool_calls_: NotGivenOr[bool] = NOT_GIVEN,
_tool_choice_: NotGivenOr[ToolChoice] = NOT_GIVEN,
_extra_kwargs_: NotGivenOr[dict[str, Any]] = NOT_GIVEN,
) -> None:
super().__init__(llm=llm, _chat_ctx_=_chat_ctx_, tools=tools, _conn_options_=_conn_options_)
self._llm: CustomLLM = llm
self._parallel_tool_calls = _parallel_tool_calls_
self._tool_choice = _tool_choice_
self._extra_kwargs = _extra_kwargs_
def _convert_chat_context_to_messages(self, _chat_ctx_: ChatContext) -> list[dict[str, Any]]:
"""Convert ChatContext to OpenAI messages format"""
_try_:
# Use the official method like OpenAI LLM does
messages, _ = _chat_ctx_.to_provider_format(format="openai")
return messages
except Exception:
# Fallback: Ensure we always have at least a system message
return [
{
"role": "system",
"content": "You are a helpful voice AI assistant that answers questions and helps with tasks in japanese."
}
]
async def _run(self) -> None:
"""
Make HTTP request to GPT-4o mini API (non-streaming).
"""
# Convert chat context to OpenAI format
messages = self._convert_chat_context_to_messages(self._chat_ctx)
# Prepare the API request payload (no streaming)
payload = {
"model": self._llm._opts.model,
"messages": messages,
"temperature": self._llm._opts.temperature,
"max_tokens": self._llm._opts.max_tokens,
"stream": False, # Disable streaming for simplicity
}
# Add tools if provided
if self._tools:
# Convert tools to OpenAI format (simplified)
tools_payload = []
for tool in self._tools:
if isinstance(tool, FunctionTool):
tools_payload.append({
"type": "function",
"function": {
"name": tool.name,
"description": tool.description or "",
"parameters": tool.parameters or {},
}
})
if tools_payload:
payload["tools"] = tools_payload
_try_:
# Get HTTP client session
session = await self._llm._get_client_session()
# Make the non-streaming API request
async with session.post(
f"{self._llm._opts.api_base_url}/chat/completions",
json=payload,
timeout=aiohttp.ClientTimeout(
total=60,
connect=self._conn_options.timeout
)
) as response:
if response.status != 200:
error_text = await response.text()
raise APIStatusError(
f"Custom LLM API error: {error_text}",
_status_code_=response.status,
_request_id_=str(response.headers.get('request-id', '')),
body=error_text
)
# Get the complete response
data = await response.json()
request_id = str(uuid.uuid4())
# Extract the response content
if 'choices' in data and len(data['choices']) > 0:
choice = data['choices'][0]
message = choice.get('message', {})
content = message.get('content', '')
# Send the complete response as a single chunk
chunk_delta = llm.ChoiceDelta(
role="assistant",
content=content,
_tool_calls_=[]
)
chunk = llm.ChatChunk(
id=request_id,
delta=chunk_delta,
usage=None # Usage info if needed
)
# Send the chunk
self._event_ch.send_nowait(chunk)
except (aiohttp.ClientError, asyncio.TimeoutError) as e:
raise APITimeoutError() from None
except APIStatusError:
raise # Re-raise our custom status errors
except Exception as e:
raise APIConnectionError() from e
t
tall-airline-16756
09/25/2025, 10:03 AMI understand that if your custom LLM or TTS implementation doesn’t currently emit metrics, you’ll need to add support for metrics collection yourself. Don’t worry, I’m here to guide you through it! Here’s a simple way to integrate metrics into your custom LLM:
• Emit Metrics: You can override methods like
chat()synthesize()ttftduration Copy code
from livekit.agents import metrics
start_time = time.monotonic()
# Perform the LLM or TTS operation
end_time = time.monotonic()
# Emit metrics
self.emit_metrics(metrics.LLMMetrics(ttft=end_time - start_time, ...))metrics_collectedf
future-shampoo-29056
09/25/2025, 10:37 AMif i use google plugin for stt how can i get the time session = AgentSession(
stt = google.STT(
model="telephony",
languages="ja-JP",
_spoken_punctuation_=False,
),
t
tall-airline-16756
09/25/2025, 10:38 AMI wanted to share a little insight about the Google STT plugin in LiveKit. It keeps track of the session connection time internally using the
_session_connected_at Copy code
import time
# Record the start time
session_start_time = time.time()
# Later, calculate the elapsed time
elapsed_time = time.time() - session_start_time
print(f"Session duration: {elapsed_time} seconds")m
most-vr-18651
10/27/2025, 2:27 PM@future-shampoo-29056
were you able to implement the tts ?
i want tp implement my own custom tts deployed on aws. How do i do that ?
t
tall-airline-16756
10/27/2025, 2:29 PMI'm here to help you implement your own custom TTS deployed on AWS in LiveKit! You can follow these steps, and if you have any questions along the way, just let me know:
• *Option 1: Use Amazon Polly TTS Plugin*`from livekit.plugins.aws import TTS
tts = TTS(
voice="Ruth",
language="en-US",
speech_engine="generative",
sample_rate=16000,
region="us-east-1"
)`
◦ First, install the plugin by running:
pip install livekit-plugins-awsAgentSession