LINEに非同期処理と画像入力対応を実装した【multi-Agent LINE bot ①】

## 概要
LINEからマルチエージェントシステムと対話できるbotを作成しました。
その際に非同期処理と画像入力を実装したので、実装ポイントをまとめます。

親記事：[マルチエージェントと対話できるLINE botを開発しよう](https://qiita.com/MokonaSato/items/62d4dc2d28bafc48eabd)
サンプルコード：https://github.com/MokonaSato/line-multi-agent-sample

## 環境・ライブラリなど
- Python: 3.12
- line-bot-sdk: 3.17.1
- fastapi: 0.115.12
- devcontainer上で開発（image: `python:3.12-slim`）

## 参考文献
- [line-bot-sdk-python (GitHub)](https://github.com/line/line-bot-sdk-python)
- [LINE Messaging APIの概要 - LINE API総復習シリーズ](https://speakerdeck.com/uezo/line-messaging-apifalsegai-yao-line-apizong-fu-xi-sirizu)

## 実装ポイント
### 非同期処理
一般的なLINE botではWebhookで受け取ったメッセージを同期的に処理します（line-bot-sdk-pythonのREADMEから抜粋）：
```python
@app.route("/callback", methods=['POST'])
def callback():
    # get X-Line-Signature header value
    signature = request.headers['X-Line-Signature']

    # get request body as text
    body = request.get_data(as_text=True)
    app.logger.info("Request body: " + body)

    # handle webhook body
    try:
        handler.handle(body, signature)
    except InvalidSignatureError:
        app.logger.info("Invalid signature. Please check your channel access token/channel secret.")
        abort(400)

    return 'OK'


@handler.add(MessageEvent, message=TextMessageContent)
def handle_message(event):
    with ApiClient(configuration) as api_client:
        line_bot_api = MessagingApi(api_client)
        line_bot_api.reply_message_with_http_info(
            ReplyMessageRequest(
                reply_token=event.reply_token,
                messages=[TextMessage(text=event.message.text)]
            )
        )
```
この実装法では、処理に時間がかかるとLINEサーバからタイムアウトがかかったり、ユーザが複数メッセージを送ると処理が詰まってしまいます。

そこで、[植木さんのSpeakerDeck](https://speakerdeck.com/uezo/line-messaging-apifalsegai-yao-line-apizong-fu-xi-sirizu?slide=11)を参考に非同期処理を実装しました：
```python:main.py
import asyncio
from concurrent.futures import ThreadPoolExecutor

import uvicorn
from dotenv import load_dotenv
from fastapi import FastAPI, Request
from fastapi.concurrency import asynccontextmanager
from fastapi.middleware.cors import CORSMiddleware
from linebot.v3.webhooks import MessageEvent

# 内部モジュールからのインポート
from src.services.agent_service import cleanup_resources, init_agent
from src.services.line_service import (
    DEFAULT_THREAD_NAME_PREFIX,
    DEFAULT_THREAD_POOL_SIZE,
    LineClient,
    LineEventHandler,
)

# .envファイルから環境変数を読み込み
load_dotenv()

# スレッドプールの作成
executor = ThreadPoolExecutor(
    max_workers=DEFAULT_THREAD_POOL_SIZE,
    thread_name_prefix=DEFAULT_THREAD_NAME_PREFIX,
)

app = FastAPI(lifespan=lifespan)

# LINEクライアントの準備
line_client = LineClient()
line_handler = LineEventHandler(line_client)


def process_events(body: str, signature: str) -> None:
    """LINE から届いたイベントをまとめて処理（スレッド内で実行）

    Args:
        body: Webhookリクエストボディ
        signature: X-Line-Signatureヘッダー値
    """
    logger.info("Processing LINE events in background thread")
    try:
        # イベントのパース
        events = line_client.parse_webhook_events(body, signature)
        # 各イベントを処理
        for event in events:
            if isinstance(event, MessageEvent):
                # メッセージイベントを非同期で処理
                asyncio.run(line_handler.handle_event(event))
    except Exception as e:
        logger.exception(f"Error in process_events: {e}")


@app.post("/callback")
async def callback(request: Request):
    """LINE Webhookハンドラー

    Args:
        request: FastAPIリクエストオブジェクト

    Returns:
        str: 応答メッセージ
    """
    # X-Line-Signatureヘッダー値を取得
    signature = request.headers.get("X-Line-Signature", "")

    # リクエストボディをテキストとして取得
    body = await request.body()
    body_text = body.decode("utf-8")
    logger.info(f"Request body: {body_text}")

    # 非同期でWebhookボディを処理
    executor.submit(process_events, body_text, signature)

    return "OK"


if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8080)
```
処理の流れとしては以下のようになっています：
```
LINEサーバー
    ↓ Webhook送信
FastAPI (/callback)
    ↓ executor.submit() ← すぐにレスポンス返却
ThreadPool (別スレッド)
    ↓ process_events()
    ↓ line_client.parse_webhook_events()
    ↓ asyncio.run(line_handler.handle_event())
LineEventHandler
    ↓ handle_text_message() / handle_image_message()
メッセージ/画像処理
    ↓ 
LINEサーバーに返信
```

### 画像入力対応
今回実装したbotではテキストと画像両方を入力として受け付けるため、下記のように実装しています。

#### 1. イベントハンドラーでの分岐処理
上記`main.py`の`process_events`関数内で`handle_event`を呼び出します。
ここでは、テキストメッセージか画像メッセージかによって処理を分岐します。
```python:src/services/line_service/handler.py
class LineEventHandler:
    """LINEイベント処理クラス

    LINEからのイベントを処理するためのクラスです。
    各種メッセージタイプに応じた処理を提供します。
    """
    async def handle_event(self, event: MessageEvent) -> None:
        """イベントハンドラのエントリーポイント

        Args:
            event: LINEメッセージイベント
        """
        try:
            # メッセージタイプに応じて処理を分岐
            if isinstance(event.message, TextMessageContent):
                await self.handle_text_message(event, event.message)

            elif isinstance(event.message, ImageMessageContent):
                await self.handle_image_message(event, event.message)

            else:
                self.line_client.reply_text(
                    event.reply_token,
                    "申し訳ございません。このメッセージタイプには対応していません。",
                )

        except Exception as e:
            logger.exception(f"Error in handle_event: {e}")
            self._handle_error_reply(event.reply_token)
```
`handle_image_message`では画像を取得してmulti-Agentシステムに画像を入力しています。
```python:src/services/line_service/handler.py
class LineEventHandler:
    async def handle_image_message(
        self, event: MessageEvent, image_content: ImageMessageContent
    ) -> None:
        """画像メッセージの処理

        Args:
            event: LINEメッセージイベント
            image_content: 画像メッセージ内容
        """
        user_id = event.source.user_id
        reply_token = event.reply_token

        try:
            # 画像データを取得
            image_data = self.line_client.get_message_content(image_content.id)

            # デフォルトメッセージ（画像のみの場合）
            default_message = (
                "この画像からレシピを抽出してNotionに登録してください"
            )

            # エージェントに問い合わせ
            reply_text = await call_agent_with_image_async(
                message=default_message,
                image_data=image_data,
                image_mime_type="image/jpeg",  # LINEは通常JPEG
                user_id=user_id,
            )
            reply_text = reply_text.rstrip("\n")

            # 返信を送信
            self.line_client.reply_text(reply_token, reply_text)

        except Exception as e:
            logger.exception(f"Error processing image message: {e}")
            self._handle_error_reply(reply_token)
```
#### 画像取得処理
上記`handle_image_message`関数で呼び出している`get_message_content`関数内では`MessagingApiBlob`(※)から`message_id`を用いて画像を取得しています。

```python:src/services/line_service/client.py
class LineClient:
    """LINE APIクライアントラッパークラス

    LINE Messaging APIへのアクセスを簡略化するためのラッパークラス。
    設定の読み込みやエラー処理を集約しています。
    """

    def get_message_content(self, message_id: str) -> bytes:
        """メッセージの画像コンテンツを取得

        Args:
            message_id: メッセージID

        Returns:
            bytes: 画像データ
        """
        try:
            with self.create_api_client() as blob_api_client:
                blob_api = MessagingApiBlob(blob_api_client)
                image_content = blob_api.get_message_content(message_id)
                return image_content
        except Exception as e:
            raise
```
※ 画像専用API。通常のテキストメッセージでは`MessagingApi`を使用する（line-bot-sdk-pythonのREADMEから抜粋）。
```python
@handler.add(MessageEvent, message=TextMessageContent)
def handle_message(event):
    with ApiClient(configuration) as api_client:
        line_bot_api = MessagingApi(api_client)
        line_bot_api.reply_message_with_http_info(
            ReplyMessageRequest(
                reply_token=event.reply_token,
                messages=[TextMessage(text=event.message.text)]
            )
        )
```