Разработка AI-системы предиктивного набора номера (Predictive Dialer)
Predictive Dialer автоматически набирает номера и соединяет с оператором только готовые звонки (когда клиент ответил). Ключевая задача: максимизировать productive talk time операторов, минимизируя abandoned calls и время ожидания.
Алгоритм предиктивного набора
from dataclasses import dataclass
from typing import Callable
import asyncio
import numpy as np
@dataclass
class DialerMetrics:
active_calls: int
available_agents: int
avg_call_duration: float # средняя продолжительность разговора
avg_answer_rate: float # 0.3 = 30% контактов отвечают
avg_ring_time: float # среднее время до ответа
abandonment_rate: float = 0 # % брошенных звонков (целевой: <3%)
class PredictiveDialer:
MAX_ABANDONMENT_RATE = 0.03 # 3% по стандарту FCC/регуляторам
PACING_ADJUSTMENT_STEP = 0.05
def __init__(self):
self.pacing_factor = 1.2 # начальный коэффициент
self.metrics_window = []
def calculate_calls_to_initiate(self, metrics: DialerMetrics) -> int:
"""Вычисляем сколько звонков инициировать"""
available = metrics.available_agents - metrics.active_calls
if available <= 0:
return 0
# Базовое предсказание: учитываем answer rate
predicted = int(available * self.pacing_factor / metrics.avg_answer_rate)
# Корректировка по abandonment rate
if metrics.abandonment_rate > self.MAX_ABANDONMENT_RATE:
self.pacing_factor = max(1.0, self.pacing_factor - self.PACING_ADJUSTMENT_STEP)
elif metrics.abandonment_rate < self.MAX_ABANDONMENT_RATE * 0.5:
self.pacing_factor = min(3.0, self.pacing_factor + self.PACING_ADJUSTMENT_STEP)
return max(0, predicted - metrics.active_calls)
async def run_dialing_loop(
self,
contacts: list[dict],
get_metrics: Callable[[], DialerMetrics],
initiate_call: Callable[[dict], None]
):
"""Основной цикл набора"""
contact_index = 0
while contact_index < len(contacts):
metrics = await get_metrics()
calls_to_make = self.calculate_calls_to_initiate(metrics)
for _ in range(calls_to_make):
if contact_index >= len(contacts):
break
contact = contacts[contact_index]
if await self.should_call(contact):
await initiate_call(contact)
contact_index += 1
await asyncio.sleep(1) # пересчёт каждую секунду
ML-предикция answer rate
class AnswerRatePredictor:
"""Предсказываем вероятность ответа для каждого контакта"""
FEATURES = [
"hour_of_day", # лучшее время суток для звонка
"day_of_week",
"previous_attempts", # число предыдущих попыток
"last_contact_days", # дней с последнего контакта
"phone_type", # мобильный/городской
"timezone_offset", # часовой пояс контакта
"segment", # сегмент клиента
]
def predict_answer_probability(self, contact: dict, now: datetime) -> float:
"""Возвращает вероятность ответа 0–1"""
features = self.extract_features(contact, now)
return self.model.predict_proba([features])[0][1]
Оптимальное время звонка
OPTIMAL_CALL_WINDOWS = {
"B2C": {
"monday-friday": [(9, 12), (17, 20)],
"saturday": [(10, 13)],
"sunday": [] # не звоним в воскресенье
},
"B2B": {
"monday-friday": [(10, 12), (14, 17)],
}
}
Сроки: базовый предиктивный дайлер — 6–8 недель. Полная система с ML-оптимизацией — 3–4 месяца.