Голосовые AI-агенты для IP-телефонии: интеграция и оптимизация latency

Интеграция AI-сотрудников с IP-телефонией

Вы внедрили голосового AI-агента, но звонки обрываются, задержка больше секунды, и операторы жалуются на эхо? Типичная ситуация, когда SIP-стек плохо стыкуется с LLM-пайплайном. Мы решаем эту проблему на уровне медиа-моста — декодируем RTP, ресемплим до 16 кГц для STT, передаём транскрипт в LLM, синтезируем ответ и возвращаем в RTP с минимальной задержкой. Наш подход проверен на 30+ проектах, где latency p99 не превышает 800 мс.

Почему latency в голосовых AI-агентах критичен?

Каждые 100 мс задержки снижают конверсию в продажах на 1-2% (CCW Digital, 2023). При длительности ответа свыше 1.5 с абонент начинает повторять фразы или кладёт трубку. Проблема часто кроется в неоптимальном аудиопотоке: использование STT с частотой 8 кГц, отсутствие параллельной обработки чанков или синхронный TTS. Мы оптимизируем каждый этап: используем глубокую буферизацию jitter-buffer, асинхронный WebSocket и аппаратное декодирование G.711. В результате экономия на операторах достигает 30-50%.

Как мы строим медиа-мост между SIP и AI

Рассмотрим реальный кейс: клиент с FreePBX (Asterisk 20) хотел автоматизировать входящие звонки. Мы развернули медиа-шлюз на Kamailio + Rtpengine, который принимает RTP, конвертирует PCMU/PCMA в PCM 16 кГц и через WebSocket передаёт в STT Deepgram Nova-2. Ответ LLM (Mistral 7B за счёт vLLM) преобразуется в речь Azure TTS и возвращается в RTP. Весь цикл занимает 700-750 мс — это на 40% быстрее, чем типовое решение с единым потоком и синхронным вызовом API. Стек: async Python с asyncio, audioop для декодирования, асинхронный AMI.

import asyncio
import audioop
import wave
from typing import Optional
import websockets

class SIPAudioBridge:
    """
    Мост между SIP RTP потоком и WebSocket для AI-агента.
    Принимает PCMU/PCMA аудио от Asterisk, передаёт PCM 16kHz в STT.
    """

    SAMPLE_RATE_RTP = 8000    # Стандарт телефонии
    SAMPLE_RATE_AI = 16000    # Deepgram Nova-2 требует 16kHz
    CODEC_PCMU = 0            # G.711 μ-law
    CODEC_PCMA = 8            # G.711 A-law

    def ulaw_to_pcm(self, ulaw_data: bytes) -> bytes:
        """Декодирование G.711 μ-law (PCMU) → PCM"""
        return audioop.ulaw2lin(ulaw_data, 2)  # 2 = 16-bit

    def alaw_to_pcm(self, alaw_data: bytes) -> bytes:
        """Декодирование G.711 A-law (PCMA) → PCM"""
        return audioop.alaw2lin(alaw_data, 2)

    def resample_8k_to_16k(self, pcm_8k: bytes) -> bytes:
        """Апсемплинг 8kHz → 16kHz для моделей STT"""
        resampled, _ = audioop.ratecv(
            pcm_8k,
            2,           # 16-bit
            1,           # mono
            self.SAMPLE_RATE_RTP,
            self.SAMPLE_RATE_AI,
            None
        )
        return resampled

    async def stream_rtp_to_stt(self, rtp_socket,
                                  stt_websocket,
                                  codec: int = CODEC_PCMU,
                                  chunk_ms: int = 20):
        """
        Потоковая передача RTP аудио → STT WebSocket.
        chunk_ms: размер чанка в миллисекундах (стандарт 20ms)
        """
        rtp_chunk_size = int(self.SAMPLE_RATE_RTP * 2 * chunk_ms / 1000)

        while True:
            rtp_data, addr = rtp_socket.recvfrom(4096)

            # RTP header — первые 12 байт
            rtp_payload = rtp_data[12:]

            if codec == self.CODEC_PCMU:
                pcm = self.ulaw_to_pcm(rtp_payload)
            elif codec == self.CODEC_PCMA:
                pcm = self.alaw_to_pcm(rtp_payload)
            else:
                continue

            pcm_16k = self.resample_8k_to_16k(pcm)
            await stt_websocket.send(pcm_16k)


class AsteriskAMIConnector:
    """
    Интеграция с Asterisk через Asterisk Manager Interface (AMI).
    AMI позволяет управлять звонками: originate, hangup, transfer, monitor.
    """

    def __init__(self, host: str, port: int,
                  username: str, password: str):
        self.host = host
        self.port = port
        self.username = username
        self.password = password
        self.reader: Optional[asyncio.StreamReader] = None
        self.writer: Optional[asyncio.StreamWriter] = None

    async def connect(self):
        self.reader, self.writer = await asyncio.open_connection(
            self.host, self.port
        )
        # Пропускаем приветствие
        await self.reader.readline()

        # Аутентификация
        await self._send_action({
            "Action": "Login",
            "Username": self.username,
            "Secret": self.password,
        })

    async def _send_action(self, action: dict) -> dict:
        """Отправка AMI action и получение ответа"""
        message = "\r\n".join(
            f"{k}: {v}" for k, v in action.items()
        ) + "\r\n\r\n"

        self.writer.write(message.encode())
        await self.writer.drain()

        response = {}
        while True:
            line = (await self.reader.readline()).decode().strip()
            if not line:
                break
            if ": " in line:
                key, value = line.split(": ", 1)
                response[key] = value

        return response

    async def originate_call_to_ai(self, phone_number: str,
                                    ai_context: str,
                                    caller_id: str = "AI Agent") -> dict:
        """
        Инициирование звонка с AI-агентом через Asterisk.
        Соединяет абонента с extension, обрабатываемым AI-агентом.
        """
        return await self._send_action({
            "Action": "Originate",
            "Channel": f"SIP/trunk/{phone_number}",
            "Context": "ai-agent",
            "Exten": "s",
            "Priority": 1,
            "CallerID": caller_id,
            "Variable": f"AI_CONTEXT={ai_context}",
            "Async": "true",
            "Timeout": 30000,
        })

    async def transfer_to_queue(self, channel: str,
                                  queue_name: str,
                                  agent_context: str) -> dict:
        """Перевод звонка в очередь живых операторов"""
        return await self._send_action({
            "Action": "Redirect",
            "Channel": channel,
            "Exten": queue_name,
            "Context": "queues",
            "Priority": 1,
            "ExtraChannel": "",
        })


class CallRecordingManager:
    """Управление записями разговоров AI-агентов"""

    def __init__(self, storage_backend: str = "s3"):
        self.storage = storage_backend

    def generate_dial_plan_snippet(self, ai_extension: str = "7000") -> str:
        """
        Фрагмент диалплана Asterisk для маршрутизации на AI-агента.
        """
        return f"""
; extensions.conf — AI Agent routing
[ai-agent]
exten => s,1,NoOp(AI Agent handling call)
 same => n,Set(CALL_ID=${{UNIQUEID}})
 same => n,Set(CALLER_NUM=${{CALLERID(num)}})
 same => n,Record(/var/spool/asterisk/recordings/${{CALL_ID}}.wav,0,300)
 same => n,AGI(ai-agent-bridge.py,${{CALL_ID}},${{CALLER_NUM}})
 same => n,GotoIf(${{AGENT_ESCALATE}}?escalate)
 same => n,Hangup()
 same => n(escalate),Queue(support-agents,t,,,300)

[inbound]
exten => _+7XXXXXXXXXX,1,Goto(ai-agent,s,1)
        """

    def save_call_metadata(self, call_id: str,
                            transcript: list[dict],
                            metadata: dict) -> dict:
        """Сохранение транскрипта и метаданных звонка"""
        record = {
            "call_id": call_id,
            "timestamp": metadata.get("start_time"),
            "duration_seconds": metadata.get("duration"),
            "caller_number": metadata.get("caller"),
            "resolution": metadata.get("resolution"),
            "escalated": metadata.get("escalated", False),
            "transcript": transcript,
            "tool_calls_made": metadata.get("tools_used", []),
        }

        # В production: сохранение в S3 + запись в PostgreSQL
        return {"status": "saved", "record_id": f"CALL-{call_id}"}

Как интегрировать AI-агента с Asterisk за 5 шагов?

1. Аудит текущей инфраструктуры: проверяем SIP-трафик, кодеки, нагрузку. Определяем точку подключения (SBC/медиа-шлюз).
2. Развёртывание медиа-шлюза: устанавливаем Kamailio + Rtpengine, настраиваем транскодинг и WebSocket-мост.
3. Интеграция с АТС: через AMI (Asterisk Manager Interface) или AGI (Asterisk Gateway Interface) подключаем управление звонками. Создаём диалплан для AI-extension.
4. Настройка LLM-агента: разворачиваем модель (Mistral 7B с vLLM), подключаем RAG для доступа к базе знаний, настраиваем guardrails.
5. Тестирование и оптимизация: симулируем 50+ одновременных звонков, замеряем latency p99, калибруем STT/TTS.

[2025-03-20 14:32:15] CALL-4521: Incoming call from +7-XXX-XXX-XX-XX
[2025-03-20 14:32:15] STT: chunk received, processing...
[2025-03-20 14:32:15] STT: duration 0.320s, text: "Здравствуйте, меня интересует..."
[2025-03-20 14:32:16] LLM: response generated in 0.250s
[2025-03-20 14:32:16] TTS: audio synthesized in 0.180s
[2025-03-20 14:32:16] RTP: audio sent to caller. Total latency: 0.750s

Диалплан и маршрутизация

Ключевые сценарии маршрутизации в IP-PBX:

• Входящие звонки — все обращения сначала идут в AI-агент. При эскалации — перевод в ACD-очередь с передачей контекста (CRM-данные + краткое резюме разговора оператору через screen pop).
• Исходящие кампании — AI-агент инициирует звонки через AMI Originate по расписанию. Прогрессивный диалер: следующий звонок начинается при завершении предыдущего.
• Hybrid routing — по часам: ночью все звонки на AI, днём — распределение по skill-based routing с AI на overflow.

Сравнение подходов: типовое решение vs наша архитектура

Требования к инфраструктуре

Что входит в работу

• Архитектурный аудит текущей телефонии (SIP-трафик, нагрузка, кодеки).
• Разработка и развёртывание медиа-моста (WebSocket + асинхронный STT/TTS).
• Интеграция с вашей АТС: Asterisk/FreePBX/3CX через AMI или AGI.
• Настройка LLM-агента (RAG, few-shot, guardrails).
• Тестирование под нагрузкой (симуляция 50+ одновременных звонков).
• Документация по эксплуатации и мониторингу, обучение ваших инженеров.
• Гарантийная поддержка в течение месяца после запуска.

Сроки и стоимость ориентировочно

Типовой проект длится от 10 до 16 недель — от аудита до приёмочного тестирования. Стоимость рассчитывается индивидуально в зависимости от сложности инфраструктуры, количества звонков и требуемой функциональности. Мы готовы оценить ваш проект после краткого собеседования. Получите консультацию по вашему проекту — мы оценим архитектуру и предложим решение. Свяжитесь с нами для демонстрации работы AI-агента на вашем сценарии.

За 5 лет мы внедрили AI-телефонию в 30+ компаниях, гарантируем снижение latency до 800 мс и полную совместимость с вашим оборудованием.

Распознавание и синтез речи: ASR, TTS, клонирование голоса

Заказчик приходит с задачей: транскрибировать 40 000 часов колл-центра за неделю. Штатный облачный ASR (Google Speech-to-Text) выдаёт WER 28% на отраслевой лексике и стоит ощутимо дорого при таких объёмах. Задача — снизить WER ниже 10% и перейти на self-hosted инференс.

Типовые проблемы, с которыми приходят

WER не сходится к нужной метрике. Чаще всего виновата не архитектура, а данные: шумные аудио без нормализации уровня (-23 LUFS вместо стандарта), смешанные языки в одном канале, акцент, специфическая доменная лексика. Whisper large-v3 из коробки даёт WER 8–12% на чистом русском и проваливается до 25–35% на записях с PSTN-артефактами и узкополосным кодеком G.711.

Диаризация ломается при больше двух спикеров. pyannote/speaker-diarization-3.1 работает стабильно при 2–3 говорящих, но DER (Diarization Error Rate) растёт с 6% до 18–22% при 5+ участниках конференции. Проблема усугубляется перекрёстными репликами: по умолчанию min_duration_on=0.1 срезает короткие вставки.

Клонирование голоса — латентность или качество. XTTS v2 (Coqui) даёт натуральный голос, но при потоковой генерации stream_chunk_size=20 первый аудиочанк прилетает через 1.4–2.0 с — неприемлемо для интерактивных сценариев. StyleTTS2 и Kokoro быстрее, но требуют точной подготовки референсного аудио.

Как это решается на практике

Базовый стек для production-пайплайна:

• ASR: openai/whisper-large-v3 или faster-whisper (CTranslate2-бэкенд, x4 скорость vs оригинал)
• Диаризация: pyannote.audio 3.x + интеграция через whisperx для выравнивания по словам
• TTS: XTTS v2 для качества, Edge-TTS или Silero для низкой латентности
• Клонирование: XTTS v2 (3–6 с референсного аудио) или OpenVoice v2

Типичный пайплайн для колл-центра выглядит так: аудио из очереди Kafka → нормализация ffmpeg -af loudnorm до -23 LUFS → faster-whisper с beam_size=5, vad_filter=True → pyannote диаризация → постпроцессинг (пунктуация через deepmultilingualpunctuation) → запись в PostgreSQL с временными метками.

Кейс из практики. Финтех-компания с 12 000 звонков/день. Исходный WER на русском с банковской лексикой — 22% (Google STT). После fine-tuning whisper-medium на 200 часах размеченных записей через Hugging Face transformers + Seq2SeqTrainer с learning_rate=1e-5, warmup_steps=500 — WER упал до 7.3%. Инференс на одной A10G через faster-whisper с compute_type=float16 обрабатывает 40-минутный звонок за 55 секунд. Итоговая стоимость инференса — $0.0008/мин против $0.016/мин у облачного провайдера.

Дообучение Whisper на доменных данных

Когда общая модель не справляется, fine-tuning — первый инструмент. Минимальный датасет для заметного улучшения — 20–30 часов размеченного аудио в целевом домене. Разметку можно получить через итеративный процесс: прогнать через базовую модель → вручную исправить 10–15% ошибок → переобучить → повторить.

training_args = Seq2SeqTrainingArguments(
    per_device_train_batch_size=16,
    gradient_accumulation_steps=2,
    learning_rate=1e-5,
    warmup_steps=500,
    max_steps=5000,
    fp16=True,
    predict_with_generate=True,
    generation_max_length=225,
)

Важно: при fine-tuning Whisper нужно замораживать encoder первые 1000 шагов (model.freeze_encoder()), иначе акустические признаки разъедутся раньше, чем decoder адаптируется к новой лексике.

Синтез речи: выбор под задачу

Для интерактивных ботов с требованием TTFB < 300 мс — Silero или Kokoro. Для озвучки контента, где важна натуральность — XTTS v2 с потоковой отдачей через WebSocket.

Процесс работы

Начинаем с аудит-сессии: берём 2–4 часа ваших записей, прогоняем через несколько моделей, замеряем WER/CER, смотрим на распределение ошибок по типам (лексические, акустические, язык). Это занимает 1–2 дня и сразу показывает, нужен ли fine-tuning или достаточно пост-обработки.

Далее — выбор архитектуры под ваш throughput: один GPU для 1000 мин/день или кластер с балансировщиком для 100 000+ мин/день. Деплой через Docker-контейнер с FastAPI или Triton Inference Server для батчированного инференса.

Сроки зависят от сложности: базовая интеграция готовой модели — 1–2 недели. Fine-tuning с подготовкой данных и валидацией — 4–8 недель. Полная разработка голосового пайплайна (ASR + диаризация + TTS + мониторинг) — 2–4 месяца.