環境構築と依存関係の解消

ローカルで動作するテキスト分類ベースのチャットボットを実装する際、まず自然言語処理と深層学習の基盤ライブラリをインストールする必要があります。以下のコマンドを実行し、必要なパッケージを環境に導入します。

pip install tensorflow numpy pandas nltk fastapi uvicorn

NLTKライブラリを使用する際、トークン化モデル（punkt）の欠如による実行時エラーが発生することがあります。ネットワーク環境の制限により自動ダウンロードが失敗する場合、以下の手順で手動配置を行います。GitHub上のリポジトリから punkt.zip を取得し、展開した内容をユーザーディレクトリ内の tokenizers/nltk_data/ に配置してください。これにより、ローカルでのトークン化処理が安定します。

学習データの定義と前処理

チャットボットの振る舞いは、意図（tag）、ユーザー入力パターン（patterns）、およびシステム応答（responses）のセットで定義します。大規模なコーパスは不要であり、特定のドメインに最適化された少数の学習サンプルで十分動作します。

import nltk
import ssl
import numpy as np
import pandas as pd
import random
import pickle
from nltk.stem import PorterStemmer
from tensorflow.keras.models import Sequential, model_from_json
from tensorflow.keras.layers import Dense, Dropout
from tensorflow.keras.optimizers import Adam
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import uvicorn

nltk.download('punkt', quiet=True)
stemmer = PorterStemmer()

chat_intents = {
    "categories": [
        {
            "tag": "greeting",
            "patterns": ["こんにちは", "おはよう", "やあ", "こんにちはか", "ねえ"],
            "responses": ["こんにちは！調子はどうですか？", "お久しぶりです", "やあ！何かお手伝いしましょうか？"]
        },
        {
            "tag": "farewell",
            "patterns": ["さようなら", "じゃあね", "またね", "バイバイ", "行ってきます"],
            "responses": ["さようなら！また会いましょう", "お気をつけて", "じゃあね！良い一日を"]
        }
    ]
}

モデルは生テキストを直接理解できないため、単語を共通の語幹に正規化し、辞書（vocabulary）とラベルリストを構築します。

word_list = []
intent_tags = []
dataset_pairs = []

for entry in chat_intents['categories']:
    for phrase in entry['patterns']:
        tokens = nltk.word_tokenize(phrase)
        word_list.extend(tokens)
        dataset_pairs.append((tokens, entry['tag']))
        if entry['tag'] not in intent_tags:
            intent_tags.append(entry['tag'])

word_list = sorted(list(set([stemmer.stem(w.lower()) for w in word_list if w.isalpha()])))
intent_tags = sorted(list(set(intent_tags)))
print(f"学習ラベル数: {len(intent_tags)}, 語彙数: {len(word_list)}")

Bag-of-Words変換と学習セットの生成

テキストを数値配列に変換するため、Bag-of-Words（BoW）アプローチを採用します。辞書内の各単語に対して、入力文に含まれていれば1、そうでなければ0を設定するバイナリベクトルを生成します。

training_input = []
output_matrix = [0] * len(intent_tags)

for tokens, tag in dataset_pairs:
    stemmed_tokens = [stemmer.stem(t.lower()) for t in tokens]
    bag = [1 if w in stemmed_tokens else 0 for w in word_list]
    target_row = list(output_matrix)
    target_row[intent_tags.index(tag)] = 1
    training_input.append([bag, target_row])

random.shuffle(training_input)
training_array = np.array(training_input)
X_train = training_array[:, 0]
y_train = training_array[:, 1]

ニューラルネットワークの構成と学習

分類タスクに対応するため、全結合層とドロップアウトを積み重ねたシークエンシャルモデルを構築します。多クラス分類のために最終層の活性化関数にsoftmaxを採用し、各意図の確率分布を出力します。最適化手法にはSGDに代わりAdamを使用し、学習の安定性を高めます。

chat_net = Sequential()
chat_net.add(Dense(64, input_shape=(len(X_train[0]),), activation='relu'))
chat_net.add(Dropout(0.4))
chat_net.add(Dense(32, activation='relu'))
chat_net.add(Dropout(0.4))
chat_net.add(Dense(len(y_train[0]), activation='softmax'))

optimizer = Adam(learning_rate=0.01)
chat_net.compile(loss='categorical_crossentropy', optimizer=optimizer, metrics=['accuracy'])
chat_net.fit(X_train, y_train, epochs=150, batch_size=4, verbose=0)
print("モデルの学習が完了しました。")

推論ロジックとモデルのシリアライズ

学習済みモデルを毎回再訓練するのではなく、構造ファイルと重みファイルに分離して保存します。推論時には、入力テキストをBoWベクトルに変換し、確率が閾値（0.25）を超えた意図のみを抽出します。

def convert_to_bow_vector(sentence, vocab):
    tokens = nltk.word_tokenize(sentence)
    stemmed = [stemmer.stem(w.lower()) for w in tokens]
    return np.array([1 if w in stemmed else 0 for w in vocab])

def get_intent_prediction(sentence, model, vocab, labels):
    threshold = 0.25
    vector = convert_to_bow_vector(sentence, vocab)
    predictions = model.predict(np.array([vector]), verbose=0)[0]
    results = [[i, prob] for i, prob in enumerate(predictions) if prob > threshold]
    results.sort(key=lambda x: x[1], reverse=True)
    return [(labels[idx], str(prob)) for idx, prob in results]

# モデルの保存
chat_net.save('chat_model.keras')
with open('chat_config.pkl', 'wb') as f:
    pickle.dump({'words': word_list, 'classes': intent_tags, 'intents': chat_intents}, f)

FastAPIによるWebインターフェースの実装

最後に、Pydanticモデルを用いたリクエストバリデーションと、FastAPIのエンドポイントを構築します。リクエストごとにモデルをロードするオーバーヘッドを避けるため、アプリケーション起動時にグローバルスコープで初期化します。

app = FastAPI(title="Local Chat API")

# アプリケーション起動時のモデル初期化
loaded_model = tf.keras.models.load_model('chat_model.keras')
with open('chat_config.pkl', 'rb') as f:
    config = pickle.load(f)

class ChatRequest(BaseModel):
    message: str

class ChatResponse(BaseModel):
    intent: str
    confidence: float
    reply: str

@app.post("/v1/chat", response_model=ChatResponse)
async def process_chat(request: ChatRequest):
    predictions = get_intent_prediction(request.message, loaded_model, config['words'], config['classes'])
    if not predictions:
        return ChatResponse(intent="unknown", confidence=0.0, reply="申し訳ありません、理解できませんでした。")
    
    best_intent, confidence = predictions[0]
    confidence_val = float(confidence)
    
    intent_data = next((item for item in config['intents']['categories'] if item['tag'] == best_intent), None)
    reply_text = random.choice(intent_data['responses']) if intent_data else "エラーが発生しました。"
    
    return ChatResponse(intent=best_intent, confidence=confidence_val, reply=reply_text)

if __name__ == "__main__":
    uvicorn.run(app, host="127.0.0.1", port=8000)

サーバーを起動後、POSTリクエストでJSON形式のメッセージを送信すると、分類された意図ラベル、確信度、および対応する応答テキストが返されます。学習データのパターンと応答を拡張することで、特定のユースケースに合わせた対話システムの動作をカスタマイズ可能です。