Интеграция Twilio Voice AI для телефонных AI-ботов

Интеграция Twilio Voice AI для телефонных AI-ботов

Twilio — ведущая облачная телефония для разработчиков. Media Streams API позволяет получать аудио звонка в реальном времени через WebSocket и отправлять обратно синтезированную речь — основа для любого голосового AI-бота.

Архитектура интеграции

Caller → Twilio PSTN → TwiML → Media Streams WebSocket → Your AI Server
                                                               ↓
                                                    STT → LLM → TTS
                                                               ↓
                                               Synthesized Audio → Twilio → Caller

TwiML webhook для входящего звонка

from fastapi import FastAPI, Request
from twilio.twiml.voice_response import VoiceResponse, Start, Stream, Say

app = FastAPI()

@app.post("/incoming-call")
async def handle_incoming_call(request: Request):
    response = VoiceResponse()

    # Запускаем Media Stream
    start = Start()
    start.stream(
        url=f"wss://api.yourapp.com/stream",
        track="both_tracks"  # входящее и исходящее аудио
    )
    response.append(start)

    # Произносим приветствие
    response.say(
        "Здравствуйте! Я голосовой ассистент. Как могу помочь?",
        voice="alice",
        language="ru-RU"
    )
    response.pause(length=30)
    return Response(content=str(response), media_type="text/xml")

WebSocket обработчик Media Streams

import asyncio
import json
import base64
from fastapi import WebSocket

@app.websocket("/stream")
async def handle_stream(websocket: WebSocket):
    await websocket.accept()
    call_sid = None
    stream_sid = None
    audio_buffer = bytearray()

    try:
        async for message in websocket.iter_text():
            data = json.loads(message)
            event = data.get("event")

            if event == "start":
                call_sid = data["start"]["callSid"]
                stream_sid = data["start"]["streamSid"]
                session = create_session(call_sid)

            elif event == "media":
                # Twilio использует mulaw 8kHz
                mulaw_audio = base64.b64decode(data["media"]["payload"])
                audio_buffer.extend(mulaw_audio)

                # Обрабатываем когда накопили 2 секунды (16000 bytes @ 8kHz)
                if len(audio_buffer) >= 16000:
                    await process_audio_chunk(
                        bytes(audio_buffer), websocket, stream_sid, session
                    )
                    audio_buffer = bytearray()

            elif event == "stop":
                break

    except Exception as e:
        logger.error(f"Stream error: {e}")

async def send_audio_to_caller(websocket: WebSocket, stream_sid: str, audio_bytes: bytes):
    """Отправляем синтезированное аудио обратно в звонок"""
    encoded = base64.b64encode(audio_bytes).decode()
    await websocket.send_json({
        "event": "media",
        "streamSid": stream_sid,
        "media": {
            "payload": encoded
        }
    })

Конвертация аудиоформатов

Twilio использует μ-law (mulaw) 8kHz. Whisper работает с PCM 16kHz:

import audioop

def mulaw_to_pcm16k(mulaw_bytes: bytes) -> bytes:
    """μ-law 8kHz → PCM 16-bit 8kHz → upsample to 16kHz"""
    pcm_8k = audioop.ulaw2lin(mulaw_bytes, 2)  # μ-law → PCM 16-bit
    pcm_16k, _ = audioop.ratecv(pcm_8k, 2, 1, 8000, 16000, None)  # 8→16kHz
    return pcm_16k

Stоимость

Twilio: $0.0085/мин входящий + $0.013/мин исходящий (США). Для РФ: ~$0.025–0.05/мин через локальных провайдеров.

Сроки: базовый бот на Twilio — 2–3 недели. Production с мониторингом — 1–1.5 месяца.