ChatGPTAdvent Calendar 2024

Vosk + gpt 4o-mini + pythonで作るリアルタイム翻訳ツール

Posted at 2024-12-21

Vosk + gpt 4o-mini + pythonで作るリアルタイム翻訳ツール

本記事では、Pythonを使用してリアルタイムで英語の音声を日本語に翻訳するツールの作成方法について解説します。このツールは、Voskを用いた音声認識、OpenAIのAPIを利用した翻訳、そしてTkinterによるGUIを組み合わせて構築されています。

概要

このツールは、ユーザーがマイクに向かって話す英語の音声をリアルタイムで日本語に翻訳し、GUI上に表示します。以下の技術を組み合わせて実現しています：

Vosk: ローカルで動作するオープンソースの音声認識ライブラリ。
OpenAI API: 高精度な翻訳を提供するために使用。
Tkinter: Python標準のGUIライブラリを使用して、ユーザーインターフェースを構築。

主な機能として、以下が挙げられます：

音声の継続的なキャプチャと認識。
一定の条件（50文字以上のテキスト蓄積、無音検出、話者変更）で翻訳処理を開始。
翻訳結果をタイピング効果で表示し、ユーザーに自然な形で提示。

必要なもの

このツールを動作させるために必要なものは以下の通りです：

Python 3.7以上
Voskモデル: 英語用のモデルをダウンロードします。例: vosk-model-en-us-0.22
OpenAI APIキー: OpenAIのアカウントから取得。
必要なPythonパッケージ:
- vosk
- sounddevice
- requests
- tkinter（Python標準ライブラリ）

インストール手順

Pythonのインストール

Python公式サイトから最新のPythonをインストールしてください。

Pythonダウンロードページ
Voskモデルのダウンロード

Voskの英語モデルをダウンロードし、適切なディレクトリに展開します。
```
wget https://alphacephei.com/vosk/models/vosk-model-en-us-0.22.zip
unzip vosk-model-en-us-0.22.zip
```
Pythonパッケージのインストール

必要なパッケージをインストールします。
```
pip install vosk sounddevice requests
```
OpenAI APIキーの取得と設定

OpenAIのアカウントを作成し、APIキーを取得します。取得したキーを環境変数に設定します。
```
export OPENAI_API_KEY='your-openai-api-key'
```

コードの構造

以下に、ツールの主要なコード部分について説明します。

1. ログ設定

ログをファイルとコンソールに出力するための設定を行います。

import logging

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s: %(levelname)s: %(message)s',
    handlers=[
        logging.FileHandler("translation_log.txt", encoding='utf-8'),
        logging.StreamHandler(sys.stdout)
    ]
)

2. グローバル設定とファイル読み込み

システムプロンプトや用語集などの設定ファイルを読み込みます。

def load_file(filename: str) -> str:
    """テキストファイル読み込み。存在しない場合は空文字列を返す。"""
    if os.path.exists(filename):
        with open(filename, "r", encoding="utf-8") as f:
            return f.read()
    return ""

SYSTEM_PROMPT = load_file("systemPrompt.txt")  # システムプロンプト
GLOSSARY_DATA = load_file("glossary.txt")      # 用語集

3. OpenAI API呼び出し関数

翻訳リクエストをOpenAI APIに送信し、レスポンスを取得します。

def call_openai_api(payload, retries=3):
    logging.info(f"REQUEST: {json.dumps(payload, ensure_ascii=False)}")

    for attempt in range(retries):
        try:
            response = requests.post(
                OPENAI_ENDPOINT,
                headers={
                    "Content-Type": "application/json",
                    "Authorization": f"Bearer {OPENAI_API_KEY}"
                },
                json=payload,
                timeout=15
            )
            response.raise_for_status()
            resp_json = response.json()
            logging.info(f"RESPONSE: {json.dumps(resp_json, ensure_ascii=False)}")
            return resp_json
        except requests.exceptions.RequestException as e:
            logging.error(f"OpenAI API attempt {attempt+1} failed: {e}")
            time.sleep(2)
    return None

4. ペイロード生成とテキスト整形

音声認識結果を元に翻訳リクエストのペイロードを生成します。

def build_payload(conversation, system_prompt: str, glossary: str, new_input: str,
                  source_lang="en", target_lang="ja") -> dict:
    full_text = ""
    for msg in conversation:
        if msg["role"] == "user":
            full_text += msg["content"]
    if len(full_text) > 1500:
        full_text = full_text[-1500:]

    context_msg = (
        f"【これまでの英語原文（最新1500文字）】\n{full_text}\n\n"
        f"元言語: {source_lang}、目標言語: {target_lang}\n\n"
        "新しく認識されたテキスト(new_input)のみを翻訳してください。"
        "過去のテキストは参考用であり、再翻訳しないでください。"
    )

    messages = [
        {"role": "system", "content": system_prompt},
        {"role": "user", "content": f"用語集:\n{glossary}"},
        {"role": "system", "content": context_msg},
        {"role": "user", "content": new_input},
    ]

    return {
        "model": MODEL_NAME,
        "messages": messages,
        "max_tokens": 1000,
        "temperature": 0.5,
    }

5. 音声認識と翻訳処理

音声データを認識し、翻訳するバックグラウンドプロセスを構築します。

VoskSpeechRecognitionThreadクラス

音声を認識し、テキストをキューに追加します。

class VoskSpeechRecognitionThread(threading.Thread):
    def __init__(self, recognized_queue: queue.Queue, source_lang="en-US"):
        super().__init__()
        self.recognized_queue = recognized_queue
        self.source_lang = source_lang
        self.running = True

        if not os.path.exists(VOSK_MODEL_PATH):
            logging.error(f"Vosk model not found at: {VOSK_MODEL_PATH}")
            self.model = None
            self.running = False
            return
        self.model = Model(VOSK_MODEL_PATH)
        logging.info("[INFO] Loaded Vosk model successfully.")

        self.sample_rate = 16000
        self.rec = KaldiRecognizer(self.model, self.sample_rate)
        self.rec.SetWords(True)

        try:
            self.stream = sd.RawInputStream(
                samplerate=self.sample_rate,
                blocksize=16000,
                dtype='int16',
                channels=1,
                callback=self.callback
            )
            self.stream.start()
            logging.info("[INFO] sounddevice stream started.")
        except Exception as e:
            logging.error(f"Could not open sounddevice stream: {e}")
            self.model = None
            self.running = False

    def callback(self, indata, frames, time_info, status):
        if not self.running:
            return
        if status:
            logging.warning(f"sounddevice status: {status}")
        try:
            waveform = bytes(indata)
            if self.rec.AcceptWaveform(waveform):
                res = self.rec.Result()
                text = self.extract_text(res)
                if text:
                    self.recognized_queue.put(text)
                    logging.info(f"Recognized text (Vosk) >> {text}")
        except Exception as e:
            logging.error(f"Error in callback: {e}")

    def run(self):
        if not self.model:
            return
        logging.info("[INFO] Vosk SpeechRecognitionThread started.")
        while self.running:
            try:
                time.sleep(0.1)
            except Exception as e:
                logging.error(f"Unexpected in Vosk thread: {e}")
                time.sleep(0.5)
        self.stream.stop()
        self.stream.close()
        logging.info("[INFO] Vosk SpeechRecognitionThread stopped.")

    def stop(self):
        self.running = False

    @staticmethod
    def extract_text(vosk_json_str: str) -> str:
        try:
            data = json.loads(vosk_json_str)
            return data.get("text", "").strip()
        except json.JSONDecodeError:
            return ""

TranslationThreadクラス

キューからテキストを取り出し、翻訳を行います。

class TranslationThread(threading.Thread):
    def __init__(self, text_queue: queue.Queue, result_queue: queue.Queue):
        super().__init__()
        self.text_queue = text_queue
        self.result_queue = result_queue
        self.running = True

    def run(self):
        logging.info("[INFO] TranslationThread started.")
        while self.running:
            try:
                text = self.text_queue.get(timeout=1)
                if text:
                    logging.info(f"TranslationThread received text: {text}")
                    translated = do_translation(text)
                    if translated:
                        self.result_queue.put(translated)
            except queue.Empty:
                continue
            except Exception as e:
                logging.error(f"TranslationThread encountered an error: {e}")
        logging.info("[INFO] TranslationThread stopped.")

    def stop(self):
        self.running = False

do_translation関数

テキストを翻訳し、整形します。

def do_translation(text: str) -> str:
    logging.info(f"Starting translation for text: {text}")
    payload = build_payload(
        conversation=app.conversation,
        system_prompt=SYSTEM_PROMPT,
        glossary=GLOSSARY_DATA,
        new_input=text,
        source_lang="en",
        target_lang="ja"
    )

    response_json = call_openai_api(payload)
    if not response_json:
        logging.error("Translation API call failed.")
        return "[翻訳エラー: 接続不可]"

    result_content = extract_response_content(response_json)
    if not result_content:
        logging.error("No valid translation in response.")
        return "[翻訳エラー: レスポンス不正]"

    cleaned = remove_forbidden_chars(result_content)
    formatted = format_translation_result(cleaned)

    logging.info(f"Translation result: {formatted}")
    return formatted

6. GUIの構築

Tkinterを使用して、ユーザーインターフェースを構築します。

class RealTimeTranslatorApp:
    def __init__(self, root):
        self.root = root
        self.root.title("リアルタイム音声翻訳 (Vosk + sounddevice) - 英語→日本語")
        self.root.geometry("800x600")
        self.root.minsize(400, 300)
        self.root.resizable(True, True)

        # 透過度の設定（0.75: 75%透明）
        self.root.attributes("-alpha", 0.75)

        # メインフレーム
        self.frame = tk.Frame(self.root, bg="#1a1a1a")
        self.frame.grid(row=0, column=0, sticky="nsew")

        # グリッドの設定
        self.root.grid_rowconfigure(0, weight=1)
        self.root.grid_columnconfigure(0, weight=1)
        self.frame.grid_rowconfigure(1, weight=1)
        self.frame.grid_columnconfigure(0, weight=1)

        # 日本語フォントの設定
        available_fonts = list(font.families())
        preferred_font = "Yu Gothic UI" if "Yu Gothic UI" in available_fonts else "Meiryo"
        self.text_font = font.Font(family=preferred_font, size=14)

        # テキストエリア
        self.text_area = tk.Text(
            self.frame,
            bg="#2d2d2d",
            fg="#f0f0f0",
            font=self.text_font,
            wrap=tk.WORD,
            insertbackground="#f0f0f0",
            highlightthickness=0,
            bd=0,
            relief=tk.FLAT
        )
        self.text_area.grid(row=1, column=0, sticky="nsew", padx=10, pady=10)

        # ステータスバー
        self.status_var = tk.StringVar()
        self.status_var.set("待機中...")
        self.status_bar = tk.Label(
            self.frame,
            textvariable=self.status_var,
            bg="#1a1a1a",
            fg="#ffffff",
            font=("Helvetica", 10),
            anchor="w"
        )
        self.status_bar.grid(row=2, column=0, sticky="ew", padx=10, pady=(0,10))
        self.frame.grid_rowconfigure(2, weight=0)
        self.frame.grid_columnconfigure(0, weight=1)

        # キューとスレッドの初期化
        self.recognized_queue = queue.Queue()
        self.translation_queue = queue.Queue()
        self.translation_result_queue = queue.Queue()

        self.sr_thread = VoskSpeechRecognitionThread(
            recognized_queue=self.recognized_queue,
            source_lang="en-US"
        )
        self.sr_thread.start()
        logging.info("[INFO] Speech recognition thread started.")

        self.translation_thread = TranslationThread(
            text_queue=self.translation_queue,
            result_queue=self.translation_result_queue
        )
        self.translation_thread.start()
        logging.info("[INFO] Translation thread started.")

        # 翻訳待ちバッファと会話履歴
        self.text_buffer = ""
        self.conversation = []

        # 翻訳のタイピング効果用のキュー
        self.typing_queue = queue.Queue()
        self.is_typing = False

        # フォントサイズ変更のバインド
        self.root.bind("<Control-MouseWheel>", self.change_font_size_windows)
        self.root.bind("<Control-Button-4>", self.change_font_size_linux)
        self.root.bind("<Control-Button-5>", self.change_font_size_linux)
        self.root.bind("<Control-MouseWheel>", self.change_font_size_mac)

        # ウィンドウの閉じるボタンにon_closeを割り当て
        self.root.protocol("WM_DELETE_WINDOW", self.on_close)

        # 無音検出用タイムスタンプと閾値
        self.last_speech_time = time.time()
        self.silence_threshold = 1.0  # 秒

        # 定期的にチェック
        self.poll_recognized_text()
        self.poll_translation_result()
        self.poll_typing_queue()
        self.poll_silence()

    def poll_recognized_text(self):
        while not self.recognized_queue.empty():
            recognized_text = self.recognized_queue.get()
            self.text_buffer += recognized_text + " "
            logging.info(f"Accumulated text: {self.text_buffer}")
            self.last_speech_time = time.time()

            if len(self.text_buffer) >= 50:
                self.translation_queue.put(self.text_buffer)
                logging.info(f"Queued for translation: {self.text_buffer}")

                self.conversation.append({"role": "user", "content": self.text_buffer})
                logging.info(f"Updated conversation: {self.conversation}")

                self.text_buffer = ""
                self.status_var.set("翻訳中...")

        self.root.after(100, self.poll_recognized_text)

    def poll_translation_result(self):
        while not self.translation_result_queue.empty():
            translation = self.translation_result_queue.get()
            if translation:
                self.typing_queue.put(translation)
                logging.info(f"Queued translation result: {translation}")
                self.status_var.set("翻訳完了")

        self.root.after(200, self.poll_translation_result)

    def poll_typing_queue(self):
        if not self.typing_queue.empty() and not self.is_typing:
            translation = self.typing_queue.get()
            self.is_typing = True
            logging.info(f"Starting to type translation: {translation}")
            self.type_text(translation)

        self.root.after(50, self.poll_typing_queue)

    def poll_silence(self):
        current_time = time.time()
        elapsed = current_time - self.last_speech_time
        if elapsed > self.silence_threshold and self.text_buffer.strip():
            self.translation_queue.put(self.text_buffer)
            logging.info(f"Queued for translation due to silence: {self.text_buffer}")

            self.conversation.append({"role": "user", "content": self.text_buffer})
            logging.info(f"Updated conversation due to silence: {self.conversation}")

            self.text_buffer = ""
            self.status_var.set("翻訳中...")

        self.root.after(500, self.poll_silence)

    def type_text(self, text: str, index: int = 0):
        if index < len(text):
            self.text_area.insert(tk.END, text[index])
            self.text_area.see(tk.END)
            index += 1
            self.root.after(15, self.type_text, text, index)
        else:
            self.is_typing = False
            logging.info("Finished typing translation.")
            self.status_var.set("待機中...")

    def on_close(self):
        try:
            logging.info("アプリケーションを終了します。")
            if self.sr_thread and self.sr_thread.is_alive():
                logging.info("Stopping speech recognition thread...")
                self.sr_thread.stop()
                self.sr_thread.join()
                logging.info("Speech recognition thread stopped.")
            if self.translation_thread and self.translation_thread.is_alive():
                logging.info("Stopping translation thread...")
                self.translation_thread.stop()
                self.translation_thread.join()
                logging.info("Translation thread stopped.")
            self.root.destroy()
            logging.info("Tkinter root destroyed.")
        except Exception as e:
            messagebox.showerror("エラー", f"アプリケーションの終了中にエラーが発生しました: {e}")
            logging.error(f"Error during on_close: {e}")

    def change_font_size_windows(self, event):
        if event.delta > 0:
            self.adjust_font_size(1)
        else:
            self.adjust_font_size(-1)

    def change_font_size_linux(self, event):
        if event.num == 4:
            self.adjust_font_size(1)
        elif event.num == 5:
            self.adjust_font_size(-1)

    def change_font_size_mac(self, event):
        if event.delta > 0:
            self.adjust_font_size(1)
        else:
            self.adjust_font_size(-1)

    def adjust_font_size(self, delta):
        current_size = self.text_font.actual()["size"]
        new_size = current_size + delta
        if 8 <= new_size <= 36:
            self.text_font.configure(size=new_size)
            logging.info(f"Font size adjusted to: {new_size}")

7. メイン関数

アプリケーションを初期化し、メインループを開始します。

def main():
    global app
    app = RealTimeTranslatorApp(tk.Tk())
    logging.info("[INFO] Application initialized.")

    def signal_handler(sig, frame):
        logging.info("Ctrl + C が押されました。アプリを終了します。")
        app.on_close()
        sys.exit(0)

    signal.signal(signal.SIGINT, signal_handler)

    try:
        logging.info("[INFO] Starting main loop.")
        app.root.mainloop()
    except KeyboardInterrupt:
        logging.info("KeyboardInterrupt発生。アプリを終了します。")
        app.on_close()

    logging.info("===== アプリ終了 =====")

if __name__ == "__main__":
    main()

シーケンス図

以下のシーケンス図は、ツールの全体的な流れと各コンポーネントの役割を視覚的に表現しています。参加者を統合し、シンプルにまとめています。

シーケンスフローの詳細:

ユーザーが話す
- ユーザーが音声入力デバイスに向かって話し始めます。
音声のキャプチャ
- 音声入力デバイスがユーザーの音声をキャプチャし、バックグラウンドプロセスに送信します。
音声認識とテキスト蓄積
- バックグラウンドプロセスが音声を認識し、50文字以上のテキストが蓄積されるか、無音が検出されたり話者が変更されたりした場合に翻訳処理を開始します。
翻訳リクエストの送信
- バックグラウンドプロセスが蓄積されたテキストをOpenAI APIに送信し、翻訳を依頼します。
翻訳結果の受信と通知
- OpenAI APIが翻訳されたテキストを返し、バックグラウンドプロセスがGUIに翻訳完了を通知します。
翻訳結果の表示
- GUIが翻訳されたテキストをフェッチし、ユーザーにタイピング効果で表示します。

使い方

環境変数の設定

OpenAI APIキーを環境変数に設定します。
```
export OPENAI_API_KEY='your-openai-api-key'
```
システムプロンプトと用語集の準備
- systemPrompt.txt: システムプロンプトの内容を記述。
- glossary.txt: 用語集を記述。
スクリプトの実行

上記のコードをreal_time_translator.pyとして保存し、実行します。
```
python real_time_translator.py
```
使用方法
- アプリケーションを起動すると、GUIが表示されます。
- ユーザーがマイクに向かって英語で話すと、リアルタイムで日本語に翻訳されて表示されます。
- 翻訳結果はタイピング効果で表示され、自然な形でユーザーに提示されます。

まとめ

本記事では、Pythonを使用してリアルタイムで英語音声を日本語に翻訳するツールの作成方法について解説しました。Voskを用いた音声認識、OpenAI APIを利用した翻訳、TkinterによるGUIを組み合わせることで、効率的かつユーザーフレンドリーな翻訳ツールを構築することができます。

このツールは、ビジネスミーティングや国際交流など、リアルタイムでのコミュニケーションを支援する場面で活用できます。さらに、シーケンス図を用いてツールの仕組みを視覚的に理解することで、カスタマーへの説明やチーム内での共有がスムーズになります。

ご参考になりましたら「♡いいね」を押して頂けると嬉しいです。

You get articles that match your needs
You can efficiently read back useful information
You can use dark theme

What you can do with signing up

Vosk + gpt 4o-mini + pythonで作るリアルタイム翻訳ツール

Vosk + gpt 4o-mini + pythonで作るリアルタイム翻訳ツール

目次

概要

必要なもの

インストール手順

コードの構造

1. ログ設定

2. グローバル設定とファイル読み込み

3. OpenAI API呼び出し関数

4. ペイロード生成とテキスト整形

5. 音声認識と翻訳処理

VoskSpeechRecognitionThreadクラス

TranslationThreadクラス

do_translation関数

6. GUIの構築

7. メイン関数

シーケンス図

使い方

まとめ