Repository: livekit/agents
File: livekit-agents/livekit/agents/voice/agent_activity.py (Python)

Code before:
        if self._rt_session is not None:
            self._rt_session.interrupt()

        if not interrupted_speeches:
            future.set_result(None)
        else:

            def on_playout_done(_: SpeechHandle) -> None:
                if not future.done() and all(speech.done() for speech in interrupted_speeches):
                    future.set_result(None)

            for speech in interrupted_speeches:
                speech.add_done_callback(on_playout_done)

        return future

    def clear_user_turn(self) -> None:
        if self._audio_recognition:
            self._audio_recognition.clear_user_turn()

        if self._rt_session is not None:
            self._rt_session.clear_audio()

    def commit_user_turn(self, *, transcript_timeout: float, stt_flush_duration: float) -> None:
        assert self._audio_recognition is not None
        self._audio_recognition.commit_user_turn(
            audio_detached=not self._session.input.audio_enabled,
            transcript_timeout=transcript_timeout,
            stt_flush_duration=stt_flush_duration,
        )

    def _schedule_speech(self, speech: SpeechHandle, priority: int, force: bool = False) -> None:
        # when force=True, we still allow to schedule a new speech even if
        # `pause_speech_scheduling` is waiting for the schedule_task to drain.
        # This allows for tool responses to be generated before the AgentActivity is finalized.

        if self._scheduling_paused and not force:
            raise RuntimeError(
                "cannot schedule new speech, the speech scheduling is draining/pausing"
            )

        if self._scheduling_atask and self._scheduling_atask.done():
            logger.warning(
                "attempting to schedule a new SpeechHandle, but the scheduling_task is not running."
            )
            return

        while True:
            try:
                # negate the priority to make it a max heap
                heapq.heappush(self._speech_q, (-priority, time.perf_counter_ns(), speech))

Review comment (suggestion):
Maybe this can be simplified to 
```suggestion
            self._rt_session.commit_audio()
            self._rt_session.generate_reply()
```

Code after:
        if self._rt_session is not None:
            self._rt_session.interrupt()

        if not interrupted_speeches:
            future.set_result(None)
        else:

            def on_playout_done(_: SpeechHandle) -> None:
                if not future.done() and all(speech.done() for speech in interrupted_speeches):
                    future.set_result(None)

            for speech in interrupted_speeches:
                speech.add_done_callback(on_playout_done)

        return future

    def clear_user_turn(self) -> None:
        if self._audio_recognition:
            self._audio_recognition.clear_user_turn()

        if self._rt_session is not None:
            self._rt_session.clear_audio()

    def commit_user_turn(self, *, transcript_timeout: float, stt_flush_duration: float) -> None:
        if self._rt_session is not None:
            self._rt_session.commit_user_turn()

        assert self._audio_recognition is not None
        self._audio_recognition.commit_user_turn(
            audio_detached=not self._session.input.audio_enabled,
            transcript_timeout=transcript_timeout,
            stt_flush_duration=stt_flush_duration,
        )

    def _schedule_speech(self, speech: SpeechHandle, priority: int, force: bool = False) -> None:
        # when force=True, we still allow to schedule a new speech even if
        # `pause_speech_scheduling` is waiting for the schedule_task to drain.
        # This allows for tool responses to be generated before the AgentActivity is finalized.

        if self._scheduling_paused and not force:
            raise RuntimeError(
                "cannot schedule new speech, the speech scheduling is draining/pausing"
            )

        if self._scheduling_atask and self._scheduling_atask.done():
            logger.warning(
                "attempting to schedule a new SpeechHandle, but the scheduling_task is not running."
            )
            return

        while True: