【Python】DeepL API を介して GPT4All と日本語で対話する

　題名の通りです。DeepL API による翻訳を用いて、オープンソースのチャットAIである GPT4All と対話する方法を紹介します。GPT4All の導入については次の記事や公式リポジトリがたよりになります。

グラボ非搭載の低スペックPCでも使える軽量チャットAI「GPT4ALL」の使い方まとめ – GIGAZINE
nomic-ai/gpt4all： gpt4all： a chatbot trained on a massive collection of clean assistant data including code, stories and dialogue
　
　対話プログラムと対話する際の中間処理として Python を用いる方法を紹介します。最近で言えば、そのままでは英語でしか会話できないチャットAIをモデルを変えることなく日本語で会話する際に使えます。

　具体的なコードと実行例は次の通りです。

import signal
import subprocess
from time import sleep

from deepl import Translator
import re

# DeepL APIキーを設定
translator = Translator('xxxxxxxx')


# テキストの翻訳を行う関数
def translate(text: str, target_lang='JA') -> str:
    text = remove_ansi_escape_sequence(text)
    if text == '' or text is None:
        return ''
    translation = translator.translate_text(text=text, target_lang=target_lang)
    if translation.text == '' or translation.text is None:
        return text
    return translation.text


# ANSIエスケープシーケンスを削除する関数
def remove_ansi_escape_sequence(text: str) -> str:
    return re \
        .compile(r'\x1B(?:[@-Z\\-_]|\[[0-?]*[ -/]*[@-~])') \
        .sub('', text)


if __name__ == '__main__':
    # 対話するプログラムを起動する
    process = subprocess.Popen(
        [r'C:\xxxx\gpt4all\chat\gpt4all-lora-quantized-win64.exe'],
        stdin=subprocess.PIPE,
        stdout=subprocess.PIPE,
        cwd=r'C:\xxxx\gpt4all\chat'
    )


    # プロセスをクリーンアップする関数
    # Python終了時にこの関数からプロセスをキルすることで、メモリやCPUをがっつり食べるプロセスが残るのを防ぐ
    def cleanup():
        process.terminate()
        sleep(5)


    signal.signal(signal.SIGTERM, cleanup)
    init_load = True
    try:
        output = ''
        while True:
            # プログラムからの出力を読めるだけ読む
            output += process.stdout.readline().decode('utf-8')
            if output == '' and process.poll() is not None:
                break
            # 起動時のメッセージはフィルタリングしない
            if init_load:
                print(output)
                init_load = False
            else:
                # 出力をフィルタリングする
                filtered_output = translate(output, target_lang='JA')
                # フィルタリングされた出力を表示
                print(filtered_output)
            output = ''

            # ユーザー入力を読み取る
            user_input = input('> ')
            user_input = translate(user_input, target_lang='EN-US')
            # 入力をプログラムに渡す
            process.stdin.write((user_input + '\n').encode('utf-8'))
            process.stdin.flush()
    finally:
        # プログラムの終了
        signal.signal(signal.SIGTERM, signal.SIG_IGN)
        signal.signal(signal.SIGINT, signal.SIG_IGN)
        cleanup()
        signal.signal(signal.SIGTERM, signal.SIG_DFL)
        signal.signal(signal.SIGINT, signal.SIG_DFL)

　やっていることとしては while ループで対話し続ける、対話の際に DeepL API とコマンドラインの修飾外しを介して GPT4All から見たら英語同士で話しているようにする、Pythonの方が終了する際に GPT4All もきっちり終わる様にする、といった具合です。

　このプログラムは大まかに対話部と変換部とプロセス終了部に分けられます。

　対話部が次の部分です。プロセスが無駄に残る危険性を度外視すれば、これくらい短くなります。ここでは GPT4All を使っていますが、GPT4All 以外でも対話するプログラムを Python を介してなんやかんやしたい時も大体同じコードで動きます。

    # 対話するプログラムを起動する
    process = subprocess.Popen(
        [r'C:\xxxx\gpt4all\chat\gpt4all-lora-quantized-win64.exe'],
        stdin=subprocess.PIPE,
        stdout=subprocess.PIPE,
        cwd=r'C:\xxxx\gpt4all\chat'
    )

        output = ''
        while True:
            # プログラムからの出力を読めるだけ読む
            output += process.stdout.readline().decode('utf-8')
            if output == '' and process.poll() is not None:
                break
            filtered_output = translate(output, target_lang='JA')
            # フィルタリングされた出力を表示
            print(filtered_output)
            output = ''

            # ユーザー入力を読み取る
            user_input = input('> ')
            user_input = translate(user_input, target_lang='EN-US')
            # 入力をプログラムに渡す
            process.stdin.write((user_input + '\n').encode('utf-8'))
            process.stdin.flush()

　変換部が次の部分です。ここではコマンドラインの修飾をしている ANSIエスケープシーケンスを外すことと翻訳をしています。やっていることを抽象化すると文字列を元に文字列を返していると言えます。翻訳、修飾以外にも不適切な発言やNGワードのフィルタリングなどを色々とここに追加で組み込むことができます。

# DeepL APIキーを設定
translator = Translator('xxxxxxxx')


# テキストの翻訳を行う関数
def translate(text: str, target_lang='JA') -> str:
    text = remove_ansi_escape_sequence(text)
    if text == '' or text is None:
        return ''
    translation = translator.translate_text(text=text, target_lang=target_lang)
    if translation.text == '' or translation.text is None:
        return text
    return translation.text


# ANSIエスケープシーケンスを削除する関数
def remove_ansi_escape_sequence(text: str) -> str:
    return re \
        .compile(r'\x1B(?:[@-Z\\-_]|\[[0-?]*[ -/]*[@-~])') \
        .sub('', text)


if __name__ == '__main__':
    try:
        output = ''
        while True:
                # 出力をフィルタリングする
                filtered_output = translate(output, target_lang='JA')
                # フィルタリングされた出力を表示
                print(filtered_output)

            # ユーザー入力を読み取る
            user_input = input('> ')
            user_input = translate(user_input, target_lang='EN-US')

　プロセス終了部は次の記事を大いに参考にしました。Pythonが終了する際、必ずサブプロセスが死ぬようにすることで無駄にマシンのリソースを食べることを防いでいます。

Python で終了時に必ず何か実行したい – Qiita#解説編