1. 初めに

前回の記事でドキュメントをグラフDB(neo4j)に登録し、ベクトルとグラフ検索で質問に関連性の高いドキュメントを提示するアプリを作ったが物足りなさを感じ、質問と検索結果を元に質問回答AIで回答する機能を追加しました。

前回の記事

2. 扱うデータの特徴

今回のアプリで扱うデータを情報セキュリティの観点で以下のように定義しました。

1. 機密性

自社や自組織のナレッジは営業秘密になる可能性もある漏洩を許されないものではと考えております。

2. 完全性

自社や自組織のナレッジは過去にどのような課題について、どのような対策をしたかという事実を検索できれば良く、AIによる推測や補完で元の事実情報を改変されないか危惧。生成AIによる推測は不要と考えています。

3. 質問回答AIを自作する理由・背景

今の生成AIならRAGも併用するので、質問と関連情報をインプットすれば、精度の高い回答をアウトプットしてくれることは認識してますが、上記2のデータを(暗号化されていても)インターネット経由の通信にも流さなくて良いようにAIは自作することに決めました。

自作のAIならアプリと同じサーバか、アプリと同じLANのサーバで稼働でき、ローカルな通信によりインターネットを経由しなくて良いメリットがあります。

しかし、私は実務でLLMの構築をした経験がないので構築が難航すると判断し、AIをPythonのスクリプトで開発することにしました。

質問回答AIとは、質問と回答のペアを教師あり学習で学習させたAIであり、質問すると回答するAIと定義します。

生成AIを否定していません。実際に使って便利であると実感もしてます。ただし、今回の試みでは、目的には合わないので使わないだけです。

4. 残りの課題

1. どのタイミングでAIに質問と回答を学習させるか？

グラフDBにデータを登録する時に、学習済みのAIに継続的追加的に新たな情報を学習させれば良いと判断しました。

2. AIに学習させる質問と回答を人が手動で作成しないといけないか？

グラフDBにデータを登録する時に、各文の主語、目的語、述語をスクリプトは認識しているので、主語についての質問・回答、目的語についての質問・回答、述語についての質問・回答の３パターンを機械的に作成できると判断しました。

3. AIに質問に回答してもらう為に、どんなパラメータを渡すか？

まずは、質問。そして、事前にグラフDBから検索した検索結果の１つ目（質問について最も類似性が高い）文脈も渡して、その２つのパラメータで回答を算出する方式としました。

5. 動作検証

1.追加学習時

2025-04-30 10:36:02.512526: I tensorflow/core/util/port.cc:153] oneDNN custom operations are on. You may see slightly different numerical results due to floating-point round-off errors from different computation orders. To turn them off, set the environment variable `TF_ENABLE_ONEDNN_OPTS=0`.
2025-04-30 10:36:08.341644: I tensorflow/core/util/port.cc:153] oneDNN custom operations are on. You may see slightly different numerical results due to floating-point round-off errors from different computation orders. To turn them off, set the environment variable `TF_ENABLE_ONEDNN_OPTS=0`.
WARNING:tensorflow:From xxx\site-packages\tf_keras\src\losses.py:2976: The name tf.losses.sparse_softmax_cross_entropy is deprecated. Please use tf.compat.v1.losses.sparse_softmax_cross_entropy instead.

Some weights of BertForQuestionAnswering were not initialized from the model checkpoint at cl-tohoku/bert-base-japanese and are newly initialized: ['qa_outputs.bias', 'qa_outputs.weight']
You should probably TRAIN this model on a down-stream task to be able to use it for predictions and inference.
モデルバージョン:xxx\model\20250429_124245
モデルとトークナイザーのロードが正常に完了しました！
[テキスト]
太郎は明子を好きです。
～省略～

[
{'context': '太郎は明子を好きです。', 'passage_answer_candidates': [{'plaintext_start_byte': 1, 'plaintext_end_byte': 7}], 'question_text': '何は明子を好き?'}, 
～省略～
]
訓練データ数: 18
Epoch 2, Loss: 2.127659320831299
追加学習を完了しました.

2.検索時

モデルとトークナイザーのロードが正常に完了しました！
-------- 「明子は誰を好きか?」による検索結果 --------
{'ファイル名': 'ner_data.txt', '何個目の文': 2, '文の内容': '明子は次郎を好きです。', '類似性': 0.83}
{'ファイル名': 'ner_data.txt', '何個目の文': 1, '文の内容': '太郎は明子を好きです。', '類似性': 0.63}
-------- 「明子は誰を好きか?」のAIによる質問への回答 --------
次郎
-------- 「明子は次郎をどうする?」による検索結果 --------
{'ファイル名': 'ner_data.txt', '何個目の文': 2, '文の内容': '明子は次郎を好きです。', '類似性': 0.73}
-------- 「明子は次郎をどうする?」のAIによる質問への回答 --------
好き
-------- 「明子は次郎をどう思ってるか?」による検索結果 --------
{'ファイル名': 'ner_data.txt', '何個目の文': 2, '文の内容': '明子は次郎を好きです。', '類似性': 0.8}
-------- 「明子は次郎をどう思ってるか?」のAIによる質問への回答 --------
好き
-------- 「太郎は誰を好きか?」による検索結果 --------
{'ファイル名': 'ner_data.txt', '何個目の文': 1, '文の内容': '太郎は明子を好きです。', '類似性': 0.86}
{'ファイル名': 'ner_data.txt', '何個目の文': 3, '文の内容': '次郎は太郎の弟です。', '類似性': 0.7}
-------- 「太郎は誰を好きか?」のAIによる質問への回答 --------
回答不可(質問を理解できない)
-------- 「誰を明子は好きか?」による検索結果 --------
{'ファイル名': 'ner_data.txt', '何個目の文': 2, '文の内容': '明子は次郎を好きです。', '類似性': 0.76}
{'ファイル名': 'ner_data.txt', '何個目の文': 1, '文の内容': '太郎は明子を好きです。', '類似性': 0.62}
-------- 「誰を明子は好きか?」のAIによる質問への回答 --------
回答不可(質問を理解できない)
-------- 「誰を太郎は好きか?」による検索結果 --------
{'ファイル名': 'ner_data.txt', '何個目の文': 1, '文の内容': '太郎は明子を好きです。', '類似性': 0.83}
{'ファイル名': 'ner_data.txt', '何個目の文': 3, '文の内容': '次郎は太郎の弟です。', '類似性': 0.67}
-------- 「誰を太郎は好きか?」のAIによる質問への回答 --------
回答不可(質問を理解できない)

6. 検索の検証結果の評価

1. 〇前半の3つの質問は素直に聞いたので、正解でした。

2. × 後半の３つの質問はチェックでエラーとなりました。

理由は主語、目的語、述語の順で学習させてますが、主語、目的語、述語の並びを変えて質問すると、別の文章として認識するようです。ちなみに、文脈・質問関連性エラーチェックを無効にすると、4つ目の回答は「明子」(正解)、5つ目の回答は「次郎」(正解)、6つ目の回答は「好き」(不正解)となることは確認済みです。

7. 今回、変更・追加したソース

sampleEntry.py(データ登録スクリプトにmakeCypherEntryAndQA、incremental_learnの呼び出しを追加)

    # 初期化
    instance_nlp = Nlp()
    instance_neo4j = Neo4jConnection()
    instance_qa_ai = QuestionAnswerAI2()
    instance_qa_ai.load(False)

    # DB接続
    if instance_neo4j.connect("bolt://localhost:7687", "neo4j", "xxxxxxxx"):
        # 入力データをファイルからリード
        with open(instance_nlp.DATA_DIR + 'ner_data.txt', 'r', encoding='utf-8') as file:
            text = file.read()

            print('[テキスト]\n' + text)

            # 形態素分析の実行
            # print('[GINZA分析結果]')
            sentences = instance_nlp.ginza(text, False)

            # 形態素分析の実行結果からグラフDB情報登録用のCypherと質問・回答AIへの質問・回答の作成
            cyphers, qas = instance_nlp.makeCypherEntryAndQA('ner_data.txt', sentences)

            # グラフDBに登録
            # print('[データ登録Cypher]')
            for c in cyphers:
                # print(c)
                instance_neo4j.queryEntry(c)

            # 質問・回答AIへの質問・回答の継続追加学習
            print(qas)
            instance_qa_ai.incremental_learn(qas)

        # DB切断
        instance_neo4j.disConnect()

sampleSearch.py(データ検索スクリプトにAIへの質問メソッド(question)の呼び出しを追加)

    # 初期化
    nlp = Nlp()
    neo4j = Neo4jConnection()
    ai = QuestionAnswerAI2()
    ai.load(False)

    # DB接続
    if neo4j.connect("bolt://localhost:7687", "neo4j", "xxxxxxxx"):
        # 検索Cypherの生成
        contexts = [
            '明子は誰を好きか?',
            '明子は次郎をどうする?',
            '明子は次郎をどう思ってるか?',
            '太郎は誰を好きか?',
            '誰を明子は好きか?',
            '誰を太郎は好きか?',
        ]

        for c in contexts:
            cypher_query = nlp.makeCypherSearch(c, 0.6)

            # 検索
            cypher_results = neo4j.querySearch(cypher_query)

            # 検索結果出力
            print('-------- 「' + c + '」による検索結果 --------')
            for cr in cypher_results:
                print(cr)

            if len(cypher_results) > 0:
                # AIへの質問
                print('-------- 「' + c + '」のAIによる質問への回答 --------')
                answer = ai.question(cypher_results[0]["文の内容"], c, False)
                print(answer)

        # DB切断
        neo4j.disConnect()

qa_ai2.py(質問回答AI)

import os
from datetime import datetime
import json
import janome
import janome.tokenizer
from sklearn.model_selection import train_test_split
from transformers import AutoTokenizer, AutoModelForQuestionAnswering
import torch
from torch.optim import AdamW
from torch.utils.data import DataLoader, Dataset
from sklearn.metrics.pairwise import cosine_similarity
import numpy as np

# 質問・回答AI(2型)
class QuestionAnswerAI2:
    # データパス
    DATA_PATH:str = os.path.dirname(__file__) + "\\data"

    # 最大サイズ
    MAX_LEN:int = 512

    def __init__(self):
        """
        コンストラクタ
        return: なし
        """
        model_name = "cl-tohoku/bert-base-japanese"
        self.model:AutoModelForQuestionAnswering = AutoModelForQuestionAnswering.from_pretrained(model_name)
        self.tokenizer:AutoTokenizer = AutoTokenizer.from_pretrained(model_name)

    # カスタムデータセットの準備
    class QADataset(Dataset):
        def __init__(self, data:list, tokenizer:AutoTokenizer, max_length:int):
            """
            コンストラクタ
            param: data 質問・回答リスト
            param: tokenizer トークナイザー
            param: max_length サイズ
            return: なし
            """
            self.data = data
            self.tokenizer = tokenizer
            self.max_length = max_length

        def __len__(self)->int:
            """
            質問・回答リストのサイズの取得
            return: 質問・回答リストのサイズ
            """
            return len(self.data)

        def __calculate_position__(self, context:str, offset:int)->int:
            """
            単語の位置の算出
            param: context コンテキスト
            param: offset バイト位置
            return: 単語の位置
            """
            tokenizer = janome.tokenizer.Tokenizer()
            words = [token.surface for token in tokenizer.tokenize(context)]

            position = -1
            byte_start = 1
            for i, w in enumerate(words):
                byte_len = len(w.encode("utf-8"))
                byte_end = byte_start + byte_len
                if byte_start <= offset < byte_end:
                    position = i
                    break
                byte_start += byte_len
                if byte_start > offset:
                    break

            return position

        def __getitem__(self, idx:int)->dict:
            """
            質問回答リストを学習データに変換
            param: idx データセットの位置
            return: 学習データ
            """
            item = self.data[idx]

            context = item.get("context", "")

            inputs = self.tokenizer.encode_plus(
                item.get("question_text", ""),
                context,
                max_length=self.max_length,
                padding="max_length",
                truncation=True,
                return_tensors="pt"
            )

            if 'start_positions' in item:
                start_positions = item["start_positions"]
            else:
                start_positions = torch.tensor(self.__calculate_position__(context, item["passage_answer_candidates"][0]['plaintext_start_byte']))

            if 'end_positions' in item:
                end_positions = item["end_positions"]
            else:
                end_positions = torch.tensor(self.__calculate_position__(context, item["passage_answer_candidates"][0]['plaintext_end_byte'] - 1))

            return {
                'input_ids': inputs['input_ids'].squeeze(0),
                'attention_mask': inputs['attention_mask'].squeeze(0),
                'start_positions': start_positions,
                'end_positions': end_positions
            }

    # データセット処理
    def __prepare_features__(self, qa):
        """ 対象のデータレイアウト(UTF-8,1起点)
        qa = {
            "context": "和歌山市は和歌山県にあり、紀ノ川の近くに位置しています。",
            "passage_answer_candidates": [
                {
                    "plaintext_start_byte": 11,
                    "plaintext_end_byte": 28
                }
            ],
            "question_text": "和歌山市はどこにありますか？"
        }
        """

        # 各要素を処理して辞書を作成
        # contexts = [" ".join([str(byte) for byte in a["plaintext_start_byte"]]) for a in examples["passage_answer_candidates"]]
        contexts = qa.get("context", qa.get("document_plaintext", ""))
        questions = qa["question_text"]

        # トークナイズ処理
        tokenized_inputs = self.tokenizer(contexts, questions, truncation=True, padding="max_length", max_length=self.MAX_LEN)

        # 必要なキーを追加
        tokenized_inputs["start_positions"] = qa["passage_answer_candidates"][0]["plaintext_start_byte"]
        tokenized_inputs["end_positions"] = qa["passage_answer_candidates"][0]["plaintext_end_byte"]

        return tokenized_inputs

    def __create_directory__(self, dir_name:str):
        """
        ディレクトリ作成
        param: dir_name 作成するディレクトリのフルパス
        return: なし
        """
        # ディレクトリを作成
        os.makedirs(dir_name, exist_ok=True)
        
        # 読み込み書き込み権限を設定（Windowsではファイル属性に依存する部分あり）
        os.chmod(dir_name, 0o777)  # Unix系でフル権限にする場合

    def __teacher_learn(self, qas:list):
        """
        教師あり学習
        param: qas 質問・回答リスト
        return: なし
        """
        # データセットサイズを確認
        print(f"訓練データ数: {len(qas)}")

        # データローダーの作成
        dataset = self.QADataset(qas, self.tokenizer, max_length=self.MAX_LEN)
        dataloader = DataLoader(dataset, batch_size=8, shuffle=True)

        # オプティマイザー設定
        optimizer = AdamW(self.model.parameters(), lr=5e-5)

        # 学習プロセス
        self.model.train()
        for epoch in range(3):
            for batch in dataloader:
                optimizer.zero_grad()
                input_ids = batch['input_ids']
                attention_mask = batch['attention_mask']
                start_positions = batch['start_positions']
                end_positions = batch['end_positions']

                outputs = self.model(
                    input_ids=input_ids,
                    attention_mask=attention_mask,
                    start_positions=start_positions,
                    end_positions=end_positions
                )

                loss = outputs.loss
                loss.backward()
                optimizer.step()

        print(f"Epoch {epoch}, Loss: {loss.item()}")

        # 更新されたモデルを保存
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        dir_name = os.path.join(self.DATA_PATH + "\\model", timestamp)
        self.__create_directory__(dir_name)
        self.model.save_pretrained(dir_name)
        self.tokenizer.save_pretrained(dir_name)

    def clean_learn(self, ratio_learn:float):
        """
        初期学習(未構築、未学習のモデルにモデルの構築と学習させる)
        param: ratio_learn データセットの内どのくらい学習するかの比率(0.0(0%)~1.0(100%)
        return: なし
        """
        # データセットのロード
        dataset_dir = self.DATA_PATH + "\\dataset\\"
        dataset_files = [os.path.join(dataset_dir, file) for file in os.listdir(dataset_dir) if file.endswith(".jsonl")]
        dataset_datas = []
        
        # 各JSONファイルを読み込み
        for file_path in dataset_files:
            with open(file_path, "r", encoding="utf-8") as f:
                temp = []
                for line in f:
                    temp.append(json.loads(line.strip()))
                dataset_datas.append(temp)

        # datasets.mapを呼び出す
        qas = [self.__prepare_features__(qa) for dataset in dataset_datas for qa in dataset]

        # 学習用のデータを抽出
        not_ratio_learn = 1.0 - ratio_learn
        if not_ratio_learn != 0.0:
            qas, _ = train_test_split(qas, test_size=not_ratio_learn, random_state=42)

        # 教師あり学習
        self.__teacher_learn(qas)

        print("初期学習を完了しました.")

    def incremental_learn(self, qas:list):
        """
        継続学習(構築済み、学習済みのモデルに追加の質問・回答を学習させる)
        param: qas 質問・回答リスト
        return: なし
        """
        # 教師あり学習
        self.__teacher_learn(qas)

        print("追加学習を完了しました.")

    def __find_latest_model__(self, parent_path)->str:
        """
        最新の学習モデルを検索
        param: parent_path 学習モデルの親ディレクトリ
        return: 最新の学習モデルのフルパス
        """
        # サブディレクトリを取得
        subdirs = [os.path.join(parent_path, d) for d in os.listdir(parent_path) if os.path.isdir(os.path.join(parent_path, d))]
        
        # ディレクトリ名が "yyyymmdd_hhmmss" の形式であることを確認し、タイムスタンプ順に並べる
        valid_dirs = []
        for subdir in subdirs:
            dir_name = os.path.basename(subdir)
            try:
                # ディレクトリ名をdatetimeとして解析
                timestamp = datetime.strptime(dir_name, "%Y%m%d_%H%M%S")
                valid_dirs.append((timestamp, subdir))
            except ValueError:
                pass  # 不正なフォーマットのディレクトリは無視
        
        # タイムスタンプ順にソートして最新を取得
        if valid_dirs:
            latest_dir = max(valid_dirs, key=lambda x: x[0])[1]  # 最も新しいディレクトリ
            return latest_dir
        else:
            raise ValueError("有効なモデルディレクトリが見つかりません。")

    def load(self, dbg:bool):
        """
        学習モデルのロード
        param: dbg デバッグ出力する？
        return: なし
        """
        # 保存したモデルとトークナイザーをロード
        model_path = self.__find_latest_model__(self.DATA_PATH + "\\model")
        print(f"モデルバージョン:{model_path}")

        self.model = AutoModelForQuestionAnswering.from_pretrained(model_path)  # モデルを読み込み
        self.tokenizer = AutoTokenizer.from_pretrained(model_path, use_fast=True) # トークナイザーを読み込み

        if dbg:
            # ロードしたモデルの情報出力
            print("モデル構造:\n")
            print(self.model)

            # モデルのパラメータ数を計算して表示
            total_params = sum(p.numel() for p in self.model.parameters())
            print(f"\n総パラメータ数: {total_params}\n")

            # 各レイヤーの詳細を表示
            print("各レイヤーの詳細:\n")
            for name, param in self.model.named_parameters():
                print(f"レイヤー名: {name}")
                print(f"パラメータ数: {param.numel()}")
                print(f"サイズ: {list(param.size())}\n")

        print("モデルとトークナイザーのロードが正常に完了しました！")

    def __check_similarity__(self, context:str, question:str)->float:
        """
        コンテキストと質問の類似性チェック
        param: context コンテキスト
        param: question 質問
        return: 類似性スコア
        """
        try:
            # トークン化してベクトル化
            context_tokens = self.tokenizer.encode(context, return_tensors="pt", max_length=self.MAX_LEN, truncation=True)
            question_tokens = self.tokenizer.encode(question, return_tensors="pt", max_length=self.MAX_LEN, truncation=True)

            # ベクトルのパディングを追加して次元を揃える
            context_vector_padded = np.pad(context_tokens.cpu().numpy(), ((0, 0), (0, self.MAX_LEN - context_tokens.shape[1])))
            question_vector_padded = np.pad(question_tokens.cpu().numpy(), ((0, 0), (0, self.MAX_LEN - question_tokens.shape[1])))

            # 類似性スコア計算
            similarity_score = cosine_similarity(context_vector_padded, question_vector_padded)[0][0]
        except Exception as e:
            print(f"類似性スコア計算でエラー: {e}")
            return -1.0

        return similarity_score

    def question(self, context:str, question:str, dbg:bool)->str:
        """
        質問への回答
        param: context コンテキスト
        param: question 質問
        param: dbg デバッグ出力する？
        return: 回答
        """
        if dbg:
            print(f"パラメータ:context={context},question={question}")

        # パラメータチェック
        if (context is None or context == "") or (question is None or question == ""):
            return "回答不可(パラメータ未入力)"

        # トークナイザーロードチェック
        if self.tokenizer is None:
            return "回答不可(トークナイザーをロードしてない)"

        # 文脈と質問の類似性チェック
        sim = self.__check_similarity__(context, question)
        if dbg:
            print(f"類似性:{sim}")
        if sim < 0.6:
            return "回答不可(質問を理解できない)"

        # トークナイズ処理
        inputs = self.tokenizer.encode_plus(question, context, return_tensors="pt")
        tokens = self.tokenizer.convert_ids_to_tokens(inputs["input_ids"][0])
        if dbg:
            print(f"Tokens: {tokens}")

        # モデルによる予測
        outputs = self.model(**inputs)
        start_scores = outputs.start_logits
        end_scores = outputs.end_logits

        if dbg:
            print(f"Start logits: {start_scores}, End logits: {end_scores}")

        # スコア範囲を取得
        start_index = torch.argmax(start_scores, dim=-1)[0]
        end_index = torch.argmax(end_scores, dim=-1)[0]

        # 範囲の補正
        # [note]上記の結果について、私はstart_indexとend_indexを独自に補正しました。

        if dbg:
            print(f"答えの位置: {start_index}, {end_index}")

        # トークンから文字列を生成
        filtered_tokens = tokens[start_index:end_index]
        answer = ""
        for token in filtered_tokens:
            if token.startswith("##"):
                answer += token[2:]
            else:
                answer += token if answer == "" else " " + token

        tokenizer = janome.tokenizer.Tokenizer()
        words = [token.surface for token in tokenizer.tokenize(question)]
        for w in words:
            if answer == w:
                return "回答不可(質問を質問で返す)"

        return answer

8. 苦労、工夫した点

1. 学習させるデータセットのレイアウトは何が正解か？

plaintext_start_byte、plaintext_end_byteの示すバイト位置について、tydiqaを参考にしたが、文字コードがUTF-8？、SJIS？起点は0？1？等に悩みましたが、暫定で、UTF-8、1起点でコーディングしました。
あと、plaintext_start_byte、plaintext_end_byteを元に回答の単語位置start_positions、end_positionsの登録も必要だと推測しました。

2. 追加学習の終了時に既存のモデルを格納するディレクトリに上書き保存できない。

一部のファイルがロックされているようで、上書き保存できないので、
「年月日_時分秒」のディレクトリを作成・格納し、ロード時はディレクトリ名が最新のディレクトリからモデルをロードするように調整しました。

3. 学習に時間がかかる

私のPC(2年前に在庫一掃セールで購入した古い世代)だとOS:Windows11 CPU:Intel Core i5 メモリ:16GBで、わずか18件の学習をするのに10分もかかります。
スペックの高いマシンで学習するなりの工夫が必要となります。

4. AIへの質問の回答のstart_index、end_indexが示すトークンの範囲が質問の領域になっている。

学習のさせ方に問題があるかもしれないですが、掲載したソースのように変数を補正することで、回答を提示できるようにしました。ソース中の[note]の個所で各自、補正をお願いします。

9. 最後に

この試みを通じて、意外とスクリプトでも単純なAIなら作成できることを実感しましたが、カスタマイズ面で直面する課題も多く、そこにこそ技術の奥深さを感じました。

そして、私は今回もAIを活用しきれなかったとも思います。もともとcl-tohoku/bert-base-japaneseやAutoModelForQuestionAnsweringは優秀な言語AIだが、私の思考力と発想力が低いせいで、このような使い方しかできませんでした。優秀なAIエンジニアだったら、もっと、このAIを高機能・高性能に使いこなせたと思います。

この経験を一つの区切りとし、次は電子回路や組み込みアプリの分野に挑戦し、新たな課題に取り組んでいく予定です。

もしこの記事がどなたかの参考になれば幸いです。ここまでお読みいただき、ありがとうございました！

以上