Building a Smart IVR with Whisper Speech-to-Text and GPT Response

# Install Python dependencies

pip install openai-whisper torch torchaudio openai asterisk-agi
Or for faster inference

pip install faster-whisper
Install FFmpeg

sudo apt install ffmpeg -y 

; /etc/asterisk/extensions.conf

[smart-ivr] ; Main entry point for incoming calls exten => 1000,1,NoOp(Smart IVR Starting) same => n,Answer() same => n,Wait(1) same => n,Set(TIMEOUT(digit)=5) same => n,Set(TIMEOUT(response)=10)
; Play greeting same => n,Playback(welcome)  ; "Welcome to our company"
; Call our Python AGI script same => n,AGI(smart-ivr.py)
; If AGI sets TARGET variable, transfer same => n,GotoIf($["${TARGET}" != ""]?transfer:fallback)
same => n(transfer),NoOp(Transferring to ${TARGET}) same => n,Goto(${TARGET})
; Fallback to operator same => n(fallback),NoOp(Routing to operator) same => n,Goto(operator,s,1)
same => n,Hangup()
; Department extensions [sales] exten => s,1,NoOp(Sales Department) same => n,Dial(SIP/sales-queue,30) same => n,Voicemail(sales@company) same => n,Hangup()
[support] exten => s,1,NoOp(Support Department) same => n,Dial(SIP/support-queue,30) same => n,Voicemail(support@company) same => n,Hangup()
[billing] exten => s,1,NoOp(Billing Department) same => n,Dial(SIP/billing-queue,30) same => n,Voicemail(billing@company) same => n,Hangup()
[operator] exten => s,1,NoOp(Operator) same => n,Dial(SIP/operator,30) same => n,Voicemail(operator@company) same => n,Hangup() 

#!/usr/bin/env python3

""" Smart IVR with Whisper + GPT """
import sys import os import subprocess import tempfile from asterisk.agi import AGI import whisper import openai
Configuration

WHISPER_MODEL = "base.en"  # or "large-v3" for better accuracy OPENAI_API_KEY = os.getenv("OPENAI_API_KEY") openai.api_key = OPENAI_API_KEY
Load Whisper model (do once)

whisper_model = whisper.load_model(WHISPER_MODEL)
Department routing rules

DEPARTMENT_MAP = { "sales": ["sales", "buy", "purchase", "pricing", "demo", "trial"], "support": ["support", "help", "problem", "issue", "broken", "not working"], "billing": ["billing", "invoice", "payment", "charge", "subscription", "refund"], "operator": ["operator", "representative", "human", "person"] }
class SmartIVR: def __init__(self): self.agi = AGI() self.caller_id = self.agi.env['agi_callerid'] self.unique_id = self.agi.env['agi_uniqueid']
def speak(self, text): """Play text-to-speech to caller""" # Simple playback (you can replace with better TTS) self.agi.verbose(f"Speaking: {text}") # For production, use Festival, Piper, or pre-recorded audio
def listen(self, max_duration=10, silence_threshold=1.5): """Record audio from caller and transcribe""" # Generate unique filename audio_file = f"/tmp/ivr_{self.unique_id}"
# Record audio self.agi.verbose(f"Recording audio to {audio_file}") self.agi.record_file( audio_file, format='wav', escape_digits='#', timeout=max_duration * 1000, beep=True )
audio_path = f"{audio_file}.wav"
# Check if file exists and has content if not os.path.exists(audio_path) or os.path.getsize(audio_path) < 1000: self.agi.verbose("No audio recorded") return None
# Transcribe with Whisper self.agi.verbose("Transcribing audio...") try: result = whisper_model.transcribe(audio_path) transcription = result['text'].strip() self.agi.verbose(f"Transcription: {transcription}")
# Cleanup os.remove(audio_path)
return transcription
except Exception as e: self.agi.verbose(f"Transcription error: {e}") return None
def understand_intent(self, text): """Use GPT to understand caller intent""" if not text: return None
prompt = f"""You are an intelligent call routing assistant. Based on what the caller said, determine which department to route them to.
Caller said: "{text}"
Departments:

sales: For inquiries about buying, pricing, demos, trials

support: For technical issues, problems, troubleshooting

billing: For payment issues, invoices, refunds, subscriptions

operator: If unclear or they explicitly ask for a human

 Respond with ONLY the department name (sales, support, billing, or operator). No explanation needed."""

try: response = openai.ChatCompletion.create( model="gpt-4", messages=[ {"role": "system", "content": "You are a call routing expert."}, {"role": "user", "content": prompt} ], temperature=0.3, max_tokens=10 )
department = response.choices[0].message.content.strip().lower() self.agi.verbose(f"GPT determined department: {department}")
# Validate department if department in DEPARTMENT_MAP.keys(): return department else: return "operator"
except Exception as e: self.agi.verbose(f"GPT error: {e}") return None
def keyword_fallback(self, text): """Fallback to keyword matching if GPT fails""" text_lower = text.lower()
for dept, keywords in DEPARTMENT_MAP.items(): for keyword in keywords: if keyword in text_lower: return dept
return "operator"
def run(self): """Main IVR flow""" try: # Greeting self.speak("How can I help you today? Please speak after the beep.")
# Listen to caller transcription = self.listen()
if not transcription: self.speak("I didn't catch that. Routing you to an operator.") self.agi.set_variable("TARGET", "operator,s,1") return
# Understand intent with GPT department = self.understand_intent(transcription)
# Fallback to keywords if GPT fails if not department: department = self.keyword_fallback(transcription)
# Set routing target self.agi.verbose(f"Routing to: {department}") self.agi.set_variable("TARGET", f"{department},s,1")
# Confirm to caller self.speak(f"Connecting you to {department}. Please hold.")
except Exception as e: self.agi.verbose(f"Error in IVR: {e}") # Always fallback to operator on error self.agi.set_variable("TARGET", "operator,s,1")
if __name__ == '__main__': ivr = SmartIVR() ivr.run() 

chmod +x /var/lib/asterisk/agi-bin/smart-ivr.py

Test Python script

python3 /var/lib/asterisk/agi-bin/smart-ivr.py 

asterisk -rx "dialplan reload"

asterisk -rx "core reload" 

from faster_whisper import WhisperModel

4x faster than standard Whisper

model = WhisperModel("base.en", device="cuda", compute_type="float16") segments, info = model.transcribe(audio_path) transcription = " ".join([segment.text for segment in segments]) 

import redis

r = redis.Redis(host='localhost', port=6379, db=0)
def get_cached_intent(text): # Check cache first cached = r.get(f"intent:{text}") if cached: return cached.decode()
# If not cached, get from GPT and cache intent = understand_intent(text) r.setex(f"intent:{text}", 3600, intent)  # Cache for 1 hour return intent