https://livekit.io logo
Join Slack
Powered by
Why is my tts and llm not in sync? and when the ll...
# ask-ai
a
Why is my tts and llm not in sync? and when the llm stops generating the text the tts model stops regardless of if it has finished everything it has to say on not. I noticed this issue in production. what could be wrong
Copy code
async def entrypoint(ctx: agents.JobContext):
    initial_ctx = ChatContext()
    
    # Get the unique room name from the job context
    room_name = ctx.room.name
    session_id = None  # Store session_id in local scope
    
    # Connect to the room FIRST
    await ctx.connect(auto_subscribe=AutoSubscribe.AUDIO_ONLY)
    
    # NOW wait for participant to join
    participant = await ctx.wait_for_participant()
    
    # Extract the Supabase JWT and conversation_id from the participant's metadata
    supabase_jwt = None
    timezone = None
    conversation_id = None
    is_reconnection = False
    user_id = None
    
    if participant and participant.metadata:
        try:
            metadata = json.loads(participant.metadata)
            supabase_jwt = metadata.get("supabase_jwt")
            timezone = metadata.get("timezone")
            conversation_id = metadata.get("conversation_id")
            
            if supabase_jwt:
                <http://logger.info|logger.info>(f"Supabase JWT successfully extracted for room {room_name}")
                user_id = participant.identity
            else:
                logger.warning(f"No supabase_jwt found in participant metadata for room {room_name}")
            if timezone:
                <http://logger.info|logger.info>(f"Timezone successfully extracted for room {room_name}: {timezone}")
            else:
                logger.warning(f"No timezone found in participant metadata for room {room_name}, defaulting to UTC")
                timezone = 'UTC'     
            if conversation_id:
                <http://logger.info|logger.info>(f"Conversation ID found: {conversation_id}. This is a reconnection.")
                is_reconnection = True
        except json.JSONDecodeError as e:
            logger.warning(f"Could not parse participant metadata in room {room_name}: {e}")
    else:
        logger.warning(f"No participant metadata available for room {room_name}")
    
    # Track call session start and store session_id
    if supabase_jwt and user_id:
        session_id = await track_call_start(room_name, user_id, supabase_jwt, conversation_id)

    # Define cleanup function
    async def cleanup():
        """Cleanup function to track call end"""
        if session_id and supabase_jwt:
            <http://logger.info|logger.info>(f"Running cleanup for room {room_name}, session_id: {session_id}")
            await track_call_end(session_id, supabase_jwt, room_name)
        else:
            logger.warning(f"Cleanup skipped for room {room_name} - missing session_id or JWT")

    # Register cleanup callback CORRECTLY
    ctx.add_shutdown_callback(cleanup)

    # If reconnection, load conversation history
    if is_reconnection and conversation_id and supabase_jwt:
        try:
            backend_url = os.getenv('BACKEND_URL') if os.getenv('ENVIRONMENT') == 'production' else '<http://localhost:8000>'
            headers = {'Authorization': f'Bearer {supabase_jwt}'}
            
            response = requests.get(
                f'{backend_url}/get-conversation-context?conversation_id={conversation_id}',
                headers=headers,
                timeout=10
            )
            
            if response.status_code == 200:
                context_data = response.json()
                
                # Add PDF content if available
                if context_data.get('pdf_content'):
                    initial_ctx.add_message(
                        role="assistant",
                        content=f"Study material ({context_data.get('pdf_name', 'document')}):\n{context_data['pdf_content']}"
                    )
                    <http://logger.info|logger.info>(f"Loaded PDF content for conversation {conversation_id}")
                
                # Add conversation history
                for msg in context_data.get('messages', []):
                    initial_ctx.add_message(
                        role=msg['role'],
                        content=msg['content']
                    )
                
                <http://logger.info|logger.info>(f"Loaded {len(context_data.get('messages', []))} messages for conversation {conversation_id}")
            else:
                logger.warning(f"Failed to load conversation context: {response.status_code}")
                is_reconnection = False
        except Exception as e:
            logger.error(f"Error loading conversation context: {e}")
            is_reconnection = False

    # If not reconnection, wait for study material
    if not is_reconnection:
        study_material = await wait_for_study_material(room_name)
        
        if study_material:
            initial_ctx.add_message(
                role="assistant",
                content=f"Study material:\n{study_material}"
            )
            print(f"Agent for room {room_name} initialized with study material")
        else:
            print(f"Agent for room {room_name} initialized without study material")

    session = AgentSession(
        stt = assemblyai.STT(
            format_turns=True,
        ),
        llm=openai.LLM(model="gpt-4o-mini"),
        tts=inworld.TTS(voice="Deborah", model="inworld-tts-1-max"),
        vad=silero.VAD.load(),
        turn_detection=MultilingualModel(),
        use_tts_aligned_transcript=True,
    )

    # Create the assistant instance with room_name and JWT for calendar access
    assistant = Assistant(chat_ctx=initial_ctx, room_name=room_name, supabase_jwt=supabase_jwt, timezone=timezone, room=ctx.room)

    # Handle data messages for interrupt functionality
    @ctx.room.on("data_received")
    def on_data_received(data):
        try:
            payload = json.loads(data.data.decode("utf-8"))
            message = payload["message"]
            if message == "__interrupt__":
                <http://logger.info|logger.info>("Interrupt command received. Stopping current TTS playback.")
                session.interrupt()
                session.interrupt()
                session.interrupt()
        except Exception as e:
            logger.error(f"Error processing data message: {e}")

    # Handle participant disconnection to delete the room immediately
    @ctx.room.on("participant_disconnected")
    def on_participant_disconnected(participant):
        async def delete_room():
            api_client = None
            try:
                <http://logger.info|logger.info>(f"Participant {participant.identity} disconnected from room {room_name}")
                
                # Delete the room immediately
                api_client = api.LiveKitAPI(
                    os.getenv("LIVEKIT_URL"),
                    os.getenv("LIVEKIT_API_KEY"),
                    os.getenv("LIVEKIT_API_SECRET"),
                )
                await api_client.room.delete_room(api.DeleteRoomRequest(
                    room=ctx.job.room.name,
                ))
                <http://logger.info|logger.info>(f"Room {room_name} deleted successfully after participant disconnect")
            except Exception as e:
                logger.error(f"Error deleting room {room_name} after participant disconnect: {e}")
            finally:
                # Properly close the API client session
                if api_client is not None:
                    await api_client.aclose()
        
        # Run the async deletion as a task
        asyncio.create_task(delete_room())

    await session.start(
        room=ctx.room,
        agent=assistant,
        room_options=room_io.RoomOptions(
            audio_input=room_io.AudioInputOptions(
                noise_cancellation=noise_cancellation.BVC(),
            ),
            delete_room_on_close=True,
        ),
    )
    
    # Choose welcome message based on reconnection status
    if is_reconnection:
        welcome_msg = "Don't answer the question you previously asked if any. Go to the page you were on and carry on the session as you were already in a call with the student."
    elif study_material:
        if len(study_material) > 10:
            welcome_msg = WELCOME_MESSAGE_1
        else:
            welcome_msg = WELCOME_MESSAGE_2
    else:
        welcome_msg = WELCOME_MESSAGE_3

    await session.generate_reply(instructions=welcome_msg)
t
Thanks for reaching out, David. I'm an AI assistant and I'm here to help. Give me a moment while I research this topic.
Thanks — I reviewed your code and the behavior you described. You're seeing TTS stop when the LLM stops because something in your setup is actively interrupting playback. The likely causes and actionable fixes: 1) Primary cause — over‑aggressive interrupt logic • Your data handler calls 
session.interrupt()
three times whenever a 
__interrupt__
message arrives. One call is enough; repeated calls are redundant and can clear the speech queue. • If the frontend or another system component sends 
__interrupt__
at turn boundaries (or accidentally), that will cancel TTS immediately even if audio hasn’t finished. Fix: • Call 
session.interrupt()
only once. • Debounce/deduplicate interrupts (e.g., ignore repeats within ~0.5s). • Validate the interrupt payload so only explicit user interrupts trigger it. Example pattern:
Copy code
last_interrupt_time = 0

@ctx.room.on("data_received")
async def on_data_received(data):
    global last_interrupt_time
    payload = json.loads(data.data.decode("utf-8"))
    if payload.get("message") == "__interrupt__":
        now = time.time()
        if now - last_interrupt_time > 0.5:
            <http://logger.info|logger.info>("Valid interrupt: stopping TTS")
            session.interrupt()
            last_interrupt_time = now
2) Verify frontend behavior • Confirm the frontend only sends 
{ "message": "__interrupt__" }
when the user explicitly interrupts (push-to-talk or UI button), not at LLM turn end or TTS start/stop. • Look for race conditions where the client emits interrupts on state transitions. 3) use_tts_aligned_transcript setting • If you don’t need word‑aligned transcripts, set 
use_tts_aligned_transcript=False
. That feature changes buffering/alignment behavior and can interact with playback timing. 4) Add logging/events to confirm flow • Listen for 
speech_created
and 
playback_finished
(and log 
ev.interrupted
) to see whether playback is being cancelled vs naturally finishing. That will tell you if interrupts are the cause. 5) Expected behavior • TTS should continue playing after the LLM finishes; the LLM ending shouldn’t by itself stop TTS. If you see otherwise, it means something (most likely an interrupt) is cancelling playback. If you’d like, I can: • Point out exact spots in your frontend to inspect for accidental interrupts. • Provide a small wrapper to centralize interrupt handling and telemetry for easier debugging. Sources: Agent speech and audio | LiveKit docs | Turn detection and interruptions | LiveKit Docs | Live Conversations with AI using ChatGPT and WebRTC | Text and transcriptions | LiveKit docs | How to detect when an agent has finished speaking
2 Views
Open in Slack
PreviousNext
Powered by