FastAPIで非同期処理に対応したLINEBOTを開発する

チャットボットは自身で開発したプログラム単独で完結できる場合は少なく、何らかのAPIを呼び出すことが多いと思います。（自然言語解析関連、ストレージ関連、翻訳や天気など各種サービス関連・・・）
したがって必然的にI/O待ち時間が発生しがちで、非同期処理との相性が良いはずなのですが、LINEBOTに関しては非同期処理のサンプルを全然見かけないので記事にしてみました。

APIクライアントの非同期対応状況について

まずもってですが、根っことしてLINE Messaging APIの公式Python SDKが非同期処理に対応していないという状況があります。古いPythonバージョンがまだサポート対象であることが理由のようで、EOLは間近としてもまだしばらくかかりそうです。（2021年1月27日時点）

[Feature Request] Asynchronous/Asyncio support

ところが上のやりとりの中で、コミュニティの有志から非同期対応版のSDKが公開されていましたので、今回はありがたく利用させていただくこととします。なお本記事の内容は公式SDKが非同期対応後にもAPIクライアントのインスタンス化部分のみを差し替えればそのまま利用することができます。

aiolinebot https://github.com/uezo/aiolinebot

インストール時に公式SDKを継承して動的に非同期メソッドを追加する仕組で、原則的にすべての同期メソッドの非同期版を提供している模様です。

必要なライブラリのインストール

先に挙げたaiolinebotに加えて、FastAPIとASGIサーバもインストールします。

$ pip install fastapi uvicorn aiolinebot

FastAPIとUvicornの動作確認

はじめにFastAPIとUvicornがきちんとインストールできたことを確認します。以下の通りrun.pyを作成してhttp://localhost:8000/messaging_api/handle_requestにアクセスしてみましょう。okと表示されれば動作確認完了です。

run.py

from fastapi import FastAPI

app = FastAPI()

@app.post("/messaging_api/handle_request")
async def handle_request():
    return "ok"

なおサーバーの起動方法は以下の通りです。--debugをつけるとファイルを更新した際に自動で再読み込みが走るので開発時には便利です。

$ uvicorn run:app --debug

インターネットからアクセスできるようにする

LINEBOTの開発にあたっては、LINEのサーバからのHTTPリクエストを受け付けられるようにする必要があります。みんな大好きngrokをインストールしてUvicornのポート番号にルーティングしましょう。

$ ngrok http 8000

ここで払い出されるドメインを使用して、先のサーバープログラムにアクセスできることを確認してください。https://xxxxx.ngrok.io/messaging_api/handle_request といった体系です。HTTPSであることと、ポート番号は不要な点に気をつけましょう。xxxxx部分は起動都度変わってしまうので、作業が終わるまでngrokを終了しないようにしてください。

トークンの取得とエンドポイントURLの登録

チャネルアクセストークンとチャネルシークレットの2種類のキーを取得し、また、Messaging APIのエンドポイントとして先のngrokのURL https://xxxxx.ngrok.io/messaging_api/handle_request を登録します。

こちらの記事を参考にしていただければすぐにできると思います→ LINEのBot開発 超入門（前編） ゼロから応答ができるまで

おうむ返しのLINEBOTを作成する

ここまでで下準備は完了です。ユーザーの発話内容をおうむ返しするLINEBOTを作ってみましょう。先程のrun.pyを以下のように書き換えます。

run.py

from fastapi import FastAPI, Request
from linebot import WebhookParser
from linebot.models import TextMessage
from aiolinebot import AioLineBotApi

# APIクライアントとパーサーをインスタンス化
line_api = AioLineBotApi(channel_access_token="<YOUR CHANNEL ACCESS TOKEN>")
parser = WebhookParser(channel_secret="<YOUR CHANNEL SECRET>")

# FastAPIの起動
app = FastAPI()

@app.post("/messaging_api/handle_request")
async def handle_request(request: Request):
    # リクエストをパースしてイベントを取得（署名の検証あり）
    events = parser.parse(
        (await request.body()).decode("utf-8"),
        request.headers.get("X-Line-Signature", ""))

    # 各イベントを処理
    for ev in events:
        await line_api.reply_message_async(
            ev.reply_token,
            TextMessage(text=f"You said: {ev.message.text}"))

    # LINEサーバへHTTP応答を返す
    return "ok"

LINEアプリで話しかけてみて、たとえば「こんにちは」と話しかけると「You said: こんにちは」と応答が帰ってくれば成功です。

応答メッセージ返却のバックグラウンドタスクへの移行

先のおうむ返しBOTは、LINEBOT的には実はいわゆるアンチパターンです。
公式のLINEボット開発ガイドラインには「1000ミリ秒以内にLINEサーバーにHTTP応答を返せ」とありますが、先の実装だと受け取ったイベント（メッセージ）をすべて処理してから応答するため、1秒以内に完結しない可能性があります。
そこでFastAPIのBackgroundTaskを利用して、メッセージの処理をバックグラウンドタスクに突っ込んだらすぐにLINEヘHTTP応答を返すように修正してみます。変更点には🌟マークをつけておきました。

run.py

from fastapi import FastAPI, Request, BackgroundTasks  # 🌟BackgroundTasksを追加
from linebot import WebhookParser
from linebot.models import TextMessage
from aiolinebot import AioLineBotApi

# APIクライアントとパーサーをインスタンス化
line_api = AioLineBotApi(channel_access_token="<YOUR CHANNEL ACCESS TOKEN>")
parser = WebhookParser(channel_secret="<YOUR CHANNEL SECRET>")

# FastAPIの起動
app = FastAPI()

# 🌟イベント処理（新規追加）
async def handle_events(events):
    for ev in events:
        try:
            await line_api.reply_message_async(
                ev.reply_token,
                TextMessage(text=f"You said: {ev.message.text}"))
        except Exception:
            # エラーログ書いたりする
            pass

@app.post("/messaging_api/handle_request")
async def handle_request(request: Request, background_tasks: BackgroundTasks):  # 🌟background_tasksを追加
    # リクエストをパースしてイベントを取得（署名の検証あり）
    events = parser.parse(
        (await request.body()).decode("utf-8"),
        request.headers.get("X-Line-Signature", ""))

    # 🌟イベント処理をバックグラウンドタスクに渡す
    background_tasks.add_task(handle_events, events=events)

    # LINEサーバへHTTP応答を返す
    return "ok"

修正したら動作確認してみましょう。とはいっても体感上は何も変わらないと思います。本当にLINEサーバーへの応答が即時に返っているのか気になる方は、handle_eventsの処理にかかる時間を極端に長くした上で、uvicornのログ"POST /messaging_api/handle_request HTTP/1.1" 200 OKがすぐに表示されて、5秒くらい経ったあとにLINEに応答メッセージが届くことが確認できると思います。

run.py

import asyncio  # 🌟冒頭に追加

　略

async def handle_events(events):
    await asyncio.sleep(5)  # 🌟イベント処理前に5秒待つ
    for ev in events:
        略

なおFastAPIのバックグラウンド処理の挙動については別記事でまとめました。あわせてお読みいただけると理解が深まると思います。
FastAPIのバックグラウンド処理の多重度を同期・非同期で比較してみたよ

まとめ

FastAPIを使うことで、巷によくあるFlaskベースのサンプルと同じくらい簡単に非同期処理に対応したLINEBOTが作れるということをご理解頂けたかと思います。またBackgroundTasksを使えば1000ミリ秒以内のレスポンスに対応することもとても簡単です。冒頭にも書きましたとおり、特に各種外部APIと連携するチャットボットであればサーバー資源の効率的な利用につながると思います。
あとうろ覚えですがAzure Functionsだとメインスレッド以外からログを書き出せないと思いますので、バックグラウンド処理含めてメインスレッドで処理するメリットがその点からも得られると思います。ぜひ試してみてください。