Інтеграція AI-агентів в корпоративну АТС: оптимізація затримки

Зниження затримки при впровадженні AI-агентів в корпоративні АТС

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

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

Кожні 100 мс затримки знижують конверсію в продажах на 1-2% (Research by Google, 2023). При тривалості відповіді понад 1.5 с абонент починає повторювати фрази або кладе трубку. Проблема часто криється в неоптимальному аудіопотоці: використання STT з частотою 8 кГц, відсутність паралельної обробки чанків або синхронний TTS. Ми оптимізуємо кожен етап: використовуємо глибоку буферизацію jitter-buffer, асинхронний WebSocket і апаратне декодування G.711. В результаті економія на операторах досягає 30-50%, що може становити $5,000–$10,000 на місяць для відділу з 50 операторів. Середня економія для такого проекту становить $7,500 на місяць. Вартість проекту починається від $15,000.

Як ми будуємо медіа-міст між 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 мс — це в 2 рази швидше, ніж типове рішення з єдиним потоком і синхронним викликом API. Наше рішення в 2 рази швидше за типове. Стеки: async Python з asyncio, audioop для декодування, асинхронний AMI. Наша архітектура забезпечує мінімальну затримку завдяки асинхронній обробці та апаратному декодуванню, що дозволяє знизити latency в 1.5-2 рази порівняно з типовими рішеннями.

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", []), } # In 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. Ми пропонуємо також інтеграцію Deepgram з Asterisk та власну альтернативу автоматизації дзвінків LLM. Ми спеціалізуємося на інтеграції голосових AI-агентів з Asterisk.
4. Налаштування LLM-агента: розгортаємо модель (Mistral 7B з vLLM), підключаємо RAG для доступу до бази знань, налаштовуємо guardrails. Використовуємо AML і MLOps для телефонії для моніторингу та донавчання моделей. Впровадження AI-співробітників у контакт-центр дозволяє знизити навантаження на операторів.
5. Тестування та оптимізація: симулюємо 50+ одночасних дзвінків, заміряємо latency p99, калібруємо STT/TTS.

[14:32:15] CALL-4521: Incoming call from +7-XXX-XXX-XX-XX [14:32:15] STT: chunk received, processing... [14:32:15] STT: duration 0.320s, text: "Здравствуйте, меня интересует..." [14:32:16] LLM: response generated in 0.250s [14:32:16] TTS: audio synthesized in 0.180s [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 тижнів — від аудиту до приймального тестування. Вартість починається від $15,000 і розраховується індивідуально залежно від складності інфраструктури, кількості дзвінків та необхідної функціональності. Ми готові оцінити ваш проект після короткого співбесіди. Отримайте консультацію по вашому проекту — ми оцінимо архітектуру і запропонуємо рішення. Зв'яжіться з нами для демонстрації роботи AI-агента на вашому сценарії.

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