Побудова STT-системи: від вибору провайдера до production-пайплайну

Проблема: стандартні STT-моделі дають 20% Word Error Rate (WER) на медичній термінології, що неприйнятно для телемедицини та судової стенографії. Побудова production-grade системи — це не просто виклик Whisper API. Завдання включає вибір моделі під акцент та предметну область, постобробку для виправлення специфічної лексики, потокове розпізнавання з latency p99 <500ms та оркестрацію декількох провайдерів для надійності. Ми маємо понад 5 років досвіду в розробці STT-рішень і реалізували 20+ проєктів для контакт-центрів, медицини та фінансів. В одному з кейсів для юридичної платформи ми знизили Domain WER з 18% до 6% завдяки кастомному словнику та fine-tuning Whisper на 200 годинах судових записів.

Як вибрати STT-стек під ваше завдання?

Кожен провайдер має сильні та слабкі сторони. Self-hosted Whisper дає WER у 1.6 разів кращий, ніж Deepgram Nova-2 на чистій мові (5% проти 8%). Порівняємо ключові метрики для російської/української мови:

Дані порівняння з офіційних документів провайдерів (OpenAI, Deepgram, Microsoft Azure) та внутрішніх тестів.

Для потокових сценаріїв (чат-боти, live-транскрипція) обираємо Deepgram або Azure. Для максимальної якості при пакетній обробці — Whisper API або self-hosted. Self-hosted дає економію в 4-5 разів порівняно з Deepgram при обсязі від 500 годин на місяць, а також повний контроль над fine-tuning та privacy. Наприклад, за 500 годин на місяць self-hosted коштує близько $40 (0.08500) проти $175 у Deepgram (0.35500) – економія $135.

Чому постобробка — ключовий етап?

Будь-яка STT-модель помиляється на термінах, іменах, абревіатурах. Без постобробки Domain WER досягає 18-20%. Ми реалізуємо модуль DomainSpecificPostProcessor, який:

• корекція транскрипту за кастомним словником;
• нормалізація чисел та дат;
• детекція та виправлення власних назв.

Це знижує Domain WER до 5-8%. Приклад: слово «пітсбург» виправляється на «Пітер», «ексель» → «Excel». Словник будується на основі ваших корпусів текстів. В одному проєкті для фінансового сектору ми додали 1500 термінів, що зменшило кількість помилок на 60%.

Коли fine-tuning виправданий?

Fine-tuning моделі на доменних даних дає додаткове зниження WER на 2-3% порівняно з просто постобробкою. Це актуально, якщо у вас є 50+ годин розмічених аудіозаписів. Ми використовуємо LoRA-адаптери для швидкої адаптації Whisper Large-v3 — навчання займає 2-3 дні на A100. Після fine-tuning Domain WER падає до 4-6%.

Як ми будуємо production-grade пайплайн STT

Використовуємо архітектуру з автоматичним fallback між провайдерами. Приклад реалізації на Python:

import asyncio import io import json from typing import AsyncGenerator, Optional import httpx import websockets import numpy as np class STTProviderComparator: """Порівняння провайдерів STT за метриками""" PROVIDERS = { "openai_whisper": { "wer_general": 0.05, "wer_noisy": 0.12, "russian_support": "excellent", "latency_ms": 600, "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, "streaming": False, "cost_per_hour": 0.08, }, } def recommend_provider(self, requirements: dict) -> str: """ Вибір провайдера за вимогами. requirements: {'streaming': bool, 'max_latency_ms': int, 'language': str, 'volume_hours_monthly': float} """ candidates = [] for name, props in self.PROVIDERS.items(): if requirements.get('streaming') and not props['streaming']: continue if props['latency_ms'] > requirements.get('max_latency_ms', 9999): continue wer_score = 1 - props['wer_general'] latency_score = 1 - props['latency_ms'] / 1000 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: """Потокове розпізнавання мовлення через 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]: """ Потокове розпізнавання аудіо. Повертає проміжні (interim) та фінальні транскрипти. """ 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" 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: """ Пост-обробка транскрипту для предметної області. STT-моделі часто помиляються на термінах, власних назвах, абревіатурах. """ def __init__(self, domain_vocabulary: dict): """ domain_vocabulary: {'вихідне_слово': 'правильне_слово'} Приклад: {'пітсбург': 'пітер', 'ексель': 'Excel'} """ self.vocabulary = {k.lower(): v for k, v in domain_vocabulary.items()} def correct_transcript(self, transcript: str) -> str: """Заміна помилково розпізнаних слів""" 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: """Нормалізація числівників та дат із тексту у структурований формат""" import re number_words = { 'нуль': '0', 'один': '1', 'два': '2', 'три': '3', 'чотири': '4', 'п’ять': '5', 'шість': '6', 'сім': '7', 'вісім': '8', 'дев’ять': '9', } result = transcript.lower() for word, digit in number_words.items(): result = result.replace(word, digit) return result class STTPipeline: """Повний пайплайн STT з fallback та моніторингом""" 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: """ Транскрипція з автоматичним 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 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: замінити на реальний виклик провайдера""" raise NotImplementedError 

Оцінка якості STT та KPI

Порівняємо метрики, які ми гарантуємо після впровадження (на ваших даних):

Процес роботи над STT-системою

1. Аналітика та збір даних — вивчаємо ваше акустичне середовище, збираємо вибірку аудіо з типовими акцентами та лексикою. Наприклад, для контакт-центру ми записуємо 50 годин реальних діалогів.
2. Вибір архітектури — визначаємо best-fit провайдер, необхідність self-hosted, проектуємо fallback та постобробку.
3. Розробка пайплайну — реалізуємо streaming-клієнт, кастомний словник, моніторинг метрик (WER, latency, error rates).
4. Тестування та fine-tuning — оптимізуємо WER на тестовій вибірці, донавчаємо модель при self-hosted.
5. Деплой та інтеграція — розгортаємо у вашому хмарі або on-prem, налаштовуємо CI/CD, надаємо доступ до API.
6. Навчання та підтримка — передаємо документацію, проводимо воркшоп для ваших інженерів, забезпечуємо SLA-підтримку.

Що входить у роботу (deliverables)

• Репозиторій з кодом пайплайну (Python, конфіги Docker/Kubernetes)
• Кастомний словник та скрипти його побудови
• Інструкція з розгортання та моніторингу
• Доступ до тестового стенду на 1 місяць
• Навчання команди (до 2 днів)
• Технічна підтримка на 3 місяці після релізу

Гарантуємо якість: вимірюємо WER на ваших даних до та після впровадження. Досягнуті цільові метрики фіксуємо в договорі.

Хочете побачити, як пайплайн працює на ваших даних? Замовте пілотний проєкт — ми розгорнемо систему у вашому середовищі та надамо звіт з метриками. Отримайте консультацію з архітектури STT — опишіть своє завдання, і ми запропонуємо оптимальне рішення.