empty_timeout

departure_timeout

empty_timeout

deleteRoom

deleteRoom

from livekit import api
import asyncio

async def delete_room_after_duration(room_name, duration):
    await asyncio.sleep(duration)  # Wait for the specified duration (in seconds)

    api_client = api.LiveKitAPI(
        "LIVEKIT_URL",  # Replace with your LiveKit server URL
        "LIVEKIT_API_KEY",  # Replace with your API key
        "LIVEKIT_API_SECRET"  # Replace with your API secret
    )

    await api_client.room.delete_room(api.DeleteRoomRequest(
        room=room_name
    ))

# Example: Delete a room named 'test-room' after 5 minutes (300 seconds)
asyncio.run(delete_room_after_duration("test-room", 300))

with_ttl()

AccessToken

from livekit.api import AccessToken
import datetime

token = AccessToken(api_key="LIVEKIT_API_KEY", api_secret="LIVEKIT_API_SECRET")
token.with_identity(identity)
token.with_metadata(metadata_token)
token.with_grants(api.VideoGrants(room_join=True, room=room))

# Set TTL to 5 minutes (300 seconds)
token.with_ttl(datetime.timedelta(minutes=5))

jwt_token = token.to_jwt()

max_endpointing_delay

AgentSession

session = voice.AgentSession(
    turn_detection=turn_detector.MultilingualModel(),
    min_endpointing_delay=0.5,  # Minimum delay after silence
    max_endpointing_delay=6.0   # Maximum delay before forcing a turn
)

commit_user_turn()

session.commit_user_turn(transcript_timeout=2.0)

class InactivityHandler:
    def __init__(self, timeout_seconds=10, message="Are you still there?"):
        self._timeout = timeout_seconds
        self._message = message

    async def _check_timeout(self):
        while True:
            await asyncio.sleep(1)
            if time.time() - self._last_active > self._timeout:
                await self._agent.say(self._message)
                break

user_away_timeout

@session.on("user_state_changed")

async def entrypoint(ctx: JobContext):
    session = AgentSession(
        vad=silero.VAD.load(),
        llm=openai.LLM(model="gpt-4o-mini"),
        stt=deepgram.STT(),
        tts=cartesia.TTS(),
        user_away_timeout=15.0  # Set timeout to 15 seconds
    )

    @session.on("user_state_changed")
    def _user_state_changed(ev: UserStateChangedEvent):
        if ev.new_state == "away":
            # Make the agent respond with specific words
            asyncio.create_task(session.generate_reply(
                instructions="Are you still there?"
            ))

    await session.start(agent=Agent(instructions="You are a helpful assistant."), room=ctx.room)

user_started_speaking

import asyncio
import time

SILENCE_THRESHOLD = 5  # seconds

async def entrypoint(ctx: JobContext):
    user_last_spoke_time = time.time()

    async def monitor_interaction():
        while True:
            if time.time() - user_last_spoke_time > SILENCE_THRESHOLD:
                await ctx.room.disconnect()  # Disconnect the room
                break
            await asyncio.sleep(1)

    @ctx.agent.on("user_started_speaking")
    def on_user_started_speaking(_msg):
        nonlocal user_last_spoke_time
        user_last_spoke_time = time.time()

    asyncio.create_task(monitor_interaction())

user_started_speaking

max_call_duration

user_started_speaking

import asyncio
import time

SILENCE_THRESHOLD = 10  # seconds

async def entrypoint(ctx: JobContext):
    user_last_spoke_time = time.time()

    async def monitor_interaction():
        while True:
            if time.time() - user_last_spoke_time > SILENCE_THRESHOLD:
                await ctx.room.disconnect()  # Disconnect the room
                break
            await asyncio.sleep(1)

    @ctx.agent.on("user_started_speaking")
    def on_user_started_speaking(_msg):
        nonlocal user_last_spoke_time
        user_last_spoke_time = time.time()

    asyncio.create_task(monitor_interaction())

user_started_speaking

asyncio.wait_for

asyncio.wait_for

import asyncio
import time

SILENCE_THRESHOLD = 10  # seconds

async def entrypoint(ctx: JobContext):
    user_last_spoke_event = asyncio.Event()

    async def monitor_inactivity():
        while True:
            try:
                # Wait for the user to speak within the threshold
                await asyncio.wait_for(user_last_spoke_event.wait(), timeout=SILENCE_THRESHOLD)
                user_last_spoke_event.clear()  # Reset the event after user speaks
            except asyncio.TimeoutError:
                # Silence threshold exceeded, disconnect the room
                await ctx.room.disconnect()
                break

    @ctx.agent.on("user_started_speaking")
    def on_user_started_speaking(_msg):
        user_last_spoke_event.set()  # Signal that the user has spoken

    asyncio.create_task(monitor_inactivity())

pip install "livekit-agents[resemble]~=1.2"

voice_uuid

from livekit.plugins import resemble
import asyncio

async def main():
    tts = resemble.TTS(
        api_key="your_api_key",
        voice_uuid="your_voice_uuid",
        sample_rate=44100,
        output_format="wav"
    )
    audio_stream = tts.synthesize("Hello, world!")
    async for chunk in audio_stream:
        print(f"Received audio chunk: {len(chunk.frame.data)} bytes")

asyncio.run(main())

tts.TTS

stream()

synthesize()

AudioFrame

AudioByteStream

AgentSession

class CustomTTSEngine(tts.TTS):
    def __init__(self, model, sample_rate, num_channels):
        super().__init__(streaming=True, sample_rate=sample_rate, num_channels=num_channels)
        self._model = model

    async def stream(self):
        # Implement streaming logic here
        pass