書き溜めた文章とChatGPTでチャットボットを作ってみる

始めに

飲んでいるときに友人に、ネットサービスに書き溜めているメモから情報を抜き出してチャットボットに話させてみたい、という希望が出てきました。chatGPT発表以降、雨後のタケノコのように出てきたチャットボットサービスを見て、何かそれっぽいものつくれないかなぁと調べたところ、材料はそこここにあり、組み合わせでできそうだったので作ってみました。データを取ってくるところは各人違うと思うので、データがcsv形式でそろっているところからご説明したいと思います。

追記2023/10/26

下記コード類をgithubにアップしましたご利用ください。

参考サイト

以下のページからコードを借用、参考にしています。ありがとうございます。

FlaskとChatGPTを使って薬剤情報サービスを作成しよう！

• 本サイトのflaskフレームワークをほぼ使用させてもらっています。

ChatGPT APIとFaissを使って長い文章から質問応答する仕組みを作ってみる

• embedingの部分とchatGPTに質問する部分について使用させてもらっています。

データ

友人のデータを掲載するわけにいかないので、自分で小さいデータを作ってみました。このcsvは１カラム目にその文章の題などのタグ、２カラム目に文章内容をそのまま入れています。例えば議事録や日記では１カラム目に日付を、２カラム目に内容を入れればいいと思います。処理するときに、１カラム目と２カラム目はくっつけて、一つの文章として扱います。

test.csv

"ねこのちゃあについて","猫のちゃあは、人懐っこいトラ猫です。大概の人が近づいても驚きません。父親にも餌をおねだりしにいきます。まるで人の気持ちを理解するかのような行動をします。"
"いぬのジョンとヨーコについて","犬のジョンと、ヨーコはトイプードルの夫婦です。２匹は散歩が大好きで毎日の散歩を楽しみにしています。途中で出くわす大きな犬にも堂々としています。"
"ねこのこむについて","猫のこむは、野良猫経験のある三毛猫です。なので、とても臆病で大概のひとを避けます。お客さんが来るといつもものかげに隠れています。"
"ぶたのももについて","子豚のモモは、まるまる太っています。野菜が大好きで毎日もりもり食べます。お散歩も大好きで、毎日犬のジョンとヨーコにくっついて一緒に散歩を楽しみます。"
"いぬのベスについて","犬のベスはオスの柴犬です。毎日散歩をしますが、出会う犬という犬と喧嘩をします。ほんとに気が小さいです。"
"ねずみのちゅー太について","ネズミのちゅー太はいつもケージで忙しくしています。お気に入りの遊具で走り回ったり、かと思えば寝ていたり。好き勝手な生活が羨ましい。"
"ねこのタンについて","猫のタンは気ままな性格です。いつもマイペースで自分の興味の赴くまま暮らしています。"
"かめの幸太郎について","かめの幸太郎は今年30歳のお年寄り。毎日野菜を食べて元気いっぱい。町の中をゆっくり散歩する姿は街の人気者です。"
"ねこのあんずについて","猫のあんずは、勝気な性格です。好奇心旺盛で、なにかあると駆けつけて見物をしています。"
"ねこのたまについて","猫のたまは、温厚でやさしいです。年寄のためいつも寝ています。人懐こく、知らない人が近づいても平気です。"

ライブラリ

本チャットボットはcsvの前処理をするスクリプトと、flaskでwebページを動作させるスクリプトに分かれており、双方で同じ関数を使用するのでその部分を共通化してライブラリにしてあります。
このサイトの関数を使用させていただいております。

token_embeding.py

import sys
import torch
from transformers import BertJapaneseTokenizer, BertForMaskedLM
from transformers import BertJapaneseTokenizer, BertModel
from janome.tokenizer import Tokenizer
import numpy as np
import re

class tokenEmbeding():

    # embedingで使いたくないワードを登録
    useless_words = ['様','さん','なに','何','匹','頭']
    # Bertのtokenizerを起動
    tokenizer = BertJapaneseTokenizer.from_pretrained('cl-tohoku/bert-base-japanese-whole-word-masking')
    model_bert = BertModel.from_pretrained('cl-tohoku/bert-base-japanese-whole-word-masking', output_hidden_states=True)
    model_bert.eval()
    # janomeのtokenizerを起動
    t = Tokenizer()

    def __init__(self):
        return

    # embedingに邪魔なのでhttp以下を削除
    def remove_http(self,textlist):
        tlist = []
        for i in textlist:
            tlist.append(re.sub('http\S+', ' ',i))
        return tlist
    
    # embedingに邪魔なので名詞句のみを抜き出し、ついでに使わない語句も消去
    def get_nouns(self,text):
        terms = [token.surface for token in self.t.tokenize(text) if token.part_of_speech.startswith('名詞')]
        terms2 = [ token for token in terms if not (token in self.useless_words) ]
        return terms2
    
    # 各文章のembedingの計算
    def calc_embedding(self,text):
        btokens = self.get_nouns(text)
        ids = self.tokenizer.convert_tokens_to_ids(["[CLS]"] + btokens[:126] + ["[SEP]"])
        tokens_tensor = torch.tensor(ids).reshape(1, -1)
        with torch.no_grad():
            output = self.model_bert(tokens_tensor)
        return output[1].numpy()

useless_wordsには、文書全体にたくさん出てくる名詞を登録しておきます。多すぎる名詞はいろいろ邪魔になるので、除外できるようにしております。適当に足し引きしてください。

前処理スクリプト

前述の文章csvを前処理します。結果はカレントのnpy/ディレクトリに保存されるので、npy/ディレクトリを作成しておいてください。前処理では、csvの読み込みで、textlistを作成し、そのembedingを計算してインデックスを算出しております。このインデックスは文書の絞り込み時にワードだけで絞り込めなかった場合に優先度をつけるために使用されます。
実行は
python3 get_embeding.py test.csv
です。ずらずらと名詞の配列が出てきて終了したら準備完了です。

get_embeding.py

import sys
import faiss
import numpy as np
import pandas as pd
import token_embeding

filename = sys.argv[1]
textlist = []
te = token_embeding.tokenEmbeding()

# textlist作成
df = pd.read_csv(filename, header=None)
for index, i in df.iterrows():
    txt = i[0] + " " + i[1]
    textlist.append(txt)
print(len(textlist))
tlist = te.remove_http(textlist)

np.save('./npy/textlist.npy',textlist,fix_imports=False)

# embedingのindex作成
tmp_embeds = list(map(te.calc_embedding, tlist))
embeddings = []
for embed in tmp_embeds:
    embeddings.append(embed[0])

index = faiss.IndexFlatIP(embeddings[0].shape[0])
index.add(np.array(embeddings))

np.save('./npy/index.npy',index,fix_imports=False)

webページおよびスクリプト本体

flaskでwebページを生成するのと、入力された質問から文章を選び出して質問プロンプトを作成してchatGPTに問い合わせる部分です。プログラムの内容はコメントで記載しています。

app.py

from flask import Flask, render_template, request
import os
import sys
import numpy as np
import openai
import token_embeding

openai.api_key = os.getenv("OPENAI_API_KEY")
te = token_embeding.tokenEmbeding()
index = np.load('./npy/index.npy',allow_pickle=True).tolist()
textlist = np.load('./npy/textlist.npy',allow_pickle=True).tolist()
prev_question = ""
prev_fact = ""
select_sentences = 4

app = Flask(__name__)

# レスポンスを整形する関数
def format_response(response):
    # レスポンス内の各部分を適切な形式に変換
    response = response.replace("・", "\n・")
    # 改行をHTML形式に変換
    response = response.replace("\n", "<br>")
    return response

# ルートURLへのリクエストを処理する関数
@app.route('/', methods=['GET', 'POST'])
def home():
    if request.method == 'POST':
        # フォームから質問を取得
        question = request.form['question']
        # 質問に対するレスポンスを生成
        response = generate_response(question)
        # レスポンスを含むページを表示
        return render_template('index.html', response=response)
    # フォームを表示
    return render_template('index.html')

# 質問に対するレスポンスを生成する関数
def generate_response(question):
    # 質問が含まれている文章を選択
    fact = get_fact_from_question(question)
    # 質問と文章から答を生成
    answer = fact_qa(fact, question)
    # レスポンスを整形
    formatted_response = format_response(answer)
    return formatted_response

# 質問が含まれている文章を選択
def get_fact_from_question(question):
    global prev_question
    global prev_fact

    toklist = te.get_nouns(question)
    question_embed = te.calc_embedding(question)
    D, I = index.search(question_embed, len(textlist))
    # 各名詞が含まれる文書の数を算,出
    cn, markcntr, mark, words = get_candidate_num(toklist,textlist)

    if markcntr > 1:
        # 名詞2つのコンビネーションを探索
        flg, cn2, word1, word2 = get_2conbination(toklist,textlist,markcntr,mark)
        # 文書数が最小の名詞を選択
        if flg == 1:
            cntr, idx = get_word_num(word1,word2,cn2,textlist)
            # 最小の文書数が3以下であればその文書を選択
            if cntr <= select_sentences:
                fact = get_text_by_words(textlist,word1[idx],word2[idx])
            # 最小の文書数が3より大きい場合はコサイン類似度で選択
            else:
                fact = get_text_by_cos(textlist,word1[idx],word2[idx],I)
        else:
            # 文書数が最小の名詞を選択
            cntr, idx = get_word_num(words,[],cn,textlist)
            # 空の質問には前回の質問を返す
            if cntr <= select_sentences:
                fact = get_text_by_words(textlist,words[idx],"")
            # 最小の文書数が3より大きい場合はコサイン類似度で選択
            else:
                fact = get_text_by_cos(textlist,words[idx],"",I)
    else:
        # 文書数が最小の名詞を選択
        cntr, idx = get_word_num(words,[],cn,textlist)
        # 空の質問には前回の質問を返す
        if cntr == len(textlist):
            question = prev_question
            fact = prev_fact
        # 最小の文書数が3以下であればその文書を選択
        elif cntr <= select_sentences:
            fact = get_text_by_words(textlist,words[idx],"")

        # 最小の文書数が3より大きい場合はコサイン類似度で選択
        else:
            fact = get_text_by_cos(textlist,words[idx],"",I)

    # 空の質問の場合の前情報を保存
    prev_question = question
    prev_fact = fact

    return fact
    
# 各名詞が含まれる文書の数を算,出
def get_candidate_num(toklist,textlist):
    cntr = 0
    words = []
    mark = {}
    markcntr = 0
    cn ={} 
    for j in range(len(toklist)):
        for i in range(len(textlist)):
            if (toklist[j] in np.array(textlist)[i]):
                cntr += 1
        if (cntr > 0):
            words.append(toklist[j])
            cn[toklist[j]] = cntr
            if (cntr < 15):
                mark[toklist[j]] = 1
                markcntr += 1
            else:
                mark[toklist[j]] = 0
            cntr = 0
        else:
            mark[toklist[j]] = 0
    return cn, markcntr, mark, words

# 名詞2つのコンビネーションを探索
def get_2conbination(toklist,textlist,markcntr,mark):
    word1 = []
    word2 = []
    cntr = 0
    cn ={} 
    flg = 0
    for k in range(len(toklist)):
        for j in range(k+1,len(toklist)):
            if (mark[toklist[k]]==1)&(mark[toklist[j]]==1):
                for i in range(len(textlist)):
                    if (toklist[j] in np.array(textlist)[i])and(toklist[k] in np.array(textlist)[i]):
                        cntr += 1

                if (cntr > 0):
                    word1.append(toklist[k])
                    word2.append(toklist[j])
                    cn[toklist[k]+toklist[j]] = cntr
                    cntr = 0
                    flg = 1
    return flg, cn, word1, word2

# 文書数が最小の名詞を選択
def get_word_num(word1,word2,cn,textlist):
    cntr = len(textlist)
    idx = -1
    for i in range(len(word1)):
        if len(word2) == 0:
            if cn[word1[i]] < cntr:
                cntr = cn[word1[i]]
                idx = i
        else:
            if cn[word1[i]+word2[i]] < cntr:
                cntr = cn[word1[i]+word2[i]]
                idx = i
    return cntr, idx

# 単語が含まれる文書を選択
def get_text_by_words(textlist,word_1,word_2):
    fact = ""
    for i in range(len(textlist)):
        if word_2 == "":
            if (word_1 in np.array(textlist)[i]):
                fact += np.array(textlist)[i]
        else:
            if (word_1 in np.array(textlist)[i])and(word_2 in np.array(textlist)[i]):
                fact += np.array(textlist)[i]
    return fact

# コサイン類似度で選択
def get_text_by_cos(textlist,word_1,word_2,I):
    cntr = 0
    fact = ""
    for i in I[0]:
        if word_2 == "":
            if (word_1 in textlist[i]):
                fact += np.array(textlist)[i]
                cntr += 1
                if (cntr > 3):
                    break
        else:
            if (word_1 in np.array(textlist)[i])and(word_2 in np.array(textlist)[i]):
                fact += np.array(textlist)[i]
                cntr += 1
                if (cntr > select_sentences):
                    break
    return fact

# openaiへの問い合わせ
def completion(new_message_text:str, settings_text:str = '', past_messages:list = []):
    if len(past_messages) == 0 and len(settings_text) != 0:
        system = {"role": "system", "content": settings_text}
        past_messages.append(system)
    new_message = {"role": "user", "content": new_message_text}
    past_messages.append(new_message)

    result = openai.ChatCompletion.create(
        model="gpt-3.5-turbo",
        messages=past_messages,
        max_tokens=512
    )
    response_message = {"role": "assistant", "content": result.choices[0].message.content}
    past_messages.append(response_message)
    response_message_text = result.choices[0].message.content
    return response_message_text, past_messages

# prompt作成
def fact_qa(fact, question):
    system_text = "あなたは参考文章をもとに質問に回答するシステムです。参考文章をもとに、段階的に考えて論理的に回答してください。"
    question_prompt = f"""## 参考文章

{fact}

## 質問

{question}"""
    answer, _ = completion(question_prompt, system_text, [])
    return answer

# メイン関数
if __name__ == '__main__':
    app.run(debug=True)

templates/index.html

<!DOCTYPE html>
<html>
<head>
    <title>チャットボット</title>
    <!-- CSSファイルを読み込む -->
    <link rel="stylesheet" type="text/css" href="{{ url_for('static', filename='styles.css') }}">
</head>
<body>
    <!-- ヘッダー部分 -->
    <header>
        <!-- サービスのタイトル -->
        <h1>○○向けチャットボット</h1>
    </header>
    <!-- 質問を入力するフォーム -->
    <form method="POST">
        <label for="question">質問前を入力してください:</label><br>
        <input type="text" size=100px id=""question" name="question"><br>
        <input type="submit" value="質問する">
    </form>
    <!-- レスポンスがある場合に表示 -->
    {% if response %}
    <div class="response">
        <!-- レスポンスを表示（HTMLタグをエスケープしない） -->
        <p>{{ response|safe }}</p>
    </div>
    {% endif %}
</body>
</html>

templates/index.css

/* 全体のフォント設定と背景色 */
body {
    font-family: 'Hiragino Kaku Gothic ProN', Meiryo, sans-serif;
    margin: 0;
    padding: 0;
    background-color: #f0f0f0;
}

/* ヘッダーのスタイル設定 */
header {
    background-color: #007BFF;
    color: white;
    text-align: center;
    padding: 20px 0;
}

/* ロゴ画像のサイズ設定 */
header img {
    height: auto;
    width: 25%;
}

/* フォームのスタイル設定 */
form {
    width: 80%;
    margin: 0 auto;
    background-color: white;
    padding: 20px;
    border-radius: 5px;
    box-shadow: 0px 0px 10px 0px rgba(0,0,0,0.1);
}

/* テキスト入力フィールドのスタイル設定 */
input[type="text"] {
    width: 50em;
    padding: 10px;
    margin: 10px 0;
    box-sizing: border-box;
}

/* 送信ボタンのスタイル設定 */
input[type="submit"] {
    padding: 10px 20px;
    background-color: #007BFF;
    color: white;
    border: none;
    cursor: pointer;
}

/* レスポンスのスタイル設定 */
.response {
    width: 80%;
    margin: 20px auto;
    padding: 20px;
    background-color: #fff;
    border-radius: 5px;
    box-shadow: 0px 0px 10px 0px rgba(0,0,0,0.1);
    max-height: 300px;
    overflow-y: auto;
}

実行は
python3 app.py
で動作します。

何をやっているかはそんなに難しくないので読んでみてください。ざっくりとは

• 質問から名詞を取り出す
• 取り出した名詞各々の含まれている文章の数を計数
• さらに２名詞のコンビネーションを計数
• コンビネーションで最小４文章に絞られたらその文章を選択
• ダメな場合は１単語で最小の文章が４文章以内なら選択
• それでも絞れない場合、embedingのインデックスで上位４文章を選択

特にこの方式が優れていると検証したわけではないので、適当に変えて遊んでみてください。
なぜこの方式にしたかというと、chatGPTに質問と一緒に食わせる文章には4K tokensという制約があり全部の文章を食わせることができません。なのでどの部分を選択するかで結果が変わります。大概は上記のように質問に含まれている単語を含む文章を選択することでうまくいきます。しかし、文章全体がないとダメな質問、今回の例でいうと全部の猫の頭数を質問しても、５つある文章全部を食わすことをしないため４頭という不正確な値になります。一応パラメータでapp.pyの
select_sentences = 4
の値を変えることで変更できます。しかし、大きくしていくといつかは4K tokenを超えることになります。また、応答時間の面でも食わす文章の大きさが小さいほうが短く済みます。というわけで、対象の文章の大きさと応答時間、正確性のトレードオフを見て調整してください。

最後に

ということで、暇を見つけて４，５日で実験したものをアップしてあります。多分専門家が見るとなんじゃこりゃというものと思いますが、手っ取り早くチャットボットをつくってみる目論見なので、ご勘弁ねがいます。なんかTokenizer２種類使ってしまっていますし。改善点などありましたら、コメント等いただけると励みになります。