livekit_agents_noise_cancellation_plugin_audio_track_processing.py

1import logging
2
3from livekit.agents import JobContext, WorkerOptions, cli
4from livekit.plugins import noise_cancellation
5
6logger = logging.getLogger("noise-cancellation-demo")
7
8async def entrypoint(ctx: JobContext):
9    # Initialize the noise cancellation plugin (Krisp)
10    # Note: This plugin requires a valid license key for production use
11    nc = noise_cancellation.NoiseCancellation.create()
12
13    @ctx.room.on("track_subscribed")
14    def on_track_subscribed(track, publication, participant):
15        if track.kind == "audio":
16            logger.info(f"Subscribed to audio track from {participant.identity}")
17            
18            # Create a processor to filter the audio track
19            processor = nc.create_processor()
20            
21            # track.add_processor attaches the noise cancellation filter to the audio stream
22            track.add_processor(processor)
23            logger.info("Noise cancellation processor attached")
24
25    await ctx.connect()
26    logger.info("Agent connected to room")
27
28if __name__ == "__main__":
29    cli.run_app(WorkerOptions(entrypoint_fnc=entrypoint))