able-jackal-84140
07/08/2025, 1:04 PMtall-airline-16756
07/08/2025, 1:04 PMtall-airline-16756
07/08/2025, 1:05 PMsession.say()
with preloaded audio frames from a fixed audio file. For example, you can play a custom audio message indicating the termination or next steps, and use the parameter allow_interruptions=False
to prevent the audio from being interrupted during playback.
Example in Python:
# Play a fixed audio message on session termination
session.say(
"I'm having trouble connecting right now. Let me transfer your call.",
audio=audio_frames_from_file(custom_error_audio),
allow_interruptions=False,
)
This approach ensures that your fixed audio is played reliably when the session ends.
If you need guidance on integrating this with session termination events specifically, please provide your programming environment or SDK details so I can offer more tailored advice.
Sources: livekit-agents/livekit/agents/tts/tts.pyable-jackal-84140
07/08/2025, 1:08 PMtall-airline-16756
07/08/2025, 1:08 PMable-jackal-84140
07/08/2025, 1:09 PMable-jackal-84140
07/08/2025, 1:09 PMfrom dotenv import load_dotenv
from livekit import agents
from livekit.agents import AgentSession, Agent, RoomInputOptions, ChatContext
from livekit.plugins import (
openai,
silero,
)
import os
from livekit.plugins.turn_detector.multilingual import MultilingualModel
from livekit.plugins.turn_detector.english import EnglishModel
from livekit.agents import metrics, MetricsCollectedEvent
from livekit import api
import logging
load_dotenv()
logger = logging.getLogger(__name__)
usage_collector = metrics.UsageCollector()
class Assistant(Agent):
def __init__(self, chat_ctx: ChatContext) -> None:
super().__init__(chat_ctx=chat_ctx, instructions="You are a helpful assistant. Provide the answer to the best of your ability to the user. Keep your responses short and clear. Reply the user in the same language in which he is responding.")
async def entrypoint(ctx: agents.JobContext):
await ctx.connect()
session = AgentSession(
llm=openai.LLM(model=os.getenv("OPENAI_MODEL"), base_url=os.getenv("OPENAI_BASE_URL"), api_key=os.getenv("OPENAI_API_KEY")),
tts=openai.TTS(model=os.getenv("TEXT_TO_SPEECH_MODEL"), api_key="websocket123", base_url=os.getenv("TEXT_TO_SPEECH_SERVER_WEBSOCKET_URL")),
vad=silero.VAD.load(),
stt=openai.STT(model=os.getenv("STT_MODEL"), api_key=os.getenv("STT_API_KEY"), base_url=os.getenv("STT_BASE_URL"))
)
@session.on("metrics_collected")
def _on_metrics_collected(ev: MetricsCollectedEvent):
metrics.log_metrics(ev.metrics)
usage_collector.collect(ev.metrics)
async def log_usage():
summary = usage_collector.get_summary()
<http://logger.info|logger.info>(f"Usage: {summary}")
initial_ctx = ChatContext()
initial_ctx.add_message(role="user", content=f"Hi there!")
await session.start(
room=ctx.room,
agent=Assistant(chat_ctx=initial_ctx),
)
ctx.add_shutdown_callback(log_usage)
await session.generate_reply(
instructions="Greet the user in short and crisp format in hindi"
)
if __name__ == "__main__":
agents.cli.run_app(agents.WorkerOptions(entrypoint_fnc=entrypoint, shutdown_process_timeout=5, agent_name="puch"))
user this exampletall-airline-16756
07/08/2025, 1:10 PMimport os
import pathlib
import logging
from dotenv import load_dotenv
from livekit.agents import JobContext, WorkerOptions, cli
from livekit.agents.voice import Agent, AgentSession
from livekit.agents.voice.events import CloseEvent
from livekit.plugins import deepgram, openai, silero
logger = logging.getLogger("session_termination_audio")
logger.setLevel(<http://logging.INFO|logging.INFO>)
load_dotenv()
async def entrypoint(ctx: JobContext):
await ctx.connect()
session = AgentSession(
stt=deepgram.STT(),
llm=openai.LLM(),
tts=silero.TTS(),
vad=silero.VAD.load(),
)
# Define termination audio file path
custom_termination_audio = os.path.join(
pathlib.Path(__file__).parent.absolute(),
"termination_message.ogg"
)
@session.on("close")
async def on_close(ev: CloseEvent):
<http://logger.info|logger.info>(f"Session is closing, reason: {ev.reason}")
# Play termination audio
async def audio_frames_from_file(file_path: str):
"""
Generates audio frames from an audio file.
"""
# Implementation example using livekit.utils.audioread
from livekit.utils import audioread
from livekit.rtc import AudioFrame
with audioread.open_file(file_path) as audio:
for frame in audio.frames():
yield AudioFrame(
data=frame.data,
sample_rate=audio.sample_rate,
num_channels=audio.num_channels,
samples_per_channel=len(frame.data) // (audio.num_channels * 2)
)
await session.say(
"Thank you for using our service. Goodbye!",
audio=audio_frames_from_file(custom_termination_audio),
allow_interruptions=False,
)
await session.start(agent=Agent(instructions="You are a helpful assistant."), room=ctx.room)
if __name__ == "__main__":
cli.run_app(WorkerOptions(entrypoint_fnc=entrypoint))
Key points:
• Place your custom audio file termination_message.ogg
in the same directory as the script.
• The on_close
handler plays a fixed audio message when the session closes.
• audio_frames_from_file
generates audio frames required by session.say()
to play the audio.
• Adjust dependencies and environment variables as needed for your LiveKit and plugin setup.
Hey there! I totally get that working with audio in LiveKit can feel a bit complex. If you're stuck on any part of this example or need a hand setting things up, I'm here to help! Feel free to reach out, and we'll work through it together. 😊 Sources: livekit-plugins/livekit-plugins-bithuman/livekit/plugins/bithuman/avatar.py