Сборка локального ASR на Python: когда облака бесят, а конфиденциальность не шутка

Забудьте про ежемесячные счета от Google Speech-to-Text и AWS Transcribe. Забудьте про нервные согласия на передачу данных. Забудьте про задержки в сети, когда нужно распознать речь в реальном времени. Если вы читаете это, значит устали от облачной зависимости в ASR (Automatic Speech Recognition).

Я собрал десятки таких систем за последние три года — для call-центров, судебных процессов, медицинских консультаций. И каждый раз одно и то же: клиенты в ужасе от того, что их разговоры улетают в неизвестность. Особенно после того скандала в 2025-м, когда одна крупная платформа "случайно" использовала аудио для тренировки своих моделей.

Сегодня соберем систему, которая работает на вашем железе. Полностью. Без единого запроса в интернет. С диарризацией (кто что сказал) и потоковой обработкой в реальном времени. И да, это будет работать на RTX 3060 или даже на CPU, если готовы ждать подольше.

Почему локальный ASR в 2026 — это не безумие, а необходимость

Вот что происходит, когда вы отправляете аудио в облако:

• Ваши данные становятся чьим-то тренировочным датасетом. Даже если в SLA написано обратное. Проверено на практике.
• Задержки убивают интерактивность. 200-300ms на сеть + обработка = пользователь уже перебивает.
• Стоимость растет непредсказуемо. Особенно если у вас всплески трафика.
• Вы не можете кастомизировать модель под свой акцент, терминологию или шумный фон.

Что нам понадобится (и почему именно это)

Если сомневаетесь между Whisper и Qwen3-ASR — посмотрите мое сравнение моделей для английского. Для русского Whisper пока стабильнее, но Qwen3-ASR догоняет и поддерживает кучу экзотических языков.

1 Ставим железо на место: от установки до первой проверки

Сначала убедимся, что у нас есть CUDA. Без нее на CPU будет больно.

# Проверяем CUDA
nvidia-smi

# Если нет — ставим (для Ubuntu 22.04+)
wget https://developer.download.nvidia.com/compute/cuda/repos/ubuntu2204/x86_64/cuda-keyring_1.1-1_all.deb
sudo dpkg -i cuda-keyring_1.1-1_all.deb
sudo apt-get update
sudo apt-get -y install cuda-toolkit-12-4

Теперь Python-окружение. Я ненавижу conda, поэтому покажу через venv:

python -m venv asr_env
source asr_env/bin/activate  # или asr_env\Scripts\activate на Windows

# Core-зависимости
pip install torch torchaudio --index-url https://download.pytorch.org/whl/cu124
pip install faster-whisper transformers pyannote.audio pyaudio sounddevice

Если pyaudio не ставится на Windows — качайте prebuilt wheels. На Linux может потребоваться libportaudio.

2 Берем модель Whisper и заставляем ее летать

Первая ошибка новичков — качать огромную large-v3 (6.7GB) для тестов. Начните с small или medium.

from faster_whisper import WhisperModel
import time

# НЕ ТАК: model = WhisperModel("large-v3", device="cuda") — будет долго грузиться
# ТАК:
model = WhisperModel("medium", device="cuda", compute_type="float16")

# Проверяем
start = time.time()
segments, info = model.transcribe("test_audio.wav", language="ru")
print(f"Detected language: {info.language}, probability: {info.language_probability:.2f}")

for segment in segments:
    print(f"[{segment.start:.2f}s -> {segment.end:.2f}s] {segment.text}")

print(f"Time: {time.time() - start:.2f}s")

compute_type="float16" ускоряет inference в 1.5-2 раза почти без потери точности на современных GPU. Если у вас карта без поддержки float16 (очень старая), используйте "int8", но будет чуть менее точно.

3 Диарризация: кто, когда и что сказал

Вот здесь начинается настоящая боль. Готовые модели pyannote часто путают голоса, особенно если в комнате эхо или несколько женщин с похожими тембрами.

Сначала базовый вариант:

from pyannote.audio import Pipeline

# Качаем модель с HuggingFace
# Требуется токен (бесплатный, но нужно зарегистрироваться)
pipeline = Pipeline.from_pretrained(
    "pyannote/speaker-diarization-3.1",
    use_auth_token="YOUR_HF_TOKEN"
)
pipeline.to("cuda")

# Применяем
diarization = pipeline("meeting.wav")

for turn, _, speaker in diarization.itertracks(yield_label=True):
    print(f"Speaker {speaker}: from {turn.start:.1f}s to {turn.end:.1f}s")

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

Альтернатива — простенькая кластеризация по embeddings:

from pyannote.audio import Inference
import numpy as np
from sklearn.cluster import AgglomerativeClustering

embedding_model = Inference(
    "pyannote/wespeaker-voxceleb-resnet34-LM",
    window="whole"
)

# Извлекаем эмбеддинги каждые 0.5 секунды
embeddings = []
timestamps = []
for start in np.arange(0, duration, 0.5):
    embedding = embedding_model({"waveform": waveform, "sample_rate": sr})
    embeddings.append(embedding)
    timestamps.append(start)

# Кластеризуем (предполагаем 2 спикера)
clustering = AgglomerativeClustering(n_clusters=2).fit(embeddings)
labels = clustering.labels_

Это грубее, но работает без токена и иногда стабильнее на нестандартных акцентах.

4 Потоковая обработка: от записи до текста в реальном времени

Вот тут Whisper показывает не самую лучшую сторону. Он обучен на 30-секундных чанках и любит контекст. Но нам нужно показывать текст по мере говорения.

Мое решение — двойной буфер:

import sounddevice as sd
import numpy as np
from collections import deque
from threading import Thread

class StreamingASR:
    def __init__(self, model_name="medium", device="cuda"):
        self.model = WhisperModel(model_name, device=device, compute_type="float16")
        self.buffer = deque(maxlen=16000*30)  # 30 секунд
        self.results = []
        self.is_recording = False
        
    def callback(self, indata, frames, time, status):
        """Вызывается при каждом заполнении буфера звуковой карты"""
        if status:
            print(f"Audio error: {status}")
        self.buffer.extend(indata[:, 0])  # берем один канал
        
    def process_chunk(self):
        """Обрабатывает накопленный буфер раз в 5 секунд"""
        while self.is_recording:
            if len(self.buffer) >= 16000*5:  # 5 секунд
                audio_np = np.array(self.buffer)
                segments, _ = self.model.transcribe(audio_np, language="ru", beam_size=1)
                for seg in segments:
                    self.results.append(seg.text)
                    print(f"Live: {seg.text}")
                # Очищаем обработанное
                for _ in range(16000*4):  # оставляем 1 секунду контекста
                    if self.buffer:
                        self.buffer.popleft()
            time.sleep(0.1)
    
    def start(self):
        self.is_recording = True
        # Поток для обработки
        process_thread = Thread(target=self.process_chunk)
        process_thread.start()
        
        # Начинаем запись
        with sd.InputStream(callback=self.callback, channels=1, samplerate=16000):
            while self.is_recording:
                sd.sleep(1000)
                # Здесь можно добавить условие остановки по кнопке

Почему буфер на 30 секунд, а обрабатываем по 5? Whisper работает лучше с контекстом. Мы всегда подаем последние 5 секунд + предыдущие 25 как историю. Это уменьшает ошибки на границах предложений.

Собираем все вместе: production-ready система

Теперь склеим ASR, диарризацию и потоковую обработку в одну систему. Цель: запись совещания, где в реальном времени показывается, кто что говорит.

import queue
import threading
from dataclasses import dataclass
from typing import List

@dataclass
class Utterance:
    speaker: str
    text: str
    start: float
    end: float

class MeetingTranscriber:
    def __init__(self):
        self.asr_model = WhisperModel("medium", device="cuda", compute_type="float16")
        self.diarization_model = Pipeline.from_pretrained(...)
        self.audio_queue = queue.Queue()
        self.results: List[Utterance] = []
        
    def audio_capture_thread(self):
        """Захватывает аудио и кладет в очередь"""
        def callback(indata, frames, time, status):
            self.audio_queue.put(indata.copy())
        
        with sd.InputStream(callback=callback, channels=1, samplerate=16000,
                           blocksize=1600):  # 100ms chunks
            while True:
                sd.sleep(1000)
    
    def processing_thread(self):
        """Обрабатывает аудио, делает диарризацию и ASR"""
        buffer = np.array([])
        
        while True:
            # Собираем 10 секунд аудио
            while len(buffer) < 16000*10:
                chunk = self.audio_queue.get()
                buffer = np.concatenate([buffer, chunk[:, 0]])
            
            # Диарризация
            diarization = self.diarization_model({"waveform": buffer, "sample_rate": 16000})
            # ASR
            segments, _ = self.asr_model.transcribe(buffer, language="ru", beam_size=1)
            
            # Сопоставляем спикеров и текст
            for segment in segments:
                # Находим, какой спикер говорил в это время
                speaker = self.find_speaker_for_interval(diarization, segment.start, segment.end)
                utterance = Utterance(
                    speaker=speaker,
                    text=segment.text,
                    start=segment.start,
                    end=segment.end
                )
                self.results.append(utterance)
                print(f"{speaker}: {segment.text}")
            
            # Оставляем 2 секунды контекста для следующего чанка
            buffer = buffer[-16000*2:]
    
    def find_speaker_for_interval(self, diarization, start, end):
        """Находит спикера, который говорил больше всего в интервале"""
        # Упрощенная логика сопоставления
        speaker_times = {}
        for turn, _, speaker in diarization.itertracks(yield_label=True):
            overlap = min(turn.end, end) - max(turn.start, start)
            if overlap > 0:
                speaker_times[speaker] = speaker_times.get(speaker, 0) + overlap
        
        if speaker_times:
            return max(speaker_times, key=speaker_times.get)
        return "UNKNOWN"

Оптимизации, которые реально работают

1. Квантование модели. Если GPU мало памяти (менее 8GB), конвертируйте Whisper в INT8:

model = WhisperModel("medium", device="cuda", compute_type="int8")

Точность упадет на 5-10%, но память сократится в 4 раза.

2. Асинхронная обработка. Не ждите окончания ASR, чтобы начать диарризацию:

import asyncio
import concurrent.futures

async def process_audio_chunk(audio_np):
    loop = asyncio.get_event_loop()
    with concurrent.futures.ThreadPoolExecutor() as pool:
        # Запускаем ASR и диарризацию параллельно
        asr_task = loop.run_in_executor(pool, model.transcribe, audio_np)
        diarization_task = loop.run_in_executor(pool, pipeline, audio_np)
        
        asr_result, diarization_result = await asyncio.gather(asr_task, diarization_task)
        return asr_result, diarization_result

3. Кэширование эмбеддингов. Если у вас много повторяющихся голосов (сотрудники компании), сохраните их voiceprints и сравнивайте с новыми, а не кластеризуйте заново.

Чего ждать в ближайшем будущем

1. Whisper-large-v4 обещают на 2026 год. По слухам, будет лучше работать с шумным аудио и поддерживать streaming из коробки.

2. Qwen3-ASR активно развивается. Если нужна поддержка множества языков в одной модели — посмотрите мой обзор Qwen3-ASR. Для продакшена уже есть готовый Docker-сервис.

3. Аппаратное ускорение на CPU через OpenVINO и ONNX Runtime. Для серверов без GPU это может стать спасением.

Что делать, если все тормозит

1. Проверьте compute_type. "float32" на GPU — это убийство производительности. Всегда используйте "float16" или "int8".

2. Уменьшайте beam_size. С beam_size=5 точность лучше на 3-5%, но скорость ниже в 3 раза. Для диалогов beam_size=1 часто достаточно.

3. Бач-обработка. Если обрабатываете записи постфактум, подавайте по несколько файлов одновременно:

# Не так
for file in audio_files:
    result = model.transcribe(file)

# А так
from concurrent.futures import ThreadPoolExecutor

def transcribe_file(file):
    return model.transcribe(file)

with ThreadPoolExecutor(max_workers=4) as executor:
    results = list(executor.map(transcribe_file, audio_files))

4. Используйте GPU с большим объемом памяти. Очевидно? Но многие пытаются запихать large-v3 в 8GB. Не выйдет. Medium требует ~4GB, small — ~2GB.

Финальный совет: когда все-таки стоит использовать облака

Да, я только что рассказывал, как от них уйти. Но есть случаи, когда облачный ASR оправдан:

• Мобильные приложения. Тащить 2GB модели в апп — самоубийство.
• Очень редкие языки, которые ваша модель не поддерживает.
• Пилотные проекты, где нужно проверить гипотезу без инвестиций в железо.
• Сезонные нагрузки, когда 95% времени система простаивает.

Но как только объемы растут, а требования к конфиденциальности ужесточаются — локальное решение окупается за 6-12 месяцев. И вы получаете полный контроль. Что в 2026 году дороже денег.

Следующий шаг — добавить LLM для суммаризации встреч или собрать голосового ассистента целиком на одной карте. Но это уже другая история.