from livekit can i get 16khz audio in opus format??

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

this si my sample code in this can you make that chages?? import logging import asyncio import os import json from dotenv import load_dotenv from livekit.agents import ( NOT_GIVEN, Agent, AgentFalseInterruptionEvent, AgentSession, JobContext, JobProcess, AgentStateChangedEvent, UserInputTranscribedEvent, SpeechCreatedEvent, UserStateChangedEvent, AgentHandoffEvent, MetricsCollectedEvent, RoomInputOptions, RunContext, WorkerOptions, function_tool, cli, metrics, ) from livekit.plugins import silero import livekit.plugins.groq as groq from livekit.plugins.turn_detector.multilingual import MultilingualModel from livekit.plugins import noise_cancellation logger = logging.getLogger("agent") load_dotenv(".env") class Assistant(Agent): def __init__(self) -> None: super().__init__( instructions="""You are a helpful voice AI assistant. You eagerly assist users with their questions by providing information from your extensive knowledge. Your responses are concise, to the point, and without any complex formatting or punctuation including emojis, asterisks, or other symbols. You are curious, friendly, and have a sense of humor.""", ) @function_tool async def lookup_weather(self, context: RunContext, location: str): logger.info(f"Looking up weather for {location}") return "sunny with a temperature of 70 degrees." def prewarm(proc: JobProcess): proc.userdata["vad"] = silero.VAD.load() async def entrypoint(ctx: JobContext): ctx.log_context_fields = {"room": ctx.room.name} print(f"Starting agent in room: {ctx.room.name}") # Set up voice AI pipeline session = AgentSession( llm=groq.LLM(model="openai/gpt-oss-20b"), stt=groq.STT(model="whisper-large-v3-turbo", language="en"), tts=groq.TTS(model="playai-tts", voice="Aaliyah-PlayAI"), turn_detection=MultilingualModel(), vad=ctx.proc.userdata["vad"], preemptive_generation=False, ) @session.on("agent_false_interruption") def _on_agent_false_interruption(ev: AgentFalseInterruptionEvent): logger.info("False positive interruption, resuming") session.generate_reply(instructions=ev.extra_instructions or NOT_GIVEN) payload = json.dumps({ "type": "agent_false_interruption", "data": ev.dict() }) asyncio.create_task(ctx.room.local_participant.publish_data(payload.encode("utf-8"), reliable=True)) logger.info("Sent agent_false_interruption via data channel") usage_collector = metrics.UsageCollector() # @session.on("metrics_collected") # def _on_metrics_collected(ev: MetricsCollectedEvent): # metrics.log_metrics(ev.metrics) # usage_collector.collect(ev.metrics) # payload = json.dumps({ # "type": "metrics_collected", # "data": ev.metrics.dict() # }) # asyncio.create_task(ctx.room.local_participant.publish_data(payload.encode("utf-8"), reliable=True)) # logger.info("Sent metrics_collected via data channel") @session.on("agent_state_changed") def _on_agent_state_changed(ev: AgentStateChangedEvent): logger.info(f"Agent state changed: {ev}") payload = json.dumps({ "type": "agent_state_changed", "data": ev.dict() }) asyncio.create_task(ctx.room.local_participant.publish_data(payload.encode("utf-8"), reliable=True)) logger.info("Sent agent_state_changed via data channel") @session.on("user_input_transcribed") def _on_user_input_transcribed(ev: UserInputTranscribedEvent): logger.info(f"User said: {ev}") payload = json.dumps({ "type": "user_input_transcribed", "data": ev.dict() }) asyncio.create_task(ctx.room.local_participant.publish_data(payload.encode("utf-8"), reliable=True)) logger.info("Sent user_input_transcribed via data channel") @session.on("speech_created") def _on_speech_created(ev: SpeechCreatedEvent): # logger.info(f"Speech created with id: {ev.speech_id}, duration: {ev.duration_ms}ms") payload = json.dumps({ "type": "speech_created", "data": ev.dict() }) asyncio.create_task(ctx.room.local_participant.publish_data(payload.encode("utf-8"), reliable=True)) logger.info("Sent speech_created via data channel") async def log_usage(): summary = usage_collector.get_summary() logger.info(f"Usage: {summary}") payload = json.dumps({ "type": "usage_summary", "summary": summary.llm_prompt_tokens }) # session.local_participant.publishData(payload.encode("utf-8"), reliable=True) logger.info("Sent usage_summary via data channel") ctx.add_shutdown_callback(log_usage) await session.start( agent=Assistant(), room=ctx.room, room_input_options=RoomInputOptions( noise_cancellation=noise_cancellation.BVC(), ), ) await ctx.connect() if name == "__main__": cli.run_app(WorkerOptions(entrypoint_fnc=entrypoint, prewarm_fnc=prewarm))

Thank you so much for sharing your code! To help you best, could you please let me know which part you'd like to modify to enable 16kHz Opus audio? For instance, are you thinking about adjusting the audio source configuration, the resampling settings, or maybe something else? I'm here to help you through it!

i wna t he agent replay should be in 16khz opus

Thank you for reaching out! I wanted to share that the LiveKit agent framework is set up to output audio in 24kHz Opus format by default, and currently, it doesn’t support direct configuration for 16kHz output since this is built into the audio pipeline. If you need the audio at 16kHz, here’s a friendly way to approach it: 1. Capture the audio output from the agent. 2. Use an external process to resample the audio from 24kHz down to 16kHz. 3. Then, re-encode or forward the resampled audio through a custom media pipeline. I’m here to help! Just let me know if you’d like some guidance on how to set up this external processing, and we can walk through it together. Sources: Speech & Audio Processing | livekit/livekit_composite | DeepWiki | livekit-plugins/livekit-plugins-tavus/livekit/plugins/tavus/avatar.py

ok do 24kHz Opus

I'm happy to let you know that the LiveKit agent framework already provides audio in 24kHz Opus format by default, so you don't need to do any extra setup for that. If there's anything else you'd like help with or if you have any other questions, please don't hesitate to reach out. I'm here to support you!

in this code make changes

import logging

import asyncio import os import json from dotenv import load_dotenv from livekit.agents import ( NOT_GIVEN, Agent, AgentFalseInterruptionEvent, AgentSession, JobContext, JobProcess, AgentStateChangedEvent, UserInputTranscribedEvent, SpeechCreatedEvent, UserStateChangedEvent, AgentHandoffEvent, MetricsCollectedEvent, RoomInputOptions, RunContext, WorkerOptions, function_tool, cli, metrics, ) from livekit.plugins import silero import livekit.plugins.groq as groq from livekit.plugins.turn_detector.multilingual import MultilingualModel from livekit.plugins import noise_cancellation logger = logging.getLogger("agent") load_dotenv(".env") class Assistant(Agent): def __init__(self) -> None: super().__init__( instructions="""You are a helpful voice AI assistant. You eagerly assist users with their questions by providing information from your extensive knowledge. Your responses are concise, to the point, and without any complex formatting or punctuation including emojis, asterisks, or other symbols. You are curious, friendly, and have a sense of humor.""", ) @function_tool async def lookup_weather(self, context: RunContext, location: str): logger.info(f"Looking up weather for {location}") return "sunny with a temperature of 70 degrees." def prewarm(proc: JobProcess): proc.userdata["vad"] = silero.VAD.load() async def entrypoint(ctx: JobContext): ctx.log_context_fields = {"room": ctx.room.name} print(f"Starting agent in room: {ctx.room.name}") # Set up voice AI pipeline session = AgentSession( llm=groq.LLM(model="openai/gpt-oss-20b"), stt=groq.STT(model="whisper-large-v3-turbo", language="en"), tts=groq.TTS(model="playai-tts", voice="Aaliyah-PlayAI"), turn_detection=MultilingualModel(), vad=ctx.proc.userdata["vad"], preemptive_generation=False, ) @session.on("agent_false_interruption") def _on_agent_false_interruption(ev: AgentFalseInterruptionEvent): logger.info("False positive interruption, resuming") session.generate_reply(instructions=ev.extra_instructions or NOT_GIVEN) payload = json.dumps({ "type": "agent_false_interruption", "data": ev.dict() }) asyncio.create_task(ctx.room.local_participant.publish_data(payload.encode("utf-8"), reliable=True)) logger.info("Sent agent_false_interruption via data channel") usage_collector = metrics.UsageCollector() # @session.on("metrics_collected") # def _on_metrics_collected(ev: MetricsCollectedEvent): # metrics.log_metrics(ev.metrics) # usage_collector.collect(ev.metrics) # payload = json.dumps({ # "type": "metrics_collected", # "data": ev.metrics.dict() # }) # asyncio.create_task(ctx.room.local_participant.publish_data(payload.encode("utf-8"), reliable=True)) # logger.info("Sent metrics_collected via data channel") @session.on("agent_state_changed") def _on_agent_state_changed(ev: AgentStateChangedEvent): logger.info(f"Agent state changed: {ev}") payload = json.dumps({ "type": "agent_state_changed", "data": ev.dict() }) asyncio.create_task(ctx.room.local_participant.publish_data(payload.encode("utf-8"), reliable=True)) logger.info("Sent agent_state_changed via data channel") @session.on("user_input_transcribed") def _on_user_input_transcribed(ev: UserInputTranscribedEvent): logger.info(f"User said: {ev}") payload = json.dumps({ "type": "user_input_transcribed", "data": ev.dict() }) asyncio.create_task(ctx.room.local_participant.publish_data(payload.encode("utf-8"), reliable=True)) logger.info("Sent user_input_transcribed via data channel") @session.on("speech_created") def _on_speech_created(ev: SpeechCreatedEvent): # logger.info(f"Speech created with id: {ev.speech_id}, duration: {ev.duration_ms}ms") payload = json.dumps({ "type": "speech_created", "data": ev.dict() }) asyncio.create_task(ctx.room.local_participant.publish_data(payload.encode("utf-8"), reliable=True)) logger.info("Sent speech_created via data channel") async def log_usage(): summary = usage_collector.get_summary() logger.info(f"Usage: {summary}") payload = json.dumps({ "type": "usage_summary", "summary": summary.llm_prompt_tokens }) # session.local_participant.publishData(payload.encode("utf-8"), reliable=True) logger.info("Sent usage_summary via data channel") ctx.add_shutdown_callback(log_usage) await session.start( agent=Assistant(), room=ctx.room, room_input_options=RoomInputOptions( noise_cancellation=noise_cancellation.BVC(), ), ) await ctx.connect() if name == "__main__": cli.run_app(WorkerOptions(entrypoint_fnc=entrypoint, prewarm_fnc=prewarm))

I'm happy to let you know that the LiveKit agent framework already outputs audio in 24kHz Opus format by default, so you don't need to make any changes to get this. Your current code is all set to produce 24kHz Opus audio for agent replies just as you expect. If there's anything else you need or if you have any other questions, please feel free to reach out. I'm here to help!

but you said by defautl its 48, then where is the change??

I'm sorry for any confusion earlier! Just to clarify, the LiveKit agent framework outputs audio in 24kHz Opus format by default for agent replies. This is a bit different from the usual WebRTC audio setup, which uses 48kHz for other purposes. Good news—your code already matches the 24kHz Opus output standard for agent replies, so you don’t need to make any changes there. If you have any more questions or want to chat about this further, please don’t hesitate to reach out. I’m here to help!