What is the APIConnectOptions for exactly? Which API are we referring to here? Whenw ould you edit this, if ever?

Is there anyway for me to use livekit API to extend departure_timeout time? or to find out how much time left before the room closes during the departure_timeout phase? Otherwise, how should I support rejoining of room? whereby how do i check if the agent is still waiting in the room? before sending back the same livekit token to the frontend?

what is the event for on user_state_change

Is it possible to add hooks to listen to events such as participant has disconnected in the backend? Currently I am simply creating the access token for client to join, so livekit handles the creation of room. If i wish to add hooks to listen to events, does that mean i need to manually create the room myself and add the hook?

why am Igetting participantInfo: {identity: 'agent-AJ_34QGakHuyeVy'} same identity for both user and agent while getting live transcription?

Hi I am using google realtime model as LLM Do we need to add seprate TTS STT? or we can have NOT_GIVEN? If we add STT and TTS will it use that?

Assume you are expert in livekit and full knowledge of google realtime model I am using google realtime model, due to me speaking english but due to my english accent is still hindi output coming out of livekit sdk in frontend is still hindi how to ensure we only get english output of user

Hi, I'm using SSML tags in my LiveKit agent and it works fine for speech, but the tags are showing up in the live transcription on screen. I've tried overriding

transcription_node

like this:

async def transcription_node(self, text_stream: AsyncIterable[str], model_settings: ModelSettings):
    async for text in super().transcription_node(text_stream, model_settings):
        if self.filter_ssml:
            text = self.ssml_tag_pattern.sub("", text)
        yield text

This works for messages sent using session.say(), but responses generated by the LLM are still not filtered. Does anyone know a way to clean or remove the SSML tags from live transcription for LLM outputs as well? Thanks a lot!

can i use responses url in livekit openai with azure plugin, or do I need to use chat/completions url?

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

Hi everyone, I’ve built a chatbot in LiveKit Playground using Python, running locally. The problem I’m facing is that when STT (Speech-to-Text) fails, the agent message doesn’t show up in the chat at all. Has anyone run into this issue before or knows how to handle STT failures so the chatbot can still show a response or at least an error message instead of staying blank?

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))

can i get 160000 samplerate audio from livekit??

I am deploying Livekit SIP in an AWS EC2 VM. This VM has a fixed ElasticIP assigned. While testing the integration with my telephony provider (Genesys), we observe that Genesys is not able to send media (RTP packets) to my VM, where the SIP Gateway is hosted. That's because the IP that Livekit SIP advertises in the Connection Address field of the 200 OK (INVITE) signaling message is the internal IP of the AWS EC2 instance, not the public one. Has anyone encountered this issue before? How has it been solved?

This message was deleted.

how to diconnrct call in voice agent without funtion calliong

is there any recommended way to truncate the conversation history before we generate the llm response?

i got this error ✓ Compiled /api/record/start in 807ms (998 modules) Attempting to start recording for room: vercel-570ce5b0-2d6a-465e-bf37-916896b943c2 LiveKit URL: wss://XXXXXXX.ngrok-free.app Egress URL: https://XXXXXXX.ngrok-free.app Room vercel-570ce5b0-2d6a-465e-bf37-916896b943c2 found with 0 participants Generated filename: livekit/record/vercel-570ce5b0-2d6a-465e-bf37-916896b943c2.mp4 File output configuration: { "filepath": "livekit/record/vercel-570ce5b0-2d6a-465e-bf37-916896b943c2.mp4", "outputType": "s3", "region": "XXXXXX", "bucket": "XXXXXX" } Starting room composite egress... Error starting recording: twirp error unknown: request has missing or invalid field: ws_url GET /api/record/start?roomName=vercel-570ce5b0-2d6a-465e-bf37-916896b943c2 500 in 1768ms this is my code import { EgressClient, EncodedFileOutput, S3Upload, RoomServiceClient } from 'livekit-server-sdk'; import { NextRequest, NextResponse } from 'next/server'; // import { getMeetingByRoomName, updateMeetingRecordingLink } from '@/lib/mongodb'; // import { BlobServiceClient, StorageSharedKeyCredential } from '@azure/storage-blob'; // Function to generate versioned filename based on existing recordings // COMMENTED OUT: Using simple filename for S3 upload /* async function generateVersionedFilename( roomName: string, storageAccountName: string, storageAccountKey: string, containerName: string ): Promise<string> { try { // Create blob service client const sharedKeyCredential = new StorageSharedKeyCredential(storageAccountName, storageAccountKey); const blobServiceClient = new BlobServiceClient(

https://${storageAccountName}.<http://blob.core.windows.net|blob.core.windows.net>

, sharedKeyCredential ); const containerClient = blobServiceClient.getContainerClient(containerName); // List all blobs that start with the room name pattern const prefix = `livekit/record/${roomName}`; const existingFiles: string[] = []; for await (const blob of containerClient.listBlobsFlat({ prefix })) { existingFiles.push(blob.name); } // Extract version numbers from existing files const versions: number[] = []; const basePattern = `livekit/record/${roomName}`; for (const fileName of existingFiles) { if (fileName ===

${basePattern}.mp4

) { // This is the base file (no version number) versions.push(0); } else { _// Try to extract version number from pattern: roomName_X.mp4_ _const match = fileName.match(new RegExp(

^${basePattern.replace(/[.*+?^${}()|[\]\\]/g, '\\$&')}_([0-9]+)\\.mp4$

));_ if (match) { versions.push(parseInt(match[1], 10)); } } } // Determine the next version number let nextVersion = 0; if (versions.length > 0) { const maxVersion = Math.max(...versions); nextVersion = maxVersion + 1; } // console.log(

Room: ${roomName}, Found versions: [${versions.join(', ')}], Next version: ${nextVersion}

); // Generate filename based on version if (nextVersion === 0) { return `livekit/record/${roomName}.mp4`; } else if (nextVersion === 1) { _return `livekit/record/${roomName}_1.mp4`;_ } else { _return `livekit/record/${roomName}_${nextVersion}.mp4`;_ } } catch (error) { console.error('Error checking existing files:', error); // Fallback to base filename if there's an error return `livekit/record/${roomName}.mp4`; } } */ export async function *GET*(_req_: NextRequest) { try { const roomName = req.nextUrl.searchParams.get('roomName'); if (roomName === null) { return new NextResponse('Missing roomName parameter', { status: 403 }); } const { LIVEKIT_API_KEY, LIVEKIT_API_SECRET, LIVEKIT_URL, } = process.env; // S3 configuration const S3_BUCKET_NAME = 'XXXXXXXXXXXXX'; const S3_REGION = 'XXXXXXXXXXXXX'; // Debug logging console.log(

Attempting to start recording for room: ${roomName}

); console.log(

LiveKit URL: ${LIVEKIT_URL}

); // Use the same URL conversion as the original working code const hostURL = new URL(LIVEKIT_URL!); hostURL.protocol = 'https:'; console.log(

Egress URL: ${hostURL.origin}

); const egressClient = new EgressClient(hostURL.origin, LIVEKIT_API_KEY, LIVEKIT_API_SECRET); // Check if room exists first try { const roomClient = new RoomServiceClient(LIVEKIT_URL!, LIVEKIT_API_KEY, LIVEKIT_API_SECRET); const rooms = await roomClient.listRooms([roomName]); if (rooms.length === 0) { console.log(

Room ${roomName} does not exist

); return new NextResponse('Room does not exist. Please ensure participants have joined the room before starting recording.', { status: 404 }); } console.log(

Room ${roomName} found with ${rooms[0].numParticipants} participants

); } catch (roomError) { console.error('Error checking room existence:', roomError); // Continue anyway, let egress handle it } const existingEgresses = await egressClient.listEgress({ roomName }); if (existingEgresses.length > 0 && existingEgresses.some((e) => e.status < 2)) { return new NextResponse('Meeting is already being recorded', { status: 409 }); } // Generate simple filename for S3 upload const fileName = `livekit/record/${roomName}.mp4`; console.log(

Generated filename: ${fileName}

); const fileOutput = new EncodedFileOutput({ filepath: fileName, output: { case: 's3', value: new S3Upload({ _// accessKey: AWS_ACCESS_KEY_ID!,_ _// secret: AWS_SECRET_ACCESS_KEY!,_ region: S3_REGION, bucket: S3_BUCKET_NAME, }), }, }); console.log('File output configuration:', JSON.stringify({ filepath: fileName, outputType: 's3', region: S3_REGION, bucket: S3_BUCKET_NAME }, null, 2)); // Use the exact method signature from your original working code console.log('Starting room composite egress...'); await egressClient.startRoomCompositeEgress( roomName, { file: fileOutput, }, { layout: 'speaker', }, ); console.log('Recording started successfully'); // Construct the URL of the saved file in S3 const fileUrl = `https://${S3_BUCKET_NAME}.s3.${S3_REGION}.amazonaws.com/${fileName}`; console.log(

Recording started. File will be saved to: ${fileUrl}

); // save the meeting record try { const response = await fetch(

${process.env.PYTHON_DISPATCHER_URL}/recordings

, { method: 'POST', headers: { 'Content-Type': 'application/json', }, body: JSON.stringify({ room_name: roomName, recording_url: fileUrl, }), }); if (!response.ok) { console.error(

Failed to save recording link. Status: ${response.status}

); } } catch (saveError) { console.error(

Error saving recording link for room: ${roomName}

, saveError); } return new NextResponse(null, { status: 200 }); } catch (error) { if (error instanceof Error) { console.error('Error starting recording:', error.message); return new NextResponse(error.message, { status: 500 }); } console.error('Unknown error starting recording:', error); return new NextResponse('Internal server error', { status: 500 }); } }

Is it a bad idea to call send/send_nowait ourselves when implementing our custom llm adapter? when would you use send vs. send_nowait? from future import annotations import asyncio import contextlib from collections import deque from collections.abc import AsyncIterator from typing import Generic, Protocol, TypeVar T = TypeVar("T") T_co = TypeVar("T_co", covariant=True) T_contra = TypeVar("T_contra", contravariant=True) # Based on asyncio.Queue, see https://github.com/python/cpython/blob/main/Lib/asyncio/queues.py class ChanClosed(Exception): pass class ChanFull(Exception): pass class ChanEmpty(Exception): pass class ChanSender(Protocol[T_contra]): async def send(self, value: T_contra) -> None: ... def send_nowait(self, value: T_contra) -> None: ... def close(self) -> None: ... class ChanReceiver(Protocol[T_co]): async def recv(self) -> T_co: ... def recv_nowait(self) -> T_co: ... def close(self) -> None: ... def __aiter__(self) -> AsyncIterator[T_co]: ... async def __anext__(self) -> T_co: ... class Chan(Generic[T]): def __init__( self, maxsize: int = 0, loop: asyncio.AbstractEventLoop | None = None, ) -> None: self._loop = loop or asyncio.get_event_loop() self._maxsize = max(maxsize, 0) # self._finished_ev = asyncio.Event() self._close_ev = asyncio.Event() self._closed = False self._gets: deque[asyncio.Future[T | None]] = deque() self._puts: deque[asyncio.Future[T | None]] = deque() self._queue: deque[T] = deque() def _wakeup_next(self, waiters: deque[asyncio.Future[T | None]]) -> None: while waiters: waiter = waiters.popleft() if not waiter.done(): waiter.set_result(None) break async def send(self, value: T) -> None: while self.full() and not self._close_ev.is_set(): p = self._loop.create_future() self._puts.append(p) try: await p except ChanClosed: raise except: p.cancel() with contextlib.suppress(ValueError): self._puts.remove(p) if not self.full() and not p.cancelled(): self._wakeup_next(self._puts) raise self.send_nowait(value) def send_nowait(self, value: T) -> None: if self.full(): raise ChanFull if self._close_ev.is_set(): raise ChanClosed self._queue.append(value) self._wakeup_next(self._gets) async def recv(self) -> T: while self.empty() and not self._close_ev.is_set(): g = self._loop.create_future() self._gets.append(g) try: await g except ChanClosed: raise except Exception: g.cancel() with contextlib.suppress(ValueError): self._gets.remove(g) if not self.empty() and not g.cancelled(): self._wakeup_next(self._gets) raise return self.recv_nowait() def recv_nowait(self) -> T: if self.empty(): if self._close_ev.is_set(): raise ChanClosed else: raise ChanEmpty item = self._queue.popleft() # if self.empty() and self._close_ev.is_set(): # self._finished_ev.set() self._wakeup_next(self._puts) return item def close(self) -> None: self._closed = True self._close_ev.set() for putter in self._puts: if not putter.cancelled(): putter.set_exception(ChanClosed()) while len(self._gets) > self.qsize(): getter = self._gets.pop() if not getter.cancelled(): getter.set_exception(ChanClosed()) while self._gets: self._wakeup_next(self._gets) # if self.empty(): # self._finished_ev.set() @property def closed(self) -> bool: return self._closed # async def join(self) -> None: # await self._finished_ev.wait() def qsize(self) -> int: """the number of elements queued (unread) in the channel buffer""" return len(self._queue) def full(self) -> bool: if self._maxsize <= 0: return False else: return self.qsize() >= self._maxsize def empty(self) -> bool: return not self._queue def __aiter__(self) -> AsyncIterator[T]: return self async def __anext__(self) -> T: try: return await self.recv() except ChanClosed: raise StopAsyncIteration from None

This message was deleted.

iam getting this error twirp error unknown: request has missing or invalid field: ws_url when trying to upload the recording to s3 storage const fileOutput = new EncodedFileOutput({ filepath: fileName, output: { case: 's3', value: new S3Upload({ accessKey: AWS_ACCESS_KEY_ID!, secret: AWS_SECRET_ACCESS_KEY!, region: S3_REGION, bucket: S3_BUCKET_NAME, }), }, }); console.log('File output configuration:', JSON.stringify({ filepath: fileName, outputType: 's3', region: S3_REGION, bucket: S3_BUCKET_NAME }, null, 2)); // Use the exact method signature from your original working code console.log('Starting room composite egress...'); await egressClient.startRoomCompositeEgress( roomName, { file: fileOutput, }, { layout: 'speaker', }, );

I have multiple team members which need access to multiple livekit projects. Right now, I have to give access to each team member in each project. Is there a way to give my entire team access to all projects in my livekit account?

How do I get the timestamp on the transcript when using AgentSession.history() ?

How can I get livekit api-key and livekit api-secret from livekit cloud server?

what node is called in pipeline when participants sends a text message to room

Is the TTL value used for the access token also used for the refresh token?

If I already have 1 agent deployed to cloud under the free tier and I test another by running it locally in dev mode, will the agents coincide?

This message was deleted.

I am using the flutter livekit_client sdk to integrate in app voice chat in my app. In AudioPublishOptions class, the audioBitrate can be set, and it's documentation says that it is the "Max audio bitrate". So is it correct to assume that livekit will automatically switch to a lower quality in case network conditions do not support a higher bitrate?

why am i getting this

ignoring incoming text stream due to no handler for topic lk.transcription

Is there a way to detect users bandwidth and adjust the bitrate of an ai avatar accordingly