DiscordのbotをTerraform＋API Gateway+Lambda（Python）で動かす (2):応用編 Followup Messageによる長時間推論

Posted On 2024-04-15

Pocket

30{icon} {views}

DiscordのbotをTerraform+AWSでサーバーレスで作る話の続き。DiscordのInterctionは3秒位内に返さないといけないという強い制約がある一方で、Followupのメッセージを入れればタイムアウトは15分まで延長できます。Followup周りをネイティブPythonでどう扱うかという話がメインです。

やりたいこと

前回の記事で、「echo, hello world」的な簡単なDiscordのbotを、API Gateway + Lambdaでデプロイすることができた
- 前回のやり方は、Interction APIを用いている
Interction APIは最初の返答までに3秒の制約があり、LLMのような長時間の処理を伴う返答が厳しい
- DEFERRED_CHANNEL_MESSAGE_WITH_SOURCEで一旦返答し、その裏でLambdaを非同期実行し、あとからFollowup Messageを送り、推論結果を返す
- API Gatewayのタイムアウト29秒制約もまあまあ面倒なので、Followupであとから返ってくる設計は嬉しい
今回のポイントはその一連のやり取り。Followup Messageの送り方もやや癖があるので、そこのPythonでの取り回しを理解する。
結論からいうと、Followup Messageは、DEFERRED_CHANNEL_MESSAGE_WITH_SOURCEによりDiscordによって設置された一時的なWebhookに対して、一時的な認証情報を使ってPOSTするものと考えれば良い

完成形

このように、「/long」というコマンドを実行すると、「考え中」というメッセージが表示されて、しばらくしたら返答が返ってくるというもの。公式情報によると15分まで対応できる（Lambdaの最大タイムアウトと同じ）。

この発展型は以下のようなサーバー構成になる。今回はLong Time AppsのLambdaは1個のみ。

API Gatewayの直下にあるLambdaはルーティングの役割になる。こういうルーティングはAPI Gatewayで行ったり、そうでなくてもリバースプロキシやElastic Load Balancerで行うのがいいとされるが、ロードバランサーは時間課金されて高いので、API Gatewayでごまかすことにする。「CloudFrontでいいじゃん」ってのはめっちゃあるが、API Gatewayだと時間あたりのスロットリングがやりやすい。

この内容は、誰がやっても同じ感じになるので、結構記事がある。ただ、ネイティブPython＋サーバーレス実装の情報はかなり少なめ。

一番参考になったのは、2番目のClassmethodの記事。ただこれは、Lambdaを直接公開したり、CDKを使ってたり、LambdaをNode.jsで書いているので、いろいろ取り回しが違う。

公式のAPIリファレンスのドキュメントも参考になるが、どのAPIを使っていいか明瞭に書かれていなくわかりづらい。

割とこれもハマったのだが、突破口が開けたのが「Followupは一時的なWebhook」と考えたこと。基本的にはWebhookから送るときと考え方は一緒と考えたらできた。Webhookの送り方は極めて簡単で、ネット上にいくつも記事がある。

DiscordにPythonで投稿する方法(bot/Webhook)

Webhookおさらい

DiscordをWebhookから投稿するのはとても簡単。こんなコードになるはずだ。

import requests

def main():
    url = "https://discord.com/api/webhooks/<appilication_id>/<bot_token>"

    headers = {
        "Content-Type": "application/json",
        "User-Agent": "DiscordBot (private use) Python-urllib/3.10",
    }
    data = {"content": "こんにちは！プログラムから投稿しました"}

    response = requests.post(url, headers=headers, json=data)
    print(response.status_code)

if __name__ == "__main__":
    main()

このケースの場合、application_idもbot_tokenも半永続的なものだが、Followup Messageの場合、bot_tokenの部分が一時的な認証の値になる。AWSでいうSTSのようなものに近いかもしれない。Followupの一時トークンは、リクエストで渡される。ルーティングの部分からアプリのところにそのまま渡してしまえばいい。

Lambdaのコード

ルーティング部分

lambda_entrypoint.pyのコードは以下の通り。

ポイントは、InteractionResponseType.DEFERRED_CHANNEL_MESSAGE_WITH_SOURCEで一回返答したのち、非同期でLambdaを実行する点。LambdaはInvocationType="Event"とすれば非同期で実行可能。非同期で実行しないと、ルーティング部分が長時間ロックされてしまい、3秒のタイムアウトに到達してしまう。

こうすることで「Botが考えています」→「返答」のようなUIをDiscord上で実装できる。

import json
from discord_interactions import verify_key, InteractionType, InteractionResponseType
import os
import boto3

def lambda_handler(event, context):
    # for debugging
    print(event)

    request_body = json.loads(event.get("body", "{}"))
    headers = event.get("headers", {})

    # Verify request
    signature = headers.get("x-signature-ed25519")
    timestamp = headers.get("x-signature-timestamp")
    raw_body = event.get("body", "{}").encode()
    if signature is None or timestamp is None or \
        not verify_key(raw_body, signature, timestamp, os.environ.get("DISCORD_PUBLIC_KEY")): # Authorization
        return {
            'statusCode': 401,
            'body': "Bad request signature"
        }

    # Handle request
    interaction_type = request_body.get("type")

    if interaction_type in [InteractionType.APPLICATION_COMMAND, InteractionType.MESSAGE_COMPONENT]:
        data = request_body.get("data", {})
        command_name = data.get("name")

        if command_name == "long":
            response_data = {
                "type": InteractionResponseType.DEFERRED_CHANNEL_MESSAGE_WITH_SOURCE
            }
            # Kick another Lambda async
            client = boto3.client("lambda")
            client.invoke(
                FunctionName=os.environ.get("LONG_PROCESS_FUNCTION_NAME"),
                InvocationType="Event",
                Payload=json.dumps(request_body)
            )

        else:
            raise NotImplementedError(f"Command '{command_name}' not implemented")

    else:
        response_data = {"type": InteractionResponseType.PONG}

    return {
        'statusCode': 200,
        'body': json.dumps(response_data)
    }

Followupの部分

lambda_long.pyのコード。長時間の処理を行う

メッセージの投げ方はWebhookの例と瓜二つで、interaction_tokenの部分に前のLambdaから渡されたものを入れれば良い。UserAgentは入れないと403になってしまうようで、このへんはハマりポイントかもしれない。

import json
from discord_interactions import InteractionResponseType
import os
import time
import urllib.request

def lambda_handler(event, context):
    # for debugging
    # print(event)

    data = event.get("data", {})
    interaction_token = event.get("token", "")

    time.sleep(10)

    url = f"https://discord.com/api/v10/webhooks/{os.getenv("DISCORD_APP_ID")}/{interaction_token}"

    payload = {
        "content": f"Delayed echoing: {data['options'][0]['value']}"
    }

    headers = {
        'Content-Type': 'application/json',
        "User-Agent": "DiscordBot (private use) Python-urllib/3.12"
    }

    req = urllib.request.Request(url, data=json.dumps(payload).encode(), headers=headers, method="POST")

    try:
        with urllib.request.urlopen(req) as res:
            response_body = res.read().decode("utf-8")
            status_code = res.getcode()
    except urllib.error.URLError as e:
        error_message = e.read().decode("utf-8")
        print(e, error_message)
        return {
            'statusCode': 500,
            'body': str(e) + " " + error_message
        }

    print(status_code, response_body)

    return {
        'statusCode': status_code,
        'body': response_body
    }

Terraformのコード

main.tfのコードはこちら。正直前回とあまり変わりがない。

相違点や工夫点は以下の通り。

エントリーポイントのLambdaのメモリ容量を少し増やす
- デフォルトの128MBだと、コールドスタートとLambdaの実行部分が重なるとタイムアウトしがちなので、気持ちメモリ容量を増やして処理能力を上げる
Lambdaの実行ロールは、別途Lambdaの実行権限が必要なので、AWSLambdaBasicExecutionRoleだけだと無理。Terraform上でロールを作り、インラインポリシーをアタッチした。

terraform {
  required_providers {
    discord-interactions = {
      source = "roleypoly/discord-interactions"
      version = "0.1.0"
    }
  }

  backend "local" {
    path = ".cache/terraform.tfstate"
  }
}

# AWS Provider
provider "aws" {
  region = "ap-northeast-1"
  profile = "develop"
}

data "aws_caller_identity" "current" {}

# Credentials from AWS
data "aws_ssm_parameter" "discord_application_id" {
  name            = "/discord/shikoan-app/application-id"
}

data "aws_ssm_parameter" "discord_bot_token" {
  name            = "/discord/shikoan-app/token"
  with_decryption = true
}

data "aws_ssm_parameter" "discord_public_key" {
  name            = "/discord/shikoan-app/public-key"
  with_decryption = true
}

# Discord Interaction Provider
provider "discord-interactions" {
  application_id = data.aws_ssm_parameter.discord_application_id.value
  bot_token      = data.aws_ssm_parameter.discord_bot_token.value
}

# Discord bot commands
resource "discord-interactions_global_command" "long" {
  name               = "long"
  description        = "Long time echo message back to sender"
  default_permission = false

  option {
    name        = "message"
    description = "The message to echo back."
    type        = 3
    required    = true
  }
}

# IAM Role
resource "aws_iam_role" "lambda_exec_role" {
  name = "LambdaExecutionAndInvokeRole"

  assume_role_policy = jsonencode({
    Version = "2012-10-17"
    Statement = [
      {
        Action = "sts:AssumeRole"
        Effect = "Allow"
        Principal = {
          Service = "lambda.amazonaws.com"
        }
      },
    ]
  })
}

resource "aws_iam_role_policy_attachment" "attach_policy_lambda_basic_execution" {
  role       = aws_iam_role.lambda_exec_role.name
  policy_arn = "arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole"
}

resource "aws_iam_role_policy" "attach_inline_policy_invoke_lambda" {
  name = "AllowInvokeLambda"
  role = aws_iam_role.lambda_exec_role.id

  policy = jsonencode({
    Version = "2012-10-17",
    Statement = [
      {
        Action = "lambda:InvokeFunction",
        Effect = "Allow",
        Resource = "*"
      }
    ]
  })
}

# Deploy Lambda
locals {
  lambda_configs = {
    "entrypoint": {
      timeout     = 5
      memory_size = 256
      environment = {
        DISCORD_PUBLIC_KEY         = data.aws_ssm_parameter.discord_public_key.value
        LONG_PROCESS_FUNCTION_NAME = "DiscordInteractionTest_long"
      }
    },
    "long": {
      timeout     = 60
      memory_size = 256
      environment = {
        DISCORD_APP_ID    = data.aws_ssm_parameter.discord_application_id.value
      }
    }
  }
}

data "archive_file" "lambda_archives" {
  for_each    = local.lambda_configs

  type        = "zip"
  source_file = "lambda_${each.key}.py"
  output_path = ".cache/lambda_${each.key}.zip"
}

resource "aws_lambda_layer_version" "discord_interactions_layer" {
  filename            = "layer/discord-interactions.zip"
  layer_name          = "discord-interactions"

  compatible_runtimes = ["python3.12"]
}

resource "aws_lambda_function" "discord_lambdas" {
  for_each         = local.lambda_configs

  function_name    = "DiscordInteractionTest_${each.key}"
  handler          = "lambda_${each.key}.lambda_handler"
  runtime          = "python3.12"
  role             = aws_iam_role.lambda_exec_role.arn
  layers           = [ aws_lambda_layer_version.discord_interactions_layer.arn ]
  timeout          = each.value.timeout
  memory_size      = each.value.memory_size

  filename         = data.archive_file.lambda_archives[each.key].output_path
  source_code_hash = data.archive_file.lambda_archives[each.key].output_base64sha256

  environment {
    variables = each.value.environment
  }
}

# API Gateway
resource "aws_api_gateway_rest_api" "discord_api" {
  name        = "DiscordLongTimeExampleAPI"
  description = "Example Discord API Gateway for a single Lambda"
}

resource "aws_api_gateway_resource" "interaction_resource" {
  rest_api_id = aws_api_gateway_rest_api.discord_api.id
  parent_id   = aws_api_gateway_rest_api.discord_api.root_resource_id
  path_part   = "interaction"
}

resource "aws_api_gateway_method" "interaction_method" {
  rest_api_id   = aws_api_gateway_rest_api.discord_api.id
  resource_id   = aws_api_gateway_resource.interaction_resource.id
  http_method   = "POST"
  authorization = "NONE"
}

resource "aws_api_gateway_integration" "interaction_integration" {
  rest_api_id             = aws_api_gateway_rest_api.discord_api.id
  resource_id             = aws_api_gateway_resource.interaction_resource.id
  http_method             = aws_api_gateway_method.interaction_method.http_method
  integration_http_method = "POST"
  type                    = "AWS_PROXY"
  uri                     = aws_lambda_function.discord_lambdas["entrypoint"].invoke_arn
}

resource "aws_api_gateway_deployment" "sample_deployment" {
  depends_on = [
    aws_api_gateway_integration.interaction_integration
  ]

  rest_api_id = aws_api_gateway_rest_api.discord_api.id
}

resource "aws_api_gateway_stage" "dev_stage" {
  stage_name    = "dev"
  rest_api_id   = aws_api_gateway_rest_api.discord_api.id
  deployment_id = aws_api_gateway_deployment.sample_deployment.id
}

resource "aws_api_gateway_method_settings" "stage_setting" {
  rest_api_id = aws_api_gateway_rest_api.discord_api.id
  stage_name  = aws_api_gateway_stage.dev_stage.stage_name
  method_path = "*/*"

  settings {
    throttling_rate_limit  = 5
    throttling_burst_limit = 2
  }
}

resource "aws_lambda_permission" "api_gateway_permission" {
  statement_id  = "AllowAPIGatewayInvoke"
  action        = "lambda:InvokeFunction"
  function_name = aws_lambda_function.discord_lambdas["entrypoint"].function_name
  principal     = "apigateway.amazonaws.com"

  source_arn = "${aws_api_gateway_rest_api.discord_api.execution_arn}/*/${aws_api_gateway_resource.interaction_resource.path_part}"
}

所感

DiscordのbotはWeb Socketが必要とか、常駐サーバーが必要とかいろいろ言われているが、プリミティブなことをするだけならこれでも結構いけそう。このレベルができれば、ChatGPTと組み合わせていろいろいけるはず。
絞るところまで絞れば、必要なライブラリは認証用のdiscord-interactions-pythonだけでもどうにかなるんだなという発見。LambdaにDockerをいちいち使わなくてもいいのが嬉しい。
チャット履歴の取得が少し癖がありそうだが、AWS側でさばくだけならSQSなりDynamoDBなりいろいろやり方はありそう。