vidconf/docs/plugins/transcriber.md

Плагин Transcriber — транскрибация аудио в текст

Обзор

VidConf использует Strategy pattern для подключаемых реализаций транскрибации речи. Это позволяет менять провайдеров (faster-whisper на CPU, OpenAI Whisper, Vosk и т.д.) через конфигурацию без изменения ядра.

Контракт Transcriber

Интерфейс

Все реализации наследуют абстрактный класс Transcriber из backend/core/plugins/transcriber.py:

from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import ClassVar

@dataclass(frozen=True, slots=True)
class Segment:
    """Один транскрибированный сегмент трека."""
    start: float      # Начало в секундах от старта файла
    end: float        # Конец в секундах от старта файла
    text: str         # Распознанный текст

class Transcriber(ABC):
    """Интерфейс Strategy для реализаций преобразования речи в текст."""
    
    provider: ClassVar[str]  # Уникальный идентификатор провайдера
    
    @abstractmethod
    def transcribe(self, audio_path: str, language: str = "ru") -> list[Segment]:
        """Транскрибировать аудиофайл в список сегментов."""
        ...

Сигнатура метода transcribe

Параметр	Тип	Описание
audio_path	str	Абсолютный путь к аудиофайлу на диске
language	str	Код языка (ISO 639-1), дефолт "ru"
Возврат	list[Segment]	Упорядоченный список сегментов, отсортированный по start

Требования:

• Если файл не найден → исключение FileNotFoundError
• Сегменты должны быть отсортированы по start в возрастающем порядке
• Дублирующиеся/перекрывающиеся сегменты допустимы (алгоритм фраз их обработает)

Реализация по умолчанию: FasterWhisperCPU

Описание

FasterWhisperCPU — оптимизированная реализация faster-whisper (CTranslate2-бэкэнд) для транскрибации на CPU с int8-квантизацией. Она встроена в проект как реализация по умолчанию.

Файл: backend/core/plugins/faster_whisper.py

Параметры конструктора

@register_transcriber
class FasterWhisperCPU(Transcriber):
    provider: ClassVar[str] = "faster_whisper_cpu"
    
    def __init__(
        self,
        model: str | None = None,
        language: str = "ru",
        download_root: str | None = None,
    ) -> None:
        """Инициализировать транскрибер.
        
        Args:
            model: Имя модели Whisper (small/base/medium/etc);
                   дефолт 'small' (оптимум для CPU).
            language: Код языка (ISO 639-1), дефолт 'ru'.
            download_root: Папка для кэша загруженных моделей;
                          дефолт ~/.cache/huggingface/hub.
        """

Параметр	Тип	Дефолт	Описание
model	str	"small"	Модель Whisper: tiny, base, small, medium, large
language	str	"ru"	Язык распознавания (ISO 639-1)
download_root	str	~/.cache/	Папка для кэша моделей

Особенности

VAD (Voice Activity Detection)

FasterWhisperCPU включает встроенный Silero VAD для автоматического разбиения аудио на речевые сегменты и подавления молчания. Это улучшает качество транскрибации и снижает галлюцинации Whisper на тишине/шуме.

• Параметр: vad_filter=True
• Порог молчания: vad_min_silence_duration_ms=500 (0.5 сек)

Отбрасывание коротких сегментов

Whisper часто халлюцинирует на очень коротких сегментах (шум, клики, тишина). FasterWhisperCPU автоматически отбрасывает сегменты, короче 0.3 секунды:

MIN_SEGMENT_DURATION_S = 0.3

# Фильтрация в transcribe()
return [
    Segment(start=segment.start, end=segment.end, text=segment.text)
    for segment in raw_segments
    if (segment.end - segment.start) >= MIN_SEGMENT_DURATION_S
]

Ленивый импорт

Зависимость faster-whisper импортируется лениво — только при первом вызове transcribe(). Это позволяет API-процессу, который не использует транскрибацию, избежать загрузки тяжёлой библиотеки в память.

def _get_model(self) -> "WhisperModel":
    if FasterWhisperCPU._model is None:
        from faster_whisper import WhisperModel  # Импорт здесь, не в начале файла
        FasterWhisperCPU._model = WhisperModel(...)
    return FasterWhisperCPU._model

Синглтон модели на процесс

Модель Whisper создаётся один раз при первом вызове transcribe() и переиспользуется всеми последующими вызовами в пределах одного процесса воркера. Это снижает нагрузку на память и CPU (загрузка модели — дорогая операция).

Примечание: Celery-воркер транскрибации запускается с флагом --pool=solo --concurrency=1, поэтому гонок за синглтоном не бывает.

Конфигурация

В config/plugins.yaml:

transcriber:
  enabled: true
  provider: "faster_whisper_cpu"
  model: "small"
  language: "ru"
  options:
    download_root: "/models/whisper"

Переменные окружения (если нужны):

# Опционально: переопределить папку кэша моделей
export HF_HOME=/models

Производительность

• Модель small: ~8 минут аудио за 1 минуту на современном CPU (зависит от CPU)
• Память: ~1.5 ГБ на процесс (с моделью и бэкэндом)
• Формат входа: WAV, MP3, OGG, FLAC, M4A и др. (поддерживает ffmpeg-compatible форматы)

Добавление нового Transcriber

Пошаговая инструкция

Шаг 1: Создать файл реализации

Создайте новый файл в backend/core/plugins/, например my_transcriber.py:

"""Плагин Transcriber на основе MyService."""

from typing import ClassVar
from core.plugins.factory import register_transcriber
from core.plugins.transcriber import Segment, Transcriber

@register_transcriber
class MyTranscriber(Transcriber):
    """Транскрибер, использующий MyService API."""
    
    provider: ClassVar[str] = "my_service"
    
    def __init__(
        self,
        model: str | None = None,
        language: str = "ru",
        api_key: str | None = None,
        **options,
    ) -> None:
        """Инициализировать транскрибер.
        
        Args:
            model: Имя модели (опционально).
            language: Язык распознавания.
            api_key: API-ключ для сервиса.
            **options: Дополнительные параметры из config/plugins.yaml.
        """
        self.model = model or "default-model"
        self.language = language
        self.api_key = api_key or os.getenv("MY_SERVICE_API_KEY")
        self.options = options
    
    def transcribe(self, audio_path: str, language: str = "ru") -> list[Segment]:
        """Транскрибировать аудиофайл через MyService API.
        
        Args:
            audio_path: Путь к аудиофайлу на диске.
            language: Язык распознавания (может переопределить конструктор).
        
        Returns:
            Упорядоченный список сегментов, отсортированный по start.
        """
        # 1. Прочитать аудиофайл
        with open(audio_path, "rb") as f:
            audio_data = f.read()
        
        # 2. Отправить на API (пример)
        response = requests.post(
            "https://api.myservice.com/transcribe",
            files={"audio": audio_data},
            json={"language": language, "model": self.model},
            headers={"Authorization": f"Bearer {self.api_key}"},
        )
        response.raise_for_status()
        
        # 3. Распарсить ответ в list[Segment]
        result = response.json()
        segments = [
            Segment(
                start=float(item["start"]),
                end=float(item["end"]),
                text=item["text"],
            )
            for item in result.get("segments", [])
        ]
        
        # 4. Отсортировать по start (важно!)
        segments.sort(key=lambda s: s.start)
        
        return segments

Важные правила:

• Класс должен наследовать Transcriber
• Класс должен иметь декоратор @register_transcriber (регистрирует в реестре)
• Класс-переменная provider: ClassVar[str] должна быть уникальным идентификатором
• Конструктор должен принимать model, language и **options
• Метод transcribe() должен возвращать сегменты, отсортированные по start

Шаг 2: Импортировать в __init__.py

Добавьте импорт в backend/core/plugins/__init__.py:

"""Пакет плагинов Transcriber/Summarizer."""

from core.plugins import faster_whisper as faster_whisper  # noqa: F401
from core.plugins import my_transcriber as my_transcriber  # noqa: F401  # ДОБАВИТЬ
from core.plugins import null as null  # noqa: F401

Примечание: noqa: F401 подавляет предупреждение о неиспользуемом импорте — он нужен для побочного эффекта (регистрация). Используйте as для явности.

Шаг 3: Обновить конфигурацию

Отредактируйте config/plugins.yaml:

transcriber:
  enabled: true
  provider: "my_service"           # Совпадает с MyTranscriber.provider
  model: "default-model"
  language: "ru"
  options:
    api_key: "${MY_SERVICE_API_KEY}"  # Переменная окружения
    # Дополнительные параметры, специфичные для вашего сервиса
    timeout_seconds: 300
    retry_count: 3

Шаг 4: Добавить переменные окружения

Отредактируйте .env (или .env.example):

# MyService API
MY_SERVICE_API_KEY=sk-...

Шаг 5: Написать тесты

Добавьте тесты в backend/tests/test_plugins_factory.py или новый test_my_transcriber.py:

import pytest
from core.plugins.factory import create_transcriber
from core.plugins.config import TranscriberConfig

def test_my_transcriber_registered():
    """Проверить регистрацию плагина."""
    cfg = TranscriberConfig(provider="my_service", model="test")
    transcriber = create_transcriber(cfg)
    assert transcriber.provider == "my_service"
    assert transcriber.model == "test"

def test_my_transcriber_transcribe(tmp_path):
    """Тест транскрибации с моком API."""
    # Создать тестовый WAV-файл
    audio_file = tmp_path / "test.wav"
    # ... создать корректный WAV-файл ...
    
    cfg = TranscriberConfig(provider="my_service", model="test")
    transcriber = create_transcriber(cfg)
    
    # Вызвать transcribe
    segments = transcriber.transcribe(str(audio_file), language="ru")
    
    # Проверить результат
    assert len(segments) > 0
    assert all(s.start < s.end for s in segments)
    assert segments == sorted(segments, key=lambda s: s.start)

Шаг 6: Протестировать локально

cd backend

# Убедиться, что конфиг загружается
uv run python -c "
from core.plugins.factory import _TRANSCRIBERS
print('Зарегистрированные транскрибаторы:', list(_TRANSCRIBERS.keys()))
"

# Запустить тесты
uv run pytest tests/test_my_transcriber.py -v

# Запустить пайплайн с новым провайдером (интеграционный тест)
CELERY_ALWAYS_EAGER=True uv run pytest tests/test_build_phrases.py -v

Отключение транскрибации

Если нужно полностью отключить модуль транскрибации, установите в config/plugins.yaml:

transcriber:
  enabled: false
  provider: "null"
  model: null

Поведение:

• Запись аудиотреков продолжает работать (egress в активном состоянии)
• При room_finished run_pipeline() завершится до инстанцирования плагина
• pipeline_status сеанса останется "recording" (не перейдёт в "transcribing")
• Лог-сообщение: "transcriber отключён (enabled=false) — сеанс ... пропущен"

Архитектурные решения

Идемпотентный пайплайн пост-обработки

Пайплайн пост-обработки — идемпотентная машина состояний (см. backend/models/session.py):

recording → transcribing → summarizing → notified / failed

run_pipeline() диспетчер:

1. Проверяет текущий pipeline_status сеанса
2. Продолжает работу с последнего успешного шага
3. Отслеживает статус каждого трека (session_audio_tracks.status)
4. Коммитит результаты транскрибации в session_audio_tracks.segments (JSONB) после каждого трека

Это позволяет восстановиться после сбоя воркера без потери данных или дублирования.

Атрибуция к участнику сеанса (ADR-002)

Фразы и аудиотреки атрибутированы не к users, а к conference_participants (участник сеанса). Это позволяет работать с гостями без user_id:

# Получить спикера фразы
participant = session.query(ConferenceParticipant).get(phrase.participant_id)
user = participant.user  # Может быть None для гостей
guest = participant.guest  # Может быть None для пользователей

Ссылки

• ADR-002: docs/architecture/adr/002-phrase-attribution-session-participant.md
• Контракты плагинов: docs/plugins/contracts.md
• Модели: backend/models/audio_track.py, backend/models/phrase.py
• Пайплайн: workers/tasks/pipeline.py
• Реконструкция фраз: workers/transcription/phrases.py
• Тесты: backend/tests/test_plugins_factory.py, backend/tests/test_build_phrases.py