livekit-plugins-aws

import asyncio import os from livekit.agents import Agent, JobContext, WorkerOptions, cli from livekit.agents.llm import OpenAI from livekit.agents.voice import VoiceActivityDetector from livekit.plugins import aws # --- Environment Variables Needed --- # LIVEKIT_URL, LIVEKIT_API_KEY, LIVEKIT_API_SECRET # AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY, AWS_REGION (e.g., us-east-1) # OPENAI_API_KEY (if using OpenAI LLM) # ---------------------------------- class MyAWSVoiceAgent(Agent): def __init__(self): super().__init__() # AWS STT and TTS pick up credentials and region from # environment variables or default AWS config (~/.aws/credentials). self.aws_stt = aws.STT() # Uses AWS Transcribe self.aws_tts = aws.TTS() # Uses AWS Polly self.openai_llm = OpenAI() # Example: Using OpenAI for LLM async def _on_connected(self, ctx: JobContext): print(f"Agent connected to room: {ctx.room.name}") # Initialize the agent session with AWS STT/TTS session = ctx.get_agent_session( llm=self.openai_llm, tts=self.aws_tts, stt=self.aws_stt, vad=VoiceActivityDetector(), # Recommended for robust voice interaction # preemptive_generation=False # Set to False if you want to disable the 1.5.0 default ) await session.start() print("Agent session started with AWS STT/TTS. Waiting for user input...") async for turn in session.ai_turns(): if turn.text: print(f"User (via AWS Transcribe): {turn.text}") response = await self.openai_llm.generate_reply(turn.history) # Agent speaks response via AWS Polly await turn.say(response.text) print(f"Agent (via AWS Polly): {response.text}") print("Agent session ended.") if __name__ == "__main__": cli.run_agent(MyAWSVoiceAgent(), WorkerOptions( log_level="INFO", rtc_url=os.environ.get("LIVEKIT_URL", "ws://localhost:7880"), webrtc_url=os.environ.get("LIVEKIT_WEBRTC_URL", "http://localhost:7880"), api_key=os.environ.get("LIVEKIT_API_KEY", ""), api_secret=os.environ.get("LIVEKIT_API_SECRET", ""), ))