Python + WhisperX：解锁语音识别的高效新姿势

大家好，我是烤鸭：

   最近在尝试做视频的质量分析，打算利用asr针对声音判断是否有人声，以及识别出来的文本进行进一步操作。asr看了几个开源的，最终选择了openai的whisper，后来发现性能不行，又换了whisperX。这是一篇实战和代码为主的文章。

引言

OpenAI的Whisper是一款强大的自动语音识别（ASR）模型，它支持多语种识别，包括中文，且经过大量的多语言和多任务监督数据训练，具有出色的鲁棒性和准确性。Python作为一种功能强大的编程语言，其丰富的库和简洁的语法使其成为实现语音识别功能的理想选择。本文将介绍如何利用Python集成Whisper，实现高效的语音识别。

目前一天小千的视频调用，平均时长3分钟。显卡是4090，平均识别耗时30s以内，业务无压力。

Whisper模型简介

Whisper是一个开源的语音识别模型，它基于Transformer架构，通过从网络上收集的680,000小时多语言数据进行训练，能够实现对多种语言的准确识别。此外，该模型对口音、背景噪音和技术语言具有很好的鲁棒性，使得其在实际应用中具有广泛的应用前景。

WhisperX 地址：
https://github.com/m-bain/whisperX

安装环境

linux
显卡是 4090
cuda pytorch
ffmpeg

python 需要的依赖

pip install --no-cache-dir flask -i https://mirrors.aliyun.com/pypi/simple
pip install --no-cache-dir ffmpeg-python -i https://mirrors.aliyun.com/pypi/simple
pip install --no-cache-dir wheel -i https://mirrors.aliyun.com/pypi/simple
pip install --no-cache-dir zhconv -i https://mirrors.aliyun.com/pypi/simple
pip install --no-cache-dir numpy -i https://mirrors.aliyun.com/pypi/simple
pip install --no-cache-dir openai-whisper -i https://mirrors.aliyun.com/pypi/simple
pip install --no-cache-dir kafka-python -i https://mirrors.aliyun.com/pypi/simple
pip install --no-cache-dir fastapi -i https://mirrors.aliyun.com/pypi/simple
pip install --no-cache-dir uvicorn -i https://mirrors.aliyun.com/pypi/simple
pip install --no-cache-dir psutil -i https://mirrors.aliyun.com/pypi/simple
pip install --no-cache-dir gputil -i https://mirrors.aliyun.com/pypi/simple
pip install --no-cache-dir requests -i https://mirrors.aliyun.com/pypi/simple
pip install --no-cache-dir use-nacos -i https://mirrors.aliyun.com/pypi/simple
pip install --no-cache-dir pyyaml -i https://mirrors.aliyun.com/pypi/simple
pip install --no-cache-dir rocketmq-client-python -i https://mirrors.aliyun.com/pypi/simple

预期的功能

我想实现的是单台机器性能打满，并行识别asr，接口可以无限制接收请求，异步返回结果。

接口层

使用的是 fastapi 框架

import concurrent.futures
import os
import time

import ffmpeg
import platform
import uvicorn
import asyncio
import psutil
from fastapi import FastAPI, BackgroundTasks, HTTPException, status, Query
from fastapi.responses import JSONResponse
import GPUtil
import requests
import json

from dict_time import TimedMap
from parse_video_param import VideoRequest
from parse_video_callback_param import VideoCallbackRequest
from api_result import ApiResult
from whisper_processor import video_process
from whisperX_processor20241119 import video_process_whisperX
from logging_config import KAFKA_LOGGER
from nacos_config20241119 import register_nacos

app = FastAPI()
executor = concurrent.futures.ThreadPoolExecutor(max_workers=1)  # 线程池
# 定义CPU使用率阈值
threshold_cpu_usage = 95  # 例如，你希望CPU使用率不超过95%
threshold_gpu_usage_MB = 2400  # 例如，你希望显存使用大小 MB
timed_map = TimedMap()

@app.post("/xxxx-video/whisperx")
async def parse_video(request: VideoRequest, background_tasks: BackgroundTasks):
    if not request or not request.path:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="No video URL provided")

    print(f"parse_video, params:{request}")
    # 将处理任务添加到后台任务中，以便不阻塞主线程
    background_tasks.add_task(process_video_whisperx, request, background_tasks)

    # 立即返回处理中响应，告诉客户端请求已经被接收并正在处理
    api_result = ApiResult(1, "success", "", "")
    return JSONResponse(api_result.to_dict(), status_code=status.HTTP_200_OK)
# 异步函数来下载和处理视频
async def process_video_whisperx(request: VideoRequest, background_tasks: BackgroundTasks):
    def sync_process_video_whisperx(request):
        text = ''
        try:
            # 记录方法耗时
            start_time_single = time.time()
            # 下载视频并保存到临时文件
            url = request.path
            chunk_size = request.chunk_size
            # 如果当前cpu使用率超过80%,就把该数据重新加到任务里
            # 获取当前CPU使用率
            cpu_usage = psutil.cpu_percent(interval=1, percpu=False)
            print(f"当前cpu利用率:{cpu_usage}")
            KAFKA_LOGGER.info(f"当前cpu利用率:{cpu_usage}")
            # 获取所有GPU的信息
            gpus = GPUtil.getGPUs()
            isGpuSuffiencent = True
            # 判断CPU使用率是否达到阈值
            if cpu_usage <= threshold_cpu_usage or isGpuSuffiencent:
                # 解析音频地址
                wavPath = getWav(url)
                print(f"mp3 url={wavPath}")
                # 不存在再去生成
                # 异步处理方法，解析音频这块可以忽略，也可以直接用视频地址
                if(not os.path.exists(wavPath)):
                    (
                        ffmpeg
                        .input(url)
                        .output(wavPath, acodec='mp3')
                        .global_args('-loglevel', 'quiet')
                        .run()
                    )
                # 使用whisper处理音频
                text = process_audio_with_whisperx(wavPath, chunk_size)
                end_time_single = time.time()
                # 创建任务并添加到事件循环中，通知业务方
                asyncio.run(callback_task(request, text))
                print(f"视频地址:{url}, 函数执行耗时: {end_time_single - start_time_single}秒")
                KAFKA_LOGGER.info(f"视频地址:{url}, 函数执行耗时: {end_time_single - start_time_single}秒")

                # 清理临时文件
                os.remove(wavPath)
            else:
                print(f"当前cpu已超限,该视频重新加入队列:{url}")
                KAFKA_LOGGER.info(f"当前cpu已超限,该视频重新加入队列:{url}")
                # 暂停5秒
                time.sleep(5)
                # 重新加到队列里
                # 将处理任务添加到后台任务中，以便不阻塞主线程
                background_tasks.add_task(process_video_whisperx, request, background_tasks)
        except Exception as ex:
            print(f"sync_process_video error: {str(ex)}")
            KAFKA_LOGGER.error(f"sync_process_video error: {ex}")
        return text
    loop = asyncio.get_running_loop()
    # 使用线程池运行同步函数，避免阻塞异步事件循环
    return await loop.run_in_executor(executor, sync_process_video_whisperx, request)
# 获取文件路径
def getWav(input_video):
    try:
        # 判断系统是windows还是linux
        operating_system = platform.system()
        # 判断操作系统类型
        if operating_system == 'Windows':
            print("当前系统是Windows")
            audio_path = "C:\\Users\\xxx\\Downloads\\"
        else :
            audio_path = "/tmp/"
        # 从原始路径中获取文件名
        filename = os.path.basename(input_video)

        # 生成新文件的完整路径
        filename_without_extension = os.path.splitext(filename)[0]
        # 使用ffmpeg-python提取音频
        new_filename = os.path.join(audio_path, filename_without_extension) + ".mp3"
    except Exception as ex1:
        print("getWav ex:", str(ex1))
    return new_filename
# 音频解析
def process_audio_with_whisperx(audio_file_path: str, chunk_size: int) -> str:
    text = video_process_whisperX(audio_file_path, chunk_size)
    return text
# 异步回调
async def callback_task(request: VideoRequest, text: str):
    # 创建任务并添加到事件循环中
    task = asyncio.create_task(callback(request, text))
    # 等待任务完成
    await task
# 回调请求方法
async def callback(request: VideoRequest, text: str):

    # 目标URL
    url = request.callback_url

    # JSON格式的参数
    data = {
        'id': request.id,
        'text': text,
        # 添加更多键值对...
    }

    # 设置一些键值对
    timed_map.set(request.path, json.dumps(data), timeout=1800)

    # 设置请求头，告诉服务器我们发送的是JSON数据
    headers = {'Content-Type': 'application/json'}

    # 设置超时时间，这里设置为5秒
    timeout = 5.0
    # 发送POST请求
    response = requests.post(url, data=json.dumps(data), headers=headers, timeout=timeout)
    print(f"url:{url},data: {json.dumps(data)},headers:{headers},response:{response}")
    # 检查请求是否成功
    if response.status_code == 200:
        # 请求成功，处理响应内容
        print("请求成功")
        print(response.json())  # 如果响应内容是JSON格式，可以直接解析
    else:
        # 请求失败，打印错误信息
        print(f"请求失败，状态码：{response.status_code}")
        print(response.text)  # 打印响应的文本内容
# 启动应用
if __name__ == "__main__":
    register_nacos()
    uvicorn.run(app, host="0.0.0.0", port=5000)

whisperX

import whisperx
from whisperx.asr import FasterWhisperPipeline
import time
import torch
import gc
import os

ENV = os.environ.get('ENV', 'development')
device = torch.device("cuda:0" if torch.cuda.is_available() else "cpu")
if ENV == 'production':
    batch_size = 16
    compute_type = "float16"
    model_name = "large-v2"
else:
    # reduce if low on GPU mem
    batch_size = 4
    # compute_type = "float16"  # change to "int8" if low on GPU mem (may reduce accuracy)
    # change to "int8" if low on GPU mem (may reduce accuracy)
    compute_type = "int8"
    model_name = "medium"
class WhisperXProcessor:
    fast_model: FasterWhisperPipeline
    def loadModel(self):
        # 1. Transcribe with original whisper (batched)
        self.fast_model = whisperx.load_model("medium", device.type, compute_type=compute_type)
        print("模型加载完成")

    def asr(self, filePath: str, chunk_size: int):
        print(f'asr start filePath:{filePath}')
        start = time.time()
        audio = whisperx.load_audio(filePath)
        result = self.fast_model.transcribe(audio, batch_size=batch_size, chunk_size = chunk_size)
        print(result)
        end = time.time()
        print('识别使用的时间：', end - start, 's')
        torch.cuda.empty_cache()
        gc.collect()
        return result

def video_process_whisperX(audio_path, chunk_size):
    app = WhisperXProcessor()
    app.loadModel()
    text = app.asr(audio_path, chunk_size)
    return text

结果验证

发送请求

curl -XPOST 'http://localhost:5000/xxxx-video/whisperX' -H 'Content-Type: application/json' -d '{"id":1,"path":"https://vc16-bd1-pl-agv.autohome.com.cn/video-26/0A33363922E51BDE/2025-02-10/FC68CC971BB8B9A46F15C4841F4F2CE2-200-wm.mp4?key=F77E8D3251C4560FA47E36563A5D5668&time=1739187850","callback_url":"http://localhost:8088/xxx/demo/testParseVideo"}'

结果日志，2分钟的视频，大概用了60s。

在这里插入图片描述

文章参考

ASR强力模型「Whisper」：解密Whisper

Python实现语音识别（whisperX）

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：/a/969433.html

如若内容造成侵权/违法违规/事实不符，请联系我们进行投诉反馈qq邮箱809451989@qq.com，一经查实，立即删除！

相关文章

【Linux】Ubuntu Linux 系统——Node.js 开发环境

【Linux】Ubuntu Linux 系统——Node.js 开发环境

ℹ️大家好&#xff0c;我是练小杰&#xff0c;今天星期五了&#xff0c;同时也是2025年的情人节&#xff0c;今晚又是一个人的举个爪子&#xff01;&#xff01; &#x1f642; 本文是有关Linux 操作系统中 Node.js 开发环境基础知识&#xff0c;后续我将添加更多相关知识噢&a…
阅读更多...
Oracle查看执行计划

Oracle查看执行计划

方式一&#xff08;查看的真实的使用到的索引&#xff09; 1.执行解释计划 2.查看结果 可以看到使用了RANGE SCAN范围扫描的索引 方式二&#xff08;查看的是预测的可能会用到的索引&#xff09; 1.执行解释计划sql explain plan for select * from COURSE where COURSE_…
阅读更多...
百度 AI开源！将在6月30日开源文心大模型4.5系列

百度 AI开源！将在6月30日开源文心大模型4.5系列

【大力财经】直击互联网最前线&#xff1a;百度近期动作频频&#xff0c;先是宣布将在未来数月陆续推出文心大模型4.5系列&#xff0c;并于6月30日正式开源。 据大力财经了解&#xff0c;自DeepSeek开源之风盛行全球后&#xff0c;开源闭源路径的选择就成为AI领域的热门话题&a…
阅读更多...
【DDD系列-2】风暴出的领域模型

【DDD系列-2】风暴出的领域模型

为什么使用DDD​ 三个问题​ 1.为什么我们的系统越做越多&#xff0c;越来越庞大&#xff0c;还需要不断的重构&#xff1f;​ 2.为什么我们的系统业务越来越复杂&#xff0c;服务层的代码越来越多难以维护&#xff0c;不敢维护&#xff1f;​ 3.为什么一旦业务变化或者数据…
阅读更多...
基于YALMIP和cplex工具箱的微电网最优调度算法matlab仿真

基于YALMIP和cplex工具箱的微电网最优调度算法matlab仿真

目录 1.课题概述 2.系统仿真结果 3.核心程序与模型 4.系统原理简介 4.1 系统建模 4.2 YALMIP工具箱 4.3 CPLEX工具箱 5.完整工程文件 1.课题概述 基于YALMIP和cplex工具箱的微电网最优调度算法matlab仿真。通过YALMIP和cplex这两个工具箱&#xff0c;完成微电网的最优调…
阅读更多...
visual studio导入cmake项目后打开无法删除和回车

visual studio导入cmake项目后打开无法删除和回车

通过Cmakelists.txt导入的项目做删除和回车无法响应&#xff0c;需要点击项目&#xff0c;然后选择配置项目就可以了
阅读更多...
npm安装时无法访问github域名的解决方法

npm安装时无法访问github域名的解决方法

个人博客地址&#xff1a;npm安装时无法访问github域名的解决方法 | 一张假钞的真实世界 今天在用npm install的时候出现了github项目访问不了的异常&#xff1a; npm ERR! Error while executing: npm ERR! /bin/git ls-remote -h -t https://github.com/nhn/raphael.git np…
阅读更多...
解锁豆瓣高清海报(三)从深度爬虫到URL构造，实现极速下载

解锁豆瓣高清海报(三)从深度爬虫到URL构造，实现极速下载

脚本地址: 项目地址: Gazer PosterBandit_v2.py 前瞻 之前的 PosterBandit.py 是按照深度爬虫的思路一步步进入海报界面来爬取, 是个值得学习的思路, 但缺点是它爬取慢, 仍然容易碰到豆瓣的 418 错误, 本文也会指出彻底解决旧版 418 错误的方法并提高爬取速度. 现在我将介绍…
阅读更多...
一维差分算法篇：高效处理区间加减

一维差分算法篇：高效处理区间加减

那么在正式介绍我们的一维差分的原理前&#xff0c;我们先来看一下一维差分所应用的一个场景&#xff0c;那么假设我们现在有一个区间为[L,R]的一个数组&#xff0c;那么我要在这个数组中的某个子区间比如[i,m] (L<i<m<R)进行一个加k值或者减去k值的一个操作&#xff…
阅读更多...
信息收集-Web应用JS架构URL提取数据匹配Fuzz接口WebPack分析自动化

信息收集-Web应用JS架构URL提取数据匹配Fuzz接口WebPack分析自动化

知识点&#xff1a; 1、信息收集-Web应用-JS提取分析-人工&插件&项目 2、信息收集-Web应用-JS提取分析-URL&配置&逻辑 FUZZ测试 ffuf https://github.com/ffuf/ffuf 匹配插件 Hae https://github.com/gh0stkey/HaE JS提取 JSFinder https://github.com/Threez…
阅读更多...
Python基础语法精要

Python基础语法精要

文章目录 一、Python的起源二、Python的用途三、Python的优缺点优点缺点 四、基础语法&#xff08;1&#xff09;常量和表达式&#xff08;2&#xff09;变量变量的语法&#xff08;i&#xff09;定义变量&#xff08;ii&#xff09;变量命名的规则 &#xff08;3&#xff09;变…
阅读更多...
测试方案整理

测试方案整理

搜索引擎放在那里&#xff1f;研发 查看问题样本或者在提取再批量入录等情况&#xff0c;一旦我没有勾选或者全选中已经批量入录的样本&#xff0c;那么在直接点击批量提取或查看问题样本的后&#xff0c;会自动默认为选择全选样本还是按照输入错误处理&#xff1f; 批量查看返…
阅读更多...
开启对话式智能分析新纪元——Wyn商业智能 BI 携手Deepseek 驱动数据分析变革

开启对话式智能分析新纪元——Wyn商业智能 BI 携手Deepseek 驱动数据分析变革

2月18号&#xff0c;Wyn 商业智能 V8.0Update1 版本将重磅推出对话式智能分析&#xff0c;集成Deepseek R1大模型&#xff0c;通过AI技术的深度融合&#xff0c;致力于打造"会思考的BI系统"&#xff0c;让数据价值触手可及&#xff0c;助力企业实现从数据洞察到决策执…
阅读更多...
政策赋能科技服务，CES Asia 2025将展北京科技新貌

政策赋能科技服务，CES Asia 2025将展北京科技新貌

近日&#xff0c;《北京市支持科技服务业高质量发展若干措施》正式印发&#xff0c;为首都科技服务业的腾飞注入了强大动力。 该《若干措施》提出了三方面14条政策措施。在壮大科技服务业市场主体方面&#xff0c;不仅支持科技服务业企业向平台化和综合性服务机构发展&#xf…
阅读更多...
2024春秋杯网络安全联赛冬季赛wp

2024春秋杯网络安全联赛冬季赛wp

web flask 根据题目描述&#xff0c;很容易想到ssti注入 测试一下 确实存在 直接打payload {{lipsum.globals[‘os’].popen(‘cat f*’).read()}} file_copy 看到题目名字为file_copy&#xff0c; 当输入路径时会返回目标文件的大小&#xff0c; 通过返回包&#xff0c…
阅读更多...
Mac os部署本地deepseek+open UI界面

Mac os部署本地deepseek+open UI界面

一.部署本地deepseek 使用ollama部署&#xff0c;方便快捷。 ollama介绍&#xff1a;Ollama 是一个高效、便捷的人工智能模型服务平台&#xff0c;提供多样化的预训练模型&#xff0c;涵盖自然语言处理、计算机视觉、语音识别等领域&#xff0c;并支持模型定制和微调。其简洁…
阅读更多...
分布式技术

分布式技术

一、为什么需要分布式技术&#xff1f; 1. 科学技术的发展推动下 应用和系统架构的变迁&#xff1a;单机单一架构迈向多机分布式架构 2. 数据大爆炸&#xff0c;海量数据处理场景面临问题 二、分布式系统概述 三、分布式、集群 四、负载均衡、故障转移、伸缩性 负载均衡&a…
阅读更多...
基于ssm的超市订单管理系统

基于ssm的超市订单管理系统

一、系统架构 前端&#xff1a;jsp | web components | jquery | css | ajax 后端&#xff1a;spring | springmvc | mybatis 环境&#xff1a;jdk1.8 | mysql | maven | tomcat 二、代码及数据 三、功能介绍 01. 登录 02. 首页 03. 订单管理 04. 供应…
阅读更多...
尚硅谷爬虫note004

尚硅谷爬虫note004

一、urllib库 1. python自带&#xff0c;无需安装 # _*_ coding : utf-8 _*_ # Time : 2025/2/11 09:39 # Author : 20250206-里奥 # File : demo14_urllib # Project : PythonProject10-14#导入urllib.request import urllib.request#使用urllib获取百度首页源码 #1.定义一…
阅读更多...
在nodejs中使用RabbitMQ（三）Routing、Topics、Headers

在nodejs中使用RabbitMQ（三）Routing、Topics、Headers

示例一、Routing exchange类型direct&#xff0c;根据消息的routekey将消息直接转发到指定队列。producer.ts 生产者主要发送消息&#xff0c;consumer.ts负责接收消息&#xff0c;同时也都可以创建exchange交换机&#xff0c;创建队列&#xff0c;为队列绑定exchange&#xff…
阅读更多...
最新文章

Copyright @ 2022~2023