Tutorial 40: AI-Powered Answering Machine Detection — Whisper + ML Classifier

Build a self-hosted answering machine detection (AMD) system that replaces Asterisk's built-in AMD() application with a Whisper-based speech recognition + machine learning classifier pipeline. Traditional AMD relies on energy detection and cadence analysis, achieving only 60-70% accuracy in real-world conditions — misclassifying live humans as machines (killing revenue-generating calls) and letting voicemail greetings through to agents (wasting expensive seat time). This tutorial's AI approach transcribes the first 3-5 seconds of answered audio using OpenAI's Whisper model, then feeds the transcript and audio features into a trained ML classifier that distinguishes human pickups from answering machines with 95%+ accuracy. The entire system runs on your own hardware with no per-call API costs, processes decisions in under 2 seconds, and continuously improves as you feed it new labeled data from your call center's actual traffic.

Technologies: Python, FastAPI, Faster-Whisper, scikit-learn, PyTorch, Asterisk, AGI, ViciDial, Prometheus Difficulty: Advanced Reading time: ~75 minutes Prerequisites: Working ViciDial/Asterisk installation, Python 3.11+, basic ML concepts

Table of Contents

1. Introduction — Why AMD Matters
2. How Traditional AMD Works
3. Architecture Overview
4. Prerequisites
5. Training Data Collection
6. Whisper Feature Extraction
7. ML Classifier — Train the Model
8. AMD Service — FastAPI Microservice
9. Asterisk Integration
10. Voicemail Drop — When Machine Detected
11. Performance Tuning
12. Monitoring & Analytics
13. Comparison with Commercial Solutions
14. Troubleshooting

1. Introduction — Why AMD Matters

Answering Machine Detection is arguably the single most impactful technology decision in an outbound call center. Every answered call falls into one of two categories:

• Human pickup — a live person answers and should be connected to an agent immediately
• Machine pickup — a voicemail greeting, IVR system, fax tone, or other automated response

Getting this classification wrong has direct financial consequences:

In a 50-agent call center making 10,000 outbound calls per day, roughly 20-40% of answered calls reach voicemail (varies by time of day, industry, and list quality). That is 2,000-4,000 AMD decisions daily. At 70% accuracy:

• 600-1,200 wrong decisions per day
• False positives at even 5% = 100-200 live humans hung up on daily
• False negatives at 25% = 500-1,000 voicemails wasting agent time

The math is brutal. If each false positive costs $5 in lost revenue opportunity and each false negative costs $0.50 in wasted agent time, a 70%-accurate AMD system costs $750-$1,500 per day in inefficiency. Improving to 95% accuracy saves most of that.

Why Traditional AMD Fails

Asterisk's built-in AMD() application — and every traditional AMD system — works by analyzing audio properties rather than content. It measures:

• How long the initial silence lasts
• How many words are spoken (estimated by voice energy bursts)
• How long the greeting runs
• The silence between words
• The total length of continuous speech

The fundamental assumption is: machines talk longer and more continuously than humans. A human answers "Hello?" (one short word), while a voicemail says "Hi, you've reached John Smith. I'm not available right now. Please leave a message after the tone" (many words, long continuous speech).

This assumption breaks constantly:

• Humans who answer with long greetings: "Hello, this is John Smith speaking, how can I help you?" — classified as machine
• Short voicemail greetings: "Leave a message" — classified as human
• IVR systems with pauses: "Press 1 for sales... press 2 for support..." — each pause resets the analysis
• Background noise: TV, radio, street noise creates energy patterns that confuse detection
• VoIP artifacts: Jitter, packet loss, and codec compression distort the audio envelope
• Cultural differences: Some cultures answer with longer greetings naturally

The AI approach solves this by actually understanding what is being said. When Whisper transcribes "Please leave a message after the tone," no amount of cadence variation matters — the content is unambiguously a voicemail greeting.

2. How Traditional AMD Works

Before building the replacement, you need to understand what you are replacing. Asterisk's AMD() application is defined in app_amd.c and accepts these parameters:

AMD() Parameters

AMD(initialSilence, greeting, afterGreetingSilence, totalAnalysisTime,
    minimumWordLength, betweenWordsSilence, maximumNumberOfWords,
    silenceThreshold, maximumWordLength)

The Detection Algorithm

The AMD state machine works roughly like this:

CALL ANSWERED
    │
    ▼
┌─────────────┐
│ Wait for     │──── silence > initialSilence ──── → MACHINE
│ first sound  │
└──────┬──────┘
       │ sound detected
       ▼
┌─────────────┐
│ Analyze      │──── word count > maximumNumberOfWords ── → MACHINE
│ greeting     │──── greeting duration > greeting ──────── → MACHINE
│              │──── single word > maximumWordLength ───── → MACHINE
└──────┬──────┘
       │ greeting ends (silence detected)
       ▼
┌─────────────┐
│ Wait after   │──── silence > afterGreetingSilence ────── → MACHINE
│ greeting     │──── new speech starts quickly ──────────── → HUMAN
└──────┬──────┘
       │ totalAnalysisTime exceeded
       ▼
    NOTSURE

ViciDial Default AMD Configuration

ViciDial's AMD settings are stored in the vicidial_campaigns table:

SELECT campaign_id, amd_send_to_vmx, amd_type,
       amd_inbound_group, amd_callmenu,
       use_amd, amd_ai
FROM vicidial_campaigns
WHERE campaign_id = 'YOUR_CAMPAIGN';

Typical ViciDial AMD settings:

; ViciDial default AMD parameters (in extensions.conf)
AMD(2500,1500,800,5000,100,50,3,256,5000)

These defaults are designed to be "safe" — they lean toward classifying calls as HUMAN to avoid false positives. The result is a high false negative rate: many voicemails get sent to agents.

Why the Parameters Cannot Be "Tuned" to Good Accuracy

Call center operators spend countless hours adjusting AMD parameters, and the results are always the same:

Aggressive settings (short greeting window, few max words):

AMD(2000,1000,600,4000,100,50,2,256,5000)

• Catches more voicemails ✓
• But also hangs up on humans who say more than "Hello?" ✗
• Compliance risk from dropping live calls ✗

Conservative settings (long greeting window, more max words):

AMD(3000,2500,1200,6000,100,50,5,256,5000)

• Rarely hangs up on humans ✓
• But many voicemails slip through to agents ✗
• Agents waste time on "Hi, you've reached..." ✗

The fundamental problem: you are trying to classify content by measuring the container. It is like trying to determine if a letter contains good news or bad news by weighing the envelope. Sometimes it correlates — but never reliably.

Real-World AMD Accuracy Data

Based on published ViciDial community data and real call center measurements:

3. Architecture Overview

System Architecture

                        OUTBOUND CALL FLOW

    ViciDial Dialer                    Asterisk
    ┌──────────┐                    ┌──────────────┐
    │ Campaign │───── originate ───→│  Dial()       │
    │ Hopper   │                    │  Call Answers │
    └──────────┘                    └──────┬───────┘
                                          │
                                   Answer detected
                                          │
                                          ▼
                                   ┌──────────────┐
                                   │ AGI Script    │
                                   │ capture_amd.py│
                                   │              │
                                   │ Record first │
                                   │ 3-5 sec of   │
                                   │ audio        │
                                   └──────┬───────┘
                                          │
                                    HTTP POST
                                    /amd endpoint
                                          │
                                          ▼
                              ┌────────────────────┐
                              │   AMD Service       │
                              │   (FastAPI :8190)   │
                              │                     │
                              │ ┌─────────────────┐ │
                              │ │ Faster-Whisper   │ │
                              │ │ Transcribe 3-5s  │ │
                              │ │ "Hello?"         │ │
                              │ └────────┬────────┘ │
                              │          │          │
                              │          ▼          │
                              │ ┌─────────────────┐ │
                              │ │ Feature Extract  │ │
                              │ │ text + audio     │ │
                              │ │ features         │ │
                              │ └────────┬────────┘ │
                              │          │          │
                              │          ▼          │
                              │ ┌─────────────────┐ │
                              │ │ ML Classifier    │ │
                              │ │ HUMAN: 0.97      │ │
                              │ │ MACHINE: 0.03    │ │
                              │ └────────┬────────┘ │
                              │          │          │
                              └──────────┼──────────┘
                                         │
                                    JSON response
                                    {result, confidence}
                                         │
                                         ▼
                              ┌────────────────────┐
                              │   AGI Script        │
                              │   Route Call         │
                              │                     │
                              │ HUMAN → Agent Queue │
                              │ MACHINE → VM Drop   │
                              │ NOTSURE → Agent     │
                              └────────────────────┘

Latency Budget

The entire AMD decision must happen before the caller (human or machine) finishes their initial utterance and starts waiting for a response. For humans, this is typically 2-4 seconds of patience. The budget:

Total budget:           3000-5000ms
├── Call answer detect:   200-500ms  (SIP 200 OK processing)
├── Audio capture:       1500-3000ms (record first words)
├── Network transfer:      50-100ms  (local network POST)
├── Whisper transcribe:   300-800ms  (base model, CPU)
├── Feature extraction:    10-50ms   (text processing)
├── ML classification:      5-20ms   (sklearn predict)
└── AGI routing:           50-100ms  (set channel variable)

Total processing:       2100-4500ms

The key insight: you do not need to wait for the full 5 seconds. As soon as Whisper produces a transcript with enough confidence, classify immediately. A human saying "Hello?" is transcribable in 800ms of audio. A voicemail saying "Hi, you've reached..." is identifiable within 2 seconds.

Component Overview

4. Prerequisites

Hardware Requirements

Software Requirements

Install on the server that will run the AMD service (can be the ViciDial server itself or a separate machine):

# System dependencies
apt-get update && apt-get install -y \
    python3.11 python3.11-venv python3.11-dev \
    ffmpeg sox libsox-dev \
    build-essential git curl

# Create project directory
mkdir -p /opt/amd-service
cd /opt/amd-service

# Python virtual environment
python3.11 -m venv venv
source venv/bin/activate

# Core dependencies
pip install --upgrade pip
pip install \
    fastapi==0.115.6 \
    uvicorn[standard]==0.34.0 \
    faster-whisper==1.1.0 \
    scikit-learn==1.6.1 \
    numpy==2.2.2 \
    pandas==2.2.3 \
    joblib==1.4.2 \
    python-multipart==0.0.20 \
    pydub==0.25.1 \
    prometheus-client==0.21.1 \
    httpx==0.28.1 \
    aiofiles==24.1.0

# Optional: PyTorch for DistilBERT classifier (Option C)
# pip install torch==2.5.1 transformers==4.48.1 --index-url https://download.pytorch.org/whl/cpu

# Optional: GPU support for faster-whisper
# pip install nvidia-cublas-cu12 nvidia-cudnn-cu12

Whisper Model Download

Pre-download the model to avoid first-request latency:

# Download whisper models (choose based on your hardware)
python3 -c "
from faster_whisper import WhisperModel
# Use 'tiny' for fastest AMD (39M params, ~1s inference on CPU)
# Use 'base' for better accuracy (74M params, ~2s inference on CPU)
# Use 'small' for best accuracy (244M params, ~5s inference on CPU)
model = WhisperModel('base', device='cpu', compute_type='int8')
print('Model downloaded and ready')
"

For AMD, base or base.en is the sweet spot — fast enough for real-time use with good enough transcription quality for short phrases.

Asterisk AGI Requirements

On the Asterisk/ViciDial server:

# Ensure AGI directory exists
ls -la /var/lib/asterisk/agi-bin/

# Python for AGI scripts (system python is fine)
python3 --version  # Need 3.6+

# Install requests library for AGI scripts
pip3 install requests

# Ensure audio recording tools are available
which sox
which ffmpeg

Network Requirements

If the AMD service runs on a separate server from Asterisk:

# AMD service port (default 8190)
# Ensure firewall allows Asterisk server → AMD service
iptables -A INPUT -s YOUR_ASTERISK_IP -p tcp --dport 8190 -j ACCEPT

# Test connectivity from Asterisk server
curl -s http://YOUR_AMD_SERVICE_IP:8190/health

Directory Structure

mkdir -p /opt/amd-service/{models,data,logs,scripts}
mkdir -p /opt/amd-service/data/{raw,labeled,features,augmented}
mkdir -p /opt/amd-service/data/labeled/{human,machine}

/opt/amd-service/
├── venv/                    # Python virtual environment
├── models/
│   ├── amd_classifier.joblib    # Trained ML model
│   ├── amd_vectorizer.joblib    # TF-IDF vectorizer
│   └── amd_scaler.joblib        # Feature scaler
├── data/
│   ├── raw/                 # Raw recordings from ViciDial
│   ├── labeled/
│   │   ├── human/           # Confirmed human pickups (5s clips)
│   │   └── machine/         # Confirmed machine pickups (5s clips)
│   ├── features/            # Extracted feature CSVs
│   └── augmented/           # Augmented training samples
├── scripts/
│   ├── collect_training_data.py
│   ├── extract_features.py
│   ├── train_classifier.py
│   └── evaluate_model.py
├── service.py               # FastAPI AMD service
├── config.py                # Configuration
├── amd_agi.py              # Asterisk AGI script
└── logs/
    └── amd.log              # Service logs

5. Training Data Collection

The classifier is only as good as its training data. For AMD, you need real-world examples of how calls sound in your specific operation — the phone numbers you dial, the demographics of your contacts, the carriers you use, the codecs in your trunks. Generic datasets will not give you 95% accuracy on your calls.

Step 1: Identify Calls with Known Outcomes

ViciDial records every call and tracks the disposition set by agents. Use this to build ground truth labels:

-- Query to find calls suitable for training data
-- Run on your ViciDial database (or replica)

-- HUMAN calls: agent talked to a live person
SELECT
    vl.uniqueid,
    vl.phone_number,
    vl.call_date,
    vl.status,
    vl.length_in_sec,
    vl.user AS agent,
    vr.recording_id,
    vr.filename,
    vr.location AS recording_path
FROM vicidial_log vl
JOIN recording_log vr ON vr.vicidial_id = vl.uniqueid
WHERE vl.call_date >= '2026-01-01'
  AND vl.status IN ('SALE','CALLBK','NI','XFER','A','B','CB')  -- Statuses indicating human contact
  AND vl.length_in_sec >= 10                                     -- Long enough to have a real conversation
  AND vr.filename IS NOT NULL
  AND vr.length_in_sec > 3                                       -- Recording exists and has audio
ORDER BY RAND()
LIMIT 1000;

-- MACHINE calls: agent confirmed voicemail/answering machine
SELECT
    vl.uniqueid,
    vl.phone_number,
    vl.call_date,
    vl.status,
    vl.length_in_sec,
    vl.user AS agent,
    vr.recording_id,
    vr.filename,
    vr.location AS recording_path
FROM vicidial_log vl
JOIN recording_log vr ON vr.vicidial_id = vl.uniqueid
WHERE vl.call_date >= '2026-01-01'
  AND vl.status IN ('AA','AM','AL','ADC','AMVM')  -- Answering machine statuses
  AND vl.length_in_sec >= 5
  AND vr.filename IS NOT NULL
  AND vr.length_in_sec > 3
ORDER BY RAND()
LIMIT 1000;

Note: Your status codes may differ. AA = Answering Machine Auto, AM = Answering Machine, AL = Answering Machine Left Message. Check your vicidial_statuses and vicidial_campaign_statuses tables for your specific codes.

Step 2: Extract First 5 Seconds from Recordings

ViciDial recordings contain the entire call. For AMD training, you only need the first 5 seconds — the initial pickup audio.

#!/usr/bin/env python3
"""
collect_training_data.py
Extract first 5 seconds from ViciDial recordings for AMD training.
"""

import os
import sys
import subprocess
import csv
import mysql.connector
from pathlib import Path

# Configuration
DB_CONFIG = {
    'host': 'YOUR_SERVER_IP',
    'port': 3306,
    'user': 'report_cron',
    'password': 'YOUR_DB_PASSWORD',
    'database': 'asterisk',
    'connect_timeout': 10,
}

RECORDING_BASE_PATH = '/var/spool/asterisk/monitorDONE'  # Adjust for your server
OUTPUT_DIR = '/opt/amd-service/data/labeled'
CLIP_DURATION = 5  # seconds
SAMPLE_RATE = 16000  # Whisper expects 16kHz
MIN_SAMPLES_PER_CLASS = 500


def get_training_candidates(cursor, status_list, label, limit=1000):
    """Query ViciDial for calls with known outcomes."""
    statuses = ','.join(f"'{s}'" for s in status_list)
    query = f"""
        SELECT
            vl.uniqueid,
            vl.phone_number,
            vl.call_date,
            vl.status,
            vl.length_in_sec,
            vr.recording_id,
            vr.filename,
            vr.location
        FROM vicidial_log vl
        JOIN recording_log vr ON vr.vicidial_id = vl.uniqueid
        WHERE vl.call_date >= DATE_SUB(NOW(), INTERVAL 90 DAY)
          AND vl.status IN ({statuses})
          AND vl.length_in_sec >= 5
          AND vr.filename IS NOT NULL
          AND vr.length_in_sec > 3
        ORDER BY RAND()
        LIMIT {limit}
    """
    cursor.execute(query)
    results = cursor.fetchall()
    print(f"  Found {len(results)} {label} candidates")
    return results


def extract_clip(input_path, output_path, duration=5):
    """Extract first N seconds from a recording, convert to 16kHz mono WAV."""
    cmd = [
        'ffmpeg', '-y',
        '-i', input_path,
        '-t', str(duration),
        '-ar', str(SAMPLE_RATE),
        '-ac', '1',           # mono
        '-acodec', 'pcm_s16le',
        output_path
    ]
    try:
        result = subprocess.run(
            cmd, capture_output=True, timeout=30, text=True
        )
        if result.returncode != 0:
            return False
        # Verify output file exists and has content
        if os.path.exists(output_path) and os.path.getsize(output_path) > 1000:
            return True
        return False
    except (subprocess.TimeoutExpired, Exception) as e:
        print(f"    Error extracting {input_path}: {e}")
        return False


def find_recording_file(filename, location):
    """Locate the actual recording file on disk."""
    # ViciDial stores recordings in various locations
    search_paths = [
        location,  # Full path from recording_log
        os.path.join(RECORDING_BASE_PATH, filename),
        os.path.join(RECORDING_BASE_PATH, f"{filename}.wav"),
        os.path.join(RECORDING_BASE_PATH, f"{filename}.mp3"),
        os.path.join(RECORDING_BASE_PATH, f"{filename}.gsm"),
    ]

    for path in search_paths:
        if path and os.path.isfile(path):
            return path
    return None


def main():
    # Create output directories
    os.makedirs(f"{OUTPUT_DIR}/human", exist_ok=True)
    os.makedirs(f"{OUTPUT_DIR}/machine", exist_ok=True)

    # Connect to database
    print("Connecting to ViciDial database...")
    conn = mysql.connector.connect(**DB_CONFIG)
    cursor = conn.cursor(dictionary=True)

    # Define status codes for each class
    human_statuses = ['SALE', 'CALLBK', 'NI', 'XFER', 'A', 'B', 'CB', 'DNC', 'N', 'NP']
    machine_statuses = ['AA', 'AM', 'AL', 'ADC', 'AMVM', 'VM']

    manifest = []  # Track all extracted clips

    for label, statuses in [('human', human_statuses), ('machine', machine_statuses)]:
        print(f"\nCollecting {label} samples...")
        candidates = get_training_candidates(cursor, statuses, label, limit=1500)

        extracted = 0
        for row in candidates:
            if extracted >= MIN_SAMPLES_PER_CLASS * 2:  # Collect extra for validation
                break

            # Find the recording file
            recording_path = find_recording_file(
                row['filename'], row.get('location', '')
            )
            if not recording_path:
                continue

            # Extract clip
            output_filename = f"{label}_{row['uniqueid']}_{row['recording_id']}.wav"
            output_path = os.path.join(OUTPUT_DIR, label, output_filename)

            if extract_clip(recording_path, output_path, CLIP_DURATION):
                extracted += 1
                manifest.append({
                    'filename': output_filename,
                    'label': label,
                    'uniqueid': row['uniqueid'],
                    'phone_number': row['phone_number'],
                    'status': row['status'],
                    'original_file': recording_path,
                })
                if extracted % 50 == 0:
                    print(f"  Extracted {extracted} {label} clips...")

        print(f"  Total {label} clips extracted: {extracted}")

    # Save manifest
    manifest_path = os.path.join(OUTPUT_DIR, 'manifest.csv')
    with open(manifest_path, 'w', newline='') as f:
        writer = csv.DictWriter(f, fieldnames=manifest[0].keys())
        writer.writeheader()
        writer.writerows(manifest)

    print(f"\nManifest saved to {manifest_path}")
    print(f"Total clips: {len(manifest)}")

    cursor.close()
    conn.close()


if __name__ == '__main__':
    main()

Step 3: Manual Verification

Automated labeling from ViciDial statuses is not perfect. Agents sometimes set wrong dispositions. Verify a random sample:

# Quick spot-check: listen to random samples
cd /opt/amd-service/data/labeled

# Play 10 random "human" samples
ls human/ | shuf | head -10 | while read f; do
    echo "Playing: human/$f"
    play "human/$f" 2>/dev/null
    read -p "Correct? (y/n/q): " answer
    if [ "$answer" = "n" ]; then
        echo "  → Moving to machine/"
        mv "human/$f" "machine/$f"
    elif [ "$answer" = "q" ]; then
        break
    fi
done

# Play 10 random "machine" samples
ls machine/ | shuf | head -10 | while read f; do
    echo "Playing: machine/$f"
    play "machine/$f" 2>/dev/null
    read -p "Correct? (y/n/q): " answer
    if [ "$answer" = "n" ]; then
        echo "  → Moving to human/"
        mv "machine/$f" "human/$f"
    elif [ "$answer" = "q" ]; then
        break
    fi
done

Step 4: Data Augmentation

If you have fewer than 500 samples per class, augment the data with realistic variations:

#!/usr/bin/env python3
"""
augment_data.py
Augment AMD training data with realistic audio variations.
"""

import os
import random
import subprocess
from pathlib import Path

INPUT_DIR = '/opt/amd-service/data/labeled'
OUTPUT_DIR = '/opt/amd-service/data/augmented'
SAMPLE_RATE = 16000


def augment_speed(input_path, output_path, factor):
    """Change playback speed (simulates different speaking rates)."""
    cmd = [
        'sox', input_path, output_path,
        'speed', str(factor),
        'rate', str(SAMPLE_RATE)
    ]
    subprocess.run(cmd, capture_output=True, timeout=30)


def augment_noise(input_path, output_path, noise_level=0.005):
    """Add white noise (simulates line noise)."""
    # Generate noise file first, then mix
    duration_cmd = ['soxi', '-D', input_path]
    result = subprocess.run(duration_cmd, capture_output=True, text=True, timeout=10)
    duration = float(result.stdout.strip())

    noise_path = output_path + '.noise.wav'
    cmd_noise = [
        'sox', '-n', '-r', str(SAMPLE_RATE), '-c', '1', noise_path,
        'synth', str(duration), 'whitenoise',
        'vol', str(noise_level)
    ]
    subprocess.run(cmd_noise, capture_output=True, timeout=30)

    cmd_mix = [
        'sox', '-m', input_path, noise_path, output_path
    ]
    subprocess.run(cmd_mix, capture_output=True, timeout=30)

    # Cleanup noise file
    if os.path.exists(noise_path):
        os.remove(noise_path)


def augment_volume(input_path, output_path, db_change):
    """Change volume (simulates different phone volumes)."""
    cmd = [
        'sox', input_path, output_path,
        'vol', f'{db_change}dB'
    ]
    subprocess.run(cmd, capture_output=True, timeout=30)


def augment_telephone_filter(input_path, output_path):
    """Apply telephone bandpass filter (300-3400 Hz)."""
    cmd = [
        'sox', input_path, output_path,
        'sinc', '300-3400'
    ]
    subprocess.run(cmd, capture_output=True, timeout=30)


def main():
    for label in ['human', 'machine']:
        input_dir = os.path.join(INPUT_DIR, label)
        output_dir = os.path.join(OUTPUT_DIR, label)
        os.makedirs(output_dir, exist_ok=True)

        files = [f for f in os.listdir(input_dir) if f.endswith('.wav')]
        print(f"Augmenting {len(files)} {label} samples...")

        # Copy originals
        for f in files:
            src = os.path.join(input_dir, f)
            dst = os.path.join(output_dir, f)
            subprocess.run(['cp', src, dst], capture_output=True)

        # Generate augmented versions
        for f in files:
            src = os.path.join(input_dir, f)
            base = f.replace('.wav', '')

            # Speed variations (0.9x and 1.1x)
            augment_speed(src, os.path.join(output_dir, f'{base}_slow.wav'), 0.9)
            augment_speed(src, os.path.join(output_dir, f'{base}_fast.wav'), 1.1)

            # Noise addition
            augment_noise(src, os.path.join(output_dir, f'{base}_noisy.wav'), 0.003)

            # Volume variations
            augment_volume(src, os.path.join(output_dir, f'{base}_quiet.wav'), -6)
            augment_volume(src, os.path.join(output_dir, f'{base}_loud.wav'), 3)

            # Telephone filter (if original was higher quality)
            augment_telephone_filter(
                src, os.path.join(output_dir, f'{base}_phone.wav')
            )

        augmented_count = len(os.listdir(output_dir))
        print(f"  {label}: {len(files)} originals → {augmented_count} total samples")


if __name__ == '__main__':
    main()

Dataset Size Guidelines

6. Whisper Feature Extraction

With labeled audio clips ready, the next step is to run each through Whisper and extract features that the ML classifier will use for its decision.

Feature Categories

The classifier uses two types of features:

Text-based features (from Whisper transcription):

• The transcribed text itself (for TF-IDF/embeddings)
• Word count
• Character count
• Contains question mark (humans often answer with questions)
• Contains common machine phrases ("leave a message", "not available", etc.)
• Contains common human phrases ("hello", "yes", "who is this", etc.)
• Number of sentences
• Average word length

Audio-based features (from Whisper metadata + audio analysis):

• Whisper confidence score
• Detected language probability
• Speech duration vs total duration (speech ratio)
• Words per second (speaking rate)
• Number of speech segments
• Average segment duration
• Initial silence duration (time before first speech)
• Total silence duration

Feature Extraction Script

#!/usr/bin/env python3
"""
extract_features.py
Extract text and audio features from labeled AMD training clips using Whisper.
"""

import os
import sys
import json
import csv
import re
import time
import wave
import struct
import math
from pathlib import Path
from faster_whisper import WhisperModel

# Configuration
DATA_DIR = '/opt/amd-service/data/augmented'  # or 'labeled' if no augmentation
FEATURES_DIR = '/opt/amd-service/data/features'
WHISPER_MODEL = 'base'  # Use same model you'll deploy with
WHISPER_DEVICE = 'cpu'
WHISPER_COMPUTE = 'int8'

# Known phrases for feature engineering
MACHINE_PHRASES = [
    'leave a message', 'leave your message', 'after the tone',
    'after the beep', 'not available', 'unavailable',
    'cannot take your call', 'can\'t take your call',
    'please leave', 'reached the voicemail', 'voicemail',
    'mailbox', 'press 1', 'press 2', 'press one', 'press two',
    'office hours', 'business hours', 'currently closed',
    'your call is important', 'please hold',
    'record your message', 'leave your name',
    'we will get back', 'we\'ll get back',
    'at the tone', 'not in right now', 'not here right now',
    'this is the voicemail', 'reached the mailbox',
    'number you have dialed', 'number you have called',
    'is not available', 'person you are calling',
]

HUMAN_PHRASES = [
    'hello', 'hi', 'yes', 'yeah', 'hey',
    'who is this', 'who\'s this', 'who are you',
    'speaking', 'can I help', 'how can I help',
    'good morning', 'good afternoon', 'good evening',
    'what do you want', 'what is it',
]


def analyze_audio_properties(filepath):
    """Extract basic audio properties without Whisper."""
    try:
        with wave.open(filepath, 'r') as w:
            frames = w.getnframes()
            rate = w.getframerate()
            duration = frames / float(rate)
            channels = w.getnchannels()
            sampwidth = w.getsampwidth()

            # Read raw audio data for energy analysis
            w.rewind()
            raw_data = w.readframes(frames)

            if sampwidth == 2:
                fmt = f'<{frames * channels}h'
                samples = struct.unpack(fmt, raw_data)
            else:
                return {'duration': duration, 'error': 'unsupported sample width'}

            # Calculate RMS energy in windows
            window_size = int(rate * 0.025)  # 25ms windows
            hop_size = int(rate * 0.010)     # 10ms hop

            energies = []
            for i in range(0, len(samples) - window_size, hop_size):
                window = samples[i:i + window_size]
                rms = math.sqrt(sum(s * s for s in window) / len(window))
                energies.append(rms)

            if not energies:
                return {'duration': duration}

            # Silence threshold: 10% of mean energy
            mean_energy = sum(energies) / len(energies)
            silence_threshold = mean_energy * 0.15

            # Find initial silence (frames before first speech)
            initial_silence_frames = 0
            for e in energies:
                if e > silence_threshold:
                    break
                initial_silence_frames += 1
            initial_silence_sec = initial_silence_frames * 0.010

            # Count silence vs speech frames
            speech_frames = sum(1 for e in energies if e > silence_threshold)
            silence_frames = len(energies) - speech_frames
            speech_ratio = speech_frames / len(energies) if energies else 0

            return {
                'duration': duration,
                'initial_silence': round(initial_silence_sec, 3),
                'speech_ratio': round(speech_ratio, 3),
                'mean_energy': round(mean_energy, 1),
                'max_energy': round(max(energies), 1),
            }
    except Exception as e:
        return {'duration': 0, 'error': str(e)}


def extract_text_features(text):
    """Extract features from transcribed text."""
    text_lower = text.lower().strip()
    words = text_lower.split()

    # Machine phrase matching
    machine_phrase_count = sum(
        1 for phrase in MACHINE_PHRASES if phrase in text_lower
    )
    has_machine_phrase = int(machine_phrase_count > 0)

    # Human phrase matching
    human_phrase_count = sum(
        1 for phrase in HUMAN_PHRASES if phrase in text_lower
    )
    has_human_phrase = int(human_phrase_count > 0)

    # Text structure features
    word_count = len(words)
    char_count = len(text_lower)
    has_question = int('?' in text)
    sentence_count = max(1, len(re.split(r'[.!?]+', text_lower)))
    avg_word_length = (
        sum(len(w) for w in words) / len(words) if words else 0
    )

    # Specific pattern checks
    contains_phone_number = int(bool(re.search(r'\d{3,}', text_lower)))
    contains_name_intro = int(bool(re.search(
        r"(this is|you've reached|you have reached|my name is)", text_lower
    )))
    contains_instruction = int(bool(re.search(
        r'(press|leave|record|wait|hold|dial)', text_lower
    )))

    return {
        'text': text,
        'word_count': word_count,
        'char_count': char_count,
        'has_question': has_question,
        'sentence_count': sentence_count,
        'avg_word_length': round(avg_word_length, 2),
        'machine_phrase_count': machine_phrase_count,
        'has_machine_phrase': has_machine_phrase,
        'human_phrase_count': human_phrase_count,
        'has_human_phrase': has_human_phrase,
        'contains_phone_number': contains_phone_number,
        'contains_name_intro': contains_name_intro,
        'contains_instruction': contains_instruction,
    }


def extract_whisper_features(model, filepath):
    """Run Whisper and extract transcription + metadata features."""
    try:
        segments, info = model.transcribe(
            filepath,
            beam_size=3,
            best_of=3,
            language='en',  # Set to your primary language
            vad_filter=True,
            vad_parameters=dict(
                min_silence_duration_ms=200,
                speech_pad_ms=100,
            ),
        )

        # Collect all segments
        segment_list = []
        full_text = ''
        for seg in segments:
            segment_list.append({
                'start': seg.start,
                'end': seg.end,
                'text': seg.text.strip(),
                'avg_logprob': seg.avg_logprob,
                'no_speech_prob': seg.no_speech_prob,
            })
            full_text += seg.text

        full_text = full_text.strip()

        # Whisper metadata features
        num_segments = len(segment_list)

        if segment_list:
            avg_confidence = sum(
                math.exp(s['avg_logprob']) for s in segment_list
            ) / num_segments
            avg_no_speech = sum(
                s['no_speech_prob'] for s in segment_list
            ) / num_segments
            total_speech_duration = sum(
                s['end'] - s['start'] for s in segment_list
            )
            avg_segment_duration = total_speech_duration / num_segments
            first_speech_start = segment_list[0]['start']

            # Words per second
            word_count = len(full_text.split())
            wps = word_count / total_speech_duration if total_speech_duration > 0 else 0
        else:
            avg_confidence = 0
            avg_no_speech = 1.0
            total_speech_duration = 0
            avg_segment_duration = 0
            first_speech_start = 5.0  # No speech detected
            wps = 0

        return {
            'transcript': full_text,
            'num_segments': num_segments,
            'avg_confidence': round(avg_confidence, 4),
            'avg_no_speech_prob': round(avg_no_speech, 4),
            'total_speech_duration': round(total_speech_duration, 3),
            'avg_segment_duration': round(avg_segment_duration, 3),
            'first_speech_start': round(first_speech_start, 3),
            'words_per_second': round(wps, 2),
            'language_prob': round(info.language_probability, 4),
            'detected_language': info.language,
        }
    except Exception as e:
        return {
            'transcript': '',
            'error': str(e),
            'num_segments': 0,
            'avg_confidence': 0,
            'avg_no_speech_prob': 1.0,
            'total_speech_duration': 0,
            'avg_segment_duration': 0,
            'first_speech_start': 5.0,
            'words_per_second': 0,
            'language_prob': 0,
            'detected_language': 'unknown',
        }


def main():
    os.makedirs(FEATURES_DIR, exist_ok=True)

    # Load Whisper model
    print(f"Loading Whisper model '{WHISPER_MODEL}'...")
    model = WhisperModel(WHISPER_MODEL, device=WHISPER_DEVICE, compute_type=WHISPER_COMPUTE)
    print("Model loaded.")

    all_features = []

    for label in ['human', 'machine']:
        label_dir = os.path.join(DATA_DIR, label)
        if not os.path.isdir(label_dir):
            print(f"Warning: {label_dir} not found, skipping")
            continue

        files = sorted([
            f for f in os.listdir(label_dir)
            if f.endswith('.wav')
        ])
        print(f"\nProcessing {len(files)} {label} files...")

        for i, filename in enumerate(files):
            filepath = os.path.join(label_dir, filename)

            # Extract audio properties
            audio_props = analyze_audio_properties(filepath)

            # Extract Whisper features
            whisper_feats = extract_whisper_features(model, filepath)

            # Extract text features from transcript
            text_feats = extract_text_features(whisper_feats.get('transcript', ''))

            # Combine all features
            features = {
                'filename': filename,
                'label': label,
                'label_numeric': 0 if label == 'human' else 1,
                **audio_props,
                **whisper_feats,
                **text_feats,
            }
            all_features.append(features)

            if (i + 1) % 100 == 0:
                print(f"  Processed {i + 1}/{len(files)}...")

    # Save features to CSV
    if not all_features:
        print("No features extracted!")
        return

    output_path = os.path.join(FEATURES_DIR, 'amd_features.csv')
    fieldnames = all_features[0].keys()

    with open(output_path, 'w', newline='') as f:
        writer = csv.DictWriter(f, fieldnames=fieldnames)
        writer.writeheader()
        writer.writerows(all_features)

    print(f"\nFeatures saved to {output_path}")
    print(f"Total samples: {len(all_features)}")

    # Print class distribution
    human_count = sum(1 for f in all_features if f['label'] == 'human')
    machine_count = sum(1 for f in all_features if f['label'] == 'machine')
    print(f"  Human:   {human_count}")
    print(f"  Machine: {machine_count}")

    # Print sample transcripts
    print("\nSample HUMAN transcripts:")
    for f in all_features[:5]:
        if f['label'] == 'human':
            print(f"  \"{f.get('transcript', '')}\"")

    print("\nSample MACHINE transcripts:")
    for f in all_features:
        if f['label'] == 'machine':
            print(f"  \"{f.get('transcript', '')}\"")
            break


if __name__ == '__main__':
    main()

Expected Feature Distributions

After extraction, you should see patterns like these in your feature data:

These patterns are what the classifier learns to exploit. A human "Hello?" produces: 1 word, has question mark, no machine phrases, low speech ratio, high first-speech confidence. A voicemail greeting produces: 15+ words, no question marks, contains "leave a message", high speech ratio, continuous speech segments.

7. ML Classifier — Train the Model

With features extracted, train a classifier. This section covers three options in order of complexity and accuracy.

Option A: Text-Only Classifier (TF-IDF + Logistic Regression)

The simplest approach: just use the transcribed text. TF-IDF converts text to numerical features, and Logistic Regression draws the decision boundary. Fast to train, fast to predict, ~90% accuracy.

#!/usr/bin/env python3
"""
train_classifier_text.py
Option A: Text-only AMD classifier using TF-IDF + LogisticRegression.
Simple, fast, ~90% accuracy.
"""

import os
import pandas as pd
import numpy as np
from sklearn.model_selection import cross_val_score, StratifiedKFold, train_test_split
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.linear_model import LogisticRegression
from sklearn.metrics import (
    classification_report, confusion_matrix,
    precision_score, recall_score, f1_score
)
from sklearn.pipeline import Pipeline
import joblib

FEATURES_PATH = '/opt/amd-service/data/features/amd_features.csv'
MODEL_DIR = '/opt/amd-service/models'


def main():
    os.makedirs(MODEL_DIR, exist_ok=True)

    # Load features
    print("Loading feature data...")
    df = pd.read_csv(FEATURES_PATH)
    print(f"Total samples: {len(df)}")
    print(f"Class distribution:\n{df['label'].value_counts()}")

    # Prepare data
    X_text = df['transcript'].fillna('').values
    y = df['label_numeric'].values  # 0=human, 1=machine

    # Split train/test
    X_train, X_test, y_train, y_test = train_test_split(
        X_text, y, test_size=0.2, random_state=42, stratify=y
    )

    print(f"\nTrain: {len(X_train)} samples")
    print(f"Test:  {len(X_test)} samples")

    # Build pipeline
    pipeline = Pipeline([
        ('tfidf', TfidfVectorizer(
            max_features=5000,
            ngram_range=(1, 3),       # Unigrams, bigrams, trigrams
            min_df=2,                  # Must appear in at least 2 documents
            max_df=0.95,               # Ignore terms in >95% of documents
            sublinear_tf=True,         # Apply log normalization
            strip_accents='unicode',
            lowercase=True,
        )),
        ('clf', LogisticRegression(
            C=1.0,
            class_weight='balanced',   # Handle class imbalance
            max_iter=1000,
            random_state=42,
        )),
    ])

    # Cross-validation
    print("\nRunning 5-fold cross-validation...")
    cv = StratifiedKFold(n_splits=5, shuffle=True, random_state=42)
    cv_scores = cross_val_score(pipeline, X_train, y_train, cv=cv, scoring='f1')
    print(f"CV F1 scores: {cv_scores}")
    print(f"Mean F1: {cv_scores.mean():.4f} (+/- {cv_scores.std() * 2:.4f})")

    # Train final model
    print("\nTraining final model...")
    pipeline.fit(X_train, y_train)

    # Evaluate on test set
    y_pred = pipeline.predict(X_test)
    y_prob = pipeline.predict_proba(X_test)[:, 1]

    print("\n" + "=" * 60)
    print("TEST SET RESULTS")
    print("=" * 60)
    print(classification_report(
        y_test, y_pred, target_names=['HUMAN', 'MACHINE']
    ))

    cm = confusion_matrix(y_test, y_pred)
    print(f"Confusion Matrix:")
    print(f"                Predicted")
    print(f"                HUMAN  MACHINE")
    print(f"  Actual HUMAN   {cm[0][0]:5d}  {cm[0][1]:5d}")
    print(f"  Actual MACHINE {cm[1][0]:5d}  {cm[1][1]:5d}")

    # AMD-specific metrics
    false_positive_rate = cm[0][1] / (cm[0][0] + cm[0][1])  # Humans classified as machines
    false_negative_rate = cm[1][0] / (cm[1][0] + cm[1][1])  # Machines classified as humans
    print(f"\nFalse Positive Rate (humans hung up on): {false_positive_rate:.2%}")
    print(f"False Negative Rate (machines sent to agent): {false_negative_rate:.2%}")

    # Save model
    model_path = os.path.join(MODEL_DIR, 'amd_text_pipeline.joblib')
    joblib.dump(pipeline, model_path)
    print(f"\nModel saved to {model_path}")

    # Test with example phrases
    print("\n" + "=" * 60)
    print("EXAMPLE PREDICTIONS")
    print("=" * 60)
    test_phrases = [
        "Hello?",
        "Yes?",
        "Hi, who's this?",
        "Good morning, how can I help you?",
        "Hi you've reached John Smith. I'm not available right now. Please leave a message after the tone.",
        "The person you are calling is not available. Please leave a message.",
        "Thank you for calling. Our office hours are Monday through Friday.",
        "",  # Empty/silence
        "Yeah what do you want?",
        "The number you have dialed is not in service.",
    ]

    for phrase in test_phrases:
        pred = pipeline.predict([phrase])[0]
        prob = pipeline.predict_proba([phrase])[0]
        label = 'HUMAN' if pred == 0 else 'MACHINE'
        confidence = max(prob)
        print(f"  [{label} {confidence:.0%}] \"{phrase}\"")


if __name__ == '__main__':
    main()

Option B: Multi-Feature Classifier (Text + Audio Features)

Combines text features with audio features for better accuracy. Uses Gradient Boosting which handles mixed feature types well. This is the recommended approach — ~95% accuracy with fast inference.

#!/usr/bin/env python3
"""
train_classifier_multi.py
Option B: Multi-feature AMD classifier using text + audio features.
GradientBoosting, ~95% accuracy. RECOMMENDED for production.
"""

import os
import pandas as pd
import numpy as np
from sklearn.model_selection import (
    cross_val_score, StratifiedKFold, train_test_split, GridSearchCV
)
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.ensemble import GradientBoostingClassifier
from sklearn.preprocessing import StandardScaler
from sklearn.metrics import (
    classification_report, confusion_matrix, roc_auc_score
)
from scipy.sparse import hstack, csr_matrix
import joblib

FEATURES_PATH = '/opt/amd-service/data/features/amd_features.csv'
MODEL_DIR = '/opt/amd-service/models'

# Numeric feature columns to use
NUMERIC_FEATURES = [
    'word_count', 'char_count', 'has_question', 'sentence_count',
    'avg_word_length', 'machine_phrase_count', 'has_machine_phrase',
    'human_phrase_count', 'has_human_phrase', 'contains_phone_number',
    'contains_name_intro', 'contains_instruction',
    'num_segments', 'avg_confidence', 'avg_no_speech_prob',
    'total_speech_duration', 'avg_segment_duration', 'first_speech_start',
    'words_per_second', 'language_prob',
    'initial_silence', 'speech_ratio', 'mean_energy',
]


def main():
    os.makedirs(MODEL_DIR, exist_ok=True)

    # Load features
    print("Loading feature data...")
    df = pd.read_csv(FEATURES_PATH)

    # Handle missing values in numeric columns
    for col in NUMERIC_FEATURES:
        if col in df.columns:
            df[col] = pd.to_numeric(df[col], errors='coerce').fillna(0)
        else:
            print(f"  Warning: column '{col}' not found, setting to 0")
            df[col] = 0

    df['transcript'] = df['transcript'].fillna('')

    print(f"Total samples: {len(df)}")
    print(f"Class distribution:\n{df['label'].value_counts()}")

    y = df['label_numeric'].values

    # Split train/test BEFORE fitting any transformers
    train_idx, test_idx = train_test_split(
        np.arange(len(df)), test_size=0.2, random_state=42, stratify=y
    )

    df_train = df.iloc[train_idx]
    df_test = df.iloc[test_idx]
    y_train = y[train_idx]
    y_test = y[test_idx]

    print(f"\nTrain: {len(df_train)} samples")
    print(f"Test:  {len(df_test)} samples")

    # --- Text features (TF-IDF) ---
    print("\nFitting TF-IDF on transcripts...")
    tfidf = TfidfVectorizer(
        max_features=3000,
        ngram_range=(1, 2),
        min_df=2,
        max_df=0.95,
        sublinear_tf=True,
        strip_accents='unicode',
        lowercase=True,
    )
    X_train_text = tfidf.fit_transform(df_train['transcript'])
    X_test_text = tfidf.transform(df_test['transcript'])

    # --- Numeric features (scaled) ---
    print("Scaling numeric features...")
    available_features = [c for c in NUMERIC_FEATURES if c in df.columns]
    scaler = StandardScaler()
    X_train_numeric = scaler.fit_transform(df_train[available_features].values)
    X_test_numeric = scaler.transform(df_test[available_features].values)

    # Convert to sparse and combine
    X_train_numeric_sparse = csr_matrix(X_train_numeric)
    X_test_numeric_sparse = csr_matrix(X_test_numeric)

    X_train_combined = hstack([X_train_text, X_train_numeric_sparse])
    X_test_combined = hstack([X_test_text, X_test_numeric_sparse])

    print(f"Combined feature matrix: {X_train_combined.shape[1]} features")
    print(f"  TF-IDF features: {X_train_text.shape[1]}")
    print(f"  Numeric features: {len(available_features)}")

    # --- Train Gradient Boosting ---
    print("\nTraining GradientBoosting classifier...")

    # Hyperparameter search
    param_grid = {
        'n_estimators': [200, 300],
        'max_depth': [4, 6],
        'learning_rate': [0.05, 0.1],
        'min_samples_leaf': [5, 10],
        'subsample': [0.8, 1.0],
    }

    gb = GradientBoostingClassifier(random_state=42)

    cv = StratifiedKFold(n_splits=5, shuffle=True, random_state=42)

    grid_search = GridSearchCV(
        gb, param_grid, cv=cv, scoring='f1',
        n_jobs=-1, verbose=1, refit=True
    )
    grid_search.fit(X_train_combined, y_train)

    print(f"\nBest parameters: {grid_search.best_params_}")
    print(f"Best CV F1: {grid_search.best_score_:.4f}")

    best_model = grid_search.best_estimator_

    # --- Evaluate on test set ---
    y_pred = best_model.predict(X_test_combined)
    y_prob = best_model.predict_proba(X_test_combined)[:, 1]

    print("\n" + "=" * 60)
    print("TEST SET RESULTS (Multi-Feature Gradient Boosting)")
    print("=" * 60)
    print(classification_report(
        y_test, y_pred, target_names=['HUMAN', 'MACHINE']
    ))

    auc = roc_auc_score(y_test, y_prob)
    print(f"ROC AUC: {auc:.4f}")

    cm = confusion_matrix(y_test, y_pred)
    print(f"\nConfusion Matrix:")
    print(f"                Predicted")
    print(f"                HUMAN  MACHINE")
    print(f"  Actual HUMAN   {cm[0][0]:5d}  {cm[0][1]:5d}")
    print(f"  Actual MACHINE {cm[1][0]:5d}  {cm[1][1]:5d}")

    false_positive_rate = cm[0][1] / (cm[0][0] + cm[0][1])
    false_negative_rate = cm[1][0] / (cm[1][0] + cm[1][1])
    print(f"\nFalse Positive Rate (humans hung up on): {false_positive_rate:.2%}")
    print(f"False Negative Rate (machines sent to agent): {false_negative_rate:.2%}")

    # --- Feature Importance ---
    print("\n" + "=" * 60)
    print("TOP 20 MOST IMPORTANT FEATURES")
    print("=" * 60)

    # Get feature names
    tfidf_names = [f"tfidf_{n}" for n in tfidf.get_feature_names_out()]
    all_feature_names = tfidf_names + available_features

    importances = best_model.feature_importances_
    sorted_idx = np.argsort(importances)[::-1]

    for i in range(min(20, len(sorted_idx))):
        idx = sorted_idx[i]
        name = all_feature_names[idx] if idx < len(all_feature_names) else f"feature_{idx}"
        print(f"  {i+1:2d}. {name:40s} {importances[idx]:.4f}")

    # --- Save artifacts ---
    print("\nSaving model artifacts...")
    joblib.dump(best_model, os.path.join(MODEL_DIR, 'amd_classifier.joblib'))
    joblib.dump(tfidf, os.path.join(MODEL_DIR, 'amd_vectorizer.joblib'))
    joblib.dump(scaler, os.path.join(MODEL_DIR, 'amd_scaler.joblib'))

    # Save feature list for inference
    import json
    meta = {
        'numeric_features': available_features,
        'model_type': 'GradientBoosting',
        'best_params': grid_search.best_params_,
        'test_f1': float(f1_score(y_test, y_pred)),
        'test_auc': float(auc),
        'false_positive_rate': float(false_positive_rate),
        'false_negative_rate': float(false_negative_rate),
        'train_samples': len(df_train),
        'test_samples': len(df_test),
    }
    with open(os.path.join(MODEL_DIR, 'amd_model_meta.json'), 'w') as f:
        json.dump(meta, f, indent=2)

    print(f"Model saved to {MODEL_DIR}/amd_classifier.joblib")
    print(f"Vectorizer saved to {MODEL_DIR}/amd_vectorizer.joblib")
    print(f"Scaler saved to {MODEL_DIR}/amd_scaler.joblib")
    print(f"Metadata saved to {MODEL_DIR}/amd_model_meta.json")
    print("\nDone!")


if __name__ == '__main__':
    # Import f1_score for metadata
    from sklearn.metrics import f1_score
    main()

Option C: Fine-Tuned DistilBERT (Best Accuracy, Slower)

For maximum accuracy (~97%), fine-tune a small transformer model on the transcribed text. This requires PyTorch and more compute but produces the most robust classifier.

#!/usr/bin/env python3
"""
train_classifier_bert.py
Option C: Fine-tuned DistilBERT for AMD classification.
~97% accuracy, requires PyTorch. Slower inference (~50ms vs ~5ms).
"""

import os
import pandas as pd
import numpy as np
import torch
from torch.utils.data import Dataset, DataLoader
from transformers import (
    DistilBertTokenizer, DistilBertForSequenceClassification,
    AdamW, get_linear_schedule_with_warmup
)
from sklearn.model_selection import train_test_split
from sklearn.metrics import classification_report, confusion_matrix
import json

FEATURES_PATH = '/opt/amd-service/data/features/amd_features.csv'
MODEL_DIR = '/opt/amd-service/models/bert_amd'
EPOCHS = 5
BATCH_SIZE = 16
MAX_LENGTH = 64  # AMD transcripts are short
LEARNING_RATE = 2e-5
DEVICE = 'cuda' if torch.cuda.is_available() else 'cpu'


class AMDDataset(Dataset):
    """Dataset for AMD text classification."""

    def __init__(self, texts, labels, tokenizer, max_length):
        self.texts = texts
        self.labels = labels
        self.tokenizer = tokenizer
        self.max_length = max_length

    def __len__(self):
        return len(self.texts)

    def __getitem__(self, idx):
        text = str(self.texts[idx])
        label = self.labels[idx]

        encoding = self.tokenizer(
            text,
            max_length=self.max_length,
            padding='max_length',
            truncation=True,
            return_tensors='pt'
        )

        return {
            'input_ids': encoding['input_ids'].squeeze(),
            'attention_mask': encoding['attention_mask'].squeeze(),
            'label': torch.tensor(label, dtype=torch.long)
        }


def train_epoch(model, dataloader, optimizer, scheduler, device):
    """Train for one epoch."""
    model.train()
    total_loss = 0
    correct = 0
    total = 0

    for batch in dataloader:
        input_ids = batch['input_ids'].to(device)
        attention_mask = batch['attention_mask'].to(device)
        labels = batch['label'].to(device)

        outputs = model(
            input_ids=input_ids,
            attention_mask=attention_mask,
            labels=labels
        )

        loss = outputs.loss
        total_loss += loss.item()

        preds = torch.argmax(outputs.logits, dim=1)
        correct += (preds == labels).sum().item()
        total += len(labels)

        loss.backward()
        torch.nn.utils.clip_grad_norm_(model.parameters(), 1.0)
        optimizer.step()
        scheduler.step()
        optimizer.zero_grad()

    return total_loss / len(dataloader), correct / total


def evaluate(model, dataloader, device):
    """Evaluate model on a dataset."""
    model.eval()
    all_preds = []
    all_labels = []
    all_probs = []
    total_loss = 0

    with torch.no_grad():
        for batch in dataloader:
            input_ids = batch['input_ids'].to(device)
            attention_mask = batch['attention_mask'].to(device)
            labels = batch['label'].to(device)

            outputs = model(
                input_ids=input_ids,
                attention_mask=attention_mask,
                labels=labels
            )

            total_loss += outputs.loss.item()
            probs = torch.softmax(outputs.logits, dim=1)
            preds = torch.argmax(probs, dim=1)

            all_preds.extend(preds.cpu().numpy())
            all_labels.extend(labels.cpu().numpy())
            all_probs.extend(probs[:, 1].cpu().numpy())

    return (
        np.array(all_preds),
        np.array(all_labels),
        np.array(all_probs),
        total_loss / len(dataloader)
    )


def main():
    os.makedirs(MODEL_DIR, exist_ok=True)

    print(f"Using device: {DEVICE}")

    # Load data
    df = pd.read_csv(FEATURES_PATH)
    df['transcript'] = df['transcript'].fillna('')

    texts = df['transcript'].values
    labels = df['label_numeric'].values

    # Split
    X_train, X_test, y_train, y_test = train_test_split(
        texts, labels, test_size=0.2, random_state=42, stratify=labels
    )

    X_train, X_val, y_train, y_val = train_test_split(
        X_train, y_train, test_size=0.1, random_state=42, stratify=y_train
    )

    print(f"Train: {len(X_train)}, Val: {len(X_val)}, Test: {len(X_test)}")

    # Tokenizer
    tokenizer = DistilBertTokenizer.from_pretrained('distilbert-base-uncased')

    # Datasets
    train_dataset = AMDDataset(X_train, y_train, tokenizer, MAX_LENGTH)
    val_dataset = AMDDataset(X_val, y_val, tokenizer, MAX_LENGTH)
    test_dataset = AMDDataset(X_test, y_test, tokenizer, MAX_LENGTH)

    train_loader = DataLoader(train_dataset, batch_size=BATCH_SIZE, shuffle=True)
    val_loader = DataLoader(val_dataset, batch_size=BATCH_SIZE)
    test_loader = DataLoader(test_dataset, batch_size=BATCH_SIZE)

    # Model
    model = DistilBertForSequenceClassification.from_pretrained(
        'distilbert-base-uncased',
        num_labels=2,
        problem_type='single_label_classification'
    )
    model.to(DEVICE)

    # Optimizer
    optimizer = AdamW(model.parameters(), lr=LEARNING_RATE, weight_decay=0.01)
    total_steps = len(train_loader) * EPOCHS
    scheduler = get_linear_schedule_with_warmup(
        optimizer, num_warmup_steps=total_steps // 10, num_training_steps=total_steps
    )

    # Training loop
    best_val_f1 = 0

    for epoch in range(EPOCHS):
        train_loss, train_acc = train_epoch(
            model, train_loader, optimizer, scheduler, DEVICE
        )

        val_preds, val_labels, val_probs, val_loss = evaluate(
            model, val_loader, DEVICE
        )
        val_acc = (val_preds == val_labels).mean()

        from sklearn.metrics import f1_score
        val_f1 = f1_score(val_labels, val_preds)

        print(f"Epoch {epoch+1}/{EPOCHS}")
        print(f"  Train Loss: {train_loss:.4f}, Train Acc: {train_acc:.4f}")
        print(f"  Val Loss:   {val_loss:.4f}, Val Acc:   {val_acc:.4f}, Val F1: {val_f1:.4f}")

        if val_f1 > best_val_f1:
            best_val_f1 = val_f1
            model.save_pretrained(MODEL_DIR)
            tokenizer.save_pretrained(MODEL_DIR)
            print(f"  Saved best model (F1: {val_f1:.4f})")

    # Final evaluation on test set
    print("\n" + "=" * 60)
    print("TEST SET RESULTS (Fine-tuned DistilBERT)")
    print("=" * 60)

    # Load best model
    model = DistilBertForSequenceClassification.from_pretrained(MODEL_DIR)
    model.to(DEVICE)

    test_preds, test_labels, test_probs, test_loss = evaluate(
        model, test_loader, DEVICE
    )

    print(classification_report(
        test_labels, test_preds, target_names=['HUMAN', 'MACHINE']
    ))

    cm = confusion_matrix(test_labels, test_preds)
    print(f"Confusion Matrix:")
    print(f"                Predicted")
    print(f"                HUMAN  MACHINE")
    print(f"  Actual HUMAN   {cm[0][0]:5d}  {cm[0][1]:5d}")
    print(f"  Actual MACHINE {cm[1][0]:5d}  {cm[1][1]:5d}")

    false_positive_rate = cm[0][1] / (cm[0][0] + cm[0][1])
    print(f"\nFalse Positive Rate: {false_positive_rate:.2%}")

    # Save metadata
    meta = {
        'model_type': 'DistilBERT',
        'epochs': EPOCHS,
        'best_val_f1': float(best_val_f1),
        'test_accuracy': float((test_preds == test_labels).mean()),
        'false_positive_rate': float(false_positive_rate),
    }
    with open(os.path.join(MODEL_DIR, 'training_meta.json'), 'w') as f:
        json.dump(meta, f, indent=2)

    print(f"\nModel saved to {MODEL_DIR}/")


if __name__ == '__main__':
    main()

Choosing the Right Option

Recommendation: Start with Option B. It gives the best accuracy-to-complexity ratio. The audio features compensate for Whisper transcription errors, and Gradient Boosting handles the mixed feature types naturally. Option A is useful for quick prototyping, and Option C is overkill for most AMD use cases.

Why Precision Matters More Than Recall

In AMD, not all errors are equal:

• False Positive (human classified as MACHINE) = You hang up on or play a voicemail message to a live person. This is a catastrophic error — the prospect is annoyed, may file a complaint, and you have lost a potential sale. Cost: $5-50 per incident.

• False Negative (machine classified as HUMAN) = A voicemail gets routed to an agent. The agent hears "Hi, you've reached..." and dispositions it as AM. This wastes 15-30 seconds. Cost: $0.25-0.50 per incident.

Therefore, you should optimize for high precision on the MACHINE class (when you predict MACHINE, be very sure) even if it means lower recall (some machines slip through to agents). In practice, this means setting a confidence threshold above 0.5:

# Instead of:
prediction = model.predict(features)  # Threshold at 0.5

# Use a higher threshold for MACHINE classification:
probabilities = model.predict_proba(features)[0]
machine_prob = probabilities[1]

MACHINE_THRESHOLD = 0.75  # Only classify as MACHINE if 75%+ confident

if machine_prob >= MACHINE_THRESHOLD:
    result = 'MACHINE'
elif machine_prob <= (1 - MACHINE_THRESHOLD):
    result = 'HUMAN'
else:
    result = 'NOTSURE'  # Route to agent (safe default)

This three-way classification (HUMAN / MACHINE / NOTSURE) is critical for production use. When the model is uncertain, default to routing to an agent — the safe choice.

Model Evaluation Checklist

Before deploying any model, verify:

[ ] False positive rate < 3% (humans classified as machines)
[ ] Overall accuracy > 90%
[ ] Model tested on held-out data (not training data)
[ ] Tested on your actual call center audio (not generic data)
[ ] Tested across different times of day (morning vs evening)
[ ] Tested across different campaigns/phone lists
[ ] Inference time < 100ms per prediction
[ ] Model file size < 500MB
[ ] Confidence threshold tuned on validation set

8. AMD Service — FastAPI Microservice

Now we wrap everything into a production-ready HTTP service that Asterisk can call in real-time. The service loads the Whisper model and ML classifier once at startup, then processes audio files submitted via HTTP POST and returns classification results in under 2 seconds.

Project Structure

/opt/amd-service/
├── main.py              # FastAPI application
├── config.py            # Environment-based configuration
├── classifier.py        # ML classifier (from Section 7)
├── requirements.txt     # Python dependencies
├── models/
│   ├── amd_classifier.pkl    # Trained scikit-learn model
│   └── feature_scaler.pkl    # Feature scaler
├── logs/
│   └── amd.log          # Application logs
└── systemd/
    └── amd-service.service   # Systemd unit file

Configuration Module

Create /opt/amd-service/config.py:

"""
AMD Service Configuration
All settings via environment variables with sensible defaults.
"""

import os
from dataclasses import dataclass


@dataclass
class Config:
    """Service configuration loaded from environment variables."""

    # Server settings
    HOST: str = os.getenv("AMD_HOST", "0.0.0.0")
    PORT: int = int(os.getenv("AMD_PORT", "8090"))
    WORKERS: int = int(os.getenv("AMD_WORKERS", "1"))

    # Whisper settings
    WHISPER_MODEL: str = os.getenv("AMD_WHISPER_MODEL", "tiny")
    WHISPER_DEVICE: str = os.getenv("AMD_WHISPER_DEVICE", "cpu")
    WHISPER_COMPUTE_TYPE: str = os.getenv("AMD_WHISPER_COMPUTE_TYPE", "int8")
    WHISPER_LANGUAGE: str = os.getenv("AMD_WHISPER_LANGUAGE", "en")

    # Classifier settings
    MODEL_PATH: str = os.getenv("AMD_MODEL_PATH", "/opt/amd-service/models/amd_classifier.pkl")
    SCALER_PATH: str = os.getenv("AMD_SCALER_PATH", "/opt/amd-service/models/feature_scaler.pkl")

    # Classification thresholds
    MACHINE_THRESHOLD: float = float(os.getenv("AMD_MACHINE_THRESHOLD", "0.80"))
    HUMAN_THRESHOLD: float = float(os.getenv("AMD_HUMAN_THRESHOLD", "0.80"))

    # Audio processing
    MAX_AUDIO_DURATION: float = float(os.getenv("AMD_MAX_AUDIO_DURATION", "5.0"))
    SAMPLE_RATE: int = int(os.getenv("AMD_SAMPLE_RATE", "8000"))

    # Thread pool for Whisper inference
    MAX_CONCURRENT: int = int(os.getenv("AMD_MAX_CONCURRENT", "4"))

    # Logging
    LOG_LEVEL: str = os.getenv("AMD_LOG_LEVEL", "INFO")
    LOG_FILE: str = os.getenv("AMD_LOG_FILE", "/opt/amd-service/logs/amd.log")

    # Prometheus metrics
    METRICS_ENABLED: bool = os.getenv("AMD_METRICS_ENABLED", "true").lower() == "true"


config = Config()

FastAPI Application

Create /opt/amd-service/main.py:

#!/usr/bin/env python3
"""
AI-Powered Answering Machine Detection Service

Receives audio files via HTTP POST, transcribes with Whisper,
classifies using a trained ML model, and returns HUMAN/MACHINE/NOTSURE.

Usage:
    uvicorn main:app --host 0.0.0.0 --port 8090
    # or
    python main.py
"""

import io
import os
import sys
import time
import wave
import logging
import tempfile
import asyncio
import pickle
from pathlib import Path
from typing import Optional
from concurrent.futures import ThreadPoolExecutor
from contextlib import asynccontextmanager

import numpy as np
import uvicorn
from fastapi import FastAPI, File, UploadFile, HTTPException, Query
from fastapi.responses import JSONResponse
from pydantic import BaseModel

from config import config

# ---------------------------------------------------------------------------
# Logging setup
# ---------------------------------------------------------------------------
os.makedirs(os.path.dirname(config.LOG_FILE), exist_ok=True)

logging.basicConfig(
    level=getattr(logging, config.LOG_LEVEL),
    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
    handlers=[
        logging.FileHandler(config.LOG_FILE),
        logging.StreamHandler(sys.stdout),
    ],
)
logger = logging.getLogger("amd-service")

# ---------------------------------------------------------------------------
# Prometheus metrics (optional)
# ---------------------------------------------------------------------------
try:
    from prometheus_client import Counter, Histogram, Gauge, generate_latest, CONTENT_TYPE_LATEST

    METRICS_AVAILABLE = True
    REQUEST_COUNT = Counter(
        "amd_requests_total",
        "Total AMD classification requests",
        ["result"],
    )
    REQUEST_LATENCY = Histogram(
        "amd_latency_seconds",
        "AMD classification latency in seconds",
        buckets=[0.1, 0.25, 0.5, 0.75, 1.0, 1.5, 2.0, 2.5, 3.0, 5.0],
    )
    CONFIDENCE_HISTOGRAM = Histogram(
        "amd_confidence",
        "Classification confidence scores",
        buckets=[0.5, 0.6, 0.7, 0.8, 0.85, 0.9, 0.95, 0.99],
    )
    WHISPER_LATENCY = Histogram(
        "amd_whisper_latency_seconds",
        "Whisper transcription latency",
        buckets=[0.1, 0.25, 0.5, 0.75, 1.0, 1.5, 2.0, 3.0],
    )
    MODEL_LOADED = Gauge("amd_model_loaded", "Whether the ML model is loaded")
    ACTIVE_REQUESTS = Gauge("amd_active_requests", "Currently processing requests")
except ImportError:
    METRICS_AVAILABLE = False
    logger.warning("prometheus_client not installed — metrics disabled")


# ---------------------------------------------------------------------------
# Global model holders
# ---------------------------------------------------------------------------
whisper_model = None
classifier_model = None
feature_scaler = None
thread_pool: Optional[ThreadPoolExecutor] = None


def load_whisper_model():
    """Load Whisper model into memory."""
    global whisper_model
    from faster_whisper import WhisperModel

    logger.info(f"Loading Whisper model '{config.WHISPER_MODEL}' on {config.WHISPER_DEVICE}...")
    start = time.time()
    whisper_model = WhisperModel(
        config.WHISPER_MODEL,
        device=config.WHISPER_DEVICE,
        compute_type=config.WHISPER_COMPUTE_TYPE,
    )
    elapsed = time.time() - start
    logger.info(f"Whisper model loaded in {elapsed:.1f}s")


def load_classifier():
    """Load trained ML classifier and feature scaler."""
    global classifier_model, feature_scaler

    model_path = Path(config.MODEL_PATH)
    scaler_path = Path(config.SCALER_PATH)

    if not model_path.exists():
        logger.error(f"Classifier model not found: {model_path}")
        raise FileNotFoundError(f"Classifier model not found: {model_path}")

    with open(model_path, "rb") as f:
        classifier_model = pickle.load(f)
    logger.info(f"Classifier loaded from {model_path}")

    if scaler_path.exists():
        with open(scaler_path, "rb") as f:
            feature_scaler = pickle.load(f)
        logger.info(f"Feature scaler loaded from {scaler_path}")
    else:
        logger.warning("No feature scaler found — using raw features")

    if METRICS_AVAILABLE:
        MODEL_LOADED.set(1)


# ---------------------------------------------------------------------------
# Audio processing helpers
# ---------------------------------------------------------------------------
def read_audio_file(file_bytes: bytes) -> np.ndarray:
    """
    Read audio bytes into a numpy array.
    Supports WAV (native) and other formats via ffmpeg fallback.
    Returns mono float32 audio at the configured sample rate.
    """
    try:
        # Try WAV first (most common from Asterisk)
        with wave.open(io.BytesIO(file_bytes), "rb") as wf:
            n_channels = wf.getnchannels()
            sample_width = wf.getsampwidth()
            framerate = wf.getframerate()
            n_frames = wf.getnframes()
            raw = wf.readframes(n_frames)

        # Convert to numpy
        if sample_width == 2:
            audio = np.frombuffer(raw, dtype=np.int16).astype(np.float32) / 32768.0
        elif sample_width == 1:
            audio = (np.frombuffer(raw, dtype=np.uint8).astype(np.float32) - 128) / 128.0
        else:
            raise ValueError(f"Unsupported sample width: {sample_width}")

        # Convert stereo to mono
        if n_channels == 2:
            audio = audio.reshape(-1, 2).mean(axis=1)

        # Resample if needed (simple decimation for 16kHz -> 8kHz etc.)
        if framerate != config.SAMPLE_RATE:
            import librosa
            audio = librosa.resample(audio, orig_sr=framerate, target_sr=config.SAMPLE_RATE)

        return audio

    except wave.Error:
        # Fallback: write to temp file, convert with ffmpeg
        with tempfile.NamedTemporaryFile(suffix=".audio", delete=False) as tmp_in:
            tmp_in.write(file_bytes)
            tmp_in_path = tmp_in.name

        tmp_out_path = tmp_in_path + ".wav"
        try:
            import subprocess
            subprocess.run(
                [
                    "ffmpeg", "-y", "-i", tmp_in_path,
                    "-ar", str(config.SAMPLE_RATE),
                    "-ac", "1", "-f", "wav",
                    tmp_out_path,
                ],
                capture_output=True,
                timeout=10,
            )
            with wave.open(tmp_out_path, "rb") as wf:
                raw = wf.readframes(wf.getnframes())
                audio = np.frombuffer(raw, dtype=np.int16).astype(np.float32) / 32768.0
            return audio
        finally:
            for p in (tmp_in_path, tmp_out_path):
                if os.path.exists(p):
                    os.unlink(p)


def extract_features(audio: np.ndarray, transcript: str) -> np.ndarray:
    """
    Extract features from audio and transcript for classification.
    Must match the feature set used during training (Section 6-7).
    """
    sr = config.SAMPLE_RATE
    duration = len(audio) / sr

    # --- Audio features ---
    # Energy
    rms = np.sqrt(np.mean(audio ** 2))

    # Zero crossing rate
    zcr = np.sum(np.abs(np.diff(np.sign(audio)))) / (2 * len(audio))

    # Silence ratio (frames below threshold)
    silence_threshold = 0.01
    silence_ratio = np.sum(np.abs(audio) < silence_threshold) / len(audio)

    # Peak amplitude
    peak = np.max(np.abs(audio))

    # Spectral centroid (simple FFT-based)
    fft_vals = np.abs(np.fft.rfft(audio))
    freqs = np.fft.rfftfreq(len(audio), d=1.0 / sr)
    spectral_centroid = np.sum(freqs * fft_vals) / (np.sum(fft_vals) + 1e-10)

    # --- Transcript features ---
    transcript_lower = transcript.lower().strip()
    word_count = len(transcript_lower.split()) if transcript_lower else 0
    char_count = len(transcript_lower)

    # Machine indicator phrases
    machine_phrases = [
        "leave a message", "after the tone", "after the beep",
        "not available", "cannot take your call", "voicemail",
        "press", "please hold", "office hours", "mailbox",
        "record your message", "at the tone", "currently unavailable",
        "reached the voicemail", "sorry we missed", "get back to you",
    ]
    machine_phrase_count = sum(1 for phrase in machine_phrases if phrase in transcript_lower)
    has_machine_phrase = 1.0 if machine_phrase_count > 0 else 0.0

    # Human indicator patterns
    human_patterns = ["hello", "hi", "hey", "yeah", "yes", "what", "who"]
    has_human_pattern = 1.0 if any(
        transcript_lower.startswith(p) or transcript_lower == p
        for p in human_patterns
    ) else 0.0

    # Greeting length (machines tend to be longer)
    is_short_greeting = 1.0 if word_count <= 3 else 0.0

    # Words per second (machines speak at consistent pace)
    words_per_second = word_count / max(duration, 0.1)

    features = np.array([
        duration,
        rms,
        zcr,
        silence_ratio,
        peak,
        spectral_centroid,
        word_count,
        char_count,
        machine_phrase_count,
        has_machine_phrase,
        has_human_pattern,
        is_short_greeting,
        words_per_second,
    ])

    return features


def classify_audio(file_bytes: bytes) -> dict:
    """
    Full AMD pipeline: read audio -> transcribe -> extract features -> classify.
    Runs synchronously (called from thread pool).
    """
    start_time = time.time()

    # 1. Read and preprocess audio
    audio = read_audio_file(file_bytes)

    # Truncate to max duration
    max_samples = int(config.MAX_AUDIO_DURATION * config.SAMPLE_RATE)
    if len(audio) > max_samples:
        audio = audio[:max_samples]

    duration = len(audio) / config.SAMPLE_RATE

    # 2. Transcribe with Whisper
    whisper_start = time.time()

    # Whisper expects float32 audio at 16kHz
    if config.SAMPLE_RATE != 16000:
        import librosa
        audio_16k = librosa.resample(audio, orig_sr=config.SAMPLE_RATE, target_sr=16000)
    else:
        audio_16k = audio

    segments, info = whisper_model.transcribe(
        audio_16k,
        language=config.WHISPER_LANGUAGE,
        beam_size=1,          # Greedy decoding for speed
        best_of=1,
        vad_filter=False,     # Short audio, no need for VAD
        without_timestamps=True,
    )
    transcript = " ".join(seg.text.strip() for seg in segments).strip()
    whisper_elapsed = time.time() - whisper_start

    if METRICS_AVAILABLE:
        WHISPER_LATENCY.observe(whisper_elapsed)

    logger.debug(f"Whisper transcription ({whisper_elapsed:.3f}s): '{transcript}'")

    # 3. Extract features
    features = extract_features(audio, transcript)

    # 4. Scale features if scaler is available
    if feature_scaler is not None:
        features_scaled = feature_scaler.transform(features.reshape(1, -1))
    else:
        features_scaled = features.reshape(1, -1)

    # 5. Classify
    probabilities = classifier_model.predict_proba(features_scaled)[0]
    # Assuming class order: [HUMAN, MACHINE]
    human_prob = probabilities[0]
    machine_prob = probabilities[1]

    if machine_prob >= config.MACHINE_THRESHOLD:
        result = "MACHINE"
        confidence = float(machine_prob)
    elif human_prob >= config.HUMAN_THRESHOLD:
        result = "HUMAN"
        confidence = float(human_prob)
    else:
        result = "NOTSURE"
        confidence = float(max(human_prob, machine_prob))

    processing_time_ms = int((time.time() - start_time) * 1000)

    if METRICS_AVAILABLE:
        REQUEST_COUNT.labels(result=result).inc()
        REQUEST_LATENCY.observe(processing_time_ms / 1000)
        CONFIDENCE_HISTOGRAM.observe(confidence)

    response = {
        "result": result,
        "confidence": round(confidence, 4),
        "transcript": transcript,
        "processing_time_ms": processing_time_ms,
        "audio_duration_s": round(duration, 2),
        "whisper_time_ms": int(whisper_elapsed * 1000),
        "probabilities": {
            "human": round(float(human_prob), 4),
            "machine": round(float(machine_prob), 4),
        },
    }

    logger.info(
        f"AMD result={result} confidence={confidence:.3f} "
        f"transcript='{transcript[:80]}' time={processing_time_ms}ms"
    )

    return response


# ---------------------------------------------------------------------------
# FastAPI app
# ---------------------------------------------------------------------------
@asynccontextmanager
async def lifespan(app: FastAPI):
    """Load models at startup, clean up at shutdown."""
    global thread_pool
    logger.info("Starting AMD Service...")
    load_whisper_model()
    load_classifier()
    thread_pool = ThreadPoolExecutor(max_workers=config.MAX_CONCURRENT)
    logger.info(f"AMD Service ready — thread pool size: {config.MAX_CONCURRENT}")
    yield
    logger.info("Shutting down AMD Service...")
    thread_pool.shutdown(wait=True)


app = FastAPI(
    title="AI AMD Service",
    description="AI-Powered Answering Machine Detection",
    version="1.0.0",
    lifespan=lifespan,
)


class AMDResponse(BaseModel):
    result: str
    confidence: float
    transcript: str
    processing_time_ms: int
    audio_duration_s: float
    whisper_time_ms: int
    probabilities: dict


@app.post("/amd", response_model=AMDResponse)
async def amd_classify(
    file: UploadFile = File(..., description="Audio file (WAV, 8kHz mono preferred)"),
    call_id: Optional[str] = Query(None, description="Call ID for logging"),
):
    """
    Classify an audio file as HUMAN, MACHINE, or NOTSURE.

    Upload a WAV file containing the first 3-5 seconds of answered audio.
    The service will transcribe with Whisper and classify using the trained model.
    """
    if METRICS_AVAILABLE:
        ACTIVE_REQUESTS.inc()

    try:
        file_bytes = await file.read()

        if len(file_bytes) == 0:
            raise HTTPException(status_code=400, detail="Empty audio file")

        if len(file_bytes) > 5 * 1024 * 1024:  # 5 MB limit
            raise HTTPException(status_code=400, detail="Audio file too large (max 5 MB)")

        # Run classification in thread pool (Whisper is CPU-bound)
        loop = asyncio.get_event_loop()
        response = await loop.run_in_executor(thread_pool, classify_audio, file_bytes)

        if call_id:
            response["call_id"] = call_id
            logger.info(f"Call {call_id}: {response['result']} ({response['confidence']:.3f})")

        return JSONResponse(content=response)

    except HTTPException:
        raise
    except Exception as e:
        logger.exception(f"AMD classification failed: {e}")
        raise HTTPException(status_code=500, detail=f"Classification failed: {str(e)}")
    finally:
        if METRICS_AVAILABLE:
            ACTIVE_REQUESTS.dec()


@app.get("/health")
async def health_check():
    """Health check endpoint for load balancers and monitoring."""
    status = {
        "status": "healthy",
        "whisper_model": config.WHISPER_MODEL,
        "whisper_loaded": whisper_model is not None,
        "classifier_loaded": classifier_model is not None,
        "scaler_loaded": feature_scaler is not None,
        "max_concurrent": config.MAX_CONCURRENT,
    }
    if not whisper_model or not classifier_model:
        status["status"] = "degraded"
        return JSONResponse(content=status, status_code=503)
    return status


@app.post("/reload")
async def reload_model():
    """Reload the ML classifier model without restarting the service."""
    try:
        load_classifier()
        return {"status": "reloaded", "model_path": config.MODEL_PATH}
    except Exception as e:
        logger.exception(f"Model reload failed: {e}")
        raise HTTPException(status_code=500, detail=f"Reload failed: {str(e)}")


@app.get("/metrics")
async def metrics():
    """Prometheus metrics endpoint."""
    if not METRICS_AVAILABLE or not config.METRICS_ENABLED:
        raise HTTPException(status_code=404, detail="Metrics not available")
    from starlette.responses import Response
    return Response(content=generate_latest(), media_type=CONTENT_TYPE_LATEST)


if __name__ == "__main__":
    uvicorn.run(
        "main:app",
        host=config.HOST,
        port=config.PORT,
        workers=config.WORKERS,
        log_level=config.LOG_LEVEL.lower(),
    )

Requirements

Create /opt/amd-service/requirements.txt:

fastapi==0.115.0
uvicorn[standard]==0.30.0
python-multipart==0.0.9
faster-whisper==1.0.3
numpy>=1.24.0
scikit-learn>=1.3.0
librosa>=0.10.0
prometheus-client>=0.20.0

Systemd Service

Create /etc/systemd/system/amd-service.service:

[Unit]
Description=AI AMD Classification Service
After=network.target
Wants=network-online.target

[Service]
Type=exec
User=root
WorkingDirectory=/opt/amd-service
ExecStart=/opt/amd-service/venv/bin/uvicorn main:app --host 0.0.0.0 --port 8090
Restart=on-failure
RestartSec=5

# Environment variables (override defaults here)
Environment=AMD_WHISPER_MODEL=tiny
Environment=AMD_WHISPER_DEVICE=cpu
Environment=AMD_WHISPER_COMPUTE_TYPE=int8
Environment=AMD_MAX_CONCURRENT=4
Environment=AMD_LOG_LEVEL=INFO

# Resource limits
LimitNOFILE=65535
MemoryMax=4G
CPUQuota=200%

[Install]
WantedBy=multi-user.target

Installation and Startup

# Create directory and virtual environment
mkdir -p /opt/amd-service/{models,logs}
cd /opt/amd-service
python3.11 -m venv venv
source venv/bin/activate

# Install dependencies
pip install -r requirements.txt

# Copy your trained model files (from Section 7)
cp /path/to/amd_classifier.pkl models/
cp /path/to/feature_scaler.pkl models/

# Start the service
systemctl daemon-reload
systemctl enable amd-service
systemctl start amd-service
systemctl status amd-service

# Verify it is running
curl http://localhost:8090/health

Test the Service

# Test with a sample audio file
curl -X POST http://localhost:8090/amd \
  -F "file=@/path/to/test_audio.wav" \
  -F "call_id=TEST001"

# Expected response:
# {
#   "result": "HUMAN",
#   "confidence": 0.9523,
#   "transcript": "Hello?",
#   "processing_time_ms": 850,
#   "audio_duration_s": 2.34,
#   "whisper_time_ms": 620,
#   "probabilities": {"human": 0.9523, "machine": 0.0477},
#   "call_id": "TEST001"
# }

# Test with curl and a generated sine wave (should classify as MACHINE/NOTSURE)
sox -n -r 8000 -c 1 /tmp/test_tone.wav synth 3 sine 440
curl -X POST http://localhost:8090/amd -F "file=@/tmp/test_tone.wav"

9. Asterisk Integration

With the AMD service running, we need to connect it to Asterisk so that every outbound answered call is automatically classified before being bridged to an agent.

Architecture: How It Fits

Dialer places call
        │
        ▼
   Call Answered
        │
        ▼
  Answer() in dialplan
        │
        ▼
  AGI(amd_check.agi)
   ┌────┴────────────────────┐
   │ 1. Record first 4s      │
   │ 2. POST to AMD service  │
   │ 3. Parse result          │
   └────┬────────────────────┘
        │
   ┌────┴────┐
   │ Result? │
   └─┬──┬──┬─┘
     │  │  │
  HUMAN │ MACHINE
     │  │     │
     ▼  │     ▼
  Dial  │  Voicemail Drop
(agent) │  or Hangup
        │
      NOTSURE
        │
        ▼
    Dial (agent)
  (safe default)

AGI Script — Standard Approach (Record + POST)

Create /var/lib/asterisk/agi-bin/amd_check.agi:

#!/usr/bin/env python3
"""
AGI script for AI-powered Answering Machine Detection.

Records the first few seconds of answered audio, sends to the AMD
classification service, and sets channel variables based on the result.

Channel variables set:
    AMDRESULT    - HUMAN, MACHINE, or NOTSURE
    AMDCONFIDENCE - Confidence score (0.0-1.0)
    AMDTRANSCRIPT - Whisper transcript of the greeting
    AMDTIME      - Processing time in milliseconds

Usage in dialplan:
    exten => s,n,AGI(amd_check.agi)
    exten => s,n,GotoIf($["${AMDRESULT}" = "MACHINE"]?machine:human)
"""

import sys
import os
import time
import json
import urllib.request
import urllib.error
import tempfile

# ---------------------------------------------------------------------------
# Configuration
# ---------------------------------------------------------------------------
AMD_SERVICE_URL = os.getenv("AMD_SERVICE_URL", "http://127.0.0.1:8090/amd")
RECORD_DURATION = 4        # Seconds to record
RECORD_SILENCE = 2         # Stop if 2s of silence detected
RECORD_FORMAT = "wav"
TIMEOUT_SECONDS = 8        # Total timeout for HTTP request
MAX_RETRIES = 1            # Retry once on failure

# ---------------------------------------------------------------------------
# AGI communication
# ---------------------------------------------------------------------------
class AGI:
    """Minimal AGI interface for communicating with Asterisk."""

    def __init__(self):
        self.env = {}
        self._read_env()

    def _read_env(self):
        """Read AGI environment variables from stdin."""
        while True:
            line = sys.stdin.readline().strip()
            if not line:
                break
            if ":" in line:
                key, _, value = line.partition(":")
                self.env[key.strip()] = value.strip()

    def execute(self, command):
        """Send AGI command and return the result."""
        sys.stdout.write(f"{command}\n")
        sys.stdout.flush()
        result = sys.stdin.readline().strip()
        return result

    def verbose(self, message, level=1):
        """Log a message to the Asterisk console."""
        self.execute(f'VERBOSE "{message}" {level}')

    def set_variable(self, name, value):
        """Set a channel variable."""
        self.execute(f'SET VARIABLE {name} "{value}"')

    def get_variable(self, name):
        """Get a channel variable."""
        result = self.execute(f"GET VARIABLE {name}")
        # Result format: 200 result=1 (value)
        if "(" in result and ")" in result:
            return result.split("(")[1].split(")")[0]
        return ""

    def answer(self):
        """Answer the channel."""
        self.execute("ANSWER")

    def record_file(self, filename, fmt="wav", escape_digits="",
                    timeout_ms=-1, silence_seconds=0, beep=False):
        """Record audio to a file."""
        beep_str = "BEEP" if beep else ""
        timeout = timeout_ms if timeout_ms > 0 else -1
        cmd = (
            f'RECORD FILE "{filename}" "{fmt}" "{escape_digits}" '
            f'{timeout} {silence_seconds} {beep_str}'
        )
        return self.execute(cmd)

    def stream_file(self, filename, escape_digits=""):
        """Play an audio file."""
        self.execute(f'STREAM FILE "{filename}" "{escape_digits}"')


def send_to_amd_service(audio_path: str, call_id: str = "") -> dict:
    """Send audio file to AMD service and return the classification result."""
    with open(audio_path, "rb") as f:
        audio_data = f.read()

    # Build multipart/form-data request manually (no requests library needed)
    boundary = "----AMDBoundary" + str(int(time.time() * 1000))

    body = []
    # File field
    body.append(f"--{boundary}".encode())
    body.append(
        b'Content-Disposition: form-data; name="file"; filename="audio.wav"'
    )
    body.append(b"Content-Type: audio/wav")
    body.append(b"")
    body.append(audio_data)

    # Call ID field (if provided)
    if call_id:
        body.append(f"--{boundary}".encode())
        body.append(
            b'Content-Disposition: form-data; name="call_id"'
        )
        body.append(b"")
        body.append(call_id.encode())

    body.append(f"--{boundary}--".encode())
    body.append(b"")

    body_bytes = b"\r\n".join(body)

    url = AMD_SERVICE_URL
    if call_id:
        url += f"?call_id={call_id}"

    req = urllib.request.Request(
        url,
        data=body_bytes,
        headers={
            "Content-Type": f"multipart/form-data; boundary={boundary}",
        },
        method="POST",
    )

    response = urllib.request.urlopen(req, timeout=TIMEOUT_SECONDS)
    return json.loads(response.read().decode())


def main():
    agi = AGI()
    call_id = agi.get_variable("UNIQUEID") or "unknown"
    agi.verbose(f"AI AMD: Starting classification for call {call_id}")

    start_time = time.time()

    # Default values (safe fallback = treat as HUMAN)
    result = "NOTSURE"
    confidence = 0.0
    transcript = ""

    try:
        # Record first N seconds of audio
        tmp_dir = tempfile.mkdtemp(prefix="amd_")
        record_path = os.path.join(tmp_dir, "amd_audio")
        wav_file = record_path + ".wav"

        agi.verbose(f"AI AMD: Recording {RECORD_DURATION}s of audio...")
        record_timeout_ms = RECORD_DURATION * 1000
        agi.record_file(
            record_path,
            fmt=RECORD_FORMAT,
            timeout_ms=record_timeout_ms,
            silence_seconds=RECORD_SILENCE,
        )

        # Verify recording exists and has data
        if not os.path.exists(wav_file):
            agi.verbose("AI AMD: Recording file not created — defaulting to NOTSURE")
            raise FileNotFoundError("Recording not created")

        file_size = os.path.getsize(wav_file)
        if file_size < 1000:  # Less than 1KB = probably empty
            agi.verbose(f"AI AMD: Recording too small ({file_size} bytes) — defaulting to NOTSURE")
            raise ValueError("Recording too small")

        # Send to AMD service
        agi.verbose("AI AMD: Sending to classification service...")

        for attempt in range(MAX_RETRIES + 1):
            try:
                amd_result = send_to_amd_service(wav_file, call_id)
                result = amd_result.get("result", "NOTSURE")
                confidence = amd_result.get("confidence", 0.0)
                transcript = amd_result.get("transcript", "")
                break
            except (urllib.error.URLError, urllib.error.HTTPError) as e:
                if attempt < MAX_RETRIES:
                    agi.verbose(f"AI AMD: Attempt {attempt + 1} failed, retrying: {e}")
                    time.sleep(0.5)
                else:
                    agi.verbose(f"AI AMD: All attempts failed: {e}")
                    raise

        # Clean up temp files
        try:
            os.unlink(wav_file)
            os.rmdir(tmp_dir)
        except OSError:
            pass

    except Exception as e:
        agi.verbose(f"AI AMD: Error — {e} — defaulting to NOTSURE (route to agent)")
        result = "NOTSURE"
        confidence = 0.0
        transcript = ""

    elapsed_ms = int((time.time() - start_time) * 1000)

    # Set channel variables for dialplan
    agi.set_variable("AMDRESULT", result)
    agi.set_variable("AMDCONFIDENCE", str(round(confidence, 4)))
    agi.set_variable("AMDTRANSCRIPT", transcript[:200])  # Truncate for safety
    agi.set_variable("AMDTIME", str(elapsed_ms))

    agi.verbose(
        f"AI AMD: result={result} confidence={confidence:.3f} "
        f"time={elapsed_ms}ms transcript='{transcript[:60]}'"
    )


if __name__ == "__main__":
    main()

Set permissions:

chmod 755 /var/lib/asterisk/agi-bin/amd_check.agi
chown asterisk:asterisk /var/lib/asterisk/agi-bin/amd_check.agi

EAGI Approach — Lower Latency (Stream Audio)

For even lower latency, use EAGI which gives direct access to the audio stream via file descriptor 3. Instead of recording and then sending, the EAGI script streams audio in real-time:

Create /var/lib/asterisk/agi-bin/amd_check_eagi.py:

#!/usr/bin/env python3
"""
EAGI script for real-time AI AMD.

Reads audio directly from Asterisk via fd3 (signed linear 16-bit, 8kHz mono),
accumulates enough samples, sends to AMD service, returns result.

This avoids the overhead of Record() — saves ~500ms vs standard AGI approach.

Usage in dialplan:
    exten => s,n,EAGI(amd_check_eagi.py)
"""

import sys
import os
import io
import struct
import time
import json
import wave
import urllib.request
import tempfile

AMD_SERVICE_URL = os.getenv("AMD_SERVICE_URL", "http://127.0.0.1:8090/amd")
CAPTURE_SECONDS = 4
SAMPLE_RATE = 8000
SAMPLE_WIDTH = 2  # 16-bit signed linear
TIMEOUT_SECONDS = 8


class EAGI:
    """EAGI interface — reads audio from fd3."""

    def __init__(self):
        self.env = {}
        self.audio_fd = os.fdopen(3, "rb")  # Audio stream from Asterisk
        self._read_env()

    def _read_env(self):
        while True:
            line = sys.stdin.readline().strip()
            if not line:
                break
            if ":" in line:
                key, _, value = line.partition(":")
                self.env[key.strip()] = value.strip()

    def execute(self, command):
        sys.stdout.write(f"{command}\n")
        sys.stdout.flush()
        return sys.stdin.readline().strip()

    def verbose(self, msg, level=1):
        self.execute(f'VERBOSE "{msg}" {level}')

    def set_variable(self, name, value):
        self.execute(f'SET VARIABLE {name} "{value}"')

    def get_variable(self, name):
        result = self.execute(f"GET VARIABLE {name}")
        if "(" in result and ")" in result:
            return result.split("(")[1].split(")")[0]
        return ""

    def read_audio(self, duration_seconds):
        """Read raw audio samples from fd3."""
        total_bytes = int(SAMPLE_RATE * SAMPLE_WIDTH * duration_seconds)
        audio_data = b""

        while len(audio_data) < total_bytes:
            try:
                chunk = self.audio_fd.read(min(4096, total_bytes - len(audio_data)))
                if not chunk:
                    break
                audio_data += chunk
            except (IOError, OSError):
                break

        return audio_data


def raw_to_wav(raw_data: bytes) -> bytes:
    """Convert raw signed-linear 16-bit 8kHz mono to WAV format."""
    buf = io.BytesIO()
    with wave.open(buf, "wb") as wf:
        wf.setnchannels(1)
        wf.setsampwidth(SAMPLE_WIDTH)
        wf.setframerate(SAMPLE_RATE)
        wf.writeframes(raw_data)
    return buf.getvalue()


def main():
    eagi = EAGI()
    call_id = eagi.get_variable("UNIQUEID") or "unknown"
    eagi.verbose(f"AI AMD (EAGI): Starting for call {call_id}")

    start_time = time.time()
    result = "NOTSURE"
    confidence = 0.0
    transcript = ""

    try:
        # Read audio directly from Asterisk audio stream
        eagi.verbose(f"AI AMD (EAGI): Capturing {CAPTURE_SECONDS}s from audio stream...")
        raw_audio = eagi.read_audio(CAPTURE_SECONDS)

        if len(raw_audio) < SAMPLE_RATE * SAMPLE_WIDTH:  # Less than 1 second
            eagi.verbose("AI AMD (EAGI): Insufficient audio captured")
            raise ValueError("Insufficient audio")

        # Convert to WAV
        wav_data = raw_to_wav(raw_audio)

        # Write to temp file for upload
        with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as tmp:
            tmp.write(wav_data)
            tmp_path = tmp.name

        # Send to AMD service
        eagi.verbose("AI AMD (EAGI): Sending to classification service...")
        boundary = f"----AMDBoundary{int(time.time() * 1000)}"
        body_parts = [
            f"--{boundary}".encode(),
            b'Content-Disposition: form-data; name="file"; filename="audio.wav"',
            b"Content-Type: audio/wav",
            b"",
            wav_data,
            f"--{boundary}--".encode(),
            b"",
        ]
        body = b"\r\n".join(body_parts)

        url = f"{AMD_SERVICE_URL}?call_id={call_id}"
        req = urllib.request.Request(
            url,
            data=body,
            headers={"Content-Type": f"multipart/form-data; boundary={boundary}"},
            method="POST",
        )
        resp = urllib.request.urlopen(req, timeout=TIMEOUT_SECONDS)
        amd_result = json.loads(resp.read().decode())

        result = amd_result.get("result", "NOTSURE")
        confidence = amd_result.get("confidence", 0.0)
        transcript = amd_result.get("transcript", "")

        os.unlink(tmp_path)

    except Exception as e:
        eagi.verbose(f"AI AMD (EAGI): Error — {e} — defaulting to NOTSURE")

    elapsed_ms = int((time.time() - start_time) * 1000)

    eagi.set_variable("AMDRESULT", result)
    eagi.set_variable("AMDCONFIDENCE", str(round(confidence, 4)))
    eagi.set_variable("AMDTRANSCRIPT", transcript[:200])
    eagi.set_variable("AMDTIME", str(elapsed_ms))

    eagi.verbose(
        f"AI AMD (EAGI): result={result} confidence={confidence:.3f} "
        f"time={elapsed_ms}ms"
    )


if __name__ == "__main__":
    main()

chmod 755 /var/lib/asterisk/agi-bin/amd_check_eagi.py
chown asterisk:asterisk /var/lib/asterisk/agi-bin/amd_check_eagi.py

Dialplan Integration

Add to your Asterisk dialplan (e.g., /etc/asterisk/extensions-custom.conf):

; ==========================================================================
; AI AMD Context — called for outbound answered calls
; ==========================================================================
[ai-amd]
exten => s,1,Answer()
exten => s,n,Wait(0.5)                              ; Brief pause for audio to stabilize
exten => s,n,AGI(amd_check.agi)                      ; Run AI AMD classification
exten => s,n,NoOp(AMD Result: ${AMDRESULT} Confidence: ${AMDCONFIDENCE})
exten => s,n,GotoIf($["${AMDRESULT}" = "MACHINE"]?machine)
exten => s,n,GotoIf($["${AMDRESULT}" = "NOTSURE"]?human)
; Fall through = HUMAN
exten => s,n(human),NoOp(HUMAN detected — connecting to agent)
exten => s,n,Set(CALLERID(name)=HUMAN-${AMDCONFIDENCE})  ; Tag for agent display
exten => s,n,Goto(from-internal,${EXTEN},1)               ; Route to agent queue
; Machine handling
exten => s,n(machine),NoOp(MACHINE detected — voicemail drop or hangup)
exten => s,n,GotoIf($["${CAMPAIGN_VM_DROP}" = "YES"]?vmdrop)
exten => s,n,Hangup()
; Voicemail drop (if enabled)
exten => s,n(vmdrop),AGI(voicemail_drop.agi)
exten => s,n,Hangup()

ViciDial Campaign Settings

To use the AI AMD with ViciDial, you have two main approaches:

Approach 1: Custom dialplan context (recommended)

In the ViciDial admin, set the campaign's Dial Context to your AI AMD context:

Campaign Settings:
  AMD Method: OFF                    (disable built-in AMD)
  Dial Context: ai-amd               (use custom AMD context)

Approach 2: AGI integration in carrier dialplan

Add the AGI call to the carrier's extension in extensions-vicidial.conf (after backing up):

; In the carrier dial extension, after Answer detection:
; WARNING: Modifying extensions-vicidial.conf requires careful planning.
; ViciDial regenerates parts of this file — put custom code in
; extensions-custom.conf and use a GoSub or Goto.

ViciDial database integration — log AMD results alongside ViciDial's call records:

-- Create a table to store AI AMD results
CREATE TABLE IF NOT EXISTS ai_amd_log (
    id BIGINT AUTO_INCREMENT PRIMARY KEY,
    uniqueid VARCHAR(50) NOT NULL,
    call_date DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
    amd_result ENUM('HUMAN','MACHINE','NOTSURE') NOT NULL,
    confidence DECIMAL(5,4) NOT NULL,
    transcript TEXT,
    processing_time_ms INT NOT NULL,
    campaign_id VARCHAR(20),
    phone_number VARCHAR(20),
    agent_disposition VARCHAR(10) DEFAULT NULL,
    INDEX idx_uniqueid (uniqueid),
    INDEX idx_call_date (call_date),
    INDEX idx_result (amd_result)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

10. Voicemail Drop — When Machine Detected

When the AMD service classifies a call as MACHINE, you have two options: hang up immediately (simple) or drop a pre-recorded voicemail message after the beep (better for lead generation). This section covers the voicemail drop approach.

How Voicemail Drop Works

MACHINE detected
       │
       ▼
  Wait for greeting to finish
  (monitor energy level)
       │
       ▼
  Detect beep
  (sudden tone burst)
       │
       ▼
  Play pre-recorded message
  ("Hi, this is Jane calling about...")
       │
       ▼
  Hangup

The tricky part is detecting the beep. Voicemail systems play a greeting, then a beep (usually 1-2kHz tone lasting 200-500ms), then start recording. We need to detect that beep reliably.

Voicemail Drop AGI Script

Create /var/lib/asterisk/agi-bin/voicemail_drop.agi:

#!/usr/bin/env python3
"""
Voicemail Drop AGI Script

Waits for the voicemail greeting to end, detects the beep,
then plays a pre-recorded message and hangs up.

Channel variables read:
    CAMPAIGN_ID    - Campaign ID (for selecting the VM drop message)
    VM_MESSAGE     - Specific message file to play (overrides campaign default)

Usage in dialplan:
    exten => s,n,AGI(voicemail_drop.agi)
"""

import sys
import os
import time
import struct
import math

# ---------------------------------------------------------------------------
# Configuration
# ---------------------------------------------------------------------------
# Directory containing voicemail drop recordings (WAV, 8kHz mono)
VM_MESSAGES_DIR = "/var/lib/asterisk/sounds/vmdrop"

# Default message if no campaign-specific one exists
DEFAULT_MESSAGE = "default_vmdrop"

# Beep detection parameters
BEEP_MIN_FREQ = 800       # Hz — minimum frequency to consider a beep
BEEP_MAX_FREQ = 2500      # Hz — maximum frequency to consider a beep
BEEP_MIN_DURATION = 0.1   # Seconds — minimum beep length
BEEP_MAX_DURATION = 1.5   # Seconds — maximum beep length
BEEP_ENERGY_THRESHOLD = 0.02  # Minimum energy to detect beep

# Timing
MAX_WAIT_GREETING = 30    # Seconds — max time to wait for greeting to finish
SILENCE_AFTER_GREETING = 0.5  # Seconds of silence after greeting (before beep)
POST_BEEP_DELAY = 0.3     # Seconds to wait after beep before playing


# ---------------------------------------------------------------------------
# AGI interface
# ---------------------------------------------------------------------------
class AGI:
    def __init__(self):
        self.env = {}
        self._read_env()

    def _read_env(self):
        while True:
            line = sys.stdin.readline().strip()
            if not line:
                break
            if ":" in line:
                key, _, value = line.partition(":")
                self.env[key.strip()] = value.strip()

    def execute(self, command):
        sys.stdout.write(f"{command}\n")
        sys.stdout.flush()
        return sys.stdin.readline().strip()

    def verbose(self, msg, level=1):
        self.execute(f'VERBOSE "{msg}" {level}')

    def set_variable(self, name, value):
        self.execute(f'SET VARIABLE {name} "{value}"')

    def get_variable(self, name):
        result = self.execute(f"GET VARIABLE {name}")
        if "(" in result and ")" in result:
            return result.split("(")[1].split(")")[0]
        return ""

    def stream_file(self, filename, escape_digits=""):
        """Play an audio file."""
        return self.execute(f'STREAM FILE "{filename}" "{escape_digits}"')

    def wait_for_digit(self, timeout_ms):
        """Wait for a DTMF digit (used for timing)."""
        return self.execute(f"WAIT FOR DIGIT {timeout_ms}")

    def channel_status(self):
        """Check if channel is still up."""
        result = self.execute("CHANNEL STATUS")
        # 200 result=6 means channel is up
        try:
            code = int(result.split("=")[1].split(" ")[0])
            return code
        except (IndexError, ValueError):
            return -1

    def get_data(self, filename, timeout_ms=0, max_digits=0):
        """Play file and wait — useful for waiting with audio monitoring."""
        return self.execute(f'GET DATA "{filename}" {timeout_ms} {max_digits}')


def select_vm_message(agi):
    """Select the voicemail drop message to play."""
    # Check for explicit message override
    explicit = agi.get_variable("VM_MESSAGE")
    if explicit:
        path = os.path.join(VM_MESSAGES_DIR, explicit)
        if os.path.exists(path + ".wav") or os.path.exists(path + ".sln"):
            return path
        agi.verbose(f"VM Drop: Explicit message not found: {explicit}")

    # Check for campaign-specific message
    campaign = agi.get_variable("CAMPAIGN_ID")
    if campaign:
        path = os.path.join(VM_MESSAGES_DIR, f"vmdrop_{campaign}")
        if os.path.exists(path + ".wav") or os.path.exists(path + ".sln"):
            return path
        agi.verbose(f"VM Drop: No message for campaign {campaign}, using default")

    # Fall back to default
    return os.path.join(VM_MESSAGES_DIR, DEFAULT_MESSAGE)


def wait_for_beep(agi, max_wait=30):
    """
    Wait for the voicemail greeting to finish and the beep to occur.

    Strategy: Monitor for a period of silence (greeting ended) followed
    by a brief tone burst (the beep). Since we cannot do real-time audio
    analysis from standard AGI, we use a simpler timing-based approach:

    1. Wait up to max_wait seconds
    2. Use WaitForSilence (Asterisk application) to detect end of greeting
    3. Then wait a short time for the beep to pass
    """
    # Use Asterisk's built-in silence detection
    # WaitForSilence(silencereqd, iterations, timeout)
    # Wait for 1000ms of silence, check once, timeout after max_wait seconds
    agi.verbose("VM Drop: Waiting for greeting to end (silence detection)...")

    # Execute WaitForSilence via EXEC
    result = agi.execute(
        f'EXEC WaitForSilence "1000|1|{max_wait}"'
    )

    agi.verbose(f"VM Drop: Silence detected (or timeout). Waiting for beep to pass...")

    # After silence is detected, the beep typically follows within 0-2 seconds.
    # Wait a short period for the beep to sound and finish.
    time.sleep(1.5)

    # Additional small delay to ensure beep has finished
    time.sleep(POST_BEEP_DELAY)

    agi.verbose("VM Drop: Beep window passed — ready to play message")
    return True


def main():
    agi = AGI()
    call_id = agi.get_variable("UNIQUEID") or "unknown"
    agi.verbose(f"VM Drop: Starting for call {call_id}")

    # Check channel is still up
    status = agi.channel_status()
    if status != 6:  # 6 = channel is up
        agi.verbose(f"VM Drop: Channel not up (status={status}) — aborting")
        return

    # Select message to play
    message_path = select_vm_message(agi)
    agi.verbose(f"VM Drop: Selected message: {message_path}")

    # Wait for the beep
    beep_detected = wait_for_beep(agi, MAX_WAIT_GREETING)

    # Check channel is still up after waiting
    status = agi.channel_status()
    if status != 6:
        agi.verbose("VM Drop: Channel dropped during greeting wait — aborting")
        return

    # Play the voicemail drop message
    agi.verbose("VM Drop: Playing message...")
    agi.stream_file(message_path)
    agi.verbose(f"VM Drop: Message played for call {call_id}")

    # Brief pause after message, then hangup
    time.sleep(0.5)
    agi.execute("HANGUP")


if __name__ == "__main__":
    main()

chmod 755 /var/lib/asterisk/agi-bin/voicemail_drop.agi
chown asterisk:asterisk /var/lib/asterisk/agi-bin/voicemail_drop.agi

Preparing Voicemail Drop Recordings

# Create directory for voicemail drop recordings
mkdir -p /var/lib/asterisk/sounds/vmdrop

# Record your messages as WAV files, then convert for Asterisk:
# Asterisk prefers: 8kHz, 16-bit, mono, signed linear (SLN) or WAV

# Convert an MP3/WAV recording to Asterisk-compatible format:
sox input_message.mp3 -r 8000 -c 1 -e signed-integer -b 16 \
    /var/lib/asterisk/sounds/vmdrop/default_vmdrop.wav

# Create campaign-specific versions:
sox uk_sales_message.mp3 -r 8000 -c 1 -e signed-integer -b 16 \
    /var/lib/asterisk/sounds/vmdrop/vmdrop_ukcamp.wav

sox italy_message.mp3 -r 8000 -c 1 -e signed-integer -b 16 \
    /var/lib/asterisk/sounds/vmdrop/vmdrop_italy_camp.wav

# Verify the files:
soxi /var/lib/asterisk/sounds/vmdrop/*.wav

# Set permissions
chown -R asterisk:asterisk /var/lib/asterisk/sounds/vmdrop/

Dialplan for Voicemail Drop Routing

Add to /etc/asterisk/extensions-custom.conf:

; ==========================================================================
; Voicemail Drop Context — routes MACHINE calls to voicemail drop
; ==========================================================================
[ai-amd-with-vmdrop]
exten => s,1,Answer()
exten => s,n,Wait(0.5)
exten => s,n,AGI(amd_check.agi)
exten => s,n,NoOp(AMD: ${AMDRESULT} / ${AMDCONFIDENCE} / ${AMDTRANSCRIPT})
;
; --- MACHINE: voicemail drop ---
exten => s,n,GotoIf($["${AMDRESULT}" = "MACHINE"]?machine)
;
; --- HUMAN or NOTSURE: connect to agent ---
exten => s,n,NoOp(Routing to agent — ${AMDRESULT})
exten => s,n,Goto(from-internal,${EXTEN},1)
;
; --- Machine path ---
exten => s,n(machine),NoOp(MACHINE detected — dropping voicemail)
exten => s,n,Set(CAMPAIGN_ID=${CAMPAIGN})
exten => s,n,AGI(voicemail_drop.agi)
exten => s,n,Hangup()

Multiple Messages Per Campaign

You can configure different voicemail drop messages and rotate between them:

# Create multiple messages for a campaign:
# vmdrop_ukcamp_1.wav — "Hi, this is Sarah from..."
# vmdrop_ukcamp_2.wav — "Good afternoon, I'm calling from..."
# vmdrop_ukcamp_3.wav — "Hello, this is a quick message about..."

To rotate, modify the AGI script's select_vm_message() function to pick randomly:

import glob
import random

def select_vm_message_rotating(agi):
    """Select a random voicemail drop message for the campaign."""
    campaign = agi.get_variable("CAMPAIGN_ID")
    if campaign:
        pattern = os.path.join(VM_MESSAGES_DIR, f"vmdrop_{campaign}_*.wav")
        messages = glob.glob(pattern)
        if messages:
            chosen = random.choice(messages)
            # Return without extension (Asterisk adds it)
            return chosen.rsplit(".", 1)[0]

    return os.path.join(VM_MESSAGES_DIR, DEFAULT_MESSAGE)

11. Performance Tuning

AMD latency directly impacts caller experience. Every millisecond between answer and agent connection is dead air that makes the caller say "Hello? Hello?" and potentially hang up. Target: under 2 seconds from answer to AMD decision.

Whisper Model Selection

The model you choose is the single biggest lever for latency vs accuracy:

Recommendation: Use tiny.en or tiny for production AMD. The accuracy difference between tiny and small is typically only 2-3 percentage points, but tiny is 3-5x faster.

Latency Budget Breakdown

Your total AMD latency budget is ~2,000ms. Here is how it breaks down:

┌─────────────────────────────────────────────┐
│ Component              │ Target  │ Actual   │
├────────────────────────┼─────────┼──────────┤
│ Audio capture (Record) │ ~0ms*   │ 0ms*     │
│ Audio transfer to AGI  │ <50ms   │ 10-30ms  │
│ HTTP POST to service   │ <50ms   │ 5-20ms   │
│ Audio preprocessing    │ <50ms   │ 10-30ms  │
│ Whisper transcription  │ <500ms  │ 200-500ms│
│ Feature extraction     │ <10ms   │ 2-5ms    │
│ ML classification      │ <10ms   │ 1-3ms    │
│ HTTP response           │ <50ms   │ 5-15ms   │
│ AGI variable setting   │ <50ms   │ 5-10ms   │
├────────────────────────┼─────────┼──────────┤
│ TOTAL (excl. capture)  │ <800ms  │ 250-650ms│
│ + Audio capture time   │ +3000ms │ +3000ms  │ (recording 3s)
│ TOTAL (with capture)   │ <3800ms │ 3250-3650│
└─────────────────────────────────────────────┘

* Audio capture runs in parallel — the 3-4 seconds of recording
  is "free" because the caller is talking during this time anyway.
  The real latency impact is only the processing after capture.

Optimizing Audio Capture Time

The biggest latency component is recording time. Optimize it:

# In amd_check.agi, tune these values:

RECORD_DURATION = 3        # Reduce from 4s to 3s (usually enough)
RECORD_SILENCE = 1.5       # Stop sooner on silence (1.5s instead of 2s)

# For EAGI approach, capture in parallel with analysis:
CAPTURE_SECONDS = 3        # 3 seconds is usually sufficient

Silence detection shortcut: If the audio goes silent within 2 seconds (human said "Hello?" and stopped), you can classify early without waiting for the full recording duration.

Concurrent Request Handling

Size your thread pool based on your call volume and hardware:

# In config.py / environment:

# Rule of thumb: 1 thread per CPU core for CPU-bound Whisper
# If you have 4 CPU cores: MAX_CONCURRENT=4

# For a 50-agent dialer making ~200 calls/hour with 30% answer rate:
# ~60 answered calls/hour = ~1 call/minute needing AMD
# At 1 second processing time, even 2 threads handles this easily

# For a 200-agent dialer:
# ~240 answered calls/hour = ~4 calls/minute
# MAX_CONCURRENT=4 handles this with room to spare

GPU vs CPU Comparison

For most call centers (under 100 agents), a standard 4-8 core CPU VPS is more than sufficient. GPU only makes sense at 500+ concurrent calls or if you want to use the small model.

Model Quantization

INT8 quantization reduces model size and speeds up inference with minimal accuracy loss:

# In config.py:
WHISPER_COMPUTE_TYPE = "int8"  # Options: float32, float16, int8

# Speed comparison (tiny model, 4-core CPU):
# float32: ~400ms
# int8:    ~250ms   (37% faster)
# float16: ~300ms   (GPU only)

# Accuracy impact of int8 quantization: < 0.5% WER increase
# For AMD purposes (we only need a rough transcript), this is negligible

Audio Preprocessing Optimizations

def preprocess_audio_fast(audio: np.ndarray, sr: int = 8000) -> np.ndarray:
    """
    Fast audio preprocessing for AMD.
    Skip silence at the beginning, truncate to useful portion.
    """
    # 1. Skip leading silence (ring/connect artifacts)
    energy_threshold = 0.005
    frame_size = int(sr * 0.02)  # 20ms frames

    start_idx = 0
    for i in range(0, len(audio) - frame_size, frame_size):
        frame_energy = np.sqrt(np.mean(audio[i:i + frame_size] ** 2))
        if frame_energy > energy_threshold:
            start_idx = max(0, i - frame_size)  # Keep one frame before speech
            break

    audio = audio[start_idx:]

    # 2. Truncate to 4 seconds max (after silence removal)
    max_samples = 4 * sr
    if len(audio) > max_samples:
        audio = audio[:max_samples]

    # 3. Normalize amplitude
    peak = np.max(np.abs(audio))
    if peak > 0:
        audio = audio / peak * 0.95

    return audio

Benchmark Results

Tested on a Hetzner CX31 (4 vCPU AMD EPYC, 8GB RAM, ~$15/mo):

Key finding: The tiny model handles AMD classification perfectly well. The transcript does not need to be word-perfect — it just needs to capture enough keywords ("leave a message", "voicemail", "hello") for the classifier to work.

12. Monitoring & Analytics

An AMD system without monitoring is a black box. You need to track accuracy, detect drift, and continuously improve.

Logging Every Prediction

Every AMD decision should be logged with enough context to analyze later:

# Add to main.py — structured logging for each prediction

import json
from datetime import datetime

def log_prediction(call_id: str, result: dict, campaign_id: str = ""):
    """Log AMD prediction to structured log file for later analysis."""
    log_entry = {
        "timestamp": datetime.utcnow().isoformat(),
        "call_id": call_id,
        "campaign_id": campaign_id,
        "result": result["result"],
        "confidence": result["confidence"],
        "transcript": result["transcript"],
        "processing_time_ms": result["processing_time_ms"],
        "audio_duration_s": result["audio_duration_s"],
        "probabilities": result["probabilities"],
    }

    # Write to JSON lines file (one JSON object per line)
    log_path = "/opt/amd-service/logs/predictions.jsonl"
    with open(log_path, "a") as f:
        f.write(json.dumps(log_entry) + "\n")

Feedback Loop — Compare Predictions vs Agent Dispositions

The most powerful way to measure AMD accuracy is comparing predictions against what agents actually report:

-- Query to compare AI AMD predictions with agent dispositions
-- Run this daily to calculate real-world accuracy

SELECT
    a.amd_result,
    a.confidence,
    -- Agent disposition from ViciDial
    CASE
        WHEN v.status IN ('A','SALE','CALLBK','NI','NP','DEC')
            THEN 'WAS_HUMAN'
        WHEN v.status IN ('AA','AM','AL','ADC','AFAX')
            THEN 'WAS_MACHINE'
        ELSE 'UNKNOWN'
    END AS actual_type,
    COUNT(*) AS count
FROM ai_amd_log a
LEFT JOIN vicidial_log v ON a.uniqueid = v.uniqueid
WHERE a.call_date >= DATE_SUB(NOW(), INTERVAL 1 DAY)
GROUP BY a.amd_result, actual_type
ORDER BY a.amd_result, actual_type;

-- Results look like:
-- +-----------+--------+-------------+-------+
-- | amd_result| conf   | actual_type | count |
-- +-----------+--------+-------------+-------+
-- | HUMAN     | 0.92   | WAS_HUMAN   | 850   |  <-- True Positive
-- | HUMAN     | 0.85   | WAS_MACHINE | 12    |  <-- False Negative
-- | MACHINE   | 0.94   | WAS_MACHINE | 420   |  <-- True Positive
-- | MACHINE   | 0.88   | WAS_HUMAN   | 8     |  <-- FALSE POSITIVE (bad!)
-- | NOTSURE   | 0.62   | WAS_HUMAN   | 45    |  <-- Correctly cautious
-- | NOTSURE   | 0.58   | WAS_MACHINE | 30    |  <-- Correctly cautious
-- +-----------+--------+-------------+-------+

-- Calculate accuracy metrics
SELECT
    COUNT(*) AS total_calls,
    SUM(CASE WHEN
        (a.amd_result = 'HUMAN' AND v.status IN ('A','SALE','CALLBK','NI','NP','DEC'))
        OR
        (a.amd_result = 'MACHINE' AND v.status IN ('AA','AM','AL','ADC','AFAX'))
        THEN 1 ELSE 0 END) AS correct_predictions,
    ROUND(
        SUM(CASE WHEN
            (a.amd_result = 'HUMAN' AND v.status IN ('A','SALE','CALLBK','NI','NP','DEC'))
            OR
            (a.amd_result = 'MACHINE' AND v.status IN ('AA','AM','AL','ADC','AFAX'))
            THEN 1 ELSE 0 END) * 100.0 / COUNT(*),
        1
    ) AS accuracy_pct,
    -- The dangerous metric: humans we hung up on
    SUM(CASE WHEN a.amd_result = 'MACHINE'
        AND v.status IN ('A','SALE','CALLBK','NI','NP','DEC')
        THEN 1 ELSE 0 END) AS false_positives,
    ROUND(
        SUM(CASE WHEN a.amd_result = 'MACHINE'
            AND v.status IN ('A','SALE','CALLBK','NI','NP','DEC')
            THEN 1 ELSE 0 END) * 100.0 / NULLIF(
            SUM(CASE WHEN a.amd_result = 'MACHINE' THEN 1 ELSE 0 END), 0),
        1
    ) AS false_positive_rate
FROM ai_amd_log a
JOIN vicidial_log v ON a.uniqueid = v.uniqueid
WHERE a.call_date >= DATE_SUB(NOW(), INTERVAL 1 DAY);

Prometheus Metrics

The service already exposes metrics at /metrics. Key metrics to monitor:

# prometheus.yml — add scrape config
scrape_configs:
  - job_name: 'amd-service'
    static_configs:
      - targets: ['YOUR_SERVER_IP:8090']
    scrape_interval: 15s
    metrics_path: /metrics

Metrics exposed:

Grafana Dashboard

Create a Grafana dashboard with these panels:

Panel 1: AMD Results Distribution (Pie Chart)

Query: sum by (result) (increase(amd_requests_total[24h]))

Panel 2: Classification Rate Over Time (Time Series)

Query A: rate(amd_requests_total{result="HUMAN"}[5m])
Query B: rate(amd_requests_total{result="MACHINE"}[5m])
Query C: rate(amd_requests_total{result="NOTSURE"}[5m])

Panel 3: Latency Histogram (Heatmap)

Query: rate(amd_latency_seconds_bucket[5m])

Panel 4: P95 Latency (Time Series)

Query: histogram_quantile(0.95, rate(amd_latency_seconds_bucket[5m]))

Panel 5: Confidence Score Distribution (Histogram)

Query: rate(amd_confidence_bucket[1h])

Panel 6: Active Requests (Gauge)

Query: amd_active_requests

Panel 7: Accuracy Over Time (requires feedback data)

Create a custom exporter or use a recording rule that queries the ai_amd_log table:

# /opt/amd-service/accuracy_exporter.py
"""
Prometheus exporter that queries the ai_amd_log table and publishes
accuracy metrics. Run as a cron job every 5 minutes.
"""

import mysql.connector
from prometheus_client import CollectorRegistry, Gauge, push_to_gateway

registry = CollectorRegistry()
accuracy = Gauge('amd_accuracy_rate', 'AMD accuracy over last 24h', registry=registry)
false_pos = Gauge('amd_false_positive_rate', 'False positive rate over last 24h', registry=registry)
total = Gauge('amd_decisions_total_24h', 'Total AMD decisions in last 24h', registry=registry)

conn = mysql.connector.connect(
    host="YOUR_SERVER_IP",
    user="grafana_ro",
    password="YOUR_DB_PASSWORD",
    database="asterisk",
)
cursor = conn.cursor(dictionary=True)

cursor.execute("""
    SELECT
        COUNT(*) AS total,
        SUM(CASE WHEN
            (amd_result='HUMAN' AND agent_disposition IN ('A','SALE','CALLBK','NI','NP','DEC'))
            OR (amd_result='MACHINE' AND agent_disposition IN ('AA','AM','AL','ADC','AFAX'))
            THEN 1 ELSE 0 END) AS correct,
        SUM(CASE WHEN amd_result='MACHINE'
            AND agent_disposition IN ('A','SALE','CALLBK','NI','NP','DEC')
            THEN 1 ELSE 0 END) AS false_pos
    FROM ai_amd_log
    WHERE call_date >= DATE_SUB(NOW(), INTERVAL 24 HOUR)
    AND agent_disposition IS NOT NULL
""")

row = cursor.fetchone()
if row and row['total'] > 0:
    accuracy.set(row['correct'] / row['total'])
    if row['correct'] > 0:
        false_pos.set(row['false_pos'] / row['total'])
    total.set(row['total'])

push_to_gateway('localhost:9091', job='amd_accuracy', registry=registry)
conn.close()

A/B Testing Setup

To safely roll out AI AMD, run it alongside traditional AMD:

; Dialplan: A/B test — 50% AI AMD, 50% traditional AMD
[amd-ab-test]
exten => s,1,Answer()
exten => s,n,Set(RANDOM_NUM=${RAND(1,100)})
;
; --- Group A: AI AMD ---
exten => s,n,GotoIf($[${RANDOM_NUM} <= 50]?ai_amd)
;
; --- Group B: Traditional AMD ---
exten => s,n,AMD()
exten => s,n,Set(AMDRESULT=${AMDSTATUS})
exten => s,n,GotoIf($["${AMDSTATUS}" = "MACHINE"]?machine:human)
;
; --- AI AMD path ---
exten => s,n(ai_amd),Set(AMD_GROUP=AI)
exten => s,n,AGI(amd_check.agi)
exten => s,n,GotoIf($["${AMDRESULT}" = "MACHINE"]?machine:human)
;
exten => s,n(human),NoOp(HUMAN — ${AMD_GROUP:-TRAD})
exten => s,n,Goto(from-internal,${EXTEN},1)
;
exten => s,n(machine),NoOp(MACHINE — ${AMD_GROUP:-TRAD})
exten => s,n,Hangup()

Then compare the two groups in your analytics:

-- Compare A/B test results
SELECT
    CASE WHEN amd_group = 'AI' THEN 'AI AMD' ELSE 'Traditional' END AS method,
    COUNT(*) AS total_calls,
    SUM(CASE WHEN amd_result = 'HUMAN' AND actual = 'HUMAN' THEN 1 ELSE 0 END) AS true_human,
    SUM(CASE WHEN amd_result = 'MACHINE' AND actual = 'MACHINE' THEN 1 ELSE 0 END) AS true_machine,
    SUM(CASE WHEN amd_result = 'MACHINE' AND actual = 'HUMAN' THEN 1 ELSE 0 END) AS false_positive,
    ROUND(AVG(processing_time_ms), 0) AS avg_latency_ms
FROM amd_ab_test_log
WHERE test_date >= DATE_SUB(NOW(), INTERVAL 7 DAY)
GROUP BY amd_group;

Retraining Pipeline

Set up a monthly retraining cycle:

#!/bin/bash
# /opt/amd-service/scripts/retrain.sh
# Run monthly: 0 2 1 * * /opt/amd-service/scripts/retrain.sh

set -e

cd /opt/amd-service
source venv/bin/activate

DATE=$(date +%Y%m%d)
BACKUP_DIR="models/archive/${DATE}"
mkdir -p "${BACKUP_DIR}"

echo "[$(date)] Starting monthly AMD model retraining..."

# 1. Back up current model
cp models/amd_classifier.pkl "${BACKUP_DIR}/"
cp models/feature_scaler.pkl "${BACKUP_DIR}/"

# 2. Export new labeled data from the last 30 days
python3 scripts/export_training_data.py \
    --days 30 \
    --output data/new_training_data.csv

# 3. Retrain model (combines existing + new data)
python3 scripts/train_classifier.py \
    --existing-data data/training_data.csv \
    --new-data data/new_training_data.csv \
    --output models/amd_classifier_new.pkl \
    --scaler models/feature_scaler_new.pkl

# 4. Evaluate new model against current model
python3 scripts/evaluate_model.py \
    --current models/amd_classifier.pkl \
    --new models/amd_classifier_new.pkl \
    --test-data data/test_data.csv \
    --report "${BACKUP_DIR}/comparison_report.txt"

# 5. Only deploy if new model is better
# (evaluate_model.py exits with code 0 if new model wins)
if [ $? -eq 0 ]; then
    echo "[$(date)] New model is better — deploying..."
    cp models/amd_classifier_new.pkl models/amd_classifier.pkl
    cp models/feature_scaler_new.pkl models/feature_scaler.pkl

    # Reload model in running service (no restart needed)
    curl -s -X POST http://localhost:8090/reload
    echo "[$(date)] Model reloaded successfully"
else
    echo "[$(date)] Current model is still better — keeping existing model"
fi

echo "[$(date)] Retraining complete"

13. Comparison with Commercial Solutions

*CPU-based. With GPU, scales to 1,000+ concurrent.

Cost Analysis for a 50-Agent Call Center

Assumptions: 10,000 outbound calls/day, 30% answer rate = 3,000 AMD decisions/day.

The AI AMD approach delivers cloud-level accuracy at zero marginal cost. The only cost is the initial setup time (2-4 hours) and the compute resources you are already paying for.

When to Use Each Solution

• ViciDial AMD(): You have low call volume, accuracy does not matter much, or you are just getting started and need something working immediately
• AI AMD (this tutorial): You want high accuracy, own your data, have technical staff to set it up, and want zero per-call costs
• Google Cloud Speech: You need multi-language support across many languages, do not want to maintain infrastructure, and can afford the per-call cost
• Amazon Connect: You are already on the AWS ecosystem and want deep integration with other AWS services
• Twilio AMD: You are already using Twilio as your telephony provider

14. Troubleshooting

Low Accuracy

Symptom: AMD accuracy is below 85%, or the false positive rate is above 5%.

Diagnosis and fixes:

# 1. Check class balance in training data
cd /opt/amd-service
source venv/bin/activate
python3 -c "
import pandas as pd
df = pd.read_csv('data/training_data.csv')
print('Class distribution:')
print(df['label'].value_counts())
print(f'Ratio: {df[\"label\"].value_counts().min() / df[\"label\"].value_counts().max():.2f}')
"
# If ratio < 0.5, you have class imbalance.
# Fix: Collect more samples of the minority class,
#       or use class_weight='balanced' in scikit-learn.

# 2. Check if training data matches production audio
# Compare sample rates, noise levels, and audio quality
soxi data/training/*.wav | grep "Sample Rate" | sort | uniq -c
# Production audio (Asterisk) is typically 8kHz mono.
# If your training data is 44.1kHz or 16kHz, the features will differ.

# 3. Check recent predictions for patterns
tail -100 logs/predictions.jsonl | python3 -c "
import sys, json
for line in sys.stdin:
    d = json.loads(line)
    if d['confidence'] < 0.8:
        print(f\"{d['result']:8s} conf={d['confidence']:.3f} '{d['transcript'][:60]}'\")
"
# Low-confidence predictions reveal what the model struggles with.

# 4. Review false positives specifically
# These are the most dangerous errors (humans classified as machines)
python3 -c "
import json
with open('logs/predictions.jsonl') as f:
    for line in f:
        d = json.loads(line)
        # Look for MACHINE results with short, human-like transcripts
        if d['result'] == 'MACHINE' and len(d['transcript'].split()) <= 3:
            print(f\"SUSPICIOUS: conf={d['confidence']:.3f} '{d['transcript']}'\")
"

Common root causes:

• Training data recorded at different sample rate than production (16kHz vs 8kHz)
• Training data from a different language or accent mix than production calls
• Class imbalance (too many human samples, not enough machine samples, or vice versa)
• Confidence threshold too aggressive (lower MACHINE_THRESHOLD to 0.85 or 0.90)

High Latency

Symptom: AMD processing takes more than 2 seconds (excluding recording time).

# 1. Check Whisper model size
curl -s http://localhost:8090/health | python3 -m json.tool
# If whisper_model is "base" or "small", switch to "tiny"

# 2. Check CPU usage during inference
top -bn1 | head -20
# If CPU is maxed, reduce MAX_CONCURRENT or upgrade hardware

# 3. Profile a single request
curl -w "\n\nTotal time: %{time_total}s\nConnect: %{time_connect}s\n" \
    -X POST http://localhost:8090/amd \
    -F "file=@/tmp/test_audio.wav"

# 4. Check if other processes are stealing CPU
ps aux --sort=-%cpu | head -10

# 5. Enable INT8 quantization if not already
# In /etc/systemd/system/amd-service.service:
# Environment=AMD_WHISPER_COMPUTE_TYPE=int8
systemctl daemon-reload && systemctl restart amd-service

Quick fixes:

• Switch from base to tiny model (3-5x speed improvement)
• Enable INT8 quantization (AMD_WHISPER_COMPUTE_TYPE=int8)
• Reduce recording duration from 4s to 3s
• Ensure AMD service runs on the same server as Asterisk (no network latency)
• Reduce MAX_CONCURRENT if CPU is overloaded (better to queue than to slow everything down)

Audio Format Issues

Symptom: Service returns errors about audio format, or transcriptions are empty/garbled.

# Check the audio format your Asterisk is producing:
soxi /var/spool/asterisk/monitor/some_recent_recording.wav
# Expected: 8000 Hz, 16-bit, 1 channel (mono), PCM signed integer

# If using Record() in AGI, verify format:
# Record() produces files in the format you specify.
# Use "wav" format for broadest compatibility.

# Test with a known-good file:
sox -n -r 8000 -c 1 -e signed-integer -b 16 /tmp/test_hello.wav \
    synth 2 sine 300-3000
curl -X POST http://localhost:8090/amd -F "file=@/tmp/test_hello.wav"

# If Asterisk produces .sln (signed linear) files:
# Convert SLN to WAV before sending:
sox -t raw -r 8000 -e signed-integer -b 16 -c 1 input.sln output.wav

# Common format mismatches:
# - G.711 ulaw/alaw: needs conversion (sox -t ul / -t al)
# - GSM compressed: needs conversion (sox -t gsm)
# - Wrong sample rate: 16kHz audio labeled as 8kHz sounds like chipmunks

False Positive Analysis

Symptom: Live humans are being classified as machines and getting hung up on or receiving voicemail drop messages.

The most common false positive patterns:

Mitigation rule of thumb: When in doubt, classify as NOTSURE and route to agent. A false negative (machine sent to agent) wastes 15 seconds of agent time. A false positive (human hung up on) loses a potential customer forever.

# Add this safety check to the classifier:
def safe_classify(result, confidence, transcript):
    """
    Override classification with safety checks.
    Err on the side of routing to agent.
    """
    # Empty transcripts are suspicious — could be a quiet human
    if not transcript.strip() and result == "MACHINE":
        return "NOTSURE", confidence

    # Very short audio might not have enough data
    if len(transcript.split()) <= 1 and result == "MACHINE":
        if confidence < 0.95:  # Require very high confidence for short audio
            return "NOTSURE", confidence

    return result, confidence

AGI Timeout Issues

Symptom: AGI script times out, Asterisk logs show "AGI Script amd_check.agi completed, returning 4" or similar timeout errors.

# Check Asterisk AGI timeout setting
asterisk -rx "core show settings" | grep -i agi
# Default AGI timeout is 30 seconds — usually enough

# Check if AMD service is responding
curl -s -o /dev/null -w "%{http_code}" http://localhost:8090/health
# Should return 200

# Check if AMD service is overloaded
curl -s http://localhost:8090/metrics | grep active_requests
# If active_requests equals MAX_CONCURRENT, requests are queuing

# Check AGI script permissions
ls -la /var/lib/asterisk/agi-bin/amd_check.agi
# Must be executable (755) and owned by asterisk user

# Check Python is available in AGI path
/var/lib/asterisk/agi-bin/amd_check.agi --help 2>&1 || echo "Script cannot execute"
# If Python is not in PATH for the asterisk user, use full path in shebang:
# #!/usr/bin/env python3  ->  #!/usr/local/bin/python3.11

# Check Asterisk logs for AGI errors
grep -i "agi\|amd_check" /var/log/asterisk/messages | tail -20

Model Drift Over Time

Symptom: Accuracy gradually decreases over weeks/months, even though nothing was changed.

This happens because voicemail greetings change over time — phone carriers update their default greetings, businesses change their voicemail messages, and new phone system types enter the market.

# Track accuracy trend over time
mysql -u report_cron -p'YOUR_DB_PASSWORD' -h YOUR_SERVER_IP asterisk -e "
SELECT
    DATE(call_date) AS date,
    COUNT(*) AS total,
    ROUND(
        SUM(CASE WHEN
            (amd_result='HUMAN' AND agent_disposition IN ('A','SALE','CALLBK','NI','NP','DEC'))
            OR (amd_result='MACHINE' AND agent_disposition IN ('AA','AM','AL','ADC','AFAX'))
            THEN 1 ELSE 0 END) * 100.0 / COUNT(*),
        1
    ) AS accuracy_pct
FROM ai_amd_log
WHERE agent_disposition IS NOT NULL
AND call_date >= DATE_SUB(NOW(), INTERVAL 30 DAY)
GROUP BY DATE(call_date)
ORDER BY date;
"

# If accuracy drops below 88%, trigger retraining:
# /opt/amd-service/scripts/retrain.sh

Prevention: Run the monthly retraining pipeline (Section 12). Each month, new labeled data from production calls is incorporated into the model, keeping it current with changing voicemail patterns.

Essential Debugging Commands

# Service status
systemctl status amd-service
journalctl -u amd-service --since "1 hour ago" --no-pager

# Recent predictions
tail -20 /opt/amd-service/logs/predictions.jsonl | python3 -m json.tool

# Service health
curl -s http://localhost:8090/health | python3 -m json.tool

# Test with a specific audio file
curl -X POST http://localhost:8090/amd \
    -F "file=@/path/to/audio.wav" \
    -F "call_id=DEBUG001" | python3 -m json.tool

# Prometheus metrics
curl -s http://localhost:8090/metrics

# Check Asterisk AGI logs
grep "AI AMD" /var/log/asterisk/messages | tail -20

# Check disk space (model files + logs)
du -sh /opt/amd-service/models/ /opt/amd-service/logs/

# Monitor real-time requests
tail -f /opt/amd-service/logs/amd.log | grep "AMD result="

# Restart the service (reloads models)
systemctl restart amd-service

# Reload model only (no restart, no downtime)
curl -X POST http://localhost:8090/reload

Summary

Files Created

What's Next

• Tutorial 41: Real-Time Call Transcription with Whisper — Extend the Whisper setup to transcribe entire calls in real-time for QA scoring and agent coaching
• Tutorial 42: Agent Performance Analytics with AI — Use call transcripts and AMD data to build AI-powered agent performance dashboards
• Tutorial 43: Predictive Dialer Optimization with ML — Apply machine learning to optimize dial ratios, best-time-to-call, and list penetration strategies