Как построить 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 — опишите свою задачу, и мы предложим оптимальное решение.

Распознавание и синтез речи: ASR, TTS, клонирование голоса

Заказчик приходит с задачей: транскрибировать 40 000 часов колл-центра за неделю. Штатный облачный ASR (Google Speech-to-Text) выдаёт WER 28% на отраслевой лексике и стоит ощутимо дорого при таких объёмах. Задача — снизить WER ниже 10% и перейти на self-hosted инференс.

Типовые проблемы, с которыми приходят

WER не сходится к нужной метрике. Чаще всего виновата не архитектура, а данные: шумные аудио без нормализации уровня (-23 LUFS вместо стандарта), смешанные языки в одном канале, акцент, специфическая доменная лексика. Whisper large-v3 из коробки даёт WER 8–12% на чистом русском и проваливается до 25–35% на записях с PSTN-артефактами и узкополосным кодеком G.711.

Диаризация ломается при больше двух спикеров. pyannote/speaker-diarization-3.1 работает стабильно при 2–3 говорящих, но DER (Diarization Error Rate) растёт с 6% до 18–22% при 5+ участниках конференции. Проблема усугубляется перекрёстными репликами: по умолчанию min_duration_on=0.1 срезает короткие вставки.

Клонирование голоса — латентность или качество. XTTS v2 (Coqui) даёт натуральный голос, но при потоковой генерации stream_chunk_size=20 первый аудиочанк прилетает через 1.4–2.0 с — неприемлемо для интерактивных сценариев. StyleTTS2 и Kokoro быстрее, но требуют точной подготовки референсного аудио.

Как это решается на практике

Базовый стек для production-пайплайна:

• ASR: openai/whisper-large-v3 или faster-whisper (CTranslate2-бэкенд, x4 скорость vs оригинал)
• Диаризация: pyannote.audio 3.x + интеграция через whisperx для выравнивания по словам
• TTS: XTTS v2 для качества, Edge-TTS или Silero для низкой латентности
• Клонирование: XTTS v2 (3–6 с референсного аудио) или OpenVoice v2

Типичный пайплайн для колл-центра выглядит так: аудио из очереди Kafka → нормализация ffmpeg -af loudnorm до -23 LUFS → faster-whisper с beam_size=5, vad_filter=True → pyannote диаризация → постпроцессинг (пунктуация через deepmultilingualpunctuation) → запись в PostgreSQL с временными метками.

Кейс из практики. Финтех-компания с 12 000 звонков/день. Исходный WER на русском с банковской лексикой — 22% (Google STT). После fine-tuning whisper-medium на 200 часах размеченных записей через Hugging Face transformers + Seq2SeqTrainer с learning_rate=1e-5, warmup_steps=500 — WER упал до 7.3%. Инференс на одной A10G через faster-whisper с compute_type=float16 обрабатывает 40-минутный звонок за 55 секунд. Итоговая стоимость инференса — $0.0008/мин против $0.016/мин у облачного провайдера.

Дообучение Whisper на доменных данных

Когда общая модель не справляется, fine-tuning — первый инструмент. Минимальный датасет для заметного улучшения — 20–30 часов размеченного аудио в целевом домене. Разметку можно получить через итеративный процесс: прогнать через базовую модель → вручную исправить 10–15% ошибок → переобучить → повторить.

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

Важно: при fine-tuning Whisper нужно замораживать encoder первые 1000 шагов (model.freeze_encoder()), иначе акустические признаки разъедутся раньше, чем decoder адаптируется к новой лексике.

Синтез речи: выбор под задачу

Для интерактивных ботов с требованием TTFB < 300 мс — Silero или Kokoro. Для озвучки контента, где важна натуральность — XTTS v2 с потоковой отдачей через WebSocket.

Процесс работы

Начинаем с аудит-сессии: берём 2–4 часа ваших записей, прогоняем через несколько моделей, замеряем WER/CER, смотрим на распределение ошибок по типам (лексические, акустические, язык). Это занимает 1–2 дня и сразу показывает, нужен ли fine-tuning или достаточно пост-обработки.

Далее — выбор архитектуры под ваш throughput: один GPU для 1000 мин/день или кластер с балансировщиком для 100 000+ мин/день. Деплой через Docker-контейнер с FastAPI или Triton Inference Server для батчированного инференса.

Сроки зависят от сложности: базовая интеграция готовой модели — 1–2 недели. Fine-tuning с подготовкой данных и валидацией — 4–8 недель. Полная разработка голосового пайплайна (ASR + диаризация + TTS + мониторинг) — 2–4 месяца.