polite-furniture-24847
08/18/2025, 8:22 AMtall-airline-16756
08/18/2025, 8:22 AMtall-airline-16756
08/18/2025, 8:22 AMuser_away_timeout
, you can gently handle it using LiveKit's event monitoring along with a bit of custom logic. Here’s a friendly walkthrough on how to set this up:
1. Event Monitoring: Keep an ear out for the user_state_changed
event—that’s what tells you when the user’s status switches to "away" after the timeout.
2. Custom Logic: Create a function that kicks in when the user goes inactive. This function can check in with the user, wait for their response, and if there’s no activity, gracefully close the session.
3. Configuration: Don’t forget to set the user_away_timeout
when you initialize the AgentSession
to decide how long you want to wait before considering the user away.
Here’s an example to help you get started:
from livekit.agents import Agent, AgentSession
import asyncio
async def entrypoint(ctx):
session = AgentSession(
vad=silero.VAD.load(),
llm=openai.LLM(model="gpt-4o-mini"),
stt=deepgram.STT(),
tts=cartesia.TTS(),
user_away_timeout=15.0, # We’re setting the timeout to 15 seconds here
)
inactivity_task = None
async def user_presence_task():
# Let’s try to re-engage the user a few times
for _ in range(3):
await session.generate_reply(
instructions="The user has been inactive. Please respond to continue."
)
await asyncio.sleep(10)
# If there’s still no response, we’ll close the session
await session.aclose()
ctx.delete_room()
@session.on("user_state_changed")
def user_state_changed(ev):
nonlocal inactivity_task
if ev.new_state == "away":
inactivity_task = asyncio.create_task(user_presence_task())
else:
if inactivity_task:
inactivity_task.cancel()
inactivity_task = None
await session.start(agent=Agent())
This setup listens for when the user goes "away" and then gently tries to bring them back. If they don’t respond after a few friendly nudges, it closes the session to keep things tidy.polite-furniture-24847
08/18/2025, 8:23 AMtall-airline-16756
08/18/2025, 8:24 AMuser_transcript
even when there’s no speech detected, you can take control by manually updating the transcript. This involves interrupting any ongoing speech and generating a reply with either the current or updated transcript. Here’s a simple example to help you out:
async def on_input_speech_started():
if self.llm is None:
return
# Grab the last transcript if you have one
last_transcript = "…" # Feel free to use your preferred default transcript or just an empty string
chat_ctx = self._agent.chat_ctx.copy()
# Create a new reply using the last transcript
speech_handle = self._generate_reply(
user_message=llm.ChatMessage(role="user", content=[last_transcript]),
chat_ctx=chat_ctx,
schedule_speech=False
)
# If there’s current speech, interrupt it
if self._current_speech is not None and self._current_speech.interrupted:
self._current_speech._mark_interrupted_by_user()
This way, your user_transcript
stays up to date even when no new speech comes in.
And just so you know, if you’re using a RealtimeModel, LiveKit has your back! It’ll automatically manage transcription and update the user_transcript
once the user stops speaking or the turn wraps up.
Sources: livekit-agents/livekit/agents/voice/transcription/synchronizer.py