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

[shaunck96]

yes I am getting the same problem with DDP. Horovad works for me, currently debugging DDP execution.

[abderrahmane-mhd]

Could you share your implementation of Horovod with me? It would really help me out.

By the way, I tested speeding up the audio to 1.1. It slightly speeds up transcription, but the time needed to speed up the audios is about 10 times greater than the time saved in transcription. So, I ended up using the original audios.

[shaunck96]

import os
import logging
import time
from multiprocessing import Process, Queue, set_start_method, TimeoutError
from typing import List

import torch
from faster_whisper import WhisperModel
import librosa
import pandas as pd

Initialize logger

logging.basicConfig(level=logging.INFO)

class DistributedTranscriber:
"""
A class to perform distributed transcription of audio files using Whisper models.

Attributes:
    model_size (str): The size of the Whisper model to be used.
"""

def __init__(self, model_size="large-v2"):
    """
    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.model = WhisperModel(self.model_size, device="cuda", compute_type="float16")

def transcribe(self, rank: int, world_size: int, audio_files: List[str], result_queue: Queue):
    """
    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.
    """
    transcriptions = []
    try:
        for audio_file in audio_files:
            start_time = time.time()
            segments, _ = self.model.transcribe(
                audio_file,
                beam_size=5,
                language="en",
                condition_on_previous_text=False,
                vad_filter=True,
                vad_parameters={'min_silence_duration_ms': 500}
            )
            transcription_time = time.time() - start_time
            transcriptions.extend([{
                'file_path': audio_file,
                'start': segment[2],
                'end': segment[3],
                'text': segment[4],
                'no_speech_probability': segment[9],
                'transcription_time': transcription_time
            } for segment in segments])
            logging.info(f"Successfully transcribed file {audio_file} on GPU {rank}.")

        try:
            result_queue.put(transcriptions, timeout=5)
        except TimeoutError:
            logging.error(f"Timeout while trying to put results in the queue on GPU {rank}.")
    except Exception as e:
        logging.error(f"Failed to transcribe file on GPU {rank}: {e}")
    finally:
        torch.cuda.empty_cache()

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.
"""
# Set the device as early as possible in the worker's execution
torch.cuda.set_device(rank)
logging.info(f"Setting CUDA device to {rank} for process")

transcriber = DistributedTranscriber()
transcriber.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.
"""
customer_recordings = {
    "abfs_path": "", 
    "dbfs_path": ""
}
file_names = [customer_recordings['dbfs_path'] + f'/{i}' for i in os.listdir(customer_recordings['dbfs_path'])]
ws_df = pd.DataFrame(file_names)
ws_df.columns = ['File_Path']

def get_duration(file_path):
    y, sr = librosa.load(file_path)
    return librosa.get_duration(y=y, sr=sr)

ws_df['duration'] = ws_df['File_Path'].apply(get_duration)

bins = [0, 150, 330, 570, 3187]
labels = ['0-150', '150-330', '330-570', '570+']
ws_df['duration_range'] = pd.cut(ws_df['duration'], bins=bins, labels=labels, include_lowest=True)

n = 5
sampled_dfs = [ws_df[ws_df['duration_range'] == label].sample(n, replace=True) for label in labels if ws_df[ws_df['duration_range'] == label].shape[0] >= n]
sampled_df = pd.concat(sampled_dfs, ignore_index=True)

under_two_mins = sampled_df[sampled_df['duration'] < 120].head(15)
two_mins_to_four_mins = sampled_df[(sampled_df['duration'] > 120) & (sampled_df['duration'] < 240)].head(10)
four_mins_to_six_mins = sampled_df[(sampled_df['duration'] > 240) & (sampled_df['duration'] < 360)].head(10)
six_to_eight_mins = sampled_df[(sampled_df['duration'] > 360) & (sampled_df['duration'] < 480)].head(10)
eight_to_ten_mins = sampled_df[(sampled_df['duration'] > 480) & (sampled_df['duration'] < 600)].head(10)
above_ten_mins = sampled_df[sampled_df['duration'] > 600].head(10)
audio = pd.concat([under_two_mins, two_mins_to_four_mins, four_mins_to_six_mins, six_to_eight_mins, eight_to_ten_mins, above_ten_mins], axis=0).reset_index(drop=True)
world_size = torch.cuda.device_count()
audio_files = list(audio['File_Path'])

file_batches = [audio_files[i::world_size] for i in range(world_size)]

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

results = []
while not result_queue.empty():
    try:
        results.extend(result_queue.get(timeout=5))
    except TimeoutError:
        logging.error("Timeout while trying to get results from the queue.")
        break

return results

if name == "main":
try:
set_start_method('spawn')
except RuntimeError:
pass

results = main()

# Save the results as a pandas DataFrame and export to CSV
results_df = pd.DataFrame(results)
save_path = ""
results_df.to_csv(save_path, index=False)
print(f"Transcriptions saved to {save_path}")

[shaunck96]

"""Horovad""" : ```
import horovod.torch as hvd
import torch
import time
from faster_whisper import WhisperModel

Initialize Horovod

hvd.init()
torch.cuda.set_device(hvd.local_rank())

class DistributedTranscriber:
"""
A class to perform distributed transcription of audio files using Whisper models.
"""
def init(self, model_size="medium.en"):
self.model_size = model_size
self.model = WhisperModel(self.model_size,
device="cuda",
compute_type="int8_float16")

def transcribe(self, audio_files):
    transcriptions = []
    times = []
    for audio_file in audio_files:
        start_time = time.time()
        segments, _ = self.model.transcribe(
            audio_file,
            beam_size=5,
            language="en",
            condition_on_previous_text=False,
            vad_filter=True,
            vad_parameters={'min_silence_duration_ms': 500}
        )
        transcription_time = time.time() - start_time
        times.append(transcription_time)
        transcriptions.extend([{
            'start': segment[2],
            'end': segment[3],
            'text': segment[4],
            'no_speech_probability': segment[9]
        } for segment in segments])
    return transcriptions, times

def main():
world_size = hvd.size()
audio_files = list(under_two_mins['File_Path'])
# Divide the audio files among processes
local_files = audio_files[hvd.rank()::world_size]
transcriber = DistributedTranscriber()
results, times = transcriber.transcribe(local_files)

# Calculate average time per process
average_time = sum(times) / len(times) if times else 0

# Gather results and times from all processes
all_results = hvd.allgather_object(results)
all_average_times = hvd.allgather_object(average_time)

if hvd.rank() == 0:
    # Only the master process will print the aggregated results and average times
    print("Transcription Results:", all_results)
    overall_average_time = sum(all_average_times) / len(all_average_times)
    print("Average Transcription Time:", overall_average_time)

if name == "main":
main()

[shaunck96]

import os
import pandas as pd
import time
from faster_whisper import WhisperModel
from concurrent.futures import ProcessPoolExecutor

class Transcribe:
    def __init__(self, device_index):
        self.device_index = device_index
        self.transcription_model = None
        self.load_context()

    def load_context(self):
        self.transcription_model = WhisperModel("medium.en", device="cuda", device_index=self.device_index, compute_type="float16")

    def predict(self, model_input):
        try:
            segments, _ = self.transcription_model.transcribe(
                model_input,
                beam_size=5,
                language="en",
                condition_on_previous_text=False,
                vad_filter=True,
                vad_parameters=dict(min_silence_duration_ms=500)
            )
            return segments
        except Exception as e:
            print(f"Error in transcription on GPU {self.device_index}: {str(e)}")
            return []

def transcribe_files(audio_files, device_index):
    model = Transcribe(device_index)
    transcriptions = []
    for audio_file in audio_files:
        print(f"Transcribing file {audio_file} on GPU {device_index}...")
        start_time = time.time()
        segments = model.predict(audio_file)
        end_time = time.time()
        transcription_time = end_time - start_time

        file_transcription = [{
            'start': segment.start,
            'end': segment.end,
            'text': segment.text,
            'no_speech_probability': segment.no_speech_prob
        } for segment in segments]

        transcriptions.append({
            'file': audio_file,
            'transcription': file_transcription,
            'transcription_time': transcription_time
        })
    return transcriptions

def main():
    dbfs_path = ""
    file_names = [os.path.join(dbfs_path, f) for f in os.listdir(dbfs_path)][:20]
    half_size = len(file_names) // 2

    files_gpu_0 = file_names[:half_size]
    files_gpu_1 = file_names[half_size:]

    with ProcessPoolExecutor(max_workers=2) as executor:
        futures = [
            executor.submit(transcribe_files, files_gpu_0, 0),
            executor.submit(transcribe_files, files_gpu_1, 1)
        ]
        results = [future.result() for future in futures]

    # Combine results from both GPUs
    all_transcriptions = results[0] + results[1]
    transcription_df = pd.DataFrame(all_transcriptions)
    transcription_df.to_csv('', index=False)

if __name__ == "__main__":
    main()
##PROCESSPOOL

[toanhuynhnguyen]

@shaunck96 You can try Batched Faster-Whipser https://github.com/SYSTRAN/faster-whisper?tab=readme-ov-file#batched-faster-whisper

[shaunck96]

Thank you for this, currently testing and speed ups seem promising.

[shaunck96]

some text appears to not get captured though in comparison to single file processing

[toanhuynhnguyen]

@shaunck96 Try this #1001