How to detect and handle silence during calls

1
import asyncio
2
import logging
3

4
from livekit.agents import (
5
    Agent,
6
    AgentServer,
7
    AgentSession,
8
    JobContext,
9
    UserStateChangedEvent,
10
    cli,
11
)
12
from livekit.plugins import cartesia, deepgram, openai, silero
13

14
logger = logging.getLogger("silence-detection-agent")
15

16
server = AgentServer()
17

18
@server.rtc_session(agent_name="silence-detection-agent")
19
async def entrypoint(ctx: JobContext):
20
    session = AgentSession(
21
        vad=silero.VAD.load(),
22
        llm=openai.LLM(model="gpt-4o-mini"),
23
        stt=deepgram.STT(),
24
        tts=cartesia.TTS(),
25
        user_away_timeout=5.0,  # seconds of silence before user is marked "away"
26
    )
27

28
    @session.on("user_state_changed")
29
    def _user_state_changed(ev: UserStateChangedEvent):
30
        if ev.new_state == "away":
31
            logger.info("User has been silent for too long, shutting down session")
32
            session.shutdown()
33

34
    await session.start(
35
        agent=Agent(instructions="You are a helpful assistant."),
36
        room=ctx.room
37
    )
38

39
if __name__ == "__main__":
40
    cli.run_app(server)

1
import asyncio
2
import logging
3

4
from livekit.agents import (
5
    Agent,
6
    AgentServer,
7
    AgentSession,
8
    JobContext,
9
    UserStateChangedEvent,
10
    cli,
11
)
12
from livekit.plugins import cartesia, deepgram, openai, silero
13

14
logger = logging.getLogger("silence-detection-agent")
15

16
server = AgentServer()
17

18
@server.rtc_session(agent_name="silence-detection-agent")
19
async def entrypoint(ctx: JobContext):
20
    session = AgentSession(
21
        vad=silero.VAD.load(),
22
        llm=openai.LLM(model="gpt-4o-mini"),
23
        stt=deepgram.STT(),
24
        tts=cartesia.TTS(),
25
        user_away_timeout=12.5,  # seconds before user is marked "away"
26
    )
27

28
    inactivity_task: asyncio.Task | None = None
29

30
    async def user_presence_task():
31
        # Try to ping the user 3 times, if we get no answer, close the session
32
        for _ in range(3):
33
            await session.generate_reply(
34
                instructions="The user has been inactive. Politely check if the user is still present."
35
            )
36
            await asyncio.sleep(10)
37

38
        logger.info("No response after multiple attempts, shutting down")
39
        session.shutdown()
40

41
    @session.on("user_state_changed")
42
    def _user_state_changed(ev: UserStateChangedEvent):
43
        nonlocal inactivity_task
44
        if ev.new_state == "away":
45
            inactivity_task = asyncio.create_task(user_presence_task())
46
            return
47

48
        # User is now speaking or listening, cancel the inactivity task
49
        if inactivity_task is not None:
50
            inactivity_task.cancel()
51
            inactivity_task = None
52

53
    await session.start(
54
        agent=Agent(instructions="You are a helpful assistant."),
55
        room=ctx.room
56
    )
57

58
if __name__ == "__main__":
59
    cli.run_app(server)