livekit_turn_detector_eoanet_speech_event_quickstart.py

1import asyncio
2from livekit.plugins import turn_detector
3from livekit import agents
4
5async def main():
6    # Initialize the TurnDetector with a specific provider (e.g., EOANET)
7    # Note: You can customize the detector with different models or thresholds.
8    detector = turn_detector.EOANET()
9
10    # Create a dummy or real transcription iterator to simulate input
11    # In a real agent, this would come from a TranscribedParticipant or VAD
12    async def dummy_transcription_iterator():
13        yield agents.stt.SpeechEvent(
14            type=agents.stt.SpeechEventType.INTERIM_TRANSCRIPT,
15            alternatives=[agents.stt.SpeechData(text="Hello, how can I help you today?", language="en")]
16        )
17        await asyncio.sleep(0.5)
18        yield agents.stt.SpeechEvent(
19            type=agents.stt.SpeechEventType.FINAL_TRANSCRIPT,
20            alternatives=[agents.stt.SpeechData(text="Hello, how can I help you today?", language="en")]
21        )
22
23    # Wrap the iterator with the turn detector
24    async for event in detector.detect(dummy_transcription_iterator()):
25        if event.type == turn_detector.TurnEventType.TURN_STARTED:
26            print("User started speaking...")
27        elif event.type == turn_detector.TurnEventType.TURN_COMPLETE:
28            print(f"Turn complete! Final text: {event.text}")
29        elif event.type == turn_detector.TurnEventType.TURN_INTERRUPTED:
30            print("User was interrupted.")
31
32if __name__ == "__main__":
33    asyncio.run(main())