$ mkdir voice-agent cd voice-agent python -m venv venv source venv/bin/activate # Mac/Linux # venv\Scripts\activate # Windows -weight: 500;">pip -weight: 500;">install assemblyai openai elevenlabs websockets pyaudio python-dotenv mkdir voice-agent cd voice-agent python -m venv venv source venv/bin/activate # Mac/Linux # venv\Scripts\activate # Windows -weight: 500;">pip -weight: 500;">install assemblyai openai elevenlabs websockets pyaudio python-dotenv mkdir voice-agent cd voice-agent python -m venv venv source venv/bin/activate # Mac/Linux # venv\Scripts\activate # Windows -weight: 500;">pip -weight: 500;">install assemblyai openai elevenlabs websockets pyaudio python-dotenv ASSEMBLYAI_API_KEY=your_key OPENAI_API_KEY=your_key ELEVENLABS_API_KEY=your_key ASSEMBLYAI_API_KEY=your_key OPENAI_API_KEY=your_key ELEVENLABS_API_KEY=your_key ASSEMBLYAI_API_KEY=your_key OPENAI_API_KEY=your_key ELEVENLABS_API_KEY=your_key class AudioCapture: def __init__(self, sample_rate=16000): self.sample_rate = sample_rate self.chunk_size = 8000 self.audio_queue = Queue() self.recording = False self.audio = pyaudio.PyAudio() self.stream = None def start_recording(self): self.recording = True self.stream = self.audio.open( format=pyaudio.paInt16, channels=1, rate=self.sample_rate, input=True, frames_per_buffer=self.chunk_size ) thread = threading.Thread(target=self._capture_audio, daemon=True) thread.-weight: 500;">start() class AudioCapture: def __init__(self, sample_rate=16000): self.sample_rate = sample_rate self.chunk_size = 8000 self.audio_queue = Queue() self.recording = False self.audio = pyaudio.PyAudio() self.stream = None def start_recording(self): self.recording = True self.stream = self.audio.open( format=pyaudio.paInt16, channels=1, rate=self.sample_rate, input=True, frames_per_buffer=self.chunk_size ) thread = threading.Thread(target=self._capture_audio, daemon=True) thread.-weight: 500;">start() class AudioCapture: def __init__(self, sample_rate=16000): self.sample_rate = sample_rate self.chunk_size = 8000 self.audio_queue = Queue() self.recording = False self.audio = pyaudio.PyAudio() self.stream = None def start_recording(self): self.recording = True self.stream = self.audio.open( format=pyaudio.paInt16, channels=1, rate=self.sample_rate, input=True, frames_per_buffer=self.chunk_size ) thread = threading.Thread(target=self._capture_audio, daemon=True) thread.-weight: 500;">start() def handle_transcript(self, transcript: aai.RealtimeTranscript): if not transcript.text: return if isinstance(transcript, aai.RealtimeFinalTranscript): print(f"You: {transcript.text}") self.conversation_history.append( {"role": "user", "content": transcript.text} ) def handle_transcript(self, transcript: aai.RealtimeTranscript): if not transcript.text: return if isinstance(transcript, aai.RealtimeFinalTranscript): print(f"You: {transcript.text}") self.conversation_history.append( {"role": "user", "content": transcript.text} ) def handle_transcript(self, transcript: aai.RealtimeTranscript): if not transcript.text: return if isinstance(transcript, aai.RealtimeFinalTranscript): print(f"You: {transcript.text}") self.conversation_history.append( {"role": "user", "content": transcript.text} ) def generate_and_speak_response(self): messages = [{ "role": "system", "content": "Keep responses conversational and concise—aim for 1-2 sentences." }] + self.conversation_history response = openai_client.chat.completions.create( model="gpt-4", messages=messages, temperature=0.7, max_tokens=150 ) ai_response = response.choices[0].message.content def generate_and_speak_response(self): messages = [{ "role": "system", "content": "Keep responses conversational and concise—aim for 1-2 sentences." }] + self.conversation_history response = openai_client.chat.completions.create( model="gpt-4", messages=messages, temperature=0.7, max_tokens=150 ) ai_response = response.choices[0].message.content def generate_and_speak_response(self): messages = [{ "role": "system", "content": "Keep responses conversational and concise—aim for 1-2 sentences." }] + self.conversation_history response = openai_client.chat.completions.create( model="gpt-4", messages=messages, temperature=0.7, max_tokens=150 ) ai_response = response.choices[0].message.content def speak_text(self, text): audio = elevenlabs_client.generate( text=text, voice="Rachel", model="eleven_monolingual_v1" ) stream(audio) def speak_text(self, text): audio = elevenlabs_client.generate( text=text, voice="Rachel", model="eleven_monolingual_v1" ) stream(audio) def speak_text(self, text): audio = elevenlabs_client.generate( text=text, voice="Rachel", model="eleven_monolingual_v1" ) stream(audio) import os import threading from queue import Queue from dotenv import load_dotenv import assemblyai as aai from openai import OpenAI from elevenlabs.client import ElevenLabs from elevenlabs import stream import pyaudio load_dotenv() aai.settings.api_key = os.getenv('ASSEMBLYAI_API_KEY') openai_client = OpenAI(api_key=os.getenv('OPENAI_API_KEY')) elevenlabs_client = ElevenLabs(api_key=os.getenv('ELEVENLABS_API_KEY')) class AudioCapture: def __init__(self, sample_rate=16000): self.sample_rate = sample_rate self.chunk_size = 8000 self.audio_queue = Queue() self.recording = False self.audio = pyaudio.PyAudio() self.stream = None def start_recording(self): self.recording = True self.stream = self.audio.open( format=pyaudio.paInt16, channels=1, rate=self.sample_rate, input=True, frames_per_buffer=self.chunk_size ) thread = threading.Thread(target=self._capture_audio, daemon=True) thread.-weight: 500;">start() def _capture_audio(self): while self.recording: try: self.audio_queue.put( self.stream.read(self.chunk_size, exception_on_overflow=False) ) except Exception as e: print(f"Audio error: {e}") class VoiceAgent: def __init__(self): self.conversation_history = [] self.is_processing = False self.audio_capture = AudioCapture() def handle_transcript(self, transcript: aai.RealtimeTranscript): if not transcript.text: return if isinstance(transcript, aai.RealtimeFinalTranscript): print(f"You: {transcript.text}") self.conversation_history.append( {"role": "user", "content": transcript.text} ) if not self.is_processing: self.is_processing = True threading.Thread( target=self.generate_and_speak_response, daemon=True ).-weight: 500;">start() def generate_and_speak_response(self): try: messages = [{ "role": "system", "content": "You are a helpful voice assistant. Keep responses conversational." }] + self.conversation_history response = openai_client.chat.completions.create( model="gpt-4", messages=messages, temperature=0.7, max_tokens=150 ) ai_response = response.choices[0].message.content print(f"Agent: {ai_response}") self.conversation_history.append( {"role": "assistant", "content": ai_response} ) audio = elevenlabs_client.generate( text=ai_response, voice="Rachel", model="eleven_monolingual_v1" ) stream(audio) finally: self.is_processing = False def start_conversation(self): self.transcriber = aai.RealtimeTranscriber( sample_rate=16000, on_data=self.handle_transcript, on_error=lambda e: print(f"Speech error: {e}") ) try: self.transcriber.connect() self.audio_capture.start_recording() print("Voice Agent ready - -weight: 500;">start speaking!") while True: audio_chunk = self.audio_capture.get_audio_data() if audio_chunk: self.transcriber.stream(audio_chunk) except KeyboardInterrupt: self.audio_capture.stop_recording() self.transcriber.close() if __name__ == "__main__": VoiceAgent().start_conversation() import os import threading from queue import Queue from dotenv import load_dotenv import assemblyai as aai from openai import OpenAI from elevenlabs.client import ElevenLabs from elevenlabs import stream import pyaudio load_dotenv() aai.settings.api_key = os.getenv('ASSEMBLYAI_API_KEY') openai_client = OpenAI(api_key=os.getenv('OPENAI_API_KEY')) elevenlabs_client = ElevenLabs(api_key=os.getenv('ELEVENLABS_API_KEY')) class AudioCapture: def __init__(self, sample_rate=16000): self.sample_rate = sample_rate self.chunk_size = 8000 self.audio_queue = Queue() self.recording = False self.audio = pyaudio.PyAudio() self.stream = None def start_recording(self): self.recording = True self.stream = self.audio.open( format=pyaudio.paInt16, channels=1, rate=self.sample_rate, input=True, frames_per_buffer=self.chunk_size ) thread = threading.Thread(target=self._capture_audio, daemon=True) thread.-weight: 500;">start() def _capture_audio(self): while self.recording: try: self.audio_queue.put( self.stream.read(self.chunk_size, exception_on_overflow=False) ) except Exception as e: print(f"Audio error: {e}") class VoiceAgent: def __init__(self): self.conversation_history = [] self.is_processing = False self.audio_capture = AudioCapture() def handle_transcript(self, transcript: aai.RealtimeTranscript): if not transcript.text: return if isinstance(transcript, aai.RealtimeFinalTranscript): print(f"You: {transcript.text}") self.conversation_history.append( {"role": "user", "content": transcript.text} ) if not self.is_processing: self.is_processing = True threading.Thread( target=self.generate_and_speak_response, daemon=True ).-weight: 500;">start() def generate_and_speak_response(self): try: messages = [{ "role": "system", "content": "You are a helpful voice assistant. Keep responses conversational." }] + self.conversation_history response = openai_client.chat.completions.create( model="gpt-4", messages=messages, temperature=0.7, max_tokens=150 ) ai_response = response.choices[0].message.content print(f"Agent: {ai_response}") self.conversation_history.append( {"role": "assistant", "content": ai_response} ) audio = elevenlabs_client.generate( text=ai_response, voice="Rachel", model="eleven_monolingual_v1" ) stream(audio) finally: self.is_processing = False def start_conversation(self): self.transcriber = aai.RealtimeTranscriber( sample_rate=16000, on_data=self.handle_transcript, on_error=lambda e: print(f"Speech error: {e}") ) try: self.transcriber.connect() self.audio_capture.start_recording() print("Voice Agent ready - -weight: 500;">start speaking!") while True: audio_chunk = self.audio_capture.get_audio_data() if audio_chunk: self.transcriber.stream(audio_chunk) except KeyboardInterrupt: self.audio_capture.stop_recording() self.transcriber.close() if __name__ == "__main__": VoiceAgent().start_conversation() import os import threading from queue import Queue from dotenv import load_dotenv import assemblyai as aai from openai import OpenAI from elevenlabs.client import ElevenLabs from elevenlabs import stream import pyaudio load_dotenv() aai.settings.api_key = os.getenv('ASSEMBLYAI_API_KEY') openai_client = OpenAI(api_key=os.getenv('OPENAI_API_KEY')) elevenlabs_client = ElevenLabs(api_key=os.getenv('ELEVENLABS_API_KEY')) class AudioCapture: def __init__(self, sample_rate=16000): self.sample_rate = sample_rate self.chunk_size = 8000 self.audio_queue = Queue() self.recording = False self.audio = pyaudio.PyAudio() self.stream = None def start_recording(self): self.recording = True self.stream = self.audio.open( format=pyaudio.paInt16, channels=1, rate=self.sample_rate, input=True, frames_per_buffer=self.chunk_size ) thread = threading.Thread(target=self._capture_audio, daemon=True) thread.-weight: 500;">start() def _capture_audio(self): while self.recording: try: self.audio_queue.put( self.stream.read(self.chunk_size, exception_on_overflow=False) ) except Exception as e: print(f"Audio error: {e}") class VoiceAgent: def __init__(self): self.conversation_history = [] self.is_processing = False self.audio_capture = AudioCapture() def handle_transcript(self, transcript: aai.RealtimeTranscript): if not transcript.text: return if isinstance(transcript, aai.RealtimeFinalTranscript): print(f"You: {transcript.text}") self.conversation_history.append( {"role": "user", "content": transcript.text} ) if not self.is_processing: self.is_processing = True threading.Thread( target=self.generate_and_speak_response, daemon=True ).-weight: 500;">start() def generate_and_speak_response(self): try: messages = [{ "role": "system", "content": "You are a helpful voice assistant. Keep responses conversational." }] + self.conversation_history response = openai_client.chat.completions.create( model="gpt-4", messages=messages, temperature=0.7, max_tokens=150 ) ai_response = response.choices[0].message.content print(f"Agent: {ai_response}") self.conversation_history.append( {"role": "assistant", "content": ai_response} ) audio = elevenlabs_client.generate( text=ai_response, voice="Rachel", model="eleven_monolingual_v1" ) stream(audio) finally: self.is_processing = False def start_conversation(self): self.transcriber = aai.RealtimeTranscriber( sample_rate=16000, on_data=self.handle_transcript, on_error=lambda e: print(f"Speech error: {e}") ) try: self.transcriber.connect() self.audio_capture.start_recording() print("Voice Agent ready - -weight: 500;">start speaking!") while True: audio_chunk = self.audio_capture.get_audio_data() if audio_chunk: self.transcriber.stream(audio_chunk) except KeyboardInterrupt: self.audio_capture.stop_recording() self.transcriber.close() if __name__ == "__main__": VoiceAgent().start_conversation() - Real-time streaming speech-to-text - Language model comprehension - Text-to-speech synthesis - Conversation flow orchestration - Below 90%: Users experience frustration - 90-93%: Functional but with occasional errors - 93%+: Natural conversations with rare corrections - Intent recognition - Context tracking - Function calling for system integration - Response generation - Total response time: <1000ms - Speech accuracy: 93%+ - Name recognition: 95%+ - Number accuracy: 95%+ - Voice quality: Human-like - Customer support automation: Answer FAQs, check order -weight: 500;">status, escalate complex issues - Appointment scheduling: Check availability, confirm details, send confirmations - Lead qualification: Gather information, understand needs, route appropriately - After-hours -weight: 500;">service: Extend availability beyond business hours - Telephony integration (Twilio, SIP trunking) - Concurrent conversation handling - Error recovery for network failures - Performance monitoring and metrics - Security compliance and data protection