livekit-plugins-noise-cancellation

import os from livekit.agents import AgentSession, JobContext, RoomOptions, RoomInputOptions from livekit.plugins import noise_cancellation class MyAgent: async def start(self, ctx: JobContext): # Example of applying BVC noise cancellation to inbound audio await ctx.connect( RoomOptions(auto_subscribe=RoomOptions.AutoSubscribe.AUDIO_ONLY), room_input_options=RoomInputOptions( noise_cancellation=noise_cancellation.BVC(), ), ) print("Agent connected with BVC noise cancellation enabled.") # Your agent logic here, e.g., process audio from ctx.room.local_participant.audio_tracks # Keep the agent alive (e.g., by waiting for a signal or keeping a loop) # For a simple quickstart, we'll just wait indefinitely for demonstration. # In a real agent, you'd manage session lifecycle more robustly. await ctx.room.wait_for_is_connected() async def main(): # This quickstart assumes a LiveKit Agent environment setup # In a real scenario, AgentSession would be started by the LiveKit server # with appropriate environment variables for authentication. LIVEKIT_URL = os.environ.get('LIVEKIT_URL', 'wss://your-livekit-url.cloud') LIVEKIT_API_KEY = os.environ.get('LIVEKIT_API_KEY', 'your_api_key') LIVEKIT_API_SECRET = os.environ.get('LIVEKIT_API_SECRET', 'your_api_secret') # This part typically runs on a LiveKit Agent service, not as a standalone script. # For local testing, you might mock AgentSession or connect to a local LiveKit instance. print("To run this, ensure LIVEKIT_URL, LIVEKIT_API_KEY, and LIVEKIT_API_SECRET are set.") print("This code snippet demonstrates plugin usage within a LiveKit AgentSession.") if __name__ == '__main__': # In a typical LiveKit Agents deployment, you wouldn't directly call main here. # The agent system would instantiate and manage MyAgent. # This is for illustration of the noise_cancellation plugin usage. import asyncio asyncio.run(main())