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:

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

Work Process for STT System

1. 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.
2. Architecture Selection — determine best-fit provider, need for self-hosted, design fallback and post-processing.
3. Pipeline Development — implement streaming client, custom dictionary, monitoring of metrics (WER, latency, error rates).
4. Testing and Fine-Tuning — optimize WER on test set, fine-tune model if self-hosted.
5. Deployment and Integration — deploy in your cloud or on-prem, set up CI/CD, provide API access.
6. 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.