Один STT-сервис дал 60-70% точности на специфической лексике (топонимы, названия улиц, профессиональные термины). Два сервиса параллельно + взвешенное голосование + AI-fusion для спорных случаев дали 95%+ точности. Время обработки 5-8 секунд, стоимость $70-130/месяц при 1000 сообщений в день. В статье — полный разбор архитектуры, алгоритмы scoring, примеры кода и расчёт экономики.

Содержание

1. Почему один STT оказалось недостаточно

2. Эволюция решения: от 60% к 95%

3. Архитектура Multi-API Ensemble

4. Взвешенное голосование: математика выбора

5. AI-fusion: когда голосования недостаточно

6. Постобработка: ловим систематические ошибки

7. Промпты для STT-сервисов

8. Мониторинг и graceful degradation

9. Результаты и метрики

10. Экономика решения

11. Выводы

1. Проблема: почему один STT недостаточно

Я разрабатывал систему транскрипции голосовых оповещений в ЧС для Telegram-канала. Задача казалась простой — Speech-to-Text API существуют больше десяти лет, технология зрелая.

Первая версия использовала один STT-сервис (SaluteSpeech от Сбера, он бесплатный для небольших объёмов). Запустили, протестировали на реальных данных. Результат: всего 60% точности.

Каждое второе сообщение содержало критические ошибки.

Анатомия ошибки

Типичное голосовое сообщение длиной 15 секунд:

«Внимание, оповещение для жителей Масычево, Илёк-Пеньковки и Мокрой Орловки. Просьба соблюдать осторожность на улице Заречной.»

А вот что возвращали разные STT-сервисы:

Обратите внимание: каждый сервис ошибается по-разному. SaluteSpeech превращает «Масычево» в «масло чего», Yandex - в «мамы чего», Google - в «мосычево». При этом Google правильно распознаёт «мокрой орловки», а первые два - нет.

Почему модели ошибаются на топонимах

Причина фундаментальная: STT-модели обучены на общей лексике. Датасеты для обучения содержат миллионы часов подкастов, аудиокниг, телефонных разговоров, YouTube-видео и редко названия деревень с населением 200 человек.

Как работает распознавание речи на высоком уровне:

1. Акустическая модель преобразует звуковой сигнал в вероятности фонем

2. Языковая модель выбирает наиболее вероятную последовательность слов

Проблема в языковой модели. Она обучена на текстах, где «масло» встречается миллионы раз, а «Масычево» ни разу. Когда акустическая модель даёт неоднозначный сигнал (а она всегда даёт неоднозначный), языковая модель выбирает знакомое слово.

Это классическая проблема OOV (Out-Of-Vocabulary). STT-системы плохо справляются со словами, которых не было в обучающих данных.

Проблема усугубляется для:

Составных топонимов: «Илёк-Пеньковка» - это один населённый пункт, но модель не знает этого. Она слышит «илёк» и думает: «странное слово, наверное ослышалась». Потом слышит «пеньковка» - тоже незнакомо. В итоге выдаёт «и лёгкая пеньковка», потому что «лёгкая» и «пеньковка» реальные слова.

Диалектных произношений: Местные жители произносят топонимы иначе, чем диктор на радио. «Козинка» может звучать как «Казинка» с редуцированным «о». Модель, обученная на стандартном произношении, теряется.

Редких слов в окружении частых: Контекст влияет на распознавание. «Улица Центральная» распознаётся лучше, чем «улица Заречная», потому что «центральная» частое слово, а «заречная» нет. Языковая модель «подтягивает» редкое слово к частому.

Аббревиатур и терминов: «FPV» произносится как «эф-пи-ви» или «фэ-пэ-вэ». Модель не знает, что это аббревиатура, и пытается найти похожие слова. Результат — «в пиве» или «и видео».

Цифр в контексте: «Дом 47» может стать «дом сорок семь» или «дом 4 7» или «дом 47». Модели нестабильны в форматировании чисел.

Пробовали (и почему не сработало)

Увеличить prompt/hints в Whisper:

Whisper от OpenAI поддерживает текстовый prompt до 224 токенов. Мы передавали список топонимов:

<source lang="python"> prompt = "Населённые пункты: Масычево, Илёк-Пеньковка, Мокрая Орловка, Козинка, Головчино..." </source>

Помогло частично — точность выросла с 60% до 70%. Но 224 токена — это ~50 слов. В нашем регионе 42 населённых пункта, плюс улицы, плюс термины. Всё не влезает.

Google Cloud Speech Adaptation:

Google предлагает механизм speech adaptation, можно загрузить список фраз с boost-коэффициентами. Потратил два дня. Результат: 72% точности, но стоимость $0.016/минута против $0.006 у Whisper. Прирост не оправдывает 2.5x цену.

Fine-tuning открытой модели:

Теоретически можно дообучить Whisper на своих данных. Практически это требует:

• 100+ часов размеченного аудио (у нас было 50 часов, но неразмеченных)

• GPU-инфраструктуру для обучения

• Время на эксперименты (недели)

В реальности денег на это нет и Fine-tuning оправдан при миллионах минут аудио в месяц.

2. Эволюция: от 60% к 95%

Ключевое открытие пришло из анализа ошибок: разные сервисы ошибаются на разных словах. Если SaluteSpeech не знает «Масычево», возможно Whisper или Gemini распознают его правильно.

Собрал статистику по 1000 сообщений: какой сервис на каком слове ошибся. Выяснилось, что пересечение ошибок всего 15%. То есть в 85% случаев хотя бы один сервис распознавал слово правильно.

Идея: запускаю несколько STT параллельно, сравниваем результаты, выбираем лучший.

Версия 2.0: четыре сервиса параллельно

Первая реализация была наивной: подключили все доступные API и запускали их одновременно.

Подключённые сервисы:

• SaluteSpeech (Сбер) - бесплатный до 5000 минут/месяц, поддерживает smart hints

• Yandex SpeechKit - хорошее качество на русском языке

• OpenAI Whisper - мультиязычный, стабильный

• Google Gemini - multimodal, понимает контекст, оплата по токенам

Я ввел понятие «консенсус»: когда все четыре сервиса возвращают одинаковый текст. На консенсусных фрагментах точность была 98%.

Но консенсус достигался только в 25% случаев. В остальных 75% разброс мнений.

Проблема 1: Latency

Каждый сервис отвечает за 2-5 секунд. При параллельном запуске через asyncio.gather общее время определяется самым медленным сервисом. SaluteSpeech периодически «задумывался» на 8-10 секунд. Yandex иногда отвечал за 12 секунд.

Итого: 14-20 секунд на одно сообщение. Для системы оповещений это неприемлемо. Пока система «думает», люди ждут критическую информацию.

Проблема 2: Стоимость

Четыре API-вызова на каждое сообщение = примерно 4x стоимость базового варианта.

При 1000 сообщениях в день:

• Whisper: $60/мес

• Yandex: $80/мес

• Gemini: $40/мес

• SaluteSpeech: бесплатно, но лимит быстро кончается

Прирост точности с 60% до 80% не оправдывал 4x увеличение стоимости.

Проблема 3: Сложность выбора

Когда сервисы расходятся во мнениях (а это 75% случаев), непонятно кому верить.

• Простое голосование «3 из 4»: не работает, часто расклад 2:2 или все четыре варианта разные

• Выбор самого длинного ответа: иногда модели галлюцинируют и добавляют несуществующие слова

• Выбор самого короткого: теряем информацию

• Случайный выбор: плохо

Версия 2.5: два лучших сервиса + ROVER

После нескольких недель сбора статистики я проанализировал, какие сервисы меньше ошибаются.

Методика анализа:

1. Собрал 500 размеченных примеров

2. Для каждого сервиса (с помощью Claude, дав ему контекст и данные) посчитал WER (Word Error Rate) по категориям: топонимы, термины, общая лексика

3. Построили матрицу корреляции ошибок: если сервис A ошибся, ошибся ли сервис B?

Результаты:

Корреляция ошибок Gemini и Whisper самая низкая (0.31). Это значит, что они ошибаются на разных словах и дополняют друг друга.

Yandex и SaluteSpeech сильно коррелируют с Whisper (0.67 и 0.58). Они не добавляют новой информации, если Whisper ошибся, скорее всего и они ошибутся.

Оставил только Gemini и Whisper. Два сервиса вместо четырёх.

Latency упала до 10-14 секунд (max из двух, а не четырёх). Стоимость вдвое ниже.

ROVER для объединения:

Для объединения результатов внедрил ROVER (Recognizer Output Voting Error Reduction) — классический алгоритм из speech recognition, разработанный в NIST в 1997 году.

ROVER работает так:

1. Выравнивает две транскрипции по словам (alignment)

2. Для каждой позиции голосует за вариант

3. При равенстве голосов выбирает первый вариант

Точность выросла до 90%. Но ROVER плохо справлялся со случаями, когда гипотезы сильно отличаются. Пример:

• Gemini: «Внимание, оповещение для Масычево»

• Whisper: «Внимание, оповещение для масло чего»

ROVER не понимает, что «Масычево» и «масло чего» - это одно и то же слово с ошибкой. Он видит разные слова и не может выбрать.

Версия 3.0: AI-fusion + постобработка

Финальная архитектура добавила два ключевых компонента:

1. AI-fusion: когда сервисы сильно расходятся (agreement < 70%), отправляем обе гипотезы в LLM. LLM получает контекст (список топонимов) и понимает, что «Масычево» - населённый пункт из списка, а «масло чего» - бессмыслица. LLM выбирает правильный вариант.

2. Постобработка: регулярные выражения для исправления систематических ошибок. База паттернов пополняется из логов. Если мы 10 раз видели ошибку «и лёгкая пеньковка» → «Илёк-Пеньковка», добавляем правило.

Также оптимизировал Gemini: перешли с Pro на Flash без потери качества, но с 3x ускорением.

Результат: 95%+ точности, 5-8 секунд latency, $70-130/месяц.

3. Архитектура Multi-API Ensemble

Общая схема

Почему параллельно, а не последовательно?

Latency критична. При последовательных запросах получаем сумму времени: 3-5 секунд Gemini + 3-5 секунд Whisper = 6-10 секунд. При параллельных максимум из двух: max(3-5, 3-5) = 3-5 секунд, плюс ~0.5 секунды на ensemble.

Обработка частичных отказов

Что если один из сервисов недоступен? Система должна работать с одним результатом:

# Фильтруем None (упавшие сервисы)
valid_results = {
    k: v for k, v in results.items() 
    if v is not None
}

if len(valid_results) == 0:
    raise TranscriptionError("All STT services failed")

if len(valid_results) == 1:
    # Один сервис — используем его результат
    single_result = list(valid_results.values())[0]
    logger.info(f"Single service mode: {list(valid_results.keys())[0]}")
    return self.postprocess(single_result.text)

# Два сервиса — полный ensemble pipeline
return await self.full_ensemble(valid_results)

4. Взвешенное голосование: математика выбора

Простое сравнение строк не работает. Даже на тривиальных фразах сервисы редко дают идентичный текст: разная пунктуация, регистр, пробелы.

Нужна метрика «качества» каждой транскрипции. Называем её confidence score.

Формула confidence score

    # Загружаем справочные данные
    self.toponyms = self._load_toponyms()
    self.domain_terms = self._load_domain_terms()
    self.emergency_patterns = self._compile_emergency_patterns()
    self.illogical_patterns = self._compile_illogical_patterns()

def calculate_confidence_score(
    self, 
    transcript: str, 
    service: str, 
    raw_confidence: float = 1.0
) -> float:
    """
    Рассчитывает confidence score для транскрипции.
    
    Учитывает:
    - Базовый вес сервиса
    - Количество распознанных топонимов (+2% за каждый)
    - Количество доменных терминов (+10% за каждый)
    - Экстренные фразы (+15% за каждую)
    - Нелогичные конструкции (-30% за каждую)
    
    Returns:
        float: score от 0.0 до 2.0
    """
    # Базовый score
    score = raw_confidence * self.service_weights.get(service, 1.0)
    
    # Бонус за топонимы
    toponym_count = self._count_toponyms(transcript)
    if toponym_count > 0:
        # +2% за каждый найденный топоним
        score *= (1 + 0.02 * toponym_count)
    
    # Бонус за доменные термины
    term_count = self._count_domain_terms(transcript)
    if term_count > 0:
        # +10% за каждый термин
        score *= (1 + 0.10 * term_count)
    
    # Бонус за экстренные фразы
    emergency_count = self._count_emergency_patterns(transcript)
    if emergency_count > 0:
        # +15% за каждую фразу
        score *= (1 + 0.15 * emergency_count)
    
    # Штраф за нелогичные конструкции
    illogical_count = self._count_illogical_patterns(transcript)
    if illogical_count > 0:
        # -30% за каждую (мультипликативно)
        score *= (0.7 ** illogical_count)
    
    # Потолок 2.0 чтобы один фактор не доминировал
    return min(score, 2.0)

Почему такие коэффициенты?

Коэффициенты подбирались эмпирически на размеченных данных (~500 примеров):

+2% за топоним - низкий бонус, потому что топонимы могут быть ложными срабатываниями. «Козинка» в контексте «корзинка» не должна сильно влиять на score.

+10% за доменный термин - выше, потому что термины редко появляются случайно. Если модель распознала «FPV» или «ПВО», скорее всего это правильно.

+15% за экстренную фразу - ещё выше, потому что устойчивые конструкции («отбой опасности», «просьба соблюдать осторожность») почти никогда не галлюцинируются.

-30% за нелогичную конструкцию - жёсткий штраф. «Масло чего» или «на двое суток опасности» — верный признак ошибки распознавания.

Справочные данные

def _load_toponyms(self) -> set[str]:
    """42 топонима региона"""
    return {
        "Козинка", "Казинка",  # варианты написания
        "Головчино", "Борисовка", "Масычево",
        "Мокрая Орловка", "Дорогощь", "Подол", "Гора-Подол",
        "Замостье", "Глотово", "Доброивановка", "Теребрено",
        "Безымено", "Некрасово", "Серетино", "Нехотеевка",
        "Журавлёвка", "Сподарюшино", "Новостроевка",
        "Илёк-Пеньковка", "Лёвкино", "Левкино", "Пеньково",
        # ... остальные
    }

def _load_domain_terms(self) -> set[str]:
    """Профессиональные термины и аббревиатуры"""
    return {
        "FPV", "БПЛА", "ПВО", "РЭБ", "РСЗО",
        "БТР", "БМП", "САУ", "ЗРК",
        # ... остальные
    }

def _compile_emergency_patterns(self) -> list[re.Pattern]:
    """Паттерны экстренных фраз"""
    patterns = [
        r'\bотбой\s+(опасности|тревоги|угрозы)\b',
        r'\bвнимание[,!]?\s+оповещение\b',
        r'\bпросьба\s+соблюдать\s+осторожность\b',
        r'\bопасность\s+(атаки|угрозы)\b',
        r'\bприближается\b',
        r'\bобнаружен[аы]?\b',
    ]
    return [re.compile(p, re.IGNORECASE) for p in patterns]

def _compile_illogical_patterns(self) -> list[re.Pattern]:
    """Паттерны бессмысленных конструкций (признак ошибки)"""
    patterns = [
        r'\bна\s+двое\s+суток\s+опасности\b',  # «отбой» → «на двое суток»
        r'\bотбоя\s+от\s+опасности\b',          # грамматически неверно
        r'\bлёгкая\s+пеньковка\b',              # разбитый топоним
        r'\bи\s+лёгкая\s+пеньковка\b',          # ещё вариант
        r'\bмасло\s+чего\b',                     # «Масычево» → бред
        r'\bмамы\s+чего\b',                      # ещё вариант
        r'\bмокрой\s+орлов\b',                   # разбитый топоним
    ]
    return [re.compile(p, re.IGNORECASE) for p in patterns]

Методы подсчёта

def _count_toponyms(self, text: str) -> int:
    """Считает количество топонимов в тексте"""
    text_lower = text.lower()
    count = 0
    for toponym in self.toponyms:
        if toponym.lower() in text_lower:
            count += 1
    return count

def _count_domain_terms(self, text: str) -> int:
    """Считает количество доменных терминов"""
    text_upper = text.upper()
    count = 0
    for term in self.domain_terms:
        if term.upper() in text_upper:
            count += 1
    return count

def _count_emergency_patterns(self, text: str) -> int:
    """Считает количество экстренных фраз"""
    return sum(1 for p in self.emergency_patterns if p.search(text))

def _count_illogical_patterns(self, text: str) -> int:
    """Считает количество нелогичных конструкций"""
    return sum(1 for p in self.illogical_patterns if p.search(text))

Алгоритм выбора победителя

def weighted_voting(
    self, 
    results: dict[str, TranscriptionResult]
) -> Optional[str]:
    """
    Выбирает лучшую транскрипцию на основе confidence scores.
    
    Returns:
        str: текст победителя, если есть явный лидер
        None: если нужен AI-fusion
    """
    scored = []
    for service, result in results.items():
        score = self.calculate_confidence_score(
            result.text, 
            service, 
            result.confidence
        )
        scored.append({
            'service': service,
            'text': result.text,
            'score': score,
            'raw_confidence': result.confidence,
        })
        logger.debug(f"{service}: score={score:.3f}, text={result.text[:50]}...")
    
    # Сортируем по убыванию score
    scored.sort(key=lambda x: x['score'], reverse=True)
    
    if len(scored) < 2:
        return scored[0]['text'] if scored else None
    
    top = scored[0]
    second = scored[1]
    
    # Clear winner: лидер на 15%+ выше второго
    if top['score'] > second['score'] * 1.15:
        logger.info(
            f"Clear winner: {top['service']} "
            f"(score={top['score']:.3f} vs {second['score']:.3f})"
        )
        return top['text']
    
    # Нет явного победителя — нужен AI-fusion
    logger.info(
        f"No clear winner: {top['service']}={top['score']:.3f}, "
        f"{second['service']}={second['score']:.3f}"
    )
    return None

5. AI-fusion: когда голосования недостаточно

В 30-40% случаев оба сервиса уверены в своих результатах, но результаты разные. Пример:

Разница 3% меньше порога в 15%.

Идея AI-fusion

Отправляем обе гипотезы в LLM вместе с контекстом (списком топонимов). LLM понимает:

• «Козинка» есть в списке топонимов

• «Казинка» — нет (хотя фонетически похоже)

• Следовательно, правильный вариант — «Козинка»

Реализация

python

class GeminiCombiner:
    """
    AI-fusion через Gemini для объединения спорных транскрипций.
    """
    
    def __init__(self, api_key: str):
        genai.configure(api_key=api_key)
        self.model = genai.GenerativeModel('gemini-2.5-flash')
        self.toponyms_context = self._load_toponyms_context()
    
    async def combine_transcriptions(
        self, 
        transcripts: list[str]
    ) -> str:
        """
        Объединяет несколько транскрипций в одну.
        
        Args:
            transcripts: список транскрипций от разных сервисов
            
        Returns:
            str: объединённая транскрипция
        """
        prompt = self._build_prompt(transcripts)
        
        generation_config = {
            "temperature": 0.1,  # Низкая для детерминизма
            "top_p": 0.95,
            "max_output_tokens": 1024,
        }
        
        try:
            response = await self.model.generate_content_async(
                prompt,
                generation_config=generation_config,
            )
            return self._parse_response(response.text)
        except Exception as e:
            logger.error(f"Gemini Combiner failed: {e}")
            # Fallback на ROVER
            return self.rover_fallback(transcripts)
    
    def _build_prompt(self, transcripts: list[str]) -> str:
        transcript_list = "\n".join(
            f"ВАРИАНТ {i+1}: {t}" 
            for i, t in enumerate(transcripts)
        )
        
        return f"""Ты эксперт по транскрипции аудио для службы экстренных оповещений.

Даны несколько вариантов транскрипции одного аудиофрагмента от разных сервисов:

{transcript_list}

КОНТЕКСТ — известные топонимы региона:
{self.toponyms_context}

ЗАДАЧА:
1. Сравни все варианты транскрипции слово за словом
2. Для каждого расхождения выбери наиболее вероятный вариант
3. Учитывай список топонимов — если слово есть в списке, предпочитай его
4. Составные топонимы (через дефис) не разбивай: "Илёк-Пеньковка" — одно слово

ПРАВИЛА:
- НЕ добавляй слова, которых нет ни в одном варианте
- НЕ исправляй грамматику, только выбирай между вариантами
- Если все варианты одинаковы — просто верни этот текст
- "Козинка" вероятнее чем "Казинка" (есть в базе топонимов)
- "Масычево" вероятнее чем "масло чего" (есть в базе)

Верни ТОЛЬКО итоговую транскрипцию, без объяснений:"""
    
    def _parse_response(self, response: str) -> str:
        """Извлекает текст из ответа, убирая возможные артефакты"""
        # Убираем markdown-форматирование если есть
        text = response.strip()
        text = re.sub(r'^```.*?\n', '', text)
        text = re.sub(r'\n```$', '', text)
        text = re.sub(r'^\*\*', '', text)
        text = re.sub(r'\*\*$', '', text)
        return text.strip()
    
    def rover_fallback(self, transcripts: list[str]) -> str:
        """
        ROVER алгоритм как fallback если Gemini недоступен.
        Простое word-level голосование.
        """
        # Токенизация
        token_lists = [t.split() for t in transcripts]
        
        # Находим максимальную длину
        max_len = max(len(tokens) for tokens in token_lists)
        
        # Паддинг до одинаковой длины
        padded = [
            tokens + [''] * (max_len - len(tokens)) 
            for tokens in token_lists
        ]
        
        # Голосование по позициям
        result = []
        for i in range(max_len):
            candidates = [tokens[i] for tokens in padded if tokens[i]]
            if candidates:
                # Берём самый частый вариант
                winner = max(set(candidates), key=candidates.count)
                result.append(winner)
        
        return ' '.join(result)

Когда вызывается AI-fusion

async def full_ensemble(
    self, 
    results: dict[str, TranscriptionResult]
) -> str:
    """
    Полный ensemble pipeline для двух результатов.
    """
    # Шаг 1: Пробуем weighted voting
    winner = self.weighted_voting(results)
    
    if winner is not None:
        # Есть явный победитель
        return self.postprocess(winner)
    
    # Шаг 2: Проверяем agreement
    texts = [r.text for r in results.values()]
    agreement = self._calculate_agreement(texts[0], texts[1])
    
    logger.info(f"Agreement: {agreement:.1%}")
    
    if agreement >= 0.7:
        # Высокий agreement — берём результат с лучшим score
        best = max(results.values(), key=lambda r: r.confidence)
        return self.postprocess(best.text)
    
    # Шаг 3: Низкий agreement — AI-fusion
    logger.info("Low agreement, using AI-fusion")
    combined = await self.combiner.combine_transcriptions(texts)
    
    return self.postprocess(combined)

def _calculate_agreement(self, text1: str, text2: str) -> float:
    """
    Рассчитывает степень совпадения двух текстов.
    Использует коэффициент Жаккара на уровне слов.
    """
    words1 = set(text1.lower().split())
    words2 = set(text2.lower().split())
    
    if not words1 and not words2:
        return 1.0
    
    intersection = len(words1 & words2)
    union = len(words1 | words2)
    
    return intersection / union if union > 0 else 0.0

6. Постобработка: ловим систематические ошибки

Даже после AI-fusion остаются повторяющиеся ошибки. Постобработка исправляет их детерминированно — без вызовов API, за микросекунды.

Три типа постобработки

class PostProcessor:
    """
    Постобработка транскрипций.
    Исправляет систематические ошибки без вызовов API.
    """
    
    def __init__(self):
        self.toponym_fixes = self._load_toponym_fixes()
        self.term_normalizations = self._load_term_normalizations()
        self.phonetic_fixes = self._load_phonetic_fixes()
    
    def process(self, text: str) -> str:
        """Применяет все исправления последовательно"""
        text = self._fix_split_toponyms(text)
        text = self._normalize_terms(text)
        text = self._fix_phonetic_errors(text)
        text = self._fix_punctuation(text)
        return text
    
    # === 1. Склейка разбитых топонимов ===
    
    def _load_toponym_fixes(self) -> dict[str, str]:
        return {
            r'\bлёгкая\s+пеньковка\b': 'Илёк-Пеньковка',
            r'\bи\s+лёгкая\s+пеньковка\b': 'Илёк-Пеньковка',
            r'\bилёк\s+пеньковка\b': 'Илёк-Пеньковка',
            r'\bмокрой\s+орлов\b': 'Мокрая Орловка',
            r'\bмокрая\s+орлов\b': 'Мокрая Орловка',
            r'\bгора\s+подол\b': 'Гора-Подол',
        }
    
    def _fix_split_toponyms(self, text: str) -> str:
        for pattern, replacement in self.toponym_fixes.items():
            text = re.sub(pattern, replacement, text, flags=re.IGNORECASE)
        return text
    
    # === 2. Нормализация терминов ===
    
    def _load_term_normalizations(self) -> dict[str, str]:
        return {
            # Аббревиатуры произнесённые по буквам
            r'\bэф\s*пи\s*ви\b': 'FPV',
            r'\bэфпиви\b': 'FPV',
            r'\bф\s*п\s*в\b': 'FPV',
            r'\bбэ\s*пэ\s*эл\s*а\b': 'БПЛА',
            r'\bбпэла\b': 'БПЛА',
            r'\bпэ\s*вэ\s*о\b': 'ПВО',
            r'\bрэ\s*б\b': 'РЭБ',
            r'\bэр\s*эс\s*зэ\s*о\b': 'РСЗО',
            
            # Транслитерация
            r'\bхимарс\b': 'HIMARS',
            r'\bхаймарс\b': 'HIMARS',
        }
    
    def _normalize_terms(self, text: str) -> str:
        for pattern, replacement in self.term_normalizations.items():
            text = re.sub(pattern, replacement, text, flags=re.IGNORECASE)
        return text
    
    # === 3. Исправление фонетических ошибок ===
    
    def _load_phonetic_fixes(self) -> dict[str, str]:
        return {
            # Частые фонетические ошибки
            r'\bна\s+двое\s+суток\s+опасности\b': 'отбой опасности',
            r'\bотбоя\s+от\s+опасности\b': 'отбой опасности',
            r'\bотбоя\s+опасности\b': 'отбой опасности',
            
            # Искажения топонимов
            r'\bмасло\s+чего\b': 'Масычево',
            r'\bмамы\s+чего\b': 'Масычево',
            r'\bБезыменно\b': 'Безымено',
            r'\bКазинки\b': 'Козинка',
        }
    
    def _fix_phonetic_errors(self, text: str) -> str:
        for pattern, replacement in self.phonetic_fixes.items():
            text = re.sub(pattern, replacement, text, flags=re.IGNORECASE)
        return text
    
    # === 4. Пунктуация ===
    
    def _fix_punctuation(self, text: str) -> str:
        # Убираем двойные пробелы
        text = re.sub(r'\s+', ' ', text)
        # Убираем пробелы перед запятыми
        text = re.sub(r'\s+,', ',', text)
        # Добавляем пробел после запятой если нет
        text = re.sub(r',(?=\S)', ', ', text)
        return text.strip()

Пополнение словаря исправлений

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

def log_correction(
    original: str, 
    corrected: str, 
    correction_type: str
):
    """Логирует исправление для последующего анализа"""
    entry = {
        'timestamp': datetime.utcnow().isoformat(),
        'original': original,
        'corrected': corrected,
        'type': correction_type,
    }
    
    with open('corrections_log.jsonl', 'a') as f:
        f.write(json.dumps(entry, ensure_ascii=False) + '\n')

За 4 месяца накопилось 331 запись. Пример:

---

7. Промпты для STT-сервисов

Gemini STT

Gemini поддерживает multimodal input аудио и текст в одном запросе. Это позволяет передать развёрнутый контекст.

class GeminiSTTClient:
    def __init__(self, api_key: str):
        genai.configure(api_key=api_key)
        self.model = genai.GenerativeModel('gemini-2.5-flash')
        self.toponyms = self._load_toponyms()
    
    async def transcribe_with_context(
        self, 
        audio_path: str
    ) -> TranscriptionResult:
        """
        Транскрибирует аудио с контекстным промптом.
        """
        # Читаем аудио
        with open(audio_path, 'rb') as f:
            audio_content = f.read()
        
        # Определяем формат
        audio_format = self._detect_format(audio_path)
        
        # Создаём multimodal prompt
        prompt = self._build_prompt()
        
        audio_part = {
            "inline_data": {
                "mime_type": f"audio/{audio_format}",
                "data": base64.b64encode(audio_content).decode('utf-8')
            }
        }
        
        generation_config = {
            "temperature": 0.1,
            "top_p": 0.95,
            "top_k": 40,
            "max_output_tokens": 8192,
        }
        
        response = await self.model.generate_content_async(
            [prompt, audio_part],
            generation_config=generation_config,
        )
        
        return self._parse_response(response.text)
    
    def _build_prompt(self) -> str:
        toponym_list = ', '.join(sorted(self.toponyms)[:30])
        
        return f"""Вы профессиональный транскрибатор для службы экстренных оповещений.

Транскрибируйте это аудио максимально точно, уделяя особое внимание:

1. НАЗВАНИЯМ НАСЕЛЁННЫХ ПУНКТОВ (см. список ниже)
2. Профессиональным терминам и аббревиатурам
3. Числам и направлениям
4. Названиям улиц

ВАЖНЫЕ НАСЕЛЁННЫЕ ПУНКТЫ РЕГИОНА:
{toponym_list}

ВАЖНО:
- Составные названия пишите через дефис: Илёк-Пеньковка, Гора-Подол, Мокрая Орловка
- Аббревиатуры пишите заглавными: FPV, БПЛА, ПВО, РЭБ
- Если слово похоже на топоним из списка — используйте написание из списка

Верните результат в формате JSON:
{{
  "transcript": "точная транскрипция аудио",
  "toponyms_found": ["список", "найденных", "топонимов"],
  "confidence": "high/medium/low",
  "unclear_parts": ["неразборчивые фрагменты если есть"]
}}"""
    
    def _parse_response(self, response: str) -> TranscriptionResult:
        """Парсит JSON-ответ от Gemini"""
        try:
            # Убираем markdown если есть
            clean = re.sub(r'^```json\s*', '', response)
            clean = re.sub(r'\s*```$', '', clean)
            
            data = json.loads(clean)
            
            confidence_map = {'high': 0.95, 'medium': 0.75, 'low': 0.5}
            
            return TranscriptionResult(
                text=data.get('transcript', ''),
                confidence=confidence_map.get(data.get('confidence', 'medium'), 0.75),
                service='gemini',
                raw_response=data,
            )
        except json.JSONDecodeError:
            # Если не JSON — берём как plain text
            return TranscriptionResult(
                text=response.strip(),
                confidence=0.6,
                service='gemini',
                raw_response={'raw': response},
            )

Whisper

Whisper принимает prompt до 224 токенов. Используем его для подсказок:

class WhisperSTTClient:
    def __init__(self, api_key: str):
        self.client = OpenAI(api_key=api_key)
        self.toponyms = self._load_toponyms()
    
    async def transcribe(self, audio_path: str) -> TranscriptionResult:
        """
        Транскрибирует аудио через Whisper API.
        """
        # Формируем prompt (до 224 токенов)
        toponym_sample = ', '.join(list(self.toponyms)[:15])
        
        prompt = (
            f"Экстренное оповещение для населения. "
            f"Могут упоминаться населённые пункты: {toponym_sample}. "
            f"Термины: FPV, БПЛА, ПВО, РЭБ. "
            f"Составные названия: Илёк-Пеньковка, Мокрая Орловка."
        )
        
        # Whisper API синхронный, оборачиваем в executor
        loop = asyncio.get_event_loop()
        response = await loop.run_in_executor(
            None,
            lambda: self._sync_transcribe(audio_path, prompt)
        )
        
        return response
    
    def _sync_transcribe(
        self, 
        audio_path: str, 
        prompt: str
    ) -> TranscriptionResult:
        with open(audio_path, 'rb') as audio_file:
            response = self.client.audio.transcriptions.create(
                model="whisper-1",
                file=audio_file,
                language="ru",
                prompt=prompt,
                response_format="verbose_json",
            )
        
        return TranscriptionResult(
            text=response.text,
            confidence=self._estimate_confidence(response),
            service='whisper',
            raw_response=response.model_dump(),
        )
    
    def _estimate_confidence(self, response) -> float:
        """
        Оценивает уверенность на основе метаданных ответа.
        Whisper не возвращает confidence напрямую, но можно оценить
        по наличию no_speech_prob и avg_logprob в сегментах.
        """
        if not hasattr(response, 'segments') or not response.segments:
            return 0.7  # Дефолт
        
        # Средний avg_logprob по сегментам
        logprobs = [
            s.get('avg_logprob', -0.5) 
            for s in response.segments 
            if 'avg_logprob' in s
        ]
        
        if not logprobs:
            return 0.7
        
        avg_logprob = sum(logprobs) / len(logprobs)
        
        # Преобразуем logprob в confidence [0, 1]
        # avg_logprob обычно от -1.0 (плохо) до 0 (идеально)
        confidence = max(0.0, min(1.0, 1.0 + avg_logprob))
        
        return confidence

8. Мониторинг и graceful degradation

Логирование

Каждый этап pipeline логируется для отладки:

import logging
import json
from datetime import datetime

# Настройка логгера
logger = logging.getLogger('voice_transcription')
logger.setLevel(logging.DEBUG)

# Форматтер с JSON для структурированных логов
class JsonFormatter(logging.Formatter):
    def format(self, record):
        log_data = {
            'timestamp': datetime.utcnow().isoformat(),
            'level': record.levelname,
            'message': record.getMessage(),
            'module': record.module,
            'function': record.funcName,
        }
        if hasattr(record, 'extra_data'):
            log_data.update(record.extra_data)
        return json.dumps(log_data, ensure_ascii=False)

# Пример использования
def log_transcription_result(
    message_id: str,
    results: dict,
    winner: str,
    final_text: str,
    processing_time: float
):
    logger.info(
        "Transcription completed",
        extra={'extra_data': {
            'message_id': message_id,
            'services': list(results.keys()),
            'winner': winner,
            'text_length': len(final_text),
            'processing_time_ms': int(processing_time * 1000),
            'gemini_score': results.get('gemini', {}).get('score'),
            'whisper_score': results.get('whisper', {}).get('score'),
        }}
    )

Метрики для мониторинга

python

from dataclasses import dataclass, field
from collections import defaultdict
import time

@dataclass
class TranscriptionMetrics:
    """Метрики для мониторинга качества транскрипции"""
    
    total_requests: int = 0
    successful_requests: int = 0
    failed_requests: int = 0
    
    # По сервисам
    service_calls: dict = field(default_factory=lambda: defaultdict(int))
    service_failures: dict = field(default_factory=lambda: defaultdict(int))
    service_latencies: dict = field(default_factory=lambda: defaultdict(list))
    
    # Ensemble статистика
    clear_winner_count: int = 0
    ai_fusion_count: int = 0
    rover_fallback_count: int = 0
    
    # Timing
    total_processing_time: float = 0.0
    
    def record_request(
        self, 
        success: bool,
        services_used: list[str],
        services_failed: list[str],
        latencies: dict[str, float],
        used_ai_fusion: bool,
        used_rover: bool,
        processing_time: float
    ):
        self.total_requests += 1
        
        if success:
            self.successful_requests += 1
        else:
            self.failed_requests += 1
        
        for service in services_used:
            self.service_calls[service] += 1
            if service in latencies:
                self.service_latencies[service].append(latencies[service])
        
        for service in services_failed:
            self.service_failures[service] += 1
        
        if used_ai_fusion:
            self.ai_fusion_count += 1
        elif used_rover:
            self.rover_fallback_count += 1
        else:
            self.clear_winner_count += 1
        
        self.total_processing_time += processing_time
    
    def get_summary(self) -> dict:
        avg_time = (
            self.total_processing_time / self.total_requests 
            if self.total_requests > 0 else 0
        )
        
        return {
            'total_requests': self.total_requests,
            'success_rate': self.successful_requests / max(self.total_requests, 1),
            'avg_processing_time_ms': int(avg_time * 1000),
            'clear_winner_rate': self.clear_winner_count / max(self.total_requests, 1),
            'ai_fusion_rate': self.ai_fusion_count / max(self.total_requests, 1),
            'service_failure_rates': {
                service: self.service_failures[service] / max(self.service_calls[service], 1)
                for service in self.service_calls
            },
            'avg_latencies_ms': {
                service: int(sum(lats) / len(lats) * 1000) if lats else 0
                for service, lats in self.service_latencies.items()
            },
        }

Graceful degradation

class ResilientEnsemble:
    """
    Ensemble с graceful degradation при отказах.
    """
    
    def __init__(self, gemini_client, whisper_client, combiner):
        self.gemini = gemini_client
        self.whisper = whisper_client
        self.combiner = combiner
        self.metrics = TranscriptionMetrics()
        
        # Circuit breaker state
        self.circuit_state = {
            'gemini': {'failures': 0, 'last_failure': None, 'open': False},
            'whisper': {'failures': 0, 'last_failure': None, 'open': False},
        }
        self.failure_threshold = 5
        self.recovery_timeout = 60  # секунд
    
    def _is_circuit_open(self, service: str) -> bool:
        """Проверяет, открыт ли circuit breaker для сервиса"""
        state = self.circuit_state[service]
        
        if not state['open']:
            return False
        
        # Проверяем timeout для восстановления
        if state['last_failure']:
            elapsed = time.time() - state['last_failure']
            if elapsed > self.recovery_timeout:
                # Пробуем восстановить
                state['open'] = False
                state['failures'] = 0
                logger.info(f"Circuit breaker closed for {service}")
                return False
        
        return True
    
    def _record_failure(self, service: str):
        """Записывает отказ сервиса"""
        state = self.circuit_state[service]
        state['failures'] += 1
        state['last_failure'] = time.time()
        
        if state['failures'] >= self.failure_threshold:
            state['open'] = True
            logger.warning(f"Circuit breaker opened for {service}")
    
    def _record_success(self, service: str):
        """Записывает успех — сбрасывает счётчик отказов"""
        self.circuit_state[service]['failures'] = 0
    
    async def transcribe(self, audio_path: str) -> str:
        """
        Транскрибирует с учётом состояния circuit breakers.
        """
        start_time = time.time()
        services_to_use = []
        
        # Определяем доступные сервисы
        if not self._is_circuit_open('gemini'):
            services_to_use.append(('gemini', self.gemini.transcribe_with_context))
        
        if not self._is_circuit_open('whisper'):
            services_to_use.append(('whisper', self.whisper.transcribe))
        
        if not services_to_use:
            raise TranscriptionError("All services unavailable (circuit breakers open)")
        
        # Запускаем доступные сервисы
        tasks = [
            asyncio.create_task(func(audio_path), name=name)
            for name, func in services_to_use
        ]
        
        results_raw = await asyncio.gather(*tasks, return_exceptions=True)
        
        # Обрабатываем результаты
        results = {}
        latencies = {}
        services_failed = []
        
        for task, result in zip(tasks, results_raw):
            service = task.get_name()
            
            if isinstance(result, Exception):
                logger.error(f"{service} failed: {result}")
                self._record_failure(service)
                services_failed.append(service)
            else:
                self._record_success(service)
                results[service] = result
                # Latency примерная (общее время / кол-во сервисов)
                latencies[service] = time.time() - start_time
        
        # ... остальная логика ensemble ...
        
        processing_time = time.time() - start_time
        
        # Записываем метрики
        self.metrics.record_request(
            success=len(results) > 0,
            services_used=[s for s, _ in services_to_use],
            services_failed=services_failed,
            latencies=latencies,
            used_ai_fusion=used_ai_fusion,
            used_rover=used_rover,
            processing_time=processing_time,
        )
        
        return final_text

9. Результаты и метрики

Методология измерения

Для оценки качества мы использовал 500 размеченных примеров: аудио + ручная транскрипция. Примеры собирал в течение месяца, которые покрывали разные условия:

• Разное качество микрофона (телефон, гарнитура)

• Разная длина сообщений (от 5 до 30 секунд)

• Разный фоновый шум (тихо, улица, помещение с эхом)

• Разные дикторы (мужские/женские голоса, разный темп речи)

Основная метрика: WER (Word Error Rate)

WER = (S + I + D) / N

где:
S = количество замен (слово распознано неправильно)
I = количество вставок (лишнее слово)
D = количество удалений (слово пропущено)
N = общее количество слов в ground truth

WER 5% означает, что в среднем 5 слов из 100 распознаны неправильно.

Сравнение точности по категориям

Детальные примеры распознавания

Пример 1: Топоним «Масычево»

Пример 2: Составной топоним «Илёк-Пеньковка»

Пример 3: Фраза «отбой опасности» (сложный случай)

Эта фраза интересна тем, что SaluteSpeech систематически ошибался одинаково:

Даже если бы Whisper и Gemini оба ошиблись, постобработка содержит правило:

r'\bна\s+двое\s+суток\s+опасности\b': 'отбой опасности'

Пример 4: Аббревиатура в контексте

Исходное аудио: «Обнаружен FPV дрон в районе Козинки»

Gemini понял из контекста, что «эф пи ви» - это аббревиатура FPV. Whisper распознал по буквам, но постобработка всё равно исправила бы.

Метрики production-системы (4 месяца работы)

Динамика по месяцам:

Точность растёт, latency падает система «учится» на своих ошибках через пополнение словарей постобработки.

10. Экономика решения

Расчёт для разных нагрузок

Примечание: расчёт для среднего сообщения 15-20 секунд. Для более длинных аудио стоимость выше.

Сравнение с альтернативами

ROI (если использовать для коммерции)

Ручная расшифровка одного сообщения занимает 1-2 минуты у оператора. При зарплате оператора 60,000₽/месяц и 1000 сообщениях в день:

• Время на ручную работу: 1000 × 1.5 мин × 30 дней = 750 часов/месяц

• Требуется операторов: 750 / 160 = ~4.7

• Стоимость: ~280,000₽/месяц

С этой системой:

• Автоматизировано: 95%

• Ручная работа: 50 сообщений/день × 1.5 мин = 37.5 часов/месяц

• Требуется операторов: 0.25

• Стоимость системы: ~$110 ≈ 11,000₽

• Стоимость оператора: ~15,000₽/месяц

• Итого: ~26,000₽/месяц vs 280,000₽/месяц

Экономия: 90% или ~250,000₽/месяц.

---

FAQ:

Q: Почему не использовать только Whisper с большим prompt?

A: Whisper ограничен 224 токенами в prompt - это ~50 слов. Если у вас 40+ топонимов, 20+ терминов и примеры фраз - не влезет. Кроме того, prompt в Whisper - это подсказка, а не инструкция. Модель может проигнорировать его.

Gemini позволяет передать полный контекст (тысячи токенов) и даёт структурированный ответ с confidence. Комбинация двух подходов работает лучше, чем любой из них отдельно.

Q: Сколько топонимов можно передать в контекст?

A: Для Gemini - практически без ограничений. Мы передаём 42 топонима + 15 терминов + примеры фраз. Это ~500 токенов, стоимость минимальна.

Для Whisper лучше ограничиться 15-20 самыми важными (частыми) словами.

Q: Что если один сервис постоянно недоступен?

A: Система работает в режиме graceful degradation. Если Gemini недоступен, используем только Whisper (точность падает до ~75%). Если Whisper недоступен - только Gemini (~80%). Circuit breaker автоматически отключает нестабильный сервис и пробует восстановить через минуту.

Q: Как часто нужно обновлять словарь топонимов?

A: Зависит от домена. Для географии — почти никогда (населённые пункты не появляются каждый день). Для продуктовой поддержки — чаще (новые фичи, продукты).

Q: Можно ли использовать подход для других языков?

A: Да. Whisper поддерживает 100+ языков. Gemini — основные мировые языки. Принцип ensemble работает независимо от языка.

Качество STT на разных языках разное. Для английского базовая точность выше (95%+), ensemble даст меньший прирост. Для редких языков — ensemble критичен.

Q: Как масштабируется решение?

A: Линейно. 10x сообщений = 10x стоимость.

При очень высоких нагрузках (10,000+ msg/день) имеет смысл посмотреть на self-hosted Whisper и что-то еще на GPU.

Q: Почему не fine-tuning?

A: Fine-tuning требует:

• 100+ часов размеченного аудио

• GPU-инфраструктуру

• Время на эксперименты

• Поддержку модели (ретрейнинг при изменении данных)

Денег и времени на. это нет. Ensemble даёт сравнимое качество за $100/месяц без DevOps.

Q: Как измерять качество?

A: WER (Word Error Rate) — стандартная метрика. Формула: (Substitutions + Insertions + Deletions) / Total Words.

WER отдельно по категориям:

• Общий WER

• WER на топонимах

• WER на терминах

• WER на числах

Это позволяет видеть, где система слабая.

Q: Что делать с очень длинными аудио (5+ минут)?

A: Разбивать на чанки по 30-60 секунд. Whisper имеет лимит 25MB на файл. Gemini 20MB.

Для разбиения использовать VAD (Voice Activity Detection) режем по паузам, чтобы не разрывать слова.

Ошибки при внедрении

Ошибка 1: Слепое доверие одному сервису

Whisper обучен на общих данных. Он не знает ваших топонимов, терминов, специфики. Всегда измеряйте качество на своих данных.

Ошибка 2: Слишком много сервисов

Больше сервисов = больше latency + стоимость + сложность выбора. Анализируйте корреляцию ошибок.

Ошибка 3: Игнорирование постобработки

«ML должен всё решить, regex — это прошлый век.»

ML-модели делают систематические ошибки. Если вы видите одну и ту же ошибку 10 раз, проще добавить правило замены, чем переобучать модель.

Ошибка 4: Отсутствие логирования

Без логов вы не узнаете о проблемах, пока пользователи не пожалуются. Логируйте: исходный текст, результат каждого сервиса, финальный результат, время обработки.

Ошибка 5: Одинаковые промпты для разных сервисов

«Скопирую промпт из Gemini в Whisper.»

Сервисы по-разному интерпретируют промпты. Whisper ожидает короткую подсказку (224 токена). Gemini развёрнутую инструкцию.

Ошибка 6: Фиксированные пороги

«Agreement > 70% — берём консенсус, иначе AI-fusion.»

Оптимальные пороги зависят от данных. 70% работает для нас, для вас может быть 60% или 80%. A/B тестируйте.

Ошибка 7: Игнорирование стоимости

«Gemini Pro точнее, будем использовать его.»

Gemini Pro в 4 раза дороже Flash при сопоставимом качестве для STT. Всегда считайте ROI.

Ретроспектива: что бы сделал иначе

1. Начать бы с анализа ошибок, а не с подключения сервисов.

Потратил две недели на интеграцию четырёх STT, а потом выяснили, что два из них бесполезны. Если бы начал с 500 размеченных примеров и анализа WER, сэкономили бы время.

Сначала данные, потом архитектура.

2. Раньше внедрили бы AI-fusion.

ROVER — хороший алгоритм, но он не понимает семантику. Неделю пытался его тюнить, добавлял веса, эвристики. В итоге один промпт в Gemini решил проблему лучше, чем все эвристики.

Урок: Не бойтесь использовать LLM для «склейки».

3. Раньше перешли бы на Gemini Flash.

Первые три недели использовал Gemini Pro. Он медленнее в 3 раза и дороже в 4 раза. Качество транскрипции. Думал, что «Pro лучше», но это не так для задачи STT.

Тестируйте младшие модели. Часто они достаточны.

Что дальше можно сделать

Снижение стоимости:

• Gemini 2.5 Flash-Lite ($0.10 input, $0.40 output) - потенциальная экономия 50-60%. Нужно протестировать качество на наших данных.

• Prompt caching - Gemini поддерживает кэширование промптов. Наш промпт с топонимами одинаковый для всех запросов. Кэширование сэкономит ~40% на input токенах.

• Батчинг для не-срочных — если сообщение не требует мгновенного ответа, можно накапливать и отправлять пачками. Некоторые API дают скидку на batch.

Улучшение качества:

• Автоматическое пополнение словаря - если транскрипция содержит новое слово, похожее на топоним (заглавная буква, суффикс -ово/-ино/-ка), добавлять в кандидаты для ручной проверки.

• A/B тестирование весов - автоматизировать подбор коэффициентов в scoring-формуле на основе размеченных данных.

• GPT-4o Transcribe - OpenAI выпустили новую модель с diarization. Стоит сравнить с Whisper.

Мониторинг:

• Real-time дашборд - график latency, success rate, agreement rate. Алерт если метрики падают.

• Автоматическое обнаружение новых ошибок - кластеризация транскрипций, которые пользователи исправляют вручную.

• Применимость подхода

  Описанный подход работает не только для топонимов. Multi-API Ensemble полезен везде, где есть специфическая лексика:

  • Медицина: названия препаратов, диагнозы, латинские термины

  • Юриспруденция: статьи законов, юридические термины

  • Техническая поддержка: названия продуктов, артикулы, SKU

  • Логистика: адреса, названия складов, номера заказов

  • Финансы: тикеры акций, названия фондов, финансовые термины

  Если статья была полезна поставьте плюс, это помогает другим найти материал.