Getting the top few transcription results

[shervinemami]

Hi,

Does anyone know how to find a few possible transcription results, instead of just the one transcription result? So if I transcribe some audio that speaks a sentence, I could receive a few text sentences with the top few guesses of the transcription (allowing me to manually choose which sentence is the most correct, instead of whisper determining the best sentence)?

I guess it might be a little related to word confidence scores mentioned in #284 but it would still be different.

[jongwook]

The best_of or beam_size option is designed to do something similar to this:

whisper/whisper/decoding.py

Lines 79 to 80 in 9f70a35

	best_of: Optional[int] = None # number of independent samples to collect, when t > 0
	beam_size: Optional[int] = None # number of beams in beam search, when t == 0

but these will select the one best candidate. Their difference is:

• best_of selects multiple random samples, so it only makes sense with a nonzero temperature and will tend to generate more diverse (i.e. more likely to be wrong) samples.
• beam_size selects the best candidates out of beam search, ranked by the likelihood. These candidates tend to be only slightly different.

So the easiest way would be to repeat the call to decode() multiple times with a nonzero temperature. If you'd like to select from the beam search candidates, you can try tweaking the SequenceRanker implementation:

whisper/whisper/decoding.py

Lines 160 to 166 in 9f70a35

	class SequenceRanker:
	def rank(self, tokens: List[List[Tensor]], sum_logprobs: List[List[float]]) -> List[int]:
	"""
	Given a list of groups of samples and their cumulative log probabilities,
	return the indices of the samples in each group to select as the final result
	"""
	raise NotImplementedError

This defaults to MaximumLikelihoodRanker in the code right below, but you can (for example) replace this with a new implementation that asks the user to select among the candidates in a UI.

[shervinemami]

Thanks a lot @jongwook , that's very helpful! I'll try tweaking SequenceRanger

[abarcovschi]

@shervinemami Were you able to implement this functionality? If so, could you please share how you tweaked the SequenceRanker, as this is the functionality I am looking for also.

[shervinemami]

Hi @abarcovschi , no i haven’t tried implementing this. If you do get it working and upload it, please tag me 😆

[abarcovschi]

@shervinemami I managed to implement this functionality. To make it work I had to modify the source code of whisper/transcribe.py -> transcribe() function to the following:

def transcribe(
    model: "Whisper",
    audio: Union[str, np.ndarray, torch.Tensor],
    *,
    verbose: Optional[bool] = None,
    temperature: Union[float, Tuple[float, ...]] = (0.0, 0.2, 0.4, 0.6, 0.8, 1.0),
    compression_ratio_threshold: Optional[float] = 2.4,
    logprob_threshold: Optional[float] = -1.0,
    no_speech_threshold: Optional[float] = 0.6,
    condition_on_previous_text: bool = True,
    initial_prompt: Optional[str] = None,
    word_timestamps: bool = False,
    prepend_punctuations: str = "\"'“¿([{-",
    append_punctuations: str = "\"'.。,，!！?？:：”)]}、",
    **decode_options,
):
    """
    Transcribe an audio file using Whisper

    Parameters
    ----------
    model: Whisper
        The Whisper model instance

    audio: Union[str, np.ndarray, torch.Tensor]
        The path to the audio file to open, or the audio waveform

    verbose: bool
        Whether to display the text being decoded to the console. If True, displays all the details,
        If False, displays minimal details. If None, does not display anything

    temperature: Union[float, Tuple[float, ...]]
        Temperature for sampling. It can be a tuple of temperatures, which will be successively used
        upon failures according to either `compression_ratio_threshold` or `logprob_threshold`.

    compression_ratio_threshold: float
        If the gzip compression ratio is above this value, treat as failed

    logprob_threshold: float
        If the average log probability over sampled tokens is below this value, treat as failed

    no_speech_threshold: float
        If the no_speech probability is higher than this value AND the average log probability
        over sampled tokens is below `logprob_threshold`, consider the segment as silent

    condition_on_previous_text: bool
        if True, the previous output of the model is provided as a prompt for the next window;
        disabling may make the text inconsistent across windows, but the model becomes less prone to
        getting stuck in a failure loop, such as repetition looping or timestamps going out of sync.

    word_timestamps: bool
        Extract word-level timestamps using the cross-attention pattern and dynamic time warping,
        and include the timestamps for each word in each segment.

    prepend_punctuations: str
        If word_timestamps is True, merge these punctuation symbols with the next word

    append_punctuations: str
        If word_timestamps is True, merge these punctuation symbols with the previous word

    initial_prompt: Optional[str]
        Optional text to provide as a prompt for the first window. This can be used to provide, or
        "prompt-engineer" a context for transcription, e.g. custom vocabularies or proper nouns
        to make it more likely to predict those word correctly.

    decode_options: dict
        Keyword arguments to construct `DecodingOptions` instances

    Returns
    -------
    A dictionary containing the resulting text ("text") and segment-level details ("segments"), and
    the spoken language ("language"), which is detected when `decode_options["language"]` is None.
    """
    dtype = torch.float16 if decode_options.get("fp16", True) else torch.float32
    if model.device == torch.device("cpu"):
        if torch.cuda.is_available():
            warnings.warn("Performing inference on CPU when CUDA is available")
        if dtype == torch.float16:
            warnings.warn("FP16 is not supported on CPU; using FP32 instead")
            dtype = torch.float32

    if dtype == torch.float32:
        decode_options["fp16"] = False

    # Pad 30-seconds of silence to the input audio, for slicing
    mel = log_mel_spectrogram(audio, model.dims.n_mels, padding=N_SAMPLES)
    content_frames = mel.shape[-1] - N_FRAMES

    if decode_options.get("language", None) is None:
        if not model.is_multilingual:
            decode_options["language"] = "en"
        else:
            if verbose:
                print(
                    "Detecting language using up to the first 30 seconds. Use `--language` to specify the language"
                )
            mel_segment = pad_or_trim(mel, N_FRAMES).to(model.device).to(dtype)
            _, probs = model.detect_language(mel_segment)
            decode_options["language"] = max(probs, key=probs.get)
            if verbose is not None:
                print(
                    f"Detected language: {LANGUAGES[decode_options['language']].title()}"
                )

    language: str = decode_options["language"]
    task: str = decode_options.get("task", "transcribe")
    tokenizer = get_tokenizer(
        model.is_multilingual,
        num_languages=model.num_languages,
        language=language,
        task=task,
    )

    if word_timestamps and task == "translate":
        warnings.warn("Word-level timestamps on translations may not be reliable.")

    def decode_with_fallback(segment: torch.Tensor) -> DecodingResult:
        temperatures = (
            [temperature] if isinstance(temperature, (int, float)) else temperature
        )
        decode_result = None

        for t in temperatures:
            kwargs = {**decode_options}
            if t > 0:
                # disable beam_size and patience when t > 0
                kwargs.pop("beam_size", None)
                kwargs.pop("patience", None)
            else:
                # disable best_of when t == 0
                kwargs.pop("best_of", None)

            options = DecodingOptions(**kwargs, temperature=t)
            decode_result = model.decode(segment, options)

            needs_fallback = False
            if (
                compression_ratio_threshold is not None
                and decode_result.compression_ratio > compression_ratio_threshold
            ):
                needs_fallback = True  # too repetitive
            if (
                logprob_threshold is not None
                and decode_result.avg_logprob < logprob_threshold
            ):
                needs_fallback = True  # average log probability is too low
            if (
                no_speech_threshold is not None
                and decode_result.no_speech_prob > no_speech_threshold
            ):
                needs_fallback = False  # silence
            if not needs_fallback:
                break

        return decode_result

    seek = 0
    input_stride = exact_div(
        N_FRAMES, model.dims.n_audio_ctx
    )  # mel frames per output token: 2
    time_precision = (
        input_stride * HOP_LENGTH / SAMPLE_RATE
    )  # time per output token: 0.02 (seconds)
    all_tokens = []
    prompt_reset_since = 0

    if initial_prompt is not None:
        initial_prompt_tokens = tokenizer.encode(" " + initial_prompt.strip())
        all_tokens.extend(initial_prompt_tokens)
    else:
        initial_prompt_tokens = []

    def new_segment(
        *, start: float, end: float, tokens: torch.Tensor, result: DecodingResult
    ):
        tokens = tokens.tolist()
        text_tokens = [token for token in tokens if token < tokenizer.eot]
        return {
            "seek": seek,
            "start": start,
            "end": end,
            "text": tokenizer.decode(text_tokens),
            "tokens": tokens,
            "temperature": result.temperature,
            "avg_logprob": result.avg_logprob,
            "compression_ratio": result.compression_ratio,
            "no_speech_prob": result.no_speech_prob,
        }

    # ******** calculate the first mel segment outside while loop ***********
    time_offset = float(seek * HOP_LENGTH / SAMPLE_RATE)
    mel_segment = mel[:, seek : seek + N_FRAMES]
    segment_size = min(N_FRAMES, content_frames - seek)
    segment_duration = segment_size * HOP_LENGTH / SAMPLE_RATE
    mel_segment = pad_or_trim(mel_segment, N_FRAMES).to(model.device).to(dtype)

    decode_options["prompt"] = all_tokens[prompt_reset_since:]
    result: DecodingResult = decode_with_fallback(mel_segment)

    seeks = [0] * decode_options["beam_size"] # one seek variable for each hypothesis

    if no_speech_threshold is not None:
        # no voice activity check
        should_skip = result.no_speech_prob > no_speech_threshold
        if (
            logprob_threshold is not None
            and result.avg_logprob > logprob_threshold
        ):
            # don't skip if the logprob is high enough, despite the no_speech_prob
            should_skip = False

        if should_skip:
            seeks = [seek + segment_size for seek in seeks]  # fast-forward to the next segment boundary

    current_segments_list = [] # value per each hypothesis, where value is a list of segments, where a segment is a dict
    current_tokens_list = [] # value per each hypothesis, where value is a list

    # for loop over all hypotheses outside while loop for the first mel segment
    for j in range(len(result.tokens)):
        current_segments = []
        hypothesis = torch.tensor(result.tokens[j])

        timestamp_tokens: torch.Tensor = hypothesis.ge(tokenizer.timestamp_begin)
        single_timestamp_ending = timestamp_tokens[-2:].tolist() == [False, True]

        consecutive = torch.where(timestamp_tokens[:-1] & timestamp_tokens[1:])[0]
        consecutive.add_(1)

        if len(consecutive) > 0:
            # if the output contains two consecutive timestamp tokens
            slices = consecutive.tolist()
            if single_timestamp_ending:
                slices.append(len(hypothesis))

            last_slice = 0
            for current_slice in slices:
                sliced_tokens = hypothesis[last_slice:current_slice]
                start_timestamp_pos = (
                    sliced_tokens[0].item() - tokenizer.timestamp_begin
                )
                end_timestamp_pos = (
                    sliced_tokens[-1].item() - tokenizer.timestamp_begin
                )
                current_segments.append(
                    new_segment(
                        start=time_offset + start_timestamp_pos * time_precision,
                        end=time_offset + end_timestamp_pos * time_precision,
                        tokens=sliced_tokens,
                        result=result,
                    )
                )
                last_slice = current_slice

            if single_timestamp_ending:
                # single timestamp at the end means no speech after the last timestamp.
                seeks[j] += segment_size
            else:
                # otherwise, ignore the unfinished segment and seek to the last timestamp
                last_timestamp_pos = (
                    hypothesis[last_slice - 1].item() - tokenizer.timestamp_begin
                )
                seeks[j] += last_timestamp_pos * input_stride
        else:
            duration = segment_duration
            timestamps = hypothesis[timestamp_tokens.nonzero().flatten()]
            if (
                len(timestamps) > 0
                and timestamps[-1].item() != tokenizer.timestamp_begin
            ):
                # no consecutive timestamps but it has a timestamp; use the last one.
                last_timestamp_pos = (
                    timestamps[-1].item() - tokenizer.timestamp_begin
                )
                duration = last_timestamp_pos * time_precision

            current_segments.append(
                new_segment(
                    start=time_offset,
                    end=time_offset + duration,
                    tokens=hypothesis,
                    result=result,
                )
            )
            seeks[j] += segment_size
        try:
            current_segments_list[j].extend([current_segments])
        except IndexError:
            current_segments_list.append([current_segments])

        # if a segment is instantaneous or does not contain text, clear it
        for segments in current_segments_list[j]:
            for segment in segments:
                if segment["start"] == segment["end"] or segment["text"].strip() == "":
                    segment["text"] = ""
                    segment["tokens"] = []
                    segment["words"] = []

        # populate current_tokens_list for this hypothesis 
        try:
            current_tokens_list[j].extend([token for segment in current_segments for token in segment["tokens"]])
        except IndexError:
            current_tokens_list.append([token for segment in current_segments for token in segment["tokens"]])

    # loop through seek values corresponding to hypotheses
    # s_index will have the same range as number of hypotheses
    for s_index in range(len(seeks)):
        seek = seeks[s_index]
        while seek < content_frames:
            time_offset = float(seek * HOP_LENGTH / SAMPLE_RATE)
            mel_segment = mel[:, seek : seek + N_FRAMES]
            segment_size = min(N_FRAMES, content_frames - seek)
            segment_duration = segment_size * HOP_LENGTH / SAMPLE_RATE
            mel_segment = pad_or_trim(mel_segment, N_FRAMES).to(model.device).to(dtype)

            decode_options["prompt"] = all_tokens[prompt_reset_since:]
            result: DecodingResult = decode_with_fallback(mel_segment)
            hypothesis = result.tokens[s_index] # get corresponding hypothesis
            hypothesis = torch.tensor(hypothesis)

            current_segments = []

            if no_speech_threshold is not None:
                # no voice activity check
                should_skip = result.no_speech_prob > no_speech_threshold
                if (
                    logprob_threshold is not None
                    and result.avg_logprob > logprob_threshold
                ):
                    # don't skip if the logprob is high enough, despite the no_speech_prob
                    should_skip = False

                if should_skip:
                    seek += segment_size  # fast-forward to the next segment boundary
                    continue

            timestamp_tokens: torch.Tensor = hypothesis.ge(tokenizer.timestamp_begin)
            single_timestamp_ending = timestamp_tokens[-2:].tolist() == [False, True]

            consecutive = torch.where(timestamp_tokens[:-1] & timestamp_tokens[1:])[0]
            consecutive.add_(1)
           
            if len(consecutive) > 0:
                # if the output contains two consecutive timestamp tokens
                slices = consecutive.tolist()
                if single_timestamp_ending:
                    slices.append(len(hypothesis))

                last_slice = 0
                for current_slice in slices:
                    sliced_tokens = hypothesis[last_slice:current_slice]
                    start_timestamp_pos = (
                        sliced_tokens[0].item() - tokenizer.timestamp_begin
                    )
                    end_timestamp_pos = (
                        sliced_tokens[-1].item() - tokenizer.timestamp_begin
                    )
                    current_segments.append(
                        new_segment(
                            start=time_offset + start_timestamp_pos * time_precision,
                            end=time_offset + end_timestamp_pos * time_precision,
                            tokens=sliced_tokens,
                            result=result,
                        )
                    )
                    last_slice = current_slice

                if single_timestamp_ending:
                    # single timestamp at the end means no speech after the last timestamp.
                    seek += segment_size
                else:
                    # otherwise, ignore the unfinished segment and seek to the last timestamp
                    last_timestamp_pos = (
                        hypothesis[last_slice - 1].item() - tokenizer.timestamp_begin
                    )
                    seek += last_timestamp_pos * input_stride
            else:
                duration = segment_duration
                timestamps = hypothesis[timestamp_tokens.nonzero().flatten()]
                if (
                    len(timestamps) > 0
                    and timestamps[-1].item() != tokenizer.timestamp_begin
                ):
                    # no consecutive timestamps but it has a timestamp; use the last one.
                    last_timestamp_pos = (
                        timestamps[-1].item() - tokenizer.timestamp_begin
                    )
                    duration = last_timestamp_pos * time_precision

                current_segments.append(
                    new_segment(
                        start=time_offset,
                        end=time_offset + duration,
                        tokens=hypothesis,
                        result=result,
                    )
                )
                seek += segment_size

            # if a segment is instantaneous or does not contain text, clear it
            for segments in current_segments:
                if segment["start"] == segment["end"] or segment["text"].strip() == "":
                    segment["text"] = ""
                    segment["tokens"] = []
                    segment["words"] = []
        
            current_segments_list[s_index].extend([current_segments])
            current_tokens_list[s_index].extend([token for segment in current_segments for token in segment["tokens"]])

    # loop through each hypothesis
    out_dicts = []
    for all_toks, segs in zip(current_tokens_list, current_segments_list):
        segs_list = [segment for sublist in segs for segment in sublist]
        out_dicts.append(dict(text=tokenizer.decode(all_toks[len(initial_prompt_tokens) :]), segments=segs_list, language=language))

    return out_dicts

out_dicts will now be a list of dictionary objects, where each dictionary has the following fields:
'text': the natural language transcript.
'segments': a list of dictionary objects that contain a text field with segments of the full text, depending on how the mel spec input was processed (i.e. the full text is composed of multiple segments).

So out_dicts will contain the top N hypotheses outputted by beam search decoding.

To achieve this output, I also modified whisper/decoding.py by:

1. Adding a custom decoding result class:

@dataclass(frozen=True)
class CustomDecodingResult:
    """All hypotheses from beam search"""
    audio_features: Tensor
    language: str
    language_probs: Optional[Dict[str, float]] = None
    tokens: List[List[int]] = field(default_factory=list)
    texts: List[str] = field(default_factory=list)
    avg_logprob: float = np.nan # use just the best hypothesis for this value
    no_speech_prob: float = np.nan
    temperature: float = np.nan
    compression_ratio: float = np.nan # use just the best hypothesis for this value

2. Adding a custom ranker class:

class CustomReturnAllSamplesRanker(SequenceRanker):
    """
    Return list of values, where a value is the likelihood for a hypothesis.
    """
    def __init__(self, length_penalty: Optional[float]):
        self.length_penalty = length_penalty

    def rank(self, tokens: List[List[Tensor]], sum_logprobs: List[List[float]]):
        def scores(logprobs, lengths):
            result = []
            for logprob, length in zip(logprobs, lengths):
                if self.length_penalty is None:
                    penalty = length
                else:
                    # from the Google NMT paper
                    penalty = ((5 + length) / 6) ** self.length_penalty
                result.append(logprob / penalty)
            return result

        # get the sequence with the highest score
        lengths = [[len(t) for t in s] for s in tokens]
        return [(scores(p, l)) for p, l in zip(sum_logprobs, lengths)]

3. Setting the sequence ranker to my new ranker in DecodingTask.__init__():
   self.sequence_ranker = CustomReturnAllSamplesRanker(options.length_penalty)
4. Modifying DecodingTask.run() to:

    @torch.no_grad()
    def run(self, mel: Tensor) -> List[DecodingResult]:
        self.decoder.reset()
        tokenizer: Tokenizer = self.tokenizer
        n_audio: int = mel.shape[0]

        audio_features: Tensor = self._get_audio_features(mel)  # encoder forward pass
        tokens: Tensor = torch.tensor([self.initial_tokens]).repeat(n_audio, 1)

        # detect language if requested, overwriting the language token
        languages, language_probs = self._detect_language(audio_features, tokens)
        if self.options.task == "lang_id":
            return [
                DecodingResult(
                    audio_features=features, language=language, language_probs=probs
                )
                for features, language, probs in zip(
                    audio_features, languages, language_probs
                )
            ]

        # repeat text tensors by the group size, for beam search or best-of-n sampling
        tokens = tokens.repeat_interleave(self.n_group, dim=0).to(audio_features.device)

        # call the main sampling loop
        tokens, sum_logprobs, no_speech_probs = self._main_loop(audio_features, tokens)

        # reshape the tensors to have (n_audio, n_group) as the first two dimensions
        audio_features = audio_features[:: self.n_group]
        no_speech_probs = no_speech_probs[:: self.n_group]
        assert audio_features.shape[0] == len(no_speech_probs) == n_audio

        tokens = tokens.reshape(n_audio, self.n_group, -1)
        sum_logprobs = sum_logprobs.reshape(n_audio, self.n_group)

        # get the final candidates for each group, and slice between the first sampled token and EOT
        tokens, sum_logprobs = self.decoder.finalize(tokens, sum_logprobs)
        tokens: List[List[Tensor]] = [
            [t[self.sample_begin : (t == tokenizer.eot).nonzero()[0, 0]] for t in s]
            for s in tokens
        ]

        # rerank the hypotheses by their likelihood from most likely to least likely
        probabilities = self.sequence_ranker.rank(tokens, sum_logprobs)
        tokens_ordered = [x for _, x in sorted(zip(probabilities[0], tokens[0]), reverse=True)]

        tokens: List[List[int]] = [t.tolist() for t in tokens_ordered]
        texts: List[str] = [tokenizer.decode(t).strip() for t in tokens]

        # NOTE: lengths of decoded beams can have different lengths of words!!!
        # texts_words_lengths = [len(t.split(' ')) for t in texts]
        # if len(set(texts_words_lengths)) > 1:
        #     a=1

        # order sum_logprobs of the hypotheses from most likely to least likely
        sum_logprobs: List[float] = [x for _, x in sorted(zip(probabilities[0], sum_logprobs[0]), reverse=True)]
        avg_logprobs: List[float] = [lp / (len(tok_seq) + 1) for tok_seq, lp in zip(tokens, sum_logprobs)]

        decoding_result = CustomDecodingResult(
                audio_features=audio_features,
                language=languages,
                tokens=tokens,
                texts=texts,
                avg_logprob=avg_logprobs[0],
                no_speech_prob=no_speech_probs[0],
                temperature=self.options.temperature,
                compression_ratio=compression_ratio(texts[0]))

        return [decoding_result]

Hope this helps!

[shervinemami]

That’s great! I’ll trying your code.

[aehlke]

How was your experience with it? Curious if it's better than trying other attempts at nonzero temperature or not

[EgSam]

@abarcovschi
thanks for sharing your codes. I have tried them but faced few issues. I manged to get your code run but I'm still waiting for the results to check them as it is taking a while already.

Issues I faced:
in transcribe.py:
decode_options["beam_size"] was not defined so I had to set it manually to the default value of 5

in decoding.py:
hypothesis = result.tokens[s_index] , s_index goes beyond the length of the tokens so I had to add an if statement to check it and continue if it goes beyond

[hpjang]

@abarcovschi
thanks a lot for sharing your code.
your code is work well
but your modified transcribe() function can't give batch_size as an option
so, Out of Memory problem occurred

how can i give batch_size option?