ChatGPTではじめる新しいLINE Botの作り方

ChatGPT、LINE Botとの相性抜群ですよね。流暢におしゃべりできるし、プロンプトの工夫をすればキャラ設定もできます。
でも、今日はそういうお話ではありません。何を作るのかではなく、どうやって作るかについてお話します。かといってコードを自動生成してもらうという話でもなく、もう全然作り方が変わってしまったというお話です。

チャットボットの処理フロー

LINE Botに限らずあらゆるチャットボットや対話システムは、ざっくりいうと以下のような仕組みになっています。

ユーザーの発話の意図や関連情報を読み取って（インテント判定・エンティティ抽出）、それを処理し（スキル実行）、処理結果をメッセージとして応答します。図の例だとインテントとして「天気予報」、エンティティとして「佐賀」を読み取って、天気予報APIなどで処理していることでしょう。

これまでのチャットボットの作り方

先の例をPythonで書くと以下のようになります。意図が読み取れないときはおうむ返しをします。

# スキル
def get_weather(location: str) -> str:
    # 本当はAPIなどから取得
    weather = {"weather": "晴れ", "temperature": 38.0}
    print(f"location: {location} / weather: {weather}")
    return f"{location}の天気は、{weather['weather']}。最高気温は{weather['temperature']}度の見込みだよ。"

def add_reminder(remind_at: str, note: str) -> str:
    # 本当はAPIなどを叩いてリマインダーを登録
    reminder = {"remind_at": remind_at, "note": note}
    print(f"remind_at: {remind_at} / note: {note} / reminder: {reminder}")
    return f"{reminder['note']}を、{reminder['remind_at']}にお知らせします。"

def chat(text: str) -> str:
    return text

skills = {"get_weather": get_weather, "add_reminder": add_reminder, "chat": chat}

# インテント・エンティティの抽出処理
def extract_intent(text: str) -> tuple:
    if "天気" in text:
        # 本当は自然言語解析により場所の情報を抽出
        entities = {"location": "佐賀"}
        return ("get_weather", entities)
    elif "リマインド" in text:
        # 本当は自然言語解析によりリマインド日時とリマインドすべき事項を取得
        entites = {"remind_at": "2023-07-22T12:34:56", "note": text}
        return ("add_reminder", entites)
    else:
        return ("chat", {"text": text})

# メイン
while True:
    # ユーザー入力を取得
    text = input("user> ")
    # インテント・エンティティ抽出
    intent, entities = extract_intent(text)
    # スキルの取得
    skill = skills[intent]
    # スキルの実行
    resp = skill(**entities)
    print(f"bot> {resp}")

実行結果

$ python bot.py
user> こんにちは
bot> こんにちは
user> 佐賀の天気は？
location: 佐賀 / weather: {'weather': '晴れ', 'temperature': 38.0}
bot> 佐賀の天気は、晴れ。最高気温は38.0度の見込みだよ。
user> 大判焼き買いに行くのを2時間後にリマインドして
remind_at: 2023-07-22T12:34:56 / note: 大判焼き買いに行くのを2時間後にリマインドして / reminder: {'remind_at': '2023-07-22T12:34:56', 'note': '大判焼き買いに行くのを2時間後にリマインドして'}
bot> 大判焼き買いに行くのを2時間後にリマインドしてを、2023-07-22T12:34:56にお知らせします。
user> ありがとう！
bot> ありがとう！

ここで、この仮組の状態から完成品に仕上げるために必要な対応は以下の通りです。

1. インテント判定・エンティティ抽出ロジックの作成: 今は「天気」と含まれていれば天気予報と判定するなど、かなりいい加減です。特にリマインダーは「あとで教えて」といったものも判定したいところです。また、情報抽出をどうやって行うかは、面倒すぎて考えたくもありません。基本的には、これらは形態素解析＋ロジックやAIのNLUサービスを利用して対応することになるでしょう。後者の場合、大量の例文を登録する必要もあります。

2. 対話フロー制御の作成: 一度に必要な情報が集まりきらず、ユーザーに聞き返すこともあるでしょう。たとえば場所の指定なく天気予報を聞かれたようなケースです。このとき、ユーザーに聞き返したり、次の要求には「天気」という単語が含まれていなくても天気予報スキルで処理したり、既に聞き出した情報があればそれを保持しておくといった制御が必要になります。チャットボットの使い勝手やユーザー体験はここの作り込みに大きく左右されます。

3. スキル処理の作成: 天気予報やリマインダーのAPIとの連携部分です。これは純粋な処理ですので、説明を省略します。

4. 応答メッセージの作成: スキルの処理結果をもとにわかりやすい文章を作成したり、キャラクターらしい話し方にしたりします。また、インテントが判定できなかったときに雑談することも含まれます。

だいたいこういったところでしょうか。ここから先が本番という感じですが、先に結論をいうと、ChatGPTを活用することで上記の3（スキル処理そのもの）以外は何も作らなくて良くなります。

新しいチャットボットの作り方

まず、これからやろうとしていることの概念を図にしてみました。スキルそのものの実行以外はすべてChatGPTが自律的に行うようにします。

それではChatGPTを使って先ほどの例を作り変えてみましょう。まずはChatGPTを使うための初期設定として、モジュールの冒頭に以下を追加します。事前にOpenAIのAPI Keyを取得したり、pip install openaiでモジュールをインストールするなどしておいてください。

import json
from openai import ChatCompletion

# ChatGPTの初期設定
api_key = "YOUR_API_KEY"
system_content = """あなたは私のAIアシスタントです。あなたは会話の他に以下のスキルを備えています。

# 天気予報スキル
* 指定された場所の天気を調べて回答します。
* 場所が不明な場合はユーザーに聞き返します。

# リマインダースキル
* ユーザーのToDoを指定日時に思い出させます。
* リマインドすべき日時や内容が不明な場合はユーザーに聞き返します。
"""
messages = [{"role": "system", "content": system_content}]

system_contentに設定しているのはいわゆるプロンプトエンジニアリングというものです。ここで天気予報とリマインダーについて触れていますが、これは書かなくても動くものの、情報が不足している場合の振る舞いについてはプロンプトで指定する必要があります。

次に、インテント・エンティティの抽出処理を作り替えてextract_intent_gptという関数を作成します。ここにはインテント判定やエンティティ抽出のためのロジックが一切含まれず、ChatGPTに判断を委ねている点に注目してください。

# ChatGPT版　インテント・エンティティの抽出処理
def extract_intent_gpt(text: str) -> tuple:
    messages.append({"role": "user", "content": text})

    # 天気予報とリマインダーの処理を定義
    functions = [
        {
            "name": "get_weather",
            "description": "天気予報を取得する処理",
            "parameters": {
                "type": "object",
                "properties": {
                    "location": {"type": "string"}
                }
            }
        },
        {
            "name": "add_reminder",
            "description": "リマインダーを登録する処理",
            "parameters": {
                "type": "object",
                "properties": {
                    "remind_at": {"type": "string"},
                    "note": {"type": "string"}
                }
            }
        },
    ]

    # ChatGPTの呼び出し
    resp = ChatCompletion.create(
        api_key=api_key,
        model="gpt-3.5-turbo-0613",
        messages=messages,
        functions=functions,
    )

    # 呼び出し結果の取得
    message = resp["choices"][0]["message"]
    messages.append(message)

    if "function_call" in message:
        # 天気やリマインダーなどの処理が判定された場合はスキルに対応するインテントとする
        return (message["function_call"]["name"], json.loads(message["function_call"]["arguments"]))
    else:
        # 処理が判定されなかった場合はchatインテントとする
        return ("chat", {"text": message["content"]})

ChatGPT APIの使い方そのものについては解説を省略しますが、ポイントとしては、Function Callingというユーザーの発話内容から実行すべき処理をChatGPTが認識し、定義に基づいて情報を抽出してJSON形式で応答する機能を利用しています。

メイン処理から呼び出すインテント判定処理もこれで置き換えましょう。

    # インテント・エンティティ抽出
    intent, entities = extract_intent_gpt(text)

このままでも既に自律的なインテント判定は行えますが、せっかくなのでChatGPTによって応答メッセージを生成するところまでやりましょう。天気予報とリマインダーのスキルを以下のように修正します。

def get_weather(location: str) -> str:
    # 本当はAPIなどから取得
    weather = {"weather": "晴れ", "temperature": 38.0}
    print(f"location: {location} / weather: {weather}")
    # return f"{location}の天気は、{weather['weather']}。最高気温は{weather['temperature']}度の見込みだよ。"
    messages.append({"role": "function", "name": "get_weather", "content": json.dumps(weather)})
    resp = ChatCompletion.create(
        api_key=api_key,
        model="gpt-3.5-turbo-0613",
        messages=messages,
    )
    messages.append(resp["choices"][0]["message"])
    return resp["choices"][0]["message"]["content"]


def add_reminder(remind_at: str, note: str) -> str:
    # 本当はAPIなどを叩いてリマインダーを登録
    reminder = {"remind_at": remind_at, "note": note}
    print(f"remind_at: {remind_at} / note: {note} / reminder: {reminder}")
    # return f"{reminder['note']}を、{reminder['remind_at']}にお知らせします。"
    messages.append({"role": "function", "name": "add_reminder", "content": json.dumps(reminder)})
    resp = ChatCompletion.create(
        api_key=api_key,
        model="gpt-3.5-turbo-0613",
        messages=messages,
    )
    messages.append(resp["choices"][0]["message"])
    return resp["choices"][0]["message"]["content"]

それでは、実行してみましょう！

実行結果

$ python bot.py
user> こんにちは
bot> こんにちは！おげんきですか？何かお手伝いできることはありますか？
user> 佐賀の天気は？
location: 佐賀 / weather: {'weather': '晴れ', 'temperature': 38.0}
bot> 佐賀の天気は晴れです。気温は38度です。
user> 大判焼き買いに行くのを2時間後にリマインドして
remind_at: 2 hours later / note: 大判焼き買いに行く / reminder: {'remind_at': '2 hours later', 'note': '大判焼き買いに行く'}
bot> 了解しました。2時間後に「大判焼き買いに行く」ことをリマインドします。お忘れなく！
user> ありがとう！
bot> どういたしまして！お役に立てて嬉しいです。何か他にお手伝いできることがありましたら、遠慮なくお申し付けくださいね。

いい感じです。さらに踏み込んでいきましょう。

追加の実行結果

python bot.py
user> 天気教えて
bot> どこの場所の天気を知りたいですか？
user> 横浜
location: 横浜 / weather: {'weather': '晴れ', 'temperature': 38.0}
bot> 横浜の天気は晴れで、気温は38度です。
user> 暑いね。
bot> はい、確かに暑そうですね。熱中症に気をつけてください。
user> 札幌ならどうかな？
location: 札幌 / weather: {'weather': '晴れ', 'temperature': 38.0}
bot> 札幌の天気も晴れで、気温は38度です。

文脈のある会話ができていますね？
さて、先に挙げた完成までに必要な対応について確認してみましょう。

1. インテント判定・エンティティ抽出ロジックの作成: 形態素解析もNLUも使わず完成
2. 対話フロー制御の作成: 天気の場所がわからない場合は聞き返して対応。雑談を挟んで「札幌ならどうかな？」と聞くと、文脈から天気予報と判断して対応
3. スキル処理の作成: （未対応）
4. 応答メッセージの作成: 天気やリマインド情報を自然な言葉で応答しているほか、挨拶にも対応

この通り、3.スキル処理の作成以外は全て対応できています。ロジックらしいロジックを組むことなく、NLUサービスに大量の例文を学習させることなく、ここまで柔軟で正確なものができてしまいました。

LINE Botにする

ここまでコンソールでしたので、仕上げにLINE Botにしていきましょう。

コンテキスト管理をユーザー毎に分割

LINE Botは複数のユーザーからアクセスされます。これまで作ってきたチャットボットはメッセージ履歴を全てのトランザクションで共有しているため、これをユーザー別に分割します。はじめにモジュール冒頭のmessagesをコメントアウトして、message2を宣言した上でユーザー別（user_id別）の取得処理としてget_messagesを追加しましょう。

# messages = [{"role": "system", "content": system_content}]
messages2 = {}

def get_messages(user_id: str) -> list:
    if user_id not in messages2:
        messages2[user_id] = [{"role": "system", "content": system_content}]
    return messages2[user_id]

続いて、もともとmessagesを使っていた各スキルとextract_intent_gptにパラメーターuser_idを追加し、その中でユーザー別のメッセージ履歴を取得する処理を追加します。

def get_weather(user_id: str, location: str) -> str:
    messages = get_messages(user_id)
        :略

def add_reminder(user_id: str, remind_at: str, note: str) -> str:
    messages = get_messages(user_id)
        :略

async def chat(user_id: str, text: str) -> str:
        :略

def extract_intent_gpt(user_id: str, text: str) -> tuple:
    messages = get_messages(user_id)
        :略

最後に、それらの呼び出し部分となるメイン処理で適当なユーザーIDを渡すようにします。

# メイン
while True:
    # ユーザー入力を取得
    text = input("user> ")
    # インテント・エンティティ抽出
    intent, entities = extract_intent_gpt("user1234567890", text) 🌟ここ
    # スキルの取得
    skill = skills[intent]
    # スキルの実行
    resp = skill("user1234567890", **entities) 🌟ここ
    print(f"bot> {resp}")

非同期への対応

ChatGPTのAPIやその他APIを呼び出すため、非同期に対応して資源を効率的に利用できるようにします。各スキルとextract_intent_gptにasyncキーワードをつけましょう。

async def get_weather(user_id: str, location: str) -> str:
async def add_reminder(user_id: str, remind_at: str, note: str) -> str:
async def chat(user_id: str, text: str) -> str:
async def extract_intent_gpt(user_id: str, text: str) -> tuple:

また、ChatCompletion.createを呼んでいる部分をawait ChatCompletion.acreateに変更しましょう。

最後にメイン処理を以下の通り非同期実行に変更します。

import asyncio
    :略
async def main():
    # メイン
    while True:
        # ユーザー入力を取得
        text = input("user> ")
        # インテント・エンティティ抽出
        intent, entities = await extract_intent_gpt("user1234567890", text)
        # スキルの取得
        skill = skills[intent]
        # スキルの実行
        resp = await skill("user1234567890", **entities)
        print(f"bot> {resp}")

if __name__ == "__main__":
    asyncio.run(main())

入出力をLINE APIに対応

はじめに、LINE BotのSDKやWebサービスにするのに必要なモジュールたちをインストールします。

$ pip install fastapi uvicorn aiohttp line-bot-sdk

続いて以下の通りLINE Botのコードを作成します。これまでに作成したskillsと extract_intent_gptを利用するようにしています。
また、channel_access_tokenとchannel_secretの取得とWebhookのエンドポイント設定（以下サンプルコードではhttps://????/linebot）はLINE Developersで行いましょう。

import aiohttp
import traceback
from fastapi import FastAPI, Request, BackgroundTasks
from linebot import AsyncLineBotApi, WebhookParser
from linebot.aiohttp_async_http_client import AiohttpAsyncHttpClient
from linebot.models import MessageEvent, TextMessage
from bot import skills, extract_intent_gpt

channel_access_token = "<YOUR CHANNEL ACCESS TOKEN>"
channel_secret = "<YOUR CHANNEL SECRET>"

session = aiohttp.ClientSession()
client = AiohttpAsyncHttpClient(session)
line_api = AsyncLineBotApi(
    channel_access_token=channel_access_token,
    async_http_client=client
)
parser = WebhookParser(channel_secret=channel_secret)

async def handle_events(events):
    for ev in events:
        if isinstance(ev, MessageEvent):
            try:
                intent, entities = await extract_intent_gpt(ev.source.user_id, ev.message.text)
                skill = skills[intent]
                resp = await skill(ev.source.user_id, **entities)

            except Exception as ex:
                print(f"Chat error: {ex}\n{traceback.format_exc()}")
                resp = "😣"

            try:
                await line_api.reply_message(
                    ev.reply_token,
                    TextMessage(text=resp)
                )

            except Exception as ex:
                print(f"LINE error: {ex}\n{traceback.format_exc()}")

app = FastAPI()

@app.on_event("shutdown")
async def app_shutdown():
    await session.close()

@app.post("/linebot")
async def handle_request(request: Request, background_tasks: BackgroundTasks):
    events = parser.parse(
        (await request.body()).decode("utf-8"),
        request.headers.get("X-Line-Signature", "")
    )
    background_tasks.add_task(handle_events, events=events)
    return "ok"

これで完成です！LINEで話しかけてみましょう。

できたー✌️

まとめ

この記事では「これまでの作り方」とか「新しい作り方」とか言いましたが、正直なところ完全に置き換わるものでもないと思います。特にエンタープライズでの利用では柔軟性よりもカチッとロジックでフローを制御したいことがほとんどでしょう。

ですが、これまでのようにロジックを作ったり要素技術的にAIを組み合わせるよりもはるかに簡単に、また、ロジックでは複雑になりすぎてしまうようなことまで実現出来てしまうというのは、もはや革命的と言えるのではないでしょうか。みなさんもぜひチャレンジしてみていただき、どういった使い分けやChatGPTの取り入れ方が良いか考えてみてくださいね。

それでは、Enjoy creating LINE Bot!

Appendix: bot.pyの完成コード

ぶつ切りなので、最後に出来上がりを掲載しておきます。

import asyncio
import json
from openai import ChatCompletion

# ChatGPTの初期設定
api_key = "YOUR_API_KEY"
system_content = """あなたは私のAIアシスタントです。あなたは会話の他に以下のスキルを備えています。

# 天気予報スキル
* 指定された場所の天気を調べて回答します。
* 場所が不明な場合はユーザーに聞き返します。

# リマインダースキル
* ユーザーのToDoを指定日時に思い出させます。
* リマインドすべき日時や内容が不明な場合はユーザーに聞き返します。
"""
# messages = [{"role": "system", "content": system_content}]
messages2 = {}

def get_messages(user_id: str) -> list:
    if user_id not in messages2:
        messages2[user_id] = [{"role": "system", "content": system_content}]
    return messages2[user_id]

# スキル
async def get_weather(user_id: str, location: str) -> str:
    messages = get_messages(user_id)
    # 本当はAPIなどから取得
    weather = {"weather": "晴れ", "temperature": 38.0}
    print(f"location: {location} / weather: {weather}")
    # return f"{location}の天気は、{weather['weather']}。最高気温は{weather['temperature']}度の見込みだよ。"
    messages.append({"role": "function", "name": "get_weather", "content": json.dumps(weather)})
    resp = await ChatCompletion.acreate(
        api_key=api_key,
        model="gpt-3.5-turbo-0613",
        messages=messages,
    )
    messages.append(resp["choices"][0]["message"])
    return resp["choices"][0]["message"]["content"]


async def add_reminder(user_id: str, remind_at: str, note: str) -> str:
    messages = get_messages(user_id)
    # 本当はAPIなどを叩いてリマインダーを登録
    reminder = {"remind_at": remind_at, "note": note}
    print(f"remind_at: {remind_at} / note: {note} / reminder: {reminder}")
    # return f"{reminder['note']}を、{reminder['remind_at']}にお知らせします。"
    messages.append({"role": "function", "name": "add_reminder", "content": json.dumps(reminder)})
    resp = await ChatCompletion.acreate(
        api_key=api_key,
        model="gpt-3.5-turbo-0613",
        messages=messages,
    )
    messages.append(resp["choices"][0]["message"])
    return resp["choices"][0]["message"]["content"]

async def chat(user_id: str, text: str) -> str:
    return text

skills = {"get_weather": get_weather, "add_reminder": add_reminder, "chat": chat}

# ChatGPT版　インテント・エンティティの抽出処理
async def extract_intent_gpt(user_id: str, text: str) -> tuple:
    messages = get_messages(user_id)
    messages.append({"role": "user", "content": text})

    # 天気予報とリマインダーの処理を定義
    functions = [
        {
            "name": "get_weather",
            "description": "天気予報を取得する処理",
            "parameters": {
                "type": "object",
                "properties": {
                    "location": {"type": "string"}
                }
            }
        },
        {
            "name": "add_reminder",
            "description": "リマインダーを登録する処理",
            "parameters": {
                "type": "object",
                "properties": {
                    "remind_at": {"type": "string"},
                    "note": {"type": "string"}
                }
            }
        },
    ]

    # ChatGPTの呼び出し
    resp = await ChatCompletion.acreate(
        api_key=api_key,
        model="gpt-3.5-turbo-0613",
        messages=messages,
        functions=functions,
    )

    # 呼び出し結果の取得
    message = resp["choices"][0]["message"]
    messages.append(message)

    if "function_call" in message:
        # 天気やリマインダーなどの処理が判定された場合はスキルに対応するインテントとする
        return (message["function_call"]["name"], json.loads(message["function_call"]["arguments"]))
    else:
        # 処理が判定されなかった場合はchatインテントとする
        return ("chat", {"text": message["content"]})

# メイン
async def main():
    # メイン
    while True:
        # ユーザー入力を取得
        text = input("user> ")
        # インテント・エンティティ抽出
        intent, entities = await extract_intent_gpt("user1234567890", text)
        # スキルの取得
        skill = skills[intent]
        # スキルの実行
        resp = await skill("user1234567890", **entities)
        print(f"bot> {resp}")

if __name__ == "__main__":
    asyncio.run(main())