In Production

[shaunck96]

Has anyone deployed this in production? medium version to transcribe batches of 10,000 calls/day with average duration of 6 minutes per call. Is it feasible from a cost standpoint versus a paid API like Azure STT. Accuracy is not the priority versus cost (as medium size model/distil medium can also be to used to perform transcription) and data privacy as we are dealing with very sensitive data. Leveraging distributed computing with spark capabilities and multi threading, currently with VMs in Azure cloud but limited to CPU for now (E4d_v4, E16d_v4, E32d_v4).

[shaunck96]

no need of dirization,etc .. just raw transcription is sufficient

[shaunck96]

model = None

def get_model():
    global model
    if model is None:
        from cs_pa_nlp.faster_whisper import WhisperModel
        model = WhisperModel("medium.en", 
                             device="cpu",
                             compute_type="int8"
                            ) 
                             #compute_type="float32")
    return model

def transcriber_fn(audio_file):
    from cs_pa_nlp.faster_whisper import WhisperModel
    import librosa
    import time
    import concurrent.futures

    y, sr = librosa.load(audio_file)
    transcription_model = get_model()
    start = time.time()
    segments, _ = transcription_model.transcribe(y, 
                                                 beam_size=5, 
                                                 language="en", 
                                                 condition_on_previous_text=False, 
                                                 vad_filter=True, 
                                                 vad_parameters={"min_silence_duration_ms": 500}
                                                 )
    
    def transcript_dict(segment):
        return {
            'start': segment[2],-
            'end': segment[3],
            'text': segment[4],
            'no_speech_probability': segment[9]
        }

    with concurrent.futures.ThreadPoolExecutor(max_workers=4) as executor:
        transcriptions = list(executor.map(transcript_dict, segments))

    duration = time.time() - start
    return [transcriptions, duration]

rdd = spark.sparkContext.parallelize(file_names[:5])
transcribed_texts = rdd.map(transcriber_fn).collect()
- sample on 5 files 

[shaunck96]

not much silence .. customer service conversations

[shaunck96]

also any suggestions on parameters above to speed the process up would be greatly appreciated

[gongouveia]

@shaunck96 Have you solved it?
how much time do you want it to take to transcribe the whole batch?
with some tweaks you can transcribe each 1min audio in 1 second using a good consumer GPU.
Activate VAD, decrease model size using some fine tuned language to your idiom.
Run the audio transc. in batch.

You can anchieve to transcribe all the audios in about 2.5 hours max.
WIth small investment for you company, upgrade the AI infrastructure and run it locally.

Plus it is always a good practice when you use sensitive data

[abderrahmane-mhd]

Hello @gongouveia,

I'm also interested in optimizing the processing time of the transcription. I have a NVIDIA A100 GPU (16GB VRAM, 112 GB Memory, 6 Cores), I've created batches of 12 audios and passed them using multiprocessing for transcription (6 processes). I don't know if my code really reduces the processing time (because I only have 1 GPU). I could transcribe 200 audios in 30 minutes (each audio has an average duration of 6 minutes).

1- Options used for transcription:

 model.transcribe(local_audio_path, 
            word_timestamps=True, vad_filter=True, 
            task = "transcribe",
            hallucination_silence_threshold = 2,
            initial_prompt = initial_prompt, language = language,
            vad_parameters=dict(min_silence_duration_ms=500))

2- Python code to process audios using multiprocessing:

def process_rows_in_batches(rows, params, batch_size=12):
    batches = [rows[i:i + batch_size] for i in range(0, len(rows), batch_size)]
    
    results = []

    for batch in batches:
        with ThreadPool(processes=6) as pool:
            # handle_audio uses a globally defined model and uses the transcription parameters defined in 1 (VAD, initial_prompt, etc).
            processed_results = pool.map(handle_audio, batch)
            results.extend(processed_results )
        torch.cuda.empty_cache()

    return results

3- I have tried using UDFs and Spark, but I had to initialize the model inside the udf method, it took 1 hour to transcribe 200 audios.

@udf(df_structure)
def handle_audio_udf(path, call_id, name):
   model = WhisperModel("large-v3", device = "cuda", compute_type = "float16", num_workers= 6)
  # custom_transcribe implements a simple post-processing of audio transcription  and uses the same parameters of transcription defined in 1
   test, segment_details, words = custom_transcribe(model=model, audio_path = path, initial_prompt = initial_prompt)
... remaining logic here

If you have any other suggestions to further reduce the processing time, I'd be really happy to test them!

[gongouveia]

Hello @abderrahmane-mhd ,

1. depending in your language instantiate a smaller model trained in your language from hugging face for example.
   convert it to the faster-whisper format.

2. separate the model encoder and decoder, and run more workers in encoders than in decoders (decoder is more exaustive). This is a harder implementation.

3. use whisper.cpp (https://github.com/ggerganov/whisper.cpp) and use optimization compile flags.

NOTE
As Whisper is quite noise invariant and resistant, I speed up the audio a little bit like1.1x or 1.2x. the audio clips srhink in size and inference decreases the same amount of time. Test it, let me know your results as well.

[shaunck96]

thanks for the suggestion. there already is a multi node gpu cluster setup so might leverage that. Wrote some multi gpu code :

import asyncio
import logging
from concurrent.futures import ThreadPoolExecutor
from multiprocessing import Process, Queue
from typing import List

import torch
import torch.distributed as dist

from cs_pa_nlp.faster_whisper import WhisperModel

# Initialize logger
logging.basicConfig(level=logging.INFO)

class DistributedTranscriber:
    """
    A class for performing distributed transcription of audio files using Whisper models.

    Attributes:
        model_size (str): The size of the Whisper model to be used.
        executor (ThreadPoolExecutor): An executor for running asynchronous tasks.
    """

    def __init__(self, model_size="large-v3"):
        """
        Initialize the DistributedTranscriber with a specified Whisper model size.
        
        Args:
            model_size (str): The size of the Whisper model to be used.
        """
        self.model_size = model_size
        self.executor = ThreadPoolExecutor(max_workers=10)

    def setup_ddp(self, rank: int, world_size: int):
        """
        Sets up the environment for distributed data parallel processing.
        
        Args:
            rank (int): The rank of the current process in the distributed environment.
            world_size (int): The total number of processes involved in the job.
        """
        dist.init_process_group(
            backend='nccl',
            init_method='env://',
            rank=rank,
            world_size=world_size
        )
        torch.cuda.set_device(rank)
        logging.info(f"DDP setup complete for rank {rank}")

    def cleanup_ddp(self):
        """
        Cleans up the distributed data parallel environment and clears CUDA cache.
        """
        dist.destroy_process_group()
        logging.info("DDP environment cleaned up")
        torch.cuda.empty_cache()

    async def transcribe(self, rank: int, world_size: int, audio_files: List[str], result_queue: Queue):
        """
        Asynchronously transcribes a list of audio files using distributed processing.
        
        Args:
            rank (int): The current process's rank in the distributed environment.
            world_size (int): The total number of processes in the distributed environment.
            audio_files (List[str]): List of paths to the audio files to transcribe.
            result_queue (Queue): Queue to collect transcription results from all processes.
        
        Returns:
            None: Results are put into a multiprocessing Queue.
        """
        self.setup_ddp(rank, world_size)
        transcriptions = []
        try:
            model = WhisperModel(self.model_size, device=f"cuda:{rank}", compute_type="float16")
            ddp_model = torch.nn.parallel.DistributedDataParallel(model, device_ids=[rank])
            for audio_file in audio_files:
                segments, _ = ddp_model.module.transcribe(
                    audio_file,
                    beam_size=5,
                    language="en",
                    condition_on_previous_text=False,
                    vad_filter=True,
                    vad_parameters={'min_silence_duration_ms': 500}
                )
                transcriptions.extend([{
                    'start': segment[2],
                    'end': segment[3],
                    'text': segment[4],
                    'no_speech_probability': segment[9]
                } for segment in segments])
                logging.info(f"Successfully transcribed file {audio_file} on GPU {rank}.")

            result_queue.put(transcriptions)
        except Exception as e:
            logging.error(f"Failed to transcribe file on GPU {rank}: {e}")
        finally:
            self.cleanup_ddp()

def worker(rank: int, world_size: int, audio_files: List[str], result_queue: Queue):
    """
    Worker function for distributing the transcription task.
    
    Args:
        rank (int): The current process's rank.
        world_size (int): The total number of processes in the distributed environment.
        audio_files (List[str]): List of audio files to be transcribed.
        result_queue (Queue): Queue to collect results from all processes.
    """
    asyncio.run(DistributedTranscriber().transcribe(rank, world_size, audio_files, result_queue))

def main():
    """
    Main function to orchestrate the distributed transcription across multiple processes.
    
    Returns:
        List[dict]: Aggregated list of all transcription results from all processes.
    """
    world_size = torch.cuda.device_count()
    files = ["/dbfs/mnt/audio/audio1.mp3", "/dbfs/mnt/audio/audio2.mp3", "/dbfs/mnt/audio/audio10000.mp3"]
    file_batches = [files[i::world_size] for i in range(world_size)]

    result_queue = Queue()
    processes = []
    for rank in range(world_size):
        p = Process(target=worker, args=(rank, world_size, file_batches[rank], result_queue))
        p.start()
        processes.append(p)

    for p in processes:
        p.join()

    # Collect results from the queue
    aggregated_results = []
    while not result_queue.empty():
        aggregated_results.extend(result_queue.get())

    return aggregated_results

if __name__ == "__main__":
    results = main()
    print(results)

[abderrahmane-mhd]

Thank you for your reply @shaunck96, I have tried to implement similar logic using multi-node gpu (1 driver and 1 worker), however, I have an error with

model = WhisperModel(self.model_size, device="cuda", compute_type="float16", device_index=[rank])
ddp_model = torch.nn.parallel.DistributedDataParallel(model, device_ids=[rank])

The error is 'WhisperModel' object has no attribute 'named_parameters'

I'm also using Databricks, and when I check the number of gpus I get 1 using torch.cuda.device_count(). But I have two gpus one in the worker and one driver.

Did you have the same problems?