Как построить production STT: архитектура, оптимизация и кейсы

Разработка системы Speech-to-Text: от выбора провайдера до 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-стек под вашу задачу?

Каждый провайдер имеет сильные и слабые стороны. Сравним ключевые метрики для русского языка:

Для потоковых сценариев (чат-боты, live-транскрипция) выбираем Deepgram или Azure. Для максимального качества при пакетной обработке — Whisper API или self-hosted. Self-hosted даёт экономию в 4-5 раз по сравнению с Deepgram при объёме от 500 часов в месяц, а также полный контроль над fine-tuning и privacy.

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

Любая 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, # Word Error Rate для стандартной речи "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 на A100 "streaming": False, "cost_per_hour": 0.08, # Self-hosted }, } 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(): # Фильтрация по 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 # Экономика при большом объёме 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" # ms тишины для детекции конца фразы 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 # Простые замены числительных (production: использовать pymorphy2) 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 — опишите свою задачу, и мы предложим оптимальное решение.