Voice AI Agents for IP Telephony: Latency Optimization

Voice AI Agents for IP Telephony: Integration and Latency Optimization

You deployed a voice AI agent, but calls drop, latency exceeds one second, and operators complain about echo? This is typical when the SIP stack poorly interfaces with the LLM pipeline. We solve this at the media bridge level—decoding RTP, resampling to 16 kHz for STT, passing the transcript to the LLM, synthesizing the response, and returning it to RTP with minimal latency. Our approach has been proven in 30+ projects, with p99 latency not exceeding 800 ms.

Why Latency Is Critical in Voice AI Agents

Every 100 ms of delay reduces sales conversion by 1-2% (CCW Digital, 2023). If response time exceeds 1.5 s, callers start repeating phrases or hang up. The problem often lies in suboptimal audio processing: using STT at 8 kHz, lack of parallel chunk processing, or synchronous TTS. We optimize each stage: using deep jitter buffers, asynchronous WebSocket, and hardware G.711 decoding. As a result, operator cost savings reach 30-50%. For a typical 100-agent call center, this translates to annual savings of $150,000–$250,000.

How We Build a Media Bridge Between SIP and AI

Consider a real case: a client with FreePBX (Asterisk 20) wanted to automate incoming calls. We deployed a media gateway using Kamailio + Rtpengine, which accepts RTP, converts PCMU/PCMA to PCM 16 kHz, and sends it via WebSocket to the STT engine (Deepgram Nova-2). The LLM response (Mistral 7B via vLLM) is converted to speech using Azure TTS and returned to RTP. The entire cycle takes 700-750 ms—40% faster than a typical solution with a single synchronous API call. Stack: async Python with asyncio, audioop for decoding, and asynchronous AMI.

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

class SIPAudioBridge:
    """
    Bridge between SIP RTP stream and WebSocket for AI agent.
    Receives PCMU/PCMA audio from Asterisk, transmits PCM 16kHz to STT.
    """

    SAMPLE_RATE_RTP = 8000    # Telephony standard
    SAMPLE_RATE_AI = 16000    # Deepgram Nova-2 requires 16kHz
    CODEC_PCMU = 0            # G.711 μ-law
    CODEC_PCMA = 8            # G.711 A-law

    def ulaw_to_pcm(self, ulaw_data: bytes) -> bytes:
        """Decode G.711 μ-law (PCMU) to PCM"""
        return audioop.ulaw2lin(ulaw_data, 2)  # 2 = 16-bit

    def alaw_to_pcm(self, alaw_data: bytes) -> bytes:
        """Decode G.711 A-law (PCMA) to PCM"""
        return audioop.alaw2lin(alaw_data, 2)

    def resample_8k_to_16k(self, pcm_8k: bytes) -> bytes:
        """Upsample 8kHz to 16kHz for STT models"""
        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):
        """
        Stream RTP audio to STT WebSocket.
        chunk_ms: chunk size in milliseconds (standard 20ms)
        """
        rtp_chunk_size = int(self.SAMPLE_RATE_RTP * 2 * chunk_ms / 1000)

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

            # RTP header — first 12 bytes
            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:
    """
    Integration with Asterisk via Asterisk Manager Interface (AMI).
    AMI allows call control: 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
        )
        # Skip greeting
        await self.reader.readline()

        # Authentication
        await self._send_action({
            "Action": "Login",
            "Username": self.username,
            "Secret": self.password,
        })

    async def _send_action(self, action: dict) -> dict:
        """Send AMI action and get response"""
        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:
        """
        Initiate a call with AI agent via Asterisk.
        Connects the caller to an extension handled by AI agent.
        """
        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:
        """Transfer call to a live agent queue"""
        return await self._send_action({
            "Action": "Redirect",
            "Channel": channel,
            "Exten": queue_name,
            "Context": "queues",
            "Priority": 1,
            "ExtraChannel": "",
        })


class CallRecordingManager:
    """Manage AI agent call recordings"""

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

    def generate_dial_plan_snippet(self, ai_extension: str = "7000") -> str:
        """
        Asterisk dialplan snippet for routing to AI agent.
        """
        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:
        """Save call transcript and metadata"""
        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: save to S3 + write to PostgreSQL
        return {"status": "saved", "record_id": f"CALL-{call_id}"}

How to Integrate an AI Agent with Asterisk in 5 Steps

1. Audit current infrastructure: Check SIP traffic, codecs, load. Determine connection point (SBC/media gateway).
2. Deploy media gateway: Set up Kamailio + Rtpengine, configure transcoding and WebSocket bridge.
3. Integrate with PBX: Use AMI (Asterisk Manager Interface) or AGI (Asterisk Gateway Interface) for call control. Create dialplan for AI extension.
4. Configure LLM agent: Deploy the model (Mistral 7B with vLLM), connect RAG for knowledge base access, set up guardrails.
5. Test and optimize: Simulate 50+ concurrent calls, measure p99 latency, calibrate STT/TTS.

[TIMESTAMP] CALL-4521: Incoming call from +7-XXX-XXX-XX-XX
[TIMESTAMP] STT: chunk received, processing...
[TIMESTAMP] STT: duration 0.320s, text: "Hello, I'm interested in..."
[TIMESTAMP] LLM: response generated in 0.250s
[TIMESTAMP] TTS: audio synthesized in 0.180s
[TIMESTAMP] RTP: audio sent to caller. Total latency: 0.750s

Dialplan and Routing

Key routing scenarios in IP-PBX:

• Inbound calls: All calls go to the AI agent first. On escalation, transfer to an ACD queue with context (CRM data + conversation summary via screen pop).
• Outbound campaigns: AI agent initiates calls via AMI Originate on schedule. Progressive dialer: next call starts when the previous finishes.
• Hybrid routing: By time of day: all calls to AI at night, skill-based routing with AI overflow during the day.

Comparison: Typical Solution vs Our Architecture

Infrastructure Requirements

What's Included in the Work

• Architectural audit of current telephony (SIP traffic, load, codecs).
• Development and deployment of media bridge (WebSocket + async STT/TTS).
• Integration with your PBX: Asterisk/FreePBX/3CX via AMI or AGI.
• Configuration of LLM agent (RAG, few-shot, guardrails).
• Load testing (simulation of 50+ concurrent calls).
• Operations and monitoring documentation, training for your engineers.
• Warranty support for one month after launch.

Timelines and Cost Estimate

A typical project takes 10 to 16 weeks—from audit to acceptance testing. Cost is calculated individually based on infrastructure complexity, call volume, and required functionality. Typical project cost ranges from $50,000 to $150,000 depending on scope. We are ready to assess your project after a brief discussion. Get a consultation for your project—we will evaluate the architecture and propose a solution. Contact us for a demonstration of the AI agent working on your scenario.

Over 5 years, we have implemented AI telephony in 30+ companies, guaranteeing latency below 800 ms and full compatibility with your equipment.

Speech Recognition and Synthesis: ASR, TTS, Voice Cloning

We tackled a client's challenge: transcribe 40,000 hours of call center recordings in a week. Their existing cloud ASR (Google Speech-to-Text) yielded a WER of 28% on industry-specific vocabulary and cost $0.006 per minute — prohibitively expensive at that volume. The goal was to reduce WER below 10% and switch to self-hosted inference. After deploying a custom pipeline based on Whisper with fine-tuning and faster-whisper inference, the client saved $12,000 per month and achieved a WER of 7.3%.

How does speech recognition ASR handle noisy call center recordings?

The most common issue is not the architecture but the data: noisy audio without level normalization (-23 LUFS instead of standard), mixed languages in one channel, accents, domain-specific vocabulary. Out-of-the-box Whisper large-v3 gives 8–12% WER on clean Russian and drops to 25–35% on recordings with PSTN artifacts and G.711 narrowband codec. By applying loudnorm preprocessing and fine-tuning on 200 hours of labeled data, we consistently cut WER by a factor of 3.

Typical problems we encounter

WER does not converge to the desired metric. Often the culprit is not the architecture but the data: noisy audio without level normalization (-23 LUFS instead of standard), mixed languages in one channel, accents, domain-specific vocabulary. Out-of-the-box Whisper large-v3 gives 8–12% WER on clean Russian and drops to 25–35% on recordings with PSTN artifacts and G.711 narrowband codec.

Diarization fails with more than two speakers. pyannote/speaker-diarization-3.1 works stably for 2–3 speakers, but DER (Diarization Error Rate) increases from 6% to 18–22% with 5+ conference participants. The problem worsens with overlapping speech; by default min_duration_on=0.1 cuts short interjections. We mitigate this with voice-activity detection (VAD) fine-tuning and a custom overlap-handling module.

Voice cloning — latency vs. quality. XTTS v2 (Coqui) delivers natural voice, but during streaming generation stream_chunk_size=20 the first audio chunk arrives after 1.4–2.0 seconds — unacceptable for interactive scenarios. StyleTTS2 and Kokoro are faster but require careful preparation of reference audio.

How do we solve it in practice?

The basic stack for a production pipeline:

• ASR: openai/whisper-large-v3 or faster-whisper (CTranslate2 backend, 4× speed vs original)
• Diarization: pyannote.audio 3.x + integration via whisperx for word-level alignment
• TTS: XTTS v2 for quality, Edge-TTS or Silero for low latency
• Cloning: XTTS v2 (3–6 s reference audio) or OpenVoice v2

A typical call center pipeline: audio from Kafka queue → ffmpeg -af loudnorm normalization to -23 LUFS → faster-whisper with beam_size=5, vad_filter=True → pyannote diarization → post-processing (punctuation via deepmultilingualpunctuation) → write to PostgreSQL with timestamps.

Case study from our practice. A fintech company with 12,000 calls per day. Initial WER on Russian with banking vocabulary — 22% (Google STT). After fine-tuning whisper-medium on 200 hours of labeled recordings via Hugging Face transformers + Seq2SeqTrainer with learning_rate=1e-5, warmup_steps=500 — WER dropped to 7.3%. Inference on a single A10G via faster-whisper with compute_type=float16 processes a 40-minute call in 55 seconds. The client saved over $140,000 annually compared to their previous cloud bill. Contact us for a free pilot estimate to see similar savings on your data.

How to fine-tune Whisper on domain data?

When a general model underperforms, fine-tuning is the first tool. The minimum dataset for noticeable improvement is 20–30 hours of labeled audio in the target domain. Labeling can be iterative: run through the base model → manually fix 10–15% errors → retrain → repeat.

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,
)

Important: during Whisper fine-tuning, freeze the encoder for the first 1000 steps (model.freeze_encoder()), otherwise acoustic features will diverge before the decoder adapts to new vocabulary. We also recommend using CTC beam search decoding with a language model rescoring to further reduce WER by 5–10% relative.

Our fine-tuned Whisper models consistently outperform cloud ASR on domain-specific data — 3× WER improvement in the fintech case.

Speech synthesis: How to choose a model for your task?

For interactive bots requiring TTFB < 300 ms — Silero or Kokoro. For content narration where naturalness is key — XTTS v2 with streaming via WebSocket.

Our process and deliverables

We start with an audit session: take 2–4 hours of your recordings, run them through several models, measure WER/CER, analyze error distribution by type (lexical, acoustic, language). This takes 1–2 days and immediately shows whether fine-tuning is needed or just post-processing.

Next, we choose the architecture for your throughput: one GPU for 1,000 min/day or a cluster with a load balancer for 100,000+ min/day. Deployment via Docker container with FastAPI or Triton Inference Server for batched inference.

What you get after engagement:

• Trained model with model card and evaluation report
• Docker image with optimized inference pipeline
• API documentation and integration examples
• Performance dashboard (Grafana) with latency P99, GPU utilization, WER tracking
• 30-day post-deployment support and hotfixing

Timelines depend on complexity:

• Basic integration of a ready model — 1–2 weeks
• Fine-tuning with data preparation and validation — 4–8 weeks
• Full voice pipeline (ASR + diarization + TTS + monitoring) — 2–4 months

Project investments typically range from $20,000 to $80,000. Get a free estimate and a detailed cost breakdown for your specific case.

Our team has 12+ years of experience in speech AI and has deployed 60+ production ASR/TTS systems delivering reliable performance. Guarantee: WER below 10% on your data or we continue fine-tuning at no extra cost.

Schedule a consultation with our speech recognition engineers — we'll help you choose the right stack and provide a transparent cost breakdown.