Building AI-based Answering Machine Detection Model for VICIdial, FreeSWITCH, and Asterisk

Outbound campaigns live and die on connect rate. When 40–70% of attempts land in voicemail, every wasted ring burns agent time, TPS, and carrier trust. Traditional Answering Machine Detection (AMD) leans on acoustic heuristics — silence lengths, energy thresholds, beep detectors — or tiny CNNs that overfit to one carrier, codec, or language and fall apart elsewhere.

If you’d rather skip the build, **VM Hunter **takes this further with a SpeechLLM model — delivering production-ready AI AMD without the infrastructure overhead.

This article walks through a production-grade alternative: AI-based AMD — transcribe the first seconds of audio and classify the text. On real campaign data, this approach reached up to ~98% accuracy while keeping latency low enough for live transfers. It’s open, trainable, and integrates cleanly with VICIdial, FreeSWITCH, and Asterisk over WebSocket.

Repo: github.com/rixwankhan/whisper-vm-finetune

Why acoustic AMD struggles:

• Carrier & codec variability. Different carriers normalize audio, insert tones, and compress aggressively (G.711, Opus, AMR). Hand-tuned rules don’t generalize.
• Script drift. Call-handling messages change (“please leave…”, “…after the tone…”, IVR variations). Each change breaks brittle detectors.
• Latency trade-off. Early decisions are fast but noisy; waiting for “beep” improves accuracy but hurts live-answer routing.

The most stable signal isn’t an energy threshold — it’s what is being said. Voicemails literally identify themselves.

Design goals:

• Accuracy through language. Let the model learn phrases like “please leave your name and number” directly from data.
• Low latency. Make an early decision at ~2.0 s; confirm with a final decision on the full greeting when available.
• Observability & control. Interpretable classifier, tunable thresholds, and clean JSON outputs.
• Drop-in integration. Web-Socket interface that any dialer can stream PCM into and get back a decision.

System architecture:

Streaming in (WebSocket). Dialer sends PCM16 mono frames (8 kHz or 16 kHz).

ASR: fine-tuned Whisper-small.

• Early path uses the first 2.0 s.
• Final path transcribes the full buffered greeting after a flush.

Text classification. TF-IDF (word bigrams + char 3–6) → logistic regression → calibrated probability of human vs machine.

Decision JSON. The server returns:

{
     "type": "final",
     "label": "machine",
     "confidence": 0.98,
     "proba_human": 0.02,
     "transcript": "please leave your name and number after the tone",
     "elapsed_ms": 150
 }

Key tech:

• Training: Transformers (Whisper), PyTorch, Datasets
• Inference: CTranslate2 + faster-whisper (GPU float16 or CPU int8_float16)
• Classifier: scikit-learn (TF-IDF + LogisticRegression)
• Serving: FastAPI + Uvicorn (WebSocket)

Data & training:

ASR fine-tune (Whisper-small)

• Input CSV: path,transcription (or path,text), audio normalized to 16 kHz mono.
• For AMD, train on 2.0 s clips (clipped/padded). This reduces the domain gap for early decisions.

python scripts/finetune_whisper_small.py \
  --csv /absolute/path/to/sample_stt.csv \
  --out_dir models/whisper-small-finetuned \
  --batch 4 --num_workers 4

# Export to CTranslate2 for fast inference
ct2-transformers-converter \
  --model models/whisper-small-finetuned \
  --output_dir models/whisper-ct2 \
  --copy_files tokenizer.json tokenizer_config.json preprocessor_config.json \
               special_tokens_map.json vocab.json merges.txt normalizer.json generation_config.json \
  --quantization float16   # use int8_float16 for CPU

Text classification (Human vs Machine)

• Input CSV: text,label where label ∈ {human,machine}.
• Use word n-grams (1–2) and character n-grams (3–6). Character features help with clipped words and telephony artifacts.

python scripts/train_text_classifier.py \
  --csv /absolute/path/to/sample_text_labels.csv \
  --out models/text_cls.joblib \
  --word_ngrams 1,2 --char_ngrams 3,6

Observed results (representative run on real campaign data):

• Classifier hold-out accuracy ≈ 99% (both classes F1 ≈ 0.98–0.99).
• Whisper fine-tune lowers short-clip ASR errors; combined end-to-end AMD accuracy up to ~98%.

Always evaluate on your own distributions and set thresholds to match business cost (e.g., penalize false-human vs false-machine differently).

Serving & latency

Start the WebSocket server:

export WHISPER_MODEL_DIR=models/whisper-ct2
export DEVICE=cuda                 # or cpu
export COMPUTE_TYPE=float16        # or int8_float16 for CPU
export CLASSIFIER_PATH=models/text_cls.joblib
export EARLY_SEC=2.0               # 0 to disable early decision

uvicorn server.ws_server:app --host 0.0.0.0 --port 8080

Client streams small binary frames (e.g., 20 ms each). When ready to finalize, send:

{"type":"flush"}

Typical wall-clock:

• Early decision: ~1.5–3.5 s from answer (depends on buffer length & compute).
• Final decision: shortly after flush (tens to low hundreds of ms).

Integrating with dialers:

VICIdial / Asterisk (EAGI)

Dialplan (extensions.conf):

; VICIDIAL_auto_dialer transfer script AMD (load-balanced)
exten => 8370,1,AGI(agi://127.0.0.1:4577/call_log)
exten => 8370,n,Playback(sip-silence)
exten => 8370,n,EAGI(/var/lib/asterisk/agi-bin/amd.py)
exten => 8370,n,AGI(VD_amd.agi,${EXTEN})
exten => 8370,n,AGI(agi-VDAD_ALL_outbound.agi,NORMAL-----LB-----${CONNECTEDLINE(name)})
exten => 8370,n,Hangup()

Dependencies on the dialer:

apt-get install -y python3-pip
pip3 install websocket-client asterisk-agi
install -m 0755 amd.py /var/lib/asterisk/agi-bin/amd.py
chown asterisk:asterisk /var/lib/asterisk/agi-bin/amd.py

EAGI (JSON-aware): /var/lib/asterisk/agi-bin/amd.py

#!/usr/bin/env python3
import os, fcntl, json, time
from websocket import create_connection
from asterisk.agi import AGI

AUDIO_FD = 3
WS_URL   = os.getenv("AMD_WS_URL", "ws://127.0.0.1:8080/ws/amd")
SAMPLE_RATE = int(os.getenv("AMD_SAMPLE_RATE", "8000"))      # EAGI audio usually 8kHz
EARLY_MIN_CONF = float(os.getenv("AMD_EARLY_MIN_CONF", "0.90"))

def set_human(agi, cause="HUMAN", stats=None):
    agi.set_variable("AMDSTATUS", "HUMAN")
    agi.set_variable("AMDCAUSE",  cause)
    if stats: agi.set_variable("AMDSTATS", stats)

def set_machine(agi, cause="MACHINE", stats=None):
    agi.set_variable("AMDSTATUS", "MACHINE")
    agi.set_variable("AMDCAUSE",  cause)
    if stats: agi.set_variable("AMDSTATS", stats)

def should_stop_on(msg):
    if not isinstance(msg, dict): return None
    typ   = str(msg.get("type", "")).lower()
    label = str(msg.get("label", "")).lower()
    conf  = float(msg.get("confidence", 0.0) or 0.0)
    if typ == "final":
        return label
    if typ == "early" and conf >= EARLY_MIN_CONF:
        return label
    return None

def process_json_and_set_vars(agi, msg):
    label = str(msg.get("label", "")).upper()  # HUMAN/MACHINE
    conf  = msg.get("confidence", "")
    ph    = msg.get("proba_human", "")
    tr    = msg.get("transcript", "")
    typ   = msg.get("type", "")
    cause = f"{typ}|p_human={ph}|conf={conf}"
    stats = f"{label}|{cause}|{tr[:120]}"
    (set_human if label == "HUMAN" else set_machine if label == "MACHINE" else set_human)(
        agi, cause if label in ("HUMAN","MACHINE") else "UNKNOWN", stats
    )

def startAGI():
    agi = AGI()
    chan = agi.env.get("agi_channel","")
    ani  = agi.env.get("agi_callerid","")
    did  = agi.env.get("agi_extension","")
    vid  = agi.env.get("agi_calleridname","")
    agi.verbose(f"AMD: answered from {ani} to {did} on {chan} VID={vid}")

    try:
        ws = create_connection(WS_URL, timeout=5)
        ws.send(json.dumps({"config": {"sample_rate": SAMPLE_RATE, "VID": f"{vid}"}}))
        agi.verbose(f"AMD: WS connected {WS_URL} (sr={SAMPLE_RATE})")
    except Exception as e:
        agi.verbose(f"AMD: WS connect error {e} → default HUMAN")
        set_human(agi, cause="NETERR"); return

    fcntl.fcntl(AUDIO_FD, fcntl.F_SETFL, os.O_NONBLOCK)
    last_msg = None
    total_bytes = 0
    buf = b""
    start = time.time()

    try:
        while True:
            try:
                time.sleep(0.2)
                chunk = os.read(AUDIO_FD, 9500)
                if not chunk:
                    ws.send(json.dumps({"type":"flush"}))
                    for _ in range(5):
                        try:
                            res = ws.recv()
                            msg = json.loads(res) if isinstance(res, str) else {}
                            last_msg = msg or last_msg
                            stop = should_stop_on(msg)
                            if stop:
                                process_json_and_set_vars(agi, msg); return
                        except Exception:
                            break
                    if last_msg: process_json_and_set_vars(agi, last_msg)
                    else: set_human(agi, cause="NOAUDIO")
                    return

                buf += chunk
                if len(buf) >= 6400:   # ~0.4s at 8kHz
                    ws.send_binary(buf)
                    total_bytes += len(buf)
                    buf = b""
                    try:
                        ws.settimeout(0.01)
                        res = ws.recv()
                        msg = json.loads(res) if isinstance(res, str) else {}
                        last_msg = msg or last_msg
                        stop = should_stop_on(msg)
                        if stop:
                            process_json_and_set_vars(agi, msg); return
                    except Exception:
                        pass
                    finally:
                        ws.settimeout(5)

                if (time.time() - start) > 5 and total_bytes == 0:
                    ws.send(json.dumps({"type":"flush"}))
                    try:
                        res = ws.recv()
                        msg = json.loads(res) if isinstance(res, str) else {}
                        process_json_and_set_vars(agi, msg)
                    except Exception:
                        set_human(agi, cause="TIMEOUT")
                    return

            except OSError as err:
                if getattr(err, "errno", None) == 11:
                    continue
                set_human(agi, cause="AUDIOERR"); return

    except Exception as e:
        set_human(agi, cause="NETERR")
    finally:
        try: ws.send(json.dumps({"type":"flush"}))
        except Exception: pass
        try: ws.close()
        except Exception: pass

if __name__ == "__main__":
    startAGI()

This script streams audio to your AMD server, parses the JSON decisions, and sets:

• AMDSTATUS → HUMAN or MACHINE
• AMDCAUSE → short string (e.g., final|p_human=0.02|conf=0.98)
• AMDSTATS → compact log including transcript snippet

VD_amd.agi then performs VM drop, hangup, or transfer.

Notes

• If your media path is 16 kHz, set AMD_SAMPLE_RATE=16000 in the environment.
• If your AMD server expects 16 kHz but dialer audio is 8 kHz, the server can upsample (recommended). If not, add resampling on the EAGI side.

FreeSWITCH & others

• The pattern is the same: stream PCM to WebSocket, parse JSON, and set channel variables or decision flags.
• For FreeSWITCH, you can implement the client in Lua (sofia) or Python ESL

Benchmarking & operating point:

Latency. Keep chunks small (20–40 ms). On GPU float16, early decisions typically fall in ~2s; finals are issued promptly after flush. CPU int8_float16 is viable for modest concurrency.

Thresholds. The server returns confidence and proba_human. Tune business rules:

• Favor machine recall (minimize live agent voicemail): require proba_human > 0.55–0.65 to call HUMAN; else default to MACHINE.
• Favor human precision (avoid hanging up on real people): lower threshold slightly or always wait for final before acting.

Monitoring. Log decision JSON, per-carrier confusion matrices, and low-confidence cases for active learning (human review → retraining).

Deployment & scaling:

• Scale horizontally. Run multiple Uvicorn workers/pods behind a WS-capable load balancer. Each worker loads its own models.
• GPU vs CPU. GPU wins for throughput and latency; CPU int8_float16 is robust and portable.
• Resilience. Add /healthz, structured logs, and systemd units. Consider backpressure on the client (e.g., 4–8 concurrent streams per worker to start).

Results & takeaway:

On real outbound traffic, this Whisper-only AMD reached up to ~98% accuracy on hold-out — without acoustic rules — while keeping latency competitive for live transfers. The approach is transparent (you can inspect n-gram weights), adaptable (re-train on new scripts), and easy to integrate (WebSocket + JSON).

If answering machines are eating your connect rate, move the decision to language. You’ll get better accuracy today — and a smoother path to continuous improvement tomorrow.

Project: github.com/rixwankhan/whisper-vm-finetune