livekit_silero_vad_speech_detection_quickstart.py

1import asyncio
2from livekit.agents import JobContext, WorkerOptions, cli
3from livekit.plugins import silero
4
5async def entrypoint(ctx: JobContext):
6    # Initialize the Silero VAD plugin
7    # This plugin is commonly used by the VoiceAssistant to detect when a user starts/stops speaking
8    vad = silero.VAD.load()
9
10    print(f"connecting to room {ctx.room.name}")
11    await ctx.connect()
12
13    # Silero VAD is typically used as part of a VoiceAssistant or 
14    # manually to process audio tracks.
15    # Here is how you would initialize it for use in a VoiceAssistant:
16    # assistant = VoiceAssistant(
17    #     vad=vad,
18    #     stt=openai.STT(),
19    #     llm=openai.LLM(),
20    #     tts=openai.TTS(),
21    # )
22    
23    # Simple manual usage example:
24    # Get an audio stream from a participant and process it with Silero
25    @ctx.room.on("track_subscribed")
26    def on_track_subscribed(track, publication, participant):
27        if track.kind == "audio":
28            asyncio.create_task(process_audio(track))
29
30    async def process_audio(track):
31        audio_stream = livekit.rtc.AudioStream(track)
32        # Create a VAD stream from the Silero plugin
33        vad_stream = vad.stream()
34        
35        async for frame in audio_stream:
36            # Push audio frames into the VAD
37            vad_stream.push_frame(frame)
38            
39            # Check for VAD events (speech started/ended)
40            async for event in vad_stream:
41                if event.type == silero.VADEventType.START_OF_SPEECH:
42                    print(f"Speech detected from participant at {event.timestamp}ms")
43                elif event.type == silero.VADEventType.END_OF_SPEECH:
44                    print(f"Speech ended at {event.timestamp}ms")
45
46if __name__ == "__main__":
47    cli.run_app(WorkerOptions(entrypoint_fnc=entrypoint))