openai whisper real time的进一步展开说明

在这篇文章中，我将展现怎样使用Python中的OpenAI Whisper近实时地转录现场音频流。我们这样做是为了监控音频流中特定的关键词。同时，转录的文本带有时间戳，供进一步使用。通过对转录文本进行模糊匹配，我们可以找到所需关键词的提及。然后，我们通过Signal Messenger向包括所需部份口述内容的群组或个人发送消息。

背景

这是一个周末内快速开发的原型实现：我想监控一个地方广播电台，以便在比赛中取得关键词的提及。这需要快速完成，因此选择了一个简单的解决方案。而且，它一定要尽量节俭资源，以减少基础设施本钱。虽然它并没有以稳定性为主要关注点进行构建，但实际上在数周内没有出现任何故障。因此，目标到达了！

所有的代码都可以在这个存储库中找到。接下来，我将介绍解决方案的整体结构，并解释其中一些相关的代码部份。

概览

解决方案由三个部份组成：

save_stream.py 连续保存音频流的30秒.mp3文件块的脚本
transcribe.py 延续转录每一个音频块，并使用OpenAI Whisper进行模糊匹配，以监控口述内容中会不会包括我们的关键词。如果匹配成功，则调用msg_group_via_signal.sh脚本。
msg_group_via_signal.sh将报警消息转发给signal-cli工具，然后通过Signal Messenger发送给一个群组。

我们之所以使用OpenAI的Whisper，是由于它是目前表现最好的音频转录模型之一。另外，它很容易获得，并提供区别的模型大小。使用小模型，即便在非英语音频上也能取得不错的结果。另外，它在CPU上的资源利用率足够高，不会落后于音频流。我在AWS的c5a.large EC2机器上部署时获得了良好的结果，每月的本钱约为65美元。为了应对不完善的转录质量，我们在转录中寻觅关键词时使用了模糊搜索。这样就能够减少误报的可能性，但会增加虚假报警的可能性。如果具有更好的规格/ GPU，您可以增加模型的大小以取得更好的质量转录。

详细信息

让我们来看一些代码。我们从save_stream.py开始。它主要由以下函数组成：

def record_stream_to_file ( stream : requests . Response ): “””Record stream audio to files as .mp3 in chunks during recording times Args: stream (requests.Response): Audio stream “”” start_utc = datetime . utcnow () start_local = datetime . now ( tz = LOCAL_TZ ) current_local_time = start_local . time () log . info ( “Current tz time: %s . Stream from: %s Stream until: %s ” , current_local_time , STREAM_TIME_FROM , STREAM_TIME_TO , ) if not STREAM_TIME_FROM < current_local_time < STREAM_TIME_TO : log . warning ( "Not during recording time" ) sys . exit ( 0 ) filename = DATA_PATH + "/stream_" + start_utc . isoformat ( timespec = "seconds" ) + ".mp3" log . info ( "Writing stream to: %s " , filename ) with open ( filename , "wb" ) as file : try : for block in stream . iter_content ( 1024 ): file . write ( block ) if datetime . utcnow () - start_utc > timedelta ( seconds = CHUNK_TIME_SECONDS ): file . close () record_stream_to_file ( stream ) except KeyboardInterrupt : log . info ( “Received keyboard interrupt” ) sys . exit ( 0 )

这个代码非常直观：我们首先检查当前本地时间会不会在预定义的时间段内。这是我们希望监控的时间段。在此时间段以外，我们只需退出脚本。为了部署，我创建了一个简单的bash脚本，用于检查Python脚本会不会已在运行，如果没有，则启动它。然后，我使用cron在每分钟启动bash脚本。这样，我们的save_stream.py只会在我们想要的时候进行录制。非常简单！谁需要Airflow？

在下一步中，我们逐块地对要求的Response对象（打开的音频流）进行迭代。我们将每一个块写入文件，直到到达时间限制（对我而言是30秒），然后递归调用此函数。

因此，我们将在本地文件夹中得到一系列延续30秒长的.mp3文件，这些文件已准备好进行转录。

接下来，我们将专注于transcribe.py脚本。这是我们解决方案的核心。与第一个脚本类似，它也只在我们每天的监控时间内延续运行，否则会停止。

第一步是从上游脚本中获得所有最近的.mp3文件：

def get_recent_files () -> list : “””Return file paths for recently created files Returns: list: File paths “”” log . info ( “Listing recent files” ) now = datetime . utcnow () audio_files = [] for file in sorted ( Path ( PATH_AUDIO_FILES ) . iterdir ()): if “.mp3” in file . name : file_ts = datetime . fromtimestamp ( file . stat () . st_ctime ) if now – file_ts <= timedelta ( minutes = RECENT_FILES_TIME_MIN ): audio_files . append ( file ) log . debug ( "Recent files: %s " , audio_files ) return audio_files 我们按照时间戳对音频文件夹中的文件进行排序，并检查它们是否在我们的近期时间限制内。如果是，则将它们添加到列表中，并准备进行处理。这是为了使我们的代码更加健壮：如果转录脚本在任何时间点失败并退出，cron将重新启动它。我们希望它继续转录，但仅针对最近几分钟的音频流。否则，它将无法实时进行。 转录功能本身非常简单： def transcribe_file ( model , options , file_path : str ) -> str : “””Transcribe the .mp3 file to text Args: model: Whisper Model file_path (str): File path Returns: str: Transcribed text “”” audio = whisper . load_audio ( file_path ) audio = whisper . pad_or_trim ( audio ) mel = whisper . log_mel_spectrogram ( audio ) . to ( model . device ) result = whisper . decode ( model , mel , options ) return result . text # type: ignore

Whisper库负责所有沉重的工作。因此，加载、预处理和解码我们的音频文件只需要四行代码。让我们好好欣赏一下Python的简单性，和开源开发人员的辛苦付出！如果您想了解更多关于创建模型输入所需的信号处理知识，我推荐浏览以下文章。官方博客介绍Whisper也是一个非常有趣的读物。

既然我们已将语音转换为文本，我们可以监视文本中会不会包括我们的关键词：

def search_for_text ( text : str ): “””Search for search term in text and send alarm if found” Args: text (str): Text to search “”” log . info ( “Searching in text” ) text = text . lower () for term in SEARCH_TERMS_LIVE : results = find_near_matches ( term , text , max_l_dist = 2 ) if results : log . debug ( “Search results: %s ” , results ) log . info ( “Found live term: %s ” , term ) send_alarm_to_signal ( text , live = True ) for term in SEARCH_TERMS_DEV : results = find_near_matches ( term , text , max_l_dist = 1 ) if results : log . debug ( “Search results: %s ” , results ) log . info ( “Found dev term: %s ” , term ) send_alarm_to_signal ( text , live = False )

我们使用模糊搜索包来将转录的文本与我们的关键词进行比较。这样即便它们没有完善转录，也能匹配到我们的关键词。从而下降毛病否定报警的概率。这里使用了简单而快速的Levenshtein距离算法。

最后，在关键词匹配时，我们调用了send_alarm_to_signal函数：

def send_alarm_to_signal ( text : str , live = False ): “””Send alarm via signal bash script Args: text (str): Text with match live (bool, optional): Live or test. Defaults to False. “”” message = “This is a test. I’ve picked up the following:

” if live : message = “This is a LIVE. I’ve picked up the following:

” message = message + text subprocess . Popen ([ PATH_TO_SIGNAL_SCRIPT , message ])

这只是调用一个bash脚本，并将匹配的转录文本部份发送给它。这个bash脚本也非常简单：

#!/bin/bash echo “Sending message:” echo ” $1 ” echo “” echo ” $1 ” | /usr/local/bin/signal-cli send –message-from-stdin -g “MyGroupID”

openai whisper real time的常见问答Q&A

!pip install git+https://github.com/openai/whisper.git
!pip install jiwer
!pip install datasets==1.18.3

import whisper

# 加载Whisper模型
model = whisper.load_model()

# 准备音频数据
audio_data = ...

# 使用Whisper进行语音辨认
transcription = whisper.transcribe(model, audio_data)

# 处理辨认结果
...

import whisper

# 加载Whisper模型
model = whisper.load_model()

# 延续录制音频并进行辨认
while True:
    audio_data = record_audio()
    transcription = whisper.transcribe(model, audio_data)
    
    # 处理辨认结果
    ...