Разработка AI-системы исходящих звонков (AI Outbound Dialer)

Разработка AI-системы исходящих звонков (AI Outbound Dialer)

AI Outbound Dialer автономно совершает тысячи исходящих звонков в час, ведёт полноценный диалог и передаёт на оператора только «горячих» контактов. Производительность: 100–500 одновременных звонков на один сервер.

Компоненты системы

from dataclasses import dataclass, field
from enum import Enum
from datetime import datetime

class CampaignType(Enum):
    CONFIRMATION = "confirmation"      # подтверждение заказов
    APPOINTMENT = "appointment"        # напоминание о записи
    SURVEY = "survey"                  # опросы
    WINBACK = "winback"               # возврат клиентов
    COLD_OUTREACH = "cold_outreach"    # холодные звонки
    DEBT_COLLECTION = "debt"          # взыскание

@dataclass
class OutboundCampaign:
    id: str
    type: CampaignType
    name: str
    contacts: list[dict]
    script_id: str
    schedule: dict  # временные окна для звонков
    retry_config: dict = field(default_factory=lambda: {
        "max_attempts": 3,
        "intervals_minutes": [30, 120, 240]
    })
    concurrent_calls: int = 50

class OutboundDialer:
    def __init__(self, telephony_provider: str = "twilio"):
        self.telephony = TelephonyFactory.create(telephony_provider)
        self.ai_agent = OutboundVoiceAgent()
        self.rate_limiter = RateLimiter()
        self.scheduler = CallScheduler()

    async def run_campaign(self, campaign: OutboundCampaign):
        """Запускаем кампанию с контролем одновременности"""
        semaphore = asyncio.Semaphore(campaign.concurrent_calls)
        tasks = []

        for contact in campaign.contacts:
            if await self.should_call(contact, campaign):
                task = asyncio.create_task(
                    self.process_contact(contact, campaign, semaphore)
                )
                tasks.append(task)

        results = await asyncio.gather(*tasks, return_exceptions=True)
        return self.aggregate_results(results)

    async def process_contact(
        self,
        contact: dict,
        campaign: OutboundCampaign,
        semaphore: asyncio.Semaphore
    ):
        async with semaphore:
            call = await self.telephony.initiate_call(contact["phone"])

            if call.status != "answered":
                await self.schedule_retry(contact, campaign)
                return {"status": "no_answer", "contact": contact["phone"]}

            # Запускаем AI-агент для диалога
            result = await self.ai_agent.run_dialog(
                call=call,
                contact=contact,
                script=await self.get_script(campaign.script_id),
                campaign_type=campaign.type
            )

            # Запись результата
            await self.log_call_result(contact, result)
            return result

Предиктивный набор (Predictive Dialing)

class PredictiveDialer:
    """Звоним больше контактов чем операторов, т.к. не все ответят"""

    def __init__(self, target_agents: int):
        self.agents = target_agents
        self.pacing_factor = 1.3  # на 30% больше звонков чем операторов

    def calculate_calls_to_initiate(
        self,
        current_active_calls: int,
        avg_answer_rate: float,   # 0.3 = 30% отвечают
        avg_call_duration_sec: float
    ) -> int:
        available_agents = self.agents - current_active_calls
        calls_needed = int(available_agents * self.pacing_factor / avg_answer_rate)
        return max(0, calls_needed - current_active_calls)

Метрики кампании

Сроки: базовый автодиалер с одним сценарием — 4–6 недель. Полная система с предиктивным набором — 3–4 месяца.