ChatGPT API を使った Alexa Skill を作成した

我が家には Amazon Echo が 4 台あり、主に家電の操作と音楽の再生、天気予報の確認、Amazon で注文した荷物の確認に使用しています。

あと地味に便利なのがタイマー機能で、「ラーメンタイマー 3 分開始して」「卵タイマー 9 分開始して」のように並行して複数のタイマーに名前をつけて使用できるので、料理のお供に使用しています。

それ以外だとたまーに「今日は何の日？」みたいに暇つぶしで呼びかけるくらい。でもあまり自由な会話はできないのですぐ飽きちゃうんですよね…。いや十分便利なのでいいんですが…。

そこで、以前 LINE 経由で使えるチャットボットを作成した時と同じように、Echo でも使えたら便利かもしれないと思い、やってみました。

ChatGPT API を使って LINE ボットを作成した

少し前ですが ChatGPT API が公開されました。 以前から ChatGPT をいい感じに使いたいな〜と思っていて、ちょうど実家に帰った際に最近の AI は凄いぜ！的な話をしており、LINE の家族グループに「ボットくん」として参加させてみることを思い立ったというのがやってみた経緯です。 大まかな構成 使用する主要な各サービスは以下です。 * LINE Messaging API * API Gateway * Lambda * DynamoDB * ChatGPT API ざっくりとした図ではこんな感じ。 LINE の Messaging API を使用すること…

ゆるふわキャンパーLorentzca

動作イメージ

こんな感じで使えます。Alexa スキルは実行 (Lambda 関数の実行) が 8 秒でタイムアウトしてしまうので、8 秒経つ前にその時点での ChatGPT の返答を送っちゃうようにしています。会話の履歴は Alexa スキルのセッション機能で保持しているので、続きを依頼すれば続きをしゃべってくれます。

苦労した点

色々苦労したので書いていきます。

requirements.txt が動作しない

公式ドキュメントを読むと、Alexa-hosted スキル (自分の AWS アカウントの Lambda 関数等を使わず、Alexa サービス側のリソースを使う方法) の場合、Python を選択した場合は requirements.txt に必要な追加のパッケージを書いておけばデプロイ時にインストールされ使える、と書いてあります。

Python — requirements.txtファイルのdependenciesセクションに依存関係を追加します。例：
boto3==1.9.216

[依存関係の追加](https://developer.amazon.com/ja-JP/docs/alexa/hosted-skills/alexa-hosted-skills-create.html#dependencies)

しかし、この方法は現在動作していません。

いろいろ試しましたが、ログを見るとどうやっても [ERROR] Runtime.ImportModuleError: Unable to import module 'lambda_function': No module named 'openai' Traceback (most recent call last): となります。

ググるとどうやら少なくとも 2023 年の 2 月 ~ 7 月はできていたっぽいんですよね…。

• https://qiita.com/saeki4n/items/599e5af2a280325c3225

よくわからん。

ということで自分で依存パッケージも含めたコードをデプロイして解決しました。

が、ここでも注意点があって…。

コンソールからの ZIP ファイルのアップロードが不便

一応 Alexa スキルのコンソールから ZIP ファイルをアップロードできるのですが、なぜかアップロード後、デプロイしたいファイルを選択する必要があり、しかも 100 ファイルまでという制限があります。普通の Lambda 関数みたいに丸ごとアップロードしたものを使って欲しい…。

このため、たとえば openai パッケージは数百、もしかしたら数千くらいのファイルがあり、これを 100 ファイルずつ毎回 ZIP アップロードし直すところからやっていくのは正直現実的ではありません。苦行です。

この問題はコンソールからアップロードするのではなく、ASK (Alexa Skills Kit) CLI を使用して CLI からデプロイを行うことで解決できます。

• https://developer.amazon.com/ja-JP/docs/alexa/smapi/quick-start-alexa-skills-kit-command-line-interface.html
• https://github.com/alexa/ask-cli

ASK CLI を使用したインポートができない

ASK CLI には便利なインポート機能があり、既存のスキル ID を指定するとインポートしてくれます。めっちゃ便利じゃん！このオプションを見つけたときはそう思いました。

• https://developer.amazon.com/ja-JP/docs/alexa/smapi/ask-cli-command-reference.html#init-command

が、これがうまくいかない。実行すると 500 エラーになってしまう。--debug オプションをつけて実行すると、どうやらコードをインポートするエンドポイントへのアクセスが 500 エラーになっている模様。

不思議なことに他のスキルはインポートでき、何が問題だったのか不明。強いていえば他のスキルは Node.js でエラーになったものは Python という違いくらい。

面倒になったのでインポートするのではなく、新規作成してコードはコンソールからスキルをエクスポートして手で移植しました。

なお、ASK CLI を使用したスキルのデプロイ方法などは以下に書いてあります。

• https://developer.amazon.com/ja-JP/docs/alexa/hosted-skills/alexa-hosted-skills-ask-cli.html

openai パッケージバージョンの互換性

昔 (と言っても最近ですが…) の知識をもとに openai パッケージを使用するとうまく動かず、どうやら v1 から仕様が変わったようです。

丁寧な移行ガイドがあったのでこの問題についてはあんまり困らずに済みました。

• https://github.com/openai/openai-python/discussions/742

Alexa スキルの 8 秒問題

8 秒コードを実行できれば十分そうですが、ChatGPT のレスポンスが思ったより遅く、簡単な挨拶のような会話以外はほぼ全てタイムアウトしてしまう…。

何か良い方法はないものかとググったところ、 openai にはストリーミングモードというのがあって、まとめて回答を得るのではなく、回答が断片的にストリーミングで送られてくるというモードを使用し、8 秒経つ前にその時点での回答を送る、といった方法を考えられている方がいました。

Alexaスキル用のチャットボットを8秒以内に応答させたい

ZennTAKAHASHI Taro

これを参考にさせていただき、同じような実装をして解決。

ストリーミングモードのリファレンスの通り、 for 文でチャンクから断片的なメッセージを取り出し繋げていく感じで使用します。

• https://platform.openai.com/docs/api-reference/streaming

for 文開始前とブロック中で都度時間の差分を計算し 8 秒近くなったらその時点までで組み立てた文を返す実装です。

def chatgpt(message, start_time):
    # Get message from ChatGPT.
    try:
        completion = client.chat.completions.create(
            model="gpt-3.5-turbo",
            messages=message,
            stream=True
        )

        connected_message = ''
        for chunk in completion:
            elapsed_time = time.time() - start_time
            finish_reason = chunk.choices[0].finish_reason
            if finish_reason != 'stop':
                connected_message += chunk.choices[0].delta.content
            if elapsed_time > 7.5:
                connected_message += '。すみません、タイムアウトしました。続きを聞きたい場合は「続きを教えて」のように話しかけてください。'
                break

        return connected_message

    except Exception as e:
        logger.error(f"OpenAI API request failed: {e}")
        return "すみません、エラーが発生しました。しばらく時間をおいてからもう一度お試しください。"

Alexa スキルには会話毎にセッションがあり、セッションに会話を都度保存することで、文脈を保ったまま ChatGPT とやりとりできます。

• https://developer.amazon.com/ja-JP/docs/alexa/custom-skills/manage-skill-session-and-session-attributes.html

上記のブログではこのセッションも使用しておりこの部分も参考にさせていただきました。

class input_anyIntentHandler(AbstractRequestHandler):
    def can_handle(self, handler_input):
        return ask_utils.is_intent_name("input_any")(handler_input)

    def handle(self, handler_input):
        start_time = time.time()

        # セッションに既存のメッセージがある場合はそれを chatgptMessageBuilder() に渡す
        print(handler_input.attributes_manager.session_attributes)
        if 'MESSAGES' in handler_input.attributes_manager.session_attributes:
            session_messages = handler_input.attributes_manager.session_attributes["MESSAGES"]
        else:
            session_messages = None

        user_input = ask_utils.get_slot_value(
            handler_input=handler_input, slot_name="any")

        messages = chatgptMessageBuilder(user_input, session_messages)

        chatgpt_response = chatgpt(messages, start_time)

        # ChatGPT の回答をメッセージに追加
        messages.append({
            "role": "assistant",
            "content": chatgpt_response
        })

        # セッションにこれまでのメッセージを保存
        handler_input.attributes_manager.session_attributes["MESSAGES"] = messages

        directive = ElicitSlotDirective(
            slot_to_elicit='any',
            updated_intent=Intent(
                name="input_any",
                confirmation_status=IntentConfirmationStatus.NONE,
                slots={
                    "any": Slot(name="any", value="", confirmation_status=SlotConfirmationStatus.NONE)
                }
            )
        )

        speak_output = chatgpt_response

        return (
            handler_input.response_builder
            .speak(speak_output)
            .ask("他に何かお手伝いできることはありますか？")
            .response
        )

どんな発話も受け取るインテント

基本的に、Alexa スキルはある程度決まった流れに沿って会話します。例えば天気予報スキルなら予報を聞いたりある地域の天気を聞いたりと言った風に、会話の流れがある程度決まっています。

この会話の雛形のようなものがインテントです。

インテント内には変数としてスロットが設定できます。例えば「今日の天気予報は？」の「今日」の部分は変数 (「明日」が入るかもしれないし、「週末」かもしれない) なのでこれをスロットとして設定する感じです。

上のようにある程度決まった目的をもとに会話するのが Alexa スキル作りの定石なので、ChatGPT を呼び出して後は自由に会話する、のような使い方は多分あまり想定されてないです。

ではどのようにユーザーからの自由な発話を受け取るようにするかというと、いろいろ調べている方がいて自分はカスタムスロットを使っています。

• https://note.com/torusnote/n/n77896a394177

他にも方法はある模様。

• https://qiita.com/y_fujisawa/items/b443bd6c3667d8aa8443

その後…

この記事の下書きを書いてからしばらく経っているのですが、API の応答がよくわかりませんが早くなり、8 秒問題がほぼ発生しなくなりました。ナンデ…？ (API の従量課金を有効にしたから…？)

あと地味に AI モデルを gpt-4o に変更するなどしています。

completion = client.chat.completions.create(
    model="gpt-4o",
    messages=message,
    stream=True
)

感想

自由な会話ができるので、なかなか面白いです。8 秒問題回避のために 200 文字を目安に応答するよう指示しているので、テンポよく対話でき暇つぶしになりますw

def chatgptMessageBuilder(user_input, session_messages):
    template = """あなたは音声対話型チャットボットです。以下の制約にしたがって回答してください。
    制約:
    - ユーザーのメッセージに句読点を補ってから回答します
    - 200 文字以内を目安に簡潔な短い文章で話します"""

コード

最後に、参考にした記事のものを流用している感じですが、コード全体は以下のようになります。

# -*- coding: utf-8 -*-

import logging
import time
from openai import OpenAI

import ask_sdk_core.utils as ask_utils
from ask_sdk_core.skill_builder import SkillBuilder
from ask_sdk_core.dispatch_components import AbstractRequestHandler
from ask_sdk_core.dispatch_components import AbstractExceptionHandler
from ask_sdk_core.handler_input import HandlerInput
from ask_sdk_model.dialog import ElicitSlotDirective
from ask_sdk_model import (
    Response, Intent, IntentConfirmationStatus, Slot, SlotConfirmationStatus)

logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)

CHATGPT_API_KEY = '<ChatGPT の API キー>'

client = OpenAI(
    api_key=CHATGPT_API_KEY,
)


def chatgpt(message, start_time):
    # Get message from ChatGPT.
    try:
        completion = client.chat.completions.create(
            model="gpt-4o",
            messages=message,
            stream=True
        )

        connected_message = ''
        for chunk in completion:
            elapsed_time = time.time() - start_time
            finish_reason = chunk.choices[0].finish_reason
            if finish_reason != 'stop':
                connected_message += chunk.choices[0].delta.content
            if elapsed_time > 7.5:
                connected_message += '。すみません、タイムアウトしました。続きを聞きたい場合は「続きを教えて」のように話しかけてください。'
                break

        return connected_message

    except Exception as e:
        logger.error(f"OpenAI API request failed: {e}")
        return "すみません、エラーが発生しました。しばらく時間をおいてからもう一度お試しください。"


# ユーザーからの発話と既存の対話を元にメッセージを作成
def chatgptMessageBuilder(user_input, session_messages):
    template = """あなたは音声対話型チャットボットです。以下の制約にしたがって回答してください。
    制約:
    - ユーザーのメッセージに句読点を補ってから回答します
    - 200 文字以内を目安に簡潔な短い文章で話します"""

    if session_messages:
        messages = session_messages
    else:
        messages = [{
            "role": "system",
            "content": template
        }]

    messages.append({
        "role": "user",
        "content": user_input
    })

    return messages


class LaunchRequestHandler(AbstractRequestHandler):
    """Handler for Skill Launch."""

    def can_handle(self, handler_input):
        return ask_utils.is_request_type("LaunchRequest")(handler_input)

    def handle(self, handler_input):
        speak_output = "こんにちわ！私はボットくんです。何かお手伝いできることはありますか？"

        return (
            handler_input.response_builder
            .speak(speak_output)
            .ask(speak_output)
            .response
        )


class input_anyIntentHandler(AbstractRequestHandler):
    def can_handle(self, handler_input):
        return ask_utils.is_intent_name("input_any")(handler_input)

    def handle(self, handler_input):
        start_time = time.time()

        # セッションに既存のメッセージがある場合はそれを chatgptMessageBuilder() に渡す
        print(handler_input.attributes_manager.session_attributes)
        if 'MESSAGES' in handler_input.attributes_manager.session_attributes:
            session_messages = handler_input.attributes_manager.session_attributes["MESSAGES"]
        else:
            session_messages = None

        user_input = ask_utils.get_slot_value(
            handler_input=handler_input, slot_name="any")

        messages = chatgptMessageBuilder(user_input, session_messages)

        chatgpt_response = chatgpt(messages, start_time)

        # ChatGPT の回答をメッセージに追加
        messages.append({
            "role": "assistant",
            "content": chatgpt_response
        })

        # セッションにこれまでのメッセージを保存
        handler_input.attributes_manager.session_attributes["MESSAGES"] = messages

        directive = ElicitSlotDirective(
            slot_to_elicit='any',
            updated_intent=Intent(
                name="input_any",
                confirmation_status=IntentConfirmationStatus.NONE,
                slots={
                    "any": Slot(name="any", value="", confirmation_status=SlotConfirmationStatus.NONE)
                }
            )
        )

        speak_output = chatgpt_response

        return (
            handler_input.response_builder
            .speak(speak_output)
            .ask("他に何かお手伝いできることはありますか？")
            .response
        )


# これ以降はほぼデフォルト
class HelpIntentHandler(AbstractRequestHandler):
    """Handler for Help Intent."""

    def can_handle(self, handler_input):
        # type: (HandlerInput) -> bool
        return ask_utils.is_intent_name("AMAZON.HelpIntent")(handler_input)

    def handle(self, handler_input):
        # type: (HandlerInput) -> Response
        speak_output = "You can say hello to me! How can I help?"

        return (
            handler_input.response_builder
            .speak(speak_output)
            .ask(speak_output)
            .response
        )


class CancelOrStopIntentHandler(AbstractRequestHandler):
    """Single handler for Cancel and Stop Intent."""

    def can_handle(self, handler_input):
        # type: (HandlerInput) -> bool
        return (ask_utils.is_intent_name("AMAZON.CancelIntent")(handler_input) or
                ask_utils.is_intent_name("AMAZON.StopIntent")(handler_input))

    def handle(self, handler_input):
        # type: (HandlerInput) -> Response
        speak_output = "Goodbye!"

        return (
            handler_input.response_builder
            .speak(speak_output)
            .response
        )


class FallbackIntentHandler(AbstractRequestHandler):
    """Single handler for Fallback Intent."""

    def can_handle(self, handler_input):
        # type: (HandlerInput) -> bool
        return ask_utils.is_intent_name("AMAZON.FallbackIntent")(handler_input)

    def handle(self, handler_input):
        # type: (HandlerInput) -> Response
        logger.info("In FallbackIntentHandler")
        speech = "Hmm, I'm not sure. You can say Hello or Help. What would you like to do?"
        reprompt = "I didn't catch that. What can I help you with?"

        return handler_input.response_builder.speak(speech).ask(reprompt).response


class SessionEndedRequestHandler(AbstractRequestHandler):
    """Handler for Session End."""

    def can_handle(self, handler_input):
        # type: (HandlerInput) -> bool
        return ask_utils.is_request_type("SessionEndedRequest")(handler_input)

    def handle(self, handler_input):
        # type: (HandlerInput) -> Response

        # Any cleanup logic goes here.

        return handler_input.response_builder.response


class IntentReflectorHandler(AbstractRequestHandler):
    """The intent reflector is used for interaction model testing and debugging.
    It will simply repeat the intent the user said. You can create custom handlers
    for your intents by defining them above, then also adding them to the request
    handler chain below.
    """

    def can_handle(self, handler_input):
        # type: (HandlerInput) -> bool
        return ask_utils.is_request_type("IntentRequest")(handler_input)

    def handle(self, handler_input):
        # type: (HandlerInput) -> Response
        intent_name = ask_utils.get_intent_name(handler_input)
        speak_output = "You just triggered " + intent_name + "."

        return (
            handler_input.response_builder
            .speak(speak_output)
            # .ask("add a reprompt if you want to keep the session open for the user to respond")
            .response
        )


class CatchAllExceptionHandler(AbstractExceptionHandler):
    """Generic error handling to capture any syntax or routing errors. If you receive an error
    stating the request handler chain is not found, you have not implemented a handler for
    the intent being invoked or included it in the skill builder below.
    """

    def can_handle(self, handler_input, exception):
        # type: (HandlerInput, Exception) -> bool
        return True

    def handle(self, handler_input, exception):
        # type: (HandlerInput, Exception) -> Response
        logger.error(exception, exc_info=True)

        speak_output = "Sorry, I had trouble doing what you asked. Please try again."

        return (
            handler_input.response_builder
            .speak(speak_output)
            .ask(speak_output)
            .response
        )

# The SkillBuilder object acts as the entry point for your skill, routing all request and response
# payloads to the handlers above. Make sure any new handlers or interceptors you've
# defined are included below. The order matters - they're processed top to bottom.


sb = SkillBuilder()

sb.add_request_handler(LaunchRequestHandler())
# 追加したインテントハンドラー (input_anyIntentHandler()) を追記するのを忘れずに
sb.add_request_handler(input_anyIntentHandler())
sb.add_request_handler(HelpIntentHandler())
sb.add_request_handler(CancelOrStopIntentHandler())
sb.add_request_handler(FallbackIntentHandler())
sb.add_request_handler(SessionEndedRequestHandler())
# make sure IntentReflectorHandler is last so it doesn't override your custom intent handlers
sb.add_request_handler(IntentReflectorHandler())

sb.add_exception_handler(CatchAllExceptionHandler())

lambda_handler = sb.lambda_handler()