Один STT-сервис дал 60-70% точности на специфической лексике (топонимы, названия улиц, профессиональные термины). Два сервиса параллельно + взвешенное голосование + AI-fusion для спорных случаев дали 95%+ точности. Время обработки 5-8 секунд, стоимость $70-130/месяц при 1000 сообщений в день. В статье — полный разбор архитектуры, алгоритмы scoring, примеры кода и расчёт экономики.
Содержание
Почему один STT оказалось недостаточно
Эволюция решения: от 60% к 95%
Архитектура Multi-API Ensemble
Взвешенное голосование: математика выбора
AI-fusion: когда голосования недостаточно
Постобработка: ловим систематические ошибки
Промпты для STT-сервисов
Мониторинг и graceful degradation
Результаты и метрики
Экономика решения
Выводы
1. Проблема: почему один STT недостаточно
Я разрабатывал систему транскрипции голосовых оповещений в ЧС для Telegram-канала. Задача казалась простой — Speech-to-Text API существуют больше десяти лет, технология зрелая.
Первая версия использовала один STT-сервис (SaluteSpeech от Сбера, он бесплатный для небольших объёмов). Запустили, протестировали на реальных данных. Результат: всего 60% точности.
Каждое второе сообщение содержало критические ошибки.
Анатомия ошибки
Типичное голосовое сообщение длиной 15 секунд:
«Внимание, оповещение для жителей Масычево, Илёк-Пеньковки и Мокрой Орловки. Просьба соблюдать осторожность на улице Заречной.»
А вот что возвращали разные STT-сервисы:
Сервис | Результат |
|---|---|
SaluteSpeech | «Внимание, оповещение для жителей масло чего, и лёгкая пеньковка и мокрой орлов. Просьба соблюдать осторожность на улице зеречной.» |
Yandex SpeechKit | «Внимание, оповещение для жителей мамы чего, илёгкая пеньковка и мокрой орловке. Просьба соблюдать осторожность на улице заречной.» |
Google Speech-to-Text | «Внимание, оповещение для жителей мосычево, илёк пеньковки и мокрой орловки. Просьба соблюдать осторожность на улице заречной.» |
Обратите внимание: каждый сервис ошибается по-разному. SaluteSpeech превращает «Масычево» в «масло чего», Yandex - в «мамы чего», Google - в «мосычево». При этом Google правильно распознаёт «мокрой орловки», а первые два - нет.
Почему модели ошибаются на топонимах
Причина фундаментальная: STT-модели обучены на общей лексике. Датасеты для обучения содержат миллионы часов подкастов, аудиокниг, телефонных разговоров, YouTube-видео и редко названия деревень с населением 200 человек.
Как работает распознавание речи на высоком уровне:
Акустическая модель преобразует звуковой сигнал в вероятности фонем
Языковая модель выбирает наиболее вероятную последовательность слов
Проблема в языковой модели. Она обучена на текстах, где «масло» встречается миллионы раз, а «Масычево» ни разу. Когда акустическая модель даёт неоднозначный сигнал (а она всегда даёт неоднозначный), языковая модель выбирает знакомое слово.
Это классическая проблема 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
После нескольких недель сбора статистики я проанализировал, какие сервисы меньше ошибаются.
Методика анализа:
Собрал 500 размеченных примеров
Для каждого сервиса (с помощью Claude, дав ему контекст и данные) посчитал WER (Word Error Rate) по категориям: топонимы, термины, общая лексика
Построили матрицу корреляции ошибок: если сервис A ошибся, ошибся ли сервис B?
Результаты:
Сервис | WER общий | WER топонимы | WER термины | Корреляция с Whisper |
|---|---|---|---|---|
Whisper | 12% | 35% | 8% | - |
Gemini | 14% | 22% | 15% | 0.31 |
Yandex | 18% | 38% | 12% | 0.67 |
SaluteSpeech | 22% | 45% | 18% | 0.58 |
Корреляция ошибок 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 работает так:
Выравнивает две транскрипции по словам (alignment)
Для каждой позиции голосует за вариант
При равенстве голосов выбирает первый вариант
Точность выросла до 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/месяц.
Версия | Сервисы | Время | Точность | Стоимость |
|---|---|---|---|---|
v1.0 | 1 (SaluteSpeech) | 3-5 сек | ~60% | ~$20/мес |
v2.0 | 4 (все) | 14-20 сек | ~80% | ~$350/мес |
v2.5 | 2 (Gemini + Whisper) | 10-14 сек | ~90% | ~$100/мес |
v3.0 | 2 + AI-fusion + постобработка | 5-8 сек | ~95% | ~$100/мес |
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% случаев оба сервиса уверены в своих результатах, но результаты разные. Пример:
Gemini | Whisper |
|---|---|
«Козинка, отбой опасности» | «Казинка, отбой опасности» |
score = 1.42 | score = 1.38 |
Разница 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 запись. Пример:
Оригинал | Исправление | Количество |
|---|---|---|
«Козинки» | «Козинка» | 32 |
«fpv» | «FPV» | 24 |
«бпла» | «БПЛА» | 19 |
«и лёгкая пеньковка» | «Илёк-Пеньковка» | 15 |
«масло чего» | «Масычево» | 12 |
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 распознаны неправильно.
Сравнение точности по категориям
Категория | Один сервис (Whisper) | Ensemble v3.0 | Улучшение |
|---|---|---|---|
Общая лексика | 90% | 98% | +8% |
Топонимы (простые) | 65% | 95% | +30% |
Топонимы (составные) | 20% | 90% | +70% |
Аббревиатуры | 70% | 98% | +28% |
Числа | 85% | 97% | +12% |
Экстренные фразы | 75% | 99% | +24% |
Детальные примеры распознавания
Пример 1: Топоним «Масычево»
Этап | Результат | WER | Комментарий |
|---|---|---|---|
SaluteSpeech | «масло чего» | 100% | Полная замена |
Yandex | «мамы чего» | 100% | Другая ошибка |
Whisper | «Масечево» | 50% | Близко, но неточно |
Gemini | «Масычево» | 0% | Правильно |
Ensemble | «Масычево» | 0% | Gemini победил по score |
Пример 2: Составной топоним «Илёк-Пеньковка»
Этап | Результат | WER | Комментарий |
|---|---|---|---|
SaluteSpeech | «и лёгкая пеньковка» | 200% | Разбил + ошибка |
Whisper | «Илек Пеньковка» | 50% | Без дефиса |
Gemini | «Илёк-Пеньковка» | 0% | Правильно |
PostProcess | Исправит «Илек Пеньковка» → «Илёк-Пеньковка» | - | Страховка |
Ensemble | «Илёк-Пеньковка» | 0% | - |
Пример 3: Фраза «отбой опасности» (сложный случай)
Эта фраза интересна тем, что SaluteSpeech систематически ошибался одинаково:
Этап | Результат | Комментарий |
|---|---|---|
SaluteSpeech | «на двое суток опасности» | Классическая фонетическая ошибка |
Whisper | «отбой опасности» | Правильно |
Gemini | «отбой опасности» | Правильно |
Ensemble | «отбой опасности» | Консенсус 2 из 3 |
Даже если бы Whisper и Gemini оба ошиблись, постобработка содержит правило:
r'\bна\s+двое\s+суток\s+опасности\b': 'отбой опасности'
Пример 4: Аббревиатура в контексте
Исходное аудио: «Обнаружен FPV дрон в районе Козинки»
Этап | Результат |
|---|---|
Whisper | «Обнаружен эф пи ви дрон в районе Козинки» |
Gemini | «Обнаружен FPV дрон в районе Козинки» |
Ensemble | «Обнаружен FPV дрон в районе Козинки» |
Gemini понял из контекста, что «эф пи ви» - это аббревиатура FPV. Whisper распознал по буквам, но постобработка всё равно исправила бы.
Метрики production-системы (4 месяца работы)
Метрика | Значение | Комментарий |
|---|---|---|
Обработано сообщений | 48,000+ | ~400/день в среднем |
Среднее время обработки | 6.2 сек | Включая все этапы |
Медианное время | 5.4 сек | 50-й перцентиль |
95-й перцентиль | 9.1 сек | Редкие медленные запросы |
99-й перцентиль | 12.3 сек | Очень редкие случаи |
Agreement между сервисами | 73% | Совпадение ≥70% слов |
Clear winner (без AI-fusion) | 64% | Один сервис явно лучше |
AI-fusion вызовов | 31% | Нужно объединение |
ROVER fallback | 5% | AI-fusion недоступен |
Ошибок, требующих ручной правки (было, уже нет) | 4.2% | ~17 из 400 в день |
Uptime системы | 99.7% | 2.6 часа downtime за 4 мес |
Динамика по месяцам:
Месяц | Точность | Avg latency | Комментарий |
|---|---|---|---|
1 | 91% | 8.1 сек | Начальная версия |
2 | 93% | 6.8 сек | Оптимизация промптов |
3 | 94% | 6.4 сек | Расширение постобработки |
4 | 95% | 6.2 сек | Тонкая настройка весов |
Точность растёт, latency падает система «учится» на своих ошибках через пополнение словарей постобработки.
10. Экономика решения
Расчёт для разных нагрузок
Нагрузка | Whisper | Gemini STT | AI-fusion | Итого/месяц |
|---|---|---|---|---|
500 msg/день | $30 | $20 | $3 | ~$55 |
1000 msg/день | $60 | $40 | $6 | ~$110 |
2000 msg/день | $120 | $80 | $12 | ~$215 |
Примечание: расчёт для среднего сообщения 15-20 секунд. Для более длинных аудио стоимость выше.
Сравнение с альтернативами
Подход | Стоимость/мес | Точность | Время | Комментарий |
|---|---|---|---|---|
1 STT (Whisper) | ~$60 | 60-70% | 3-5 сек | Дёшево, много ошибок |
4 STT параллельно | ~$350 | 80% | 14-20 сек | Дорого, медленно |
Fine-tuned модель | $5000+ upfront | 95%+ | 3-5 сек | Требует данных |
Наш ensemble | ~$110 | 95% | 5-8 сек | Оптимальный баланс |
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
Логистика: адреса, названия складов, номера заказов
Финансы: тикеры акций, названия фондов, финансовые термины
Если статья была полезна поставьте плюс, это помогает другим найти материал.
