Zenn
🤩

Streamlit＋Flask＋Whisperで社内オンプレ文字起こしサーバーを構築(同期処理)

に公開
AI
Python
Streamlit
文字起こし
Whisper
tech

✨ はじめに

会議の文字起こしをしてみたいけど、

  • 外部に録音データを送るのは情報漏洩が不安…
  • TeamsやZoomの文字起こしは精度がイマイチ…

そんなとき、軽量なGPU（例：RTX 1650などVRAM4GB程度）さえあれば、自社内のローカル環境で高精度な文字起こしを実現できます！

今回は、Streamlit + Flask + Whisperを使ってシンプルな文字起こしアプリを構築してみました。

📂 構成概要

このシステムは以下の3つのPythonファイルで構成されています：

ファイル名 役割
app_flask.py ユーザーが音声ファイルをアップロードするUIを提供
server_flask.py ファイルを受け取り、処理スクリプトを呼び出す
transcribe_flask.py Whisperを使って音声ファイルを文字起こし
app_flask.py
import streamlit as st
import requests
import time

# 秒を「〇分〇秒」の形式に変換する関数
def convert_seconds(seconds):
    minutes = seconds // 60  # 分を計算（整数除算）
    remaining_seconds = seconds % 60  # 残りの秒を計算
    return f"{int(minutes)}分{int(remaining_seconds)}秒"

# アプリケーションのタイトルと説明を設定
st.title("音声文字起こし")
st.write('**Whisperを利用して音声データを文字起こしすることが出来ます。**')

# サーバーのURL設定
server_url = "http://localhost:5000/"

# サーバーに接続できるか確認
with st.spinner("サーバーのチェック中..."):  
    try:
        requests.get(server_url)  # サーバーへのGETリクエスト
    except requests.ConnectionError:
        st.error('サーバーが立ち上がっていません。', icon="🚨")  # エラーメッセージを表示
        st.stop()  # アプリを停止

# モデル選択のラジオボタン
model = st.radio("model",["汎用モデル","チューニングモデル"])

# 音声ファイルのアップロード機能
audio_file = st.file_uploader("音声ファイルをアップロードしてください", type=["mp3", "wav", "m4a", "mp4"],key="audio_file_trancribe")

# ファイルがアップロードされた場合の処理
if audio_file:
    # 出力ファイル名を設定
    st.session_state.transcribe_file_name = "文字起こし結果_"+audio_file.name.replace(".mp4","").replace(".mp3","").replace(".wav","").replace(".m4a","")+".txt"
    
    # 学習用データ提供のトグルボタン
    if button_save_audio := st.toggle("学習用の為音声ファイルをフィードバックする",key="button_save_audio",
                                        help="音声文字起こしモデルを改善するために、音声ファイルを集めています。音声ファイルを学習データとして利用させていただきます。"):
        st.subheader("ご協力ありがとうございます🤗")
        st.balloons()  # 風船エフェクトを表示
    
    # 文字起こし開始ボタン
    button_trans_start = st.button("文字起こしを開始する",type="primary")
    if button_trans_start:
        start_time = time.time()  # 処理開始時間を記録
        with st.spinner("**文字起こしを実行中...**"):
            try:
                # サーバーにデータを送信して文字起こし実行をリクエスト
                response = requests.post(
                    server_url+"transcribe_server",
                    files={"audio": audio_file.getvalue()},  # 音声ファイルデータ
                    data={"model": model,"save_audio":button_save_audio,"file_name":audio_file.name},  # 追加データ
                )

                if response.status_code == 200:  # 成功の場合
                    end_time = time.time()  # 処理終了時間を記録
                    st.session_state.execution_time = end_time - start_time  # 実行時間を計算
                    st.session_state.transcribe_data = response.json()  # 結果データを保存
                else:
                    st.write("Error: ", response.text)  # エラーメッセージを表示
            except requests.ConnectionError:
                st.error('サーバーとの接続が解除されました。', icon="🚨")
                st.stop()

# 文字起こし結果の表示
if "transcribe_data" in st.session_state:
    st.write("実行時間：", convert_seconds(st.session_state.execution_time))  # 実行時間を表示
    probability = round(float(st.session_state.transcribe_data['language_probability']) * 100, 1)  # 言語検出の確信度を計算
    st.write(f"検出言語: {st.session_state.transcribe_data['language']} 信用度 {probability}%")  # 検出言語と確信度を表示
    
    # 文字起こし結果のダウンロードボタン
    st.download_button(label="文字起こし結果をダウンロードする",data=st.session_state.transcribe_data["full_text"],file_name=st.session_state.transcribe_file_name,icon=":material/download:")
    
    # 文字起こし結果の表示
    st.markdown("**文字起こし結果**")
    st.markdown(st.session_state.transcribe_data["time_line"], unsafe_allow_html=True)  # タイムスタンプ付きの文字起こし結果を表示
server_flask.py
from flask import Flask, request
import os
import logging
logging.basicConfig(level=logging.DEBUG)
app = Flask(__name__)
from transcribe_flask import transcribe

# 文字起こしAPI（POSTリクエスト用）のエンドポイント
@app.route('/transcribe_server', methods=['POST'])
def transcribe_server():
    try:
        # リクエストからデータを取得
        audio_file = request.files['audio']  # 音声ファイル
        model = request.form['model']  # 使用するモデル
        save_audio = request.form['save_audio']  # 音声保存フラグ
        file_name = request.form['file_name']  # ファイル名
        
        # 音声ファイルを一時的に保存
        audio_file.save(file_name)
        
        # 学習用に音声を保存する場合の処理
        if save_audio == "True":  
            with open(file_name, 'rb') as f_src:  # 元ファイルを読み込み
                destination_path = "path_tp_save"  # 保存先のパス
                with open(destination_path, 'wb') as f_dst:  # 保存先ファイルに書き込み
                    f_dst.write(f_src.read())
        
        # 汎用モデルの場合の処理
        if model == "汎用モデル":
            result = transcribe(audio_file=file_name)  # 文字起こし実行

        # 一時ファイルを削除
        os.remove(file_name)
        return result  # 結果を返す

    except Exception as e:
        return str(e), 500  # エラーが発生した場合は500エラーを返す

# メインプログラム（直接実行された場合）
if __name__ == "__main__":
    app.run(host="0.0.0.0", port=5000, debug=True)  # サーバーを起動（すべてのインターフェースでリッスン）
transcribe_flask.py
from flask import jsonify
from faster_whisper import WhisperModel

# 秒を「〇分〇秒」の形式に変換する関数
def convert_seconds(seconds):
    minutes = seconds // 60  # 分を計算（整数除算）
    remaining_seconds = seconds % 60  # 残りの秒を計算
    return f"{int(minutes)}分{int(remaining_seconds)}秒"

# 音声ファイルを文字起こしする関数
def transcribe(audio_file):
    # Whisperモデル（large-v3）をGPUで初期化
    model = WhisperModel("large-v3", device="cuda", compute_type="float16")
    
    # 音声ファイルの文字起こしを実行
    segments, info = model.transcribe(audio_file,
                                      language = "ja",  # 日本語を指定
                                      beam_size=5,  # ビームサーチの幅（精度向上のため）
                                      vad_filter=True,  # 音声区間検出フィルターを有効化
                                      without_timestamps=True,  # タイムスタンプなしの出力を無効化
                                      prompt_reset_on_temperature=0,  # プロンプトリセットの温度閾値
                                      # initial_prompt=""  # 初期プロンプト（今回は未使用）
                                      )

    # 結果を格納する変数を初期化
    full_text=""  # 全文テキスト
    time_line=""  # タイムスタンプ付きテキスト
    
    # 各セグメント（文章）ごとに処理
    for segment in segments:
        # タイムスタンプ付きのテキストを生成（開始時間 -> 終了時間の形式）
        time_line+="[%s -> %s] %s" % (convert_seconds(segment.start), convert_seconds(segment.end), segment.text)+"  \n"
        # 全文テキストに追加
        full_text+=segment.text+"\n"

    # 結果をJSON形式で返すためのデータ作成
    result = {
        "language": info.language,  # 検出された言語
        "language_probability": info.language_probability,  # 言語検出の確信度
        "time_line":time_line,  # タイムスタンプ付きテキスト
        "full_text":full_text  # 全文テキスト
    }

    # JSON形式で結果を返す
    return jsonify(result)

https://github.com/tsuzukia21/st-transcribe

処理フローは以下のようになります：

[ユーザー]
   ↓（音声ファイルをアップロード）
[app_flask.py]
   ↓（POST送信）
[server_flask.py]
   ↓（Whisperで文字起こし）
[transcribe_flask.py]
   ↓
[テキスト結果を表示]

📅 実行方法

1. 必要ライブラリのインストール

pip install flask faster_whisper streamlit

faster_whisperを実行するときは別途cuda、cudnn環境が必要になります。こちらは本記事では紹介しません。以下記事を参考にするかぐぐってください
https://siro-yamaneko.hatenablog.jp/entry/2024/08/03/210826
https://qiita.com/tf63/items/0c6da72fe749319423b4

2. アプリ、サーバー起動

python server_flask.py
python app_flask.py

3. ブラウザでアクセス

http://localhost:8501

🔹 各ファイルの説明

app_flask.py

app_flask.py は、ユーザーがブラウザ上で音声ファイルを選択し、サーバーにアップロードするためのUIを提供する役割を担います。今回はStreamlitを使ってシンプルに実装しています。

主な処理内容は以下の通りです：

文字起こしサーバーが起動しているかチェック

# サーバーに接続できるか確認
with st.spinner("サーバーのチェック中..."):  
    try:
        requests.get(server_url)  # サーバーへのGETリクエスト
    except requests.ConnectionError:
        st.error('サーバーが立ち上がっていません。', icon="🚨")  # エラーメッセージを表示
        st.stop()  # アプリを停止

本アプリは、Flaskで立ち上げたサーバー（server_flask.py）に対してHTTPリクエストを送信する構成となっています。
そのため、サーバー側が起動していないと文字起こし処理は正常に動作しません。

文字起こし処理を始める前に、以下のように server_flask.py を起動しておく必要があります：

python server_flask.py

app_flask.pyからのリクエストは、デフォルトで http://localhost:5000/transcribe_server に送られます。
そのため、ポートやホストを変更している場合はURLを合わせて修正する必要があります。

ファイルアップロードUI

audio_file = st.file_uploader("音声ファイルをアップロードしてください", type=["mp3", "wav", "m4a", "mp4"],key="audio_file_trancribe")

この部分では、Streamlitの file_uploader コンポーネントを使って、ユーザーにファイルを選択してもらいます。type引数で、アップロード可能なファイル形式を制限しています。

「文字起こし開始」ボタンを表示し、クリックされたら処理開始

button_trans_start = st.button("文字起こしを開始する",type="primary")

st.button はボタンUIを作成します。

Flaskサーバーにファイルを送信する

response = requests.post(
    server_url+"transcribe_server",
    files={"audio": audio_file.getvalue()},  # 音声ファイルデータ
    data={"model": model,"save_audio":button_save_audio,"file_name":audio_file.name},  # 追加データ
)

この部分で、Flaskで立ち上げた server_flask.py の /transcribe_server エンドポイントに対してファイルを送信しています。requests.post によりHTTPリクエストを送信し、ファイルは辞書形式で files に指定しています。

サーバーから返ってきたテキストを表示する

if "transcribe_data" in st.session_state:
    st.write("実行時間：", convert_seconds(st.session_state.execution_time))  # 実行時間を表示
    probability = round(float(st.session_state.transcribe_data['language_probability']) * 100, 1)  # 言語検出の確信度を計算
    st.write(f"検出言語: {st.session_state.transcribe_data['language']} 信用度 {probability}%")  # 検出言語と確信度を表示
    
    # 文字起こし結果のダウンロードボタン
    st.download_button(label="文字起こし結果をダウンロードする",data=st.session_state.transcribe_data["full_text"],file_name=st.session_state.transcribe_file_name,icon=":material/download:")
    
    # 文字起こし結果の表示
    st.markdown("**文字起こし結果**")
    st.markdown(st.session_state.transcribe_data["time_line"], unsafe_allow_html=True)  # タイムスタンプ付きの文字起こし結果を表示

サーバー側で文字起こしされた結果を、そのまま画面上に表示します。返ってくるのはJSON形式で、ユーザーがアップロードした内容の文字起こしがここに表示されるというわけです。どのようなJSON形式かは後述します。
また、文字起こし結果はダウンロードボタンでテキスト形式でダウンロードできます。

補足*ファインチューニングについて

今回は実装していませんが、公開されているwhisperモデルでは社内用語や技術用語を上手く文字起こしすることが出来ません。実際に文字起こしサーバを運用する場合、whisperモデルのファインチューニングをお勧めします。ファインチューニング方法は以下記事が参考になります。
https://zenn.dev/k_sone/articles/e0c08268986ac2
https://zenn.dev/k_sone/articles/4d137d58dd06a6

ファインチューニングするにあたって学習用データが必要になりますので許可をもらえれば集めておきたいとこです。

    # 学習用データ提供のトグルボタン
    if button_save_audio := st.toggle("学習用の為音声ファイルをフィードバックする",key="button_save_audio",
                                        help="音声文字起こしモデルを改善するために、音声ファイルを集めています。音声ファイルを学習データとして利用させていただきます。"):
        st.subheader("ご協力ありがとうございます🤗")
        st.balloons()  # 風船エフェクトを表示

ファインチューニング不要って場合はここはコメントアウトしちゃっておkです。

server_flask.py

このファイルは、ユーザーがアップロードした音声ファイルを受け取り、実際に文字起こし処理を行うプログラム（transcribe_flask.py）を呼び出して、結果をユーザーに返す役割を果たします。

主な処理内容は以下の通りです：

1. エンドポイントの作成

Flaskでは、@app.route('/transcribe_server', methods=['POST']) のように書くことで、特定のURLにアクセスがあったときの処理を定義できます。ここでは、POSTされた音声ファイルを受け取る処理をしています。

2. アップロードされたファイルを保存

アップロードされたファイルは request.files['file'] で受け取ることができます。これをローカルに一時保存します。

def transcribe_server():
    try:
        # リクエストからデータを取得
        audio_file = request.files['audio']  # 音声ファイル
        model = request.form['model']  # 使用するモデル
        save_audio = request.form['save_audio']  # 音声保存フラグ
        file_name = request.form['file_name']  # ファイル名
        
        # 音声ファイルを一時的に保存
        audio_file.save(file_name)

3. transcribe_flask.py を呼び出して変換

別のPythonスクリプト（transcribe_flask.py）を実行します。このスクリプトがWhisperを使って音声ファイルを文字に変換します。

from transcribe_flask import transcribe

result = transcribe(audio_file=file_name)

このように外部スクリプトを起動します。

4. 結果を読み込んでレスポンスとして返す

transcribe_flask.py によって生成されたJSONファイルを読み込んで、その中の文字起こし結果を表示します。

transcribe_flask.py

このファイルは、実際にWhisperを使って音声を文字起こしする処理を行うスクリプトです。

主な処理内容は以下の通りです：

1. 引数で受け取った音声ファイルを開く

Flaskサーバー側から呼び出されたときに、ファイル名（パス）が渡されるので、それを元に音声ファイルを読み込みます。

2. Whisperモデルで文字起こし

OpenAIのWhisperを使い、音声を文字起こしします。

from faster_whisper import WhisperModel
model = WhisperModel("large-v3", device="cuda", compute_type="float16")

3. テキストをJSONとして保存

文字起こし結果はresult["full_text"]の形式で取得できます。

result = {
    "language": info.language,  # 検出された言語
    "language_probability": info.language_probability,  # 言語検出の確信度
    "time_line":time_line,  # タイムスタンプ付きテキスト
    "full_text":full_text  # 全文テキスト
}

📋 おわりに

自社内で安全に音声文字起こしをしたい、議事録作成を自動化させたいなどのニーズは多くあります。
今回のようにFlaskとWhisper、streamlitを組み合わせれば、オンプレミスで手軽に文字起こし環境を構築可能です！

今回の実装は“同期処理”なので、音声を送ると変換が終わるまで画面ではクルクル⏳が回りっぱなしで、ユーザーは完了を待つしかありません。これを非同期に切り替えると、サーバーは「受け付けたよ」とすぐ返事を返し、あとから「いま30％くらい進んでるよ」の進捗状況や途中文字起こし結果の表示などを知らせてくれます。これが同期処理と非同期処理の大きな違いでUX的に全然違いますね。次記事では非同期処理を用いた音声処理アプリを作成します！
https://github.com/tsuzukia21/st-transcribe

Discussion