Реалізація розпізнавання мовлення з аудіофайлів (Batch STT) Batch STT обробляє заздалегідь записані файли без вимог до затримки - можна використовувати найякісніші моделі та вичавити максимум точності. Типове завдання: архівна обробка записів кол-центру, транскрибація подкастів, створення субтитрів до відеоконтенту. ### Архітектура batch-пайплайну```
Upload → S3/Local Storage → Queue (Celery/SQS) → Worker → STT → Post-Processing → Storage
Ключові рішення при проектуванні: - Нарізка довгих файлів на сегменти по 5-10 хвилин (покращує точність) - Паралельна обробка декількох файлів - Retry-логіка для збійних завдань - Зберігання проміжних результатів ### Повний pipeline обробкиpython
import os
from pathlib import Path
from faster_whisper import WhisperModel
from celery import Celery
import ffmpeg
app = Celery('batch_stt', broker='redis://localhost:6379/0', backend='redis://localhost:6379/1') model = WhisperModel("large-v3", device="cuda", compute_type="int8_float16")
def convert_to_wav(input_path: str) -> str: output_path = input_path.rsplit('.', 1)[0] + '_converted.wav' ffmpeg.input(input_path).output( output_path, ar=16000, ac=1, acodec='pcm_s16le' ).overwrite_output().run(quiet=True) return output_path
@app.task(bind=True, max_retries=3, time_limit=3600) def process_audio_file(self, file_path: str, options: dict = None): options = options or {} try: # Конвертация в нужный формат wav_path = convert_to_wav(file_path)
segments, info = model.transcribe(
wav_path,
language=options.get('language'),
vad_filter=True,
word_timestamps=options.get('word_timestamps', False),
beam_size=5
)
result = {
"file": file_path,
"language": info.language,
"language_probability": info.language_probability,
"duration": info.duration,
"segments": []
}
for seg in segments:
segment_data = {
"start": round(seg.start, 3),
"end": round(seg.end, 3),
"text": seg.text.strip()
}
if options.get('word_timestamps'):
segment_data["words"] = [
{"word": w.word, "start": w.start, "end": w.end, "probability": w.probability}
for w in (seg.words or [])
]
result["segments"].append(segment_data)
os.unlink(wav_path)
return result
except Exception as exc:
raise self.retry(exc=exc, countdown=60 * (self.request.retries + 1))
1 година аудіо на RTX 4090 з large-v3: ~15-20 хвилин обробки. ### Терміни реалізації - Скрипт для одиночних файлів: 1 день - Пайплайн з чергою та API: 3-5 днів - Повна система з дашбордом статусів: 1 тиждень