Голосовые 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 мс и полную совместимость с вашим оборудованием.