user_away_timeout

"away"

"listening"

user_away_timeout

user_away_timeout

user_away_timeout

SpeechHandle

handle.wait_for_playout()

agent_state_changed

"listening"

"speaking"

allow_interruptions=False

agent_state_changed

SpeechHandle

wait_for_playout()

"speaking"

"away"

SpeechHandle.wait_for_playout()

agent_state_changed

"listening"

user_away_timeout

import asyncio
from typing import Optional

# ---- Configuration ----
USER_AWAY_TIMEOUT = 3          # seconds to consider user away when both silent
CONFIRMATION_PROMPT = "Are you still there?"
MAX_NO_RESPONSE_ROUNDS = 2     # how many confirm prompts before ending call
LATENCY_OPTIMIZATIONS = {
    "streaming_mode": True,
    # add any SDK-specific low-latency flags here
}

# ---- Stubs / Integration Points (replace with your SDK calls) ----
class Session:
    # Minimal session interface used by this script
    def __init__(self):
        self.on_agent_state_changed = None
        self.on_user_speech_activity = None
        self.on_agent_speech_activity = None

    async def say(self, text: str, allow_interruptions: bool = True):
        """
        Start agent speech. Return a SpeechHandle-like object with wait_for_playout().
        Replace with actual SDK call that returns a handle.
        """
        return SpeechHandle(text)

    def subscribe_agent_state(self, callback):
        self.on_agent_state_changed = callback

    def subscribe_user_activity(self, callback):
        self.on_user_speech_activity = callback

    def subscribe_agent_activity(self, callback):
        self.on_agent_speech_activity = callback

class SpeechHandle:
    # Simulate a playout handle. Replace with actual handle type.
    def __init__(self, text: str, playout_time: float = 0.5):
        self._playout_time = playout_time
        self._done = asyncio.Event()

    async def wait_for_playout(self):
        # In real SDK this resolves when playout fully finishes
        await asyncio.sleep(self._playout_time)
        self._done.set()

# ---- Agent Implementation ----
class PresenceAgent:
    def __init__(self, session: Session):
        self.session = session
        self.user_speaking = False
        self.agent_speaking = False
        self.no_response_rounds = 0
        self._away_task: Optional[asyncio.Task] = None

        # Subscribe to events
        session.subscribe_agent_state(self._on_agent_state_changed)
        session.subscribe_user_activity(self._on_user_activity)
        session.subscribe_agent_activity(self._on_agent_activity)

    # Event handlers (hook these to the SDK events)
    def _on_agent_state_changed(self, ev):
        # ev.new_state could be "speaking", "listening", etc.
        self.agent_speaking = (ev.new_state == "speaking")
        # When agent becomes listening, we may start the away timer if user silent
        if ev.new_state == "listening":
            self._maybe_start_away_timer()

    def _on_user_activity(self, is_speaking: bool):
        self.user_speaking = is_speaking
        if is_speaking:
            # user returned — reset counters and stop away timer
            self.no_response_rounds = 0
            self._cancel_away_timer()
        else:
            # user silence — maybe start timer
            self._maybe_start_away_timer()

    def _on_agent_activity(self, is_speaking: bool):
        self.agent_speaking = is_speaking
        if is_speaking:
            self._cancel_away_timer()
        else:
            self._maybe_start_away_timer()

    def _maybe_start_away_timer(self):
        # Start the away timer only if both user and agent are silent
        if not self.user_speaking and not self.agent_speaking:
            if self._away_task is None or self._away_task.done():
                self._away_task = asyncio.create_task(self._away_countdown())
        else:
            self._cancel_away_timer()

    def _cancel_away_timer(self):
        if self._away_task and not self._away_task.done():
            self._away_task.cancel()
            self._away_task = None

    async def _away_countdown(self):
        try:
            await asyncio.sleep(USER_AWAY_TIMEOUT)
        except asyncio.CancelledError:
            return

        # If we reach here, both were silent for USER_AWAY_TIMEOUT
        await self._handle_user_away_timeout()

    async def _handle_user_away_timeout(self):
        # Confirm presence flow
        if self.no_response_rounds >= MAX_NO_RESPONSE_ROUNDS:
            await self._end_call()
            return

        # Prompt the user to confirm presence
        self.no_response_rounds += 1
        handle = await self.session.say(CONFIRMATION_PROMPT, allow_interruptions=True)
        # Wait for playout to finish so we are certain agent audio is done
        await handle.wait_for_playout()

        # After playout finishes, if user still silent, the away countdown will restart via event handlers
        # (no direct sleep here; rely on event-driven user activity hooks)
        # If we want, we can start an explicit short wait window here:
        # await asyncio.sleep(USER_AWAY_TIMEOUT)
        # But relying on the common code above avoids duplicating logic.

    async def _end_call(self):
        # End the call gracefully — replace with SDK hangup/terminate logic
        print("No response after prompts — ending call.")
        # perform hangup/cleanup here

# ---- Example wiring / main loop ----
async def main():
    # Initialize session (replace with real SDK initialization, including streaming/latency flags)
    session = Session()

    agent = PresenceAgent(session)

    # Example: connect and start streaming optimizations here
    # e.g., session.enable_streaming_mode(LATENCY_OPTIMIZATIONS)  # pseudo-code

    # The following is illustrative: simulate events
    class Ev:
        def __init__(self, new_state):
            self.new_state = new_state

    # Simulate agent speaking then listening then silence
    session.on_agent_state_changed(Ev("speaking"))
    await asyncio.sleep(0.6)
    session.on_agent_state_changed(Ev("listening"))

    # No user activity afterwards; agent listening triggers away timer
    # Wait enough time to let the agent confirm and end the call
    await asyncio.sleep(10)

if __name__ == "__main__":
    asyncio.run(main())

wait_for_playout()

allow_interruptions=True

user_away_timeout

user_state_changed

SpeechHandle.wait_for_playout()

AgentSession

import asyncio
import logging
from typing import Optional

from livekit.agents import (
    AgentServer,
    Agent,
    AgentSession,
    JobContext,
    UserStateChangedEvent,
)
# Replace these with the plugins you use:
from livekit.plugins import silero, deepgram, cartesia, openai

logger = logging.getLogger("presence-agent")
server = AgentServer()

# Configuration
USER_AWAY_TIMEOUT = 3.0          # seconds required for both parties to be silent
CONFIRM_PROMPT = "Are you still there?"
MAX_RETRIES = 2                  # total confirm attempts before hangup
CONFIRM_WAIT = 5.0               # seconds to wait for user response after prompt

@server.rtc_session(agent_name="presence-agent")
async def entrypoint(ctx: JobContext):
    # Create agent session with VAD/STT/LLM/TTS and away timeout.
    # Swap the plugin initializers for your environment.
    session = AgentSession(
        vad=silero.VAD.load(),          # VAD used to detect user speech activity
        stt=deepgram.STT(),            # speech-to-text plugin
        llm=openai.LLM(model="gpt-4o-mini"),
        tts=cartesia.TTS(),
        user_away_timeout=USER_AWAY_TIMEOUT,
    )

    # State
    inactivity_task: Optional[asyncio.Task] = None
    retry_count = 0

    async def confirm_presence_flow():
        nonlocal retry_count, inactivity_task
        retry_count = 0
        while retry_count < MAX_RETRIES:
            retry_count += 1
            # Use generate_reply to produce TTS via the agent pipeline.
            # This returns a SpeechHandle-like object in the real SDK.
            handle = await session.generate_reply(instructions=CONFIRM_PROMPT)
            # Wait for actual audio playout to finish before starting response timer.
            # The real SpeechHandle from LiveKit provides wait_for_playout().
            await handle.wait_for_playout()

            # Wait for a short window for user activity. We rely on user_state_changed events
            # to cancel this wait early if the user speaks.
            try:
                await asyncio.wait_for(_wait_for_user_speaking(), timeout=CONFIRM_WAIT)
                # If user spoke, handlers clear retry_count and we exit
                return
            except asyncio.TimeoutError:
                # No user activity in CONFIRM_WAIT seconds; loop to prompt again
                continue

        # After retries exhausted, say goodbye and shut down
        await session.generate_reply(instructions="Goodbye.")
        # Ensure final audio plays before shutdown
        # If generate_reply returns handle, await its wait_for_playout() as above.
        await asyncio.sleep(1.0)
        <http://logger.info|logger.info>("No response after retries. Shutting down session.")
        session.shutdown()

    # Helper: a coroutine that returns when user is detected speaking.
    # The real code will be event-driven; here we use an event to be set by the user_state handler.
    user_spoke_event = asyncio.Event()
    async def _wait_for_user_speaking():
        await user_spoke_event.wait()
        user_spoke_event.clear()

    # Event handler for user state changes
    @session.on("user_state_changed")
    def on_user_state_changed(ev: UserStateChangedEvent):
        nonlocal inactivity_task, retry_count
        # ev.new_state can be "speaking", "listening", "away"
        if ev.new_state == "speaking":
            # user returned — stop any inactivity flows
            if inactivity_task and not inactivity_task.done():
                inactivity_task.cancel()
                inactivity_task = None
            retry_count = 0
            user_spoke_event.set()
        elif ev.new_state == "away":
            # When user becomes away, start the confirm-presence flow if not already running
            if inactivity_task is None or inactivity_task.done():
                inactivity_task = asyncio.create_task(confirm_presence_flow())
        else:
            # "listening" or other states: no immediate action
            pass

    # Optionally listen for agent_state_changed if you need to know when agent is speaking
    @session.on("agent_state_changed")
    def on_agent_state_changed(ev):
        # ev.new_state can be used if you want to block away-timers while agent is speaking.
        # The AgentSession handles user_away_timeout only when both are silent.
        pass

    # Start the session + agent
    agent = Agent(instructions="You are a helpful assistant.")
    await session.start(agent=agent, room=ctx.room)
    await ctx.connect(auto_subscribe=True)

    # Initial greeting
    handle = await session.generate_reply(instructions="Hello! How can I help you today?")
    await handle.wait_for_playout()

    # Keep the entrypoint alive until the session ends
    await session.wait_closed()

if __name__ == "__main__":
    logging.basicConfig(level=<http://logging.INFO|logging.INFO>)
    server.run()