Developing a Speech-to-Text System: From Provider Selection to Production Pipeline
Problem: standard STT models yield 20% Word Error Rate (WER) on medical terminology, which is unacceptable for telemedicine and court stenography. Building a production-grade system goes beyond simply calling the Whisper API. The challenge involves selecting a model suited for the accent and domain, post-processing to correct specialized vocabulary, streaming recognition with p99 latency <500ms, and orchestrating multiple providers for reliability. We have 5+ years of experience in STT development and have delivered 20+ projects for contact centers, healthcare, and finance. In one case for a legal platform, we reduced Domain WER from 18% to 6% using a custom dictionary and fine-tuning Whisper on 200 hours of court recordings.
How to Choose the STT Stack for Your Task?
Each provider has strengths and weaknesses. Here is a comparison of key metrics for Russian:
| Provider | WER (clean speech) | WER (noisy speech) | Latency (ms) | Streaming | Cost Efficiency |
|---|---|---|---|---|---|
| OpenAI Whisper (API) | 5% | 12% | 600 | No | High (pay-per-use) |
| Deepgram Nova-2 | 8% | 15% | 250 | Yes | Medium |
| Azure Speech | 9% | 14% | 300 | Yes | Low |
| Whisper Large-v3 (self-hosted) | 5% | 12% | 400 | No | Very high (pays off in 3-4 months) |
For streaming scenarios (chatbots, live transcription), choose Deepgram or Azure. For maximum quality in batch processing, use Whisper API or self-hosted. Self-hosted yields 4-5x savings compared to Deepgram at volumes above 500 hours per month, plus full control over fine-tuning and privacy.
Why Post-Processing is a Key Stage?
Any STT model makes mistakes on terms, names, abbreviations. Without post-processing, Domain WER reaches 18-20%. We implement a DomainSpecificPostProcessor that:
- corrects the transcript against a custom dictionary;
- normalizes numbers and dates;
- detects and corrects proper nouns.
This reduces Domain WER to 5-8%. Example: the word "pittsburg" is corrected to "Pittsburgh", "ekcel" → "Excel". The dictionary is built from your text corpora. In one project for the financial sector, we added 1500 terms, reducing error count by 60%.
When is Fine-Tuning Justified?
Fine-tuning the model on domain data yields an additional 2-3% WER reduction over post-processing alone. This is relevant if you have 50+ hours of labeled audio recordings. We use LoRA adapters for fast adaptation of Whisper Large-v3 — training takes 2-3 days on an A100. After fine-tuning, Domain WER drops to 4-6%.
How We Build a Production-Grade STT Pipeline
We use an architecture with automatic failover between providers. Example implementation in Python:
import asyncio
import io
import json
from typing import AsyncGenerator, Optional
import httpx
import websockets
import numpy as np
class STTProviderComparator:
"""Compare STT providers by metrics"""
PROVIDERS = {
"openai_whisper": {
"wer_general": 0.05, # Word Error Rate for standard speech
"wer_noisy": 0.12,
"russian_support": "excellent",
"latency_ms": 600, # Batch mode
"streaming": False,
"cost_per_hour": 0.36,
},
"deepgram_nova2": {
"wer_general": 0.08,
"wer_noisy": 0.15,
"russian_support": "good",
"latency_ms": 250,
"streaming": True,
"cost_per_hour": 0.35,
},
"azure_speech": {
"wer_general": 0.09,
"wer_noisy": 0.14,
"russian_support": "excellent",
"latency_ms": 300,
"streaming": True,
"cost_per_hour": 0.96,
},
"whisper_selfhosted": {
"wer_general": 0.05,
"wer_noisy": 0.12,
"russian_support": "excellent",
"latency_ms": 400, # Large-v3 on A100
"streaming": False,
"cost_per_hour": 0.08, # Self-hosted
},
}
def recommend_provider(self, requirements: dict) -> str:
"""
Select provider based on requirements.
requirements: {'streaming': bool, 'max_latency_ms': int, 'language': str,
'volume_hours_monthly': float}
"""
candidates = []
for name, props in self.PROVIDERS.items():
# Filter by hard constraints
if requirements.get('streaming') and not props['streaming']:
continue
if props['latency_ms'] > requirements.get('max_latency_ms', 9999):
continue
# Scoring
wer_score = 1 - props['wer_general']
latency_score = 1 - props['latency_ms'] / 1000
# Economics at high volume
monthly_cost = props['cost_per_hour'] * requirements.get('volume_hours_monthly', 100)
cost_score = 1 / (1 + monthly_cost / 1000)
total_score = wer_score * 0.4 + latency_score * 0.3 + cost_score * 0.3
candidates.append((name, round(total_score, 3)))
return max(candidates, key=lambda x: x[1])[0] if candidates else "openai_whisper"
class StreamingSTTClient:
"""Streaming speech recognition via Deepgram WebSocket"""
def __init__(self, api_key: str, language: str = "ru"):
self.api_key = api_key
self.language = language
self.base_url = "wss://api.deepgram.com/v1/listen"
async def transcribe_stream(self, audio_chunks: AsyncGenerator[bytes, None],
sample_rate: int = 16000) -> AsyncGenerator[str, None]:
"""
Streaming audio recognition.
Returns interim and final transcripts.
"""
params = (
f"?language={self.language}"
f"&encoding=linear16"
f"&sample_rate={sample_rate}"
f"&channels=1"
f"&model=nova-2"
f"&smart_format=true"
f"&punctuate=true"
f"&endpointing=300" # ms silence to detect end of phrase
f"&interim_results=true"
)
async with websockets.connect(
self.base_url + params,
extra_headers={"Authorization": f"Token {self.api_key}"},
max_size=10_000_000
) as ws:
async def send_audio():
async for chunk in audio_chunks:
await ws.send(chunk)
await ws.send(json.dumps({"type": "CloseStream"}))
asyncio.create_task(send_audio())
async for message in ws:
data = json.loads(message)
if data.get("type") == "Results":
channel = data.get("channel", {})
alternatives = channel.get("alternatives", [])
if alternatives:
transcript = alternatives[0].get("transcript", "")
is_final = data.get("is_final", False)
if transcript:
yield transcript if is_final else f"[interim] {transcript}"
class DomainSpecificPostProcessor:
"""
Post-processing transcript for a specific domain.
STT models often make mistakes on terms, proper nouns, abbreviations.
"""
def __init__(self, domain_vocabulary: dict):
"""
domain_vocabulary: {'incorrect_word': 'correct_word'}
Example: {'pittsburg': 'Pittsburgh', 'ekcel': 'Excel'}
"""
self.vocabulary = {k.lower(): v for k, v in domain_vocabulary.items()}
def correct_transcript(self, transcript: str) -> str:
"""Replace misrecognized words"""
words = transcript.split()
corrected = []
for word in words:
clean = word.lower().rstrip('.,!?;:')
punct = word[len(clean):]
corrected.append(self.vocabulary.get(clean, word.rstrip('.,!?;:')) + punct)
return ' '.join(corrected)
def normalize_numbers_and_dates(self, transcript: str) -> str:
"""Normalize numbers and dates from text to structured format"""
import re
# Simple digit replacements (production: use pymorphy2)
number_words = {
'zero': '0', 'one': '1', 'two': '2', 'three': '3', 'four': '4',
'five': '5', 'six': '6', 'seven': '7', 'eight': '8', 'nine': '9',
}
result = transcript.lower()
for word, digit in number_words.items():
result = result.replace(word, digit)
return result
class STTPipeline:
"""Full STT pipeline with fallback and monitoring"""
def __init__(self, primary_provider, fallback_provider=None,
post_processor: Optional[DomainSpecificPostProcessor] = None):
self.primary = primary_provider
self.fallback = fallback_provider
self.post_processor = post_processor
self._error_count = 0
async def transcribe(self, audio_data: bytes,
language: str = "ru") -> dict:
"""
Transcription with automatic fallback.
"""
try:
transcript, confidence = await self._call_provider(
self.primary, audio_data, language
)
provider_used = "primary"
except Exception as e:
self._error_count += 1
if self.fallback:
transcript, confidence = await self._call_provider(
self.fallback, audio_data, language
)
provider_used = "fallback"
else:
raise
# Post-processing
if self.post_processor:
transcript = self.post_processor.correct_transcript(transcript)
return {
"transcript": transcript,
"confidence": confidence,
"provider": provider_used,
"language": language,
}
async def _call_provider(self, provider, audio: bytes, language: str) -> tuple:
"""Stub: replace with actual provider call"""
raise NotImplementedError
STT Quality Assessment and KPIs
Compare metrics we guarantee after deployment (on your data):
| Metric | Description | Target Value |
|---|---|---|
| WER (Word Error Rate) | % of words with errors | < 8% for clean speech |
| CER (Character Error Rate) | % of characters with errors | < 3% |
| RTF (Real-Time Factor) | time/audio duration | < 0.3 for streaming |
| First-word Latency | delay to first result | < 400ms |
| Domain WER | WER on specialized terms | < 12% |
Work Process for STT System
- Analytics and Data Collection — study your acoustic environment, collect a sample of audio with typical accents and vocabulary. For a contact center, we record 50 hours of real conversations.
- Architecture Selection — determine best-fit provider, need for self-hosted, design fallback and post-processing.
- Pipeline Development — implement streaming client, custom dictionary, monitoring of metrics (WER, latency, error rates).
- Testing and Fine-Tuning — optimize WER on test set, fine-tune model if self-hosted.
- Deployment and Integration — deploy in your cloud or on-prem, set up CI/CD, provide API access.
- Training and Support — deliver documentation, conduct workshop for your engineers, provide SLA support.
What’s Included (Deliverables)
- Repository with pipeline code (Python, Docker/Kubernetes configs)
- Custom dictionary and scripts to build it
- Instructions for deployment and monitoring
- Access to test environment for 1 month
- Team training (up to 2 days)
- Technical support for 3 months after release
We guarantee quality: we measure WER on your data before and after deployment. Achieved target metrics are fixed in the contract.
Want to see the pipeline work on your data? Order a pilot project — we’ll deploy the system in your environment and provide a report with metrics. Get a consultation on STT architecture — describe your task, and we’ll propose the optimal solution.