ChatGPTと会話できるAlexaスキルにFunction callingを追加してみた
はじめに
先日、「ChatGPTと会話できるAlexaスキルを作ろう 〜Node.jsとPythonに対応〜」という本を出版しましたが、その本の最後に次のような文章が載っています。
「5.4 さらなる改良へ向けて」より
今回は、ここに挙げた「Function callingを使って、会話中に別の機能を呼び出す」というアイディアを、実際に試してみようと思います。目指すのは、次のような会話です。
- ユーザー：「アレクサ、チャットボットを開始」
- （ここでChatGPTと会話できるAlexaスキルが起動する）
- アレクサ：「私はAIチャットボットです。何でも聞いてください」
- ユーザー：「こんにちは」
- アレクサ：「こんにちは。何かお手伝いできることはありますか？」
- ユーザー：「東京の天気は？」
- （ここでFunction callingにより、天気情報を取得する関数が呼び出される）
- アレクサ：「東京の天気は晴れ、最低気温は25℃、最高気温は32℃です」
ご存知かもしれませんが、ChatGPT単体ではリアルタイムの情報を調べて答えることはできません。Function callingを使うことで、文脈から関数を呼び出すべきかどうかを判断して、必要な情報を取得できます。
今回の成果物はこちらです。
実装
ソースコードは次のとおりです。長いので折りたたんでいます。
ソースコード
class ChatBotIntentHandler(AbstractRequestHandler):
"""Handler for ChatBot Intent."""
def can_handle(self, handler_input):
# type: (HandlerInput) -> bool
return ask_utils.is_intent_name("ChatBotIntent")(handler_input)
def get_weather(self, city):
BASE_URL = "http://api.openweathermap.org/data/2.5/weather"
API_KEY = "your-api-key" # OpenWeatherのAPIキーを設定
response = requests.get(BASE_URL, params={"q": city, "appid": API_KEY})
weather_data = response.json()
# すべての温度をケルビンから摂氏に変換して上書き
weather_data["main"]["temp"] = weather_data["main"]["temp"] - 273.15
weather_data["main"]["feels_like"] = weather_data["main"]["feels_like"] - 273.15
weather_data["main"]["temp_min"] = weather_data["main"]["temp_min"] - 273.15
weather_data["main"]["temp_max"] = weather_data["main"]["temp_max"] - 273.15
return json.dumps(weather_data)
def handle(self, handler_input):
# type: (HandlerInput) -> Response
# OpenAIのAPIキーを設定
openai.api_key = "your-api-key"
# プロンプトの準備
template = """あなたは音声対話型チャットボットです。以下の制約にしたがって回答してください。
制約：
- ユーザーのメッセージに句読点を補ってから回答します
- 簡潔な短い文章で話します
- 質問の答えがわからない場合は「わかりません」と答えます"""
# メッセージの初期化
messages = [
{
"role": "system",
"content": template
}
]
# セッションからメッセージを取り出す
if "MESSAGES" in handler_input.attributes_manager.session_attributes:
messages = handler_input.attributes_manager.session_attributes["MESSAGES"]
user_input = ask_utils.get_slot_value(handler_input=handler_input, slot_name="user_message")
# ユーザーのメッセージを追加
messages.append({
"role": "user",
"content": user_input if isinstance(user_input, str) else "こんにちは"
})
# Function calling用の関数定義
functions = [
{
"name": "get_weather",
"description": "天気情報を取得する",
"parameters": {
"type": "object",
"properties": {
"city": {
"type": "string",
"description": "都市名。例: Tokyo",
},
},
"required": ["city"],
},
}
]
try:
# Function callingを有効にしてOpenAIのAPIを呼び出す
response = openai.ChatCompletion.create(
model="gpt-3.5-turbo",
messages=messages,
functions=functions,
function_call="auto"
)
# ChatGPTがfunction_callを返したら、指定された関数を呼び出す
message = response['choices'][0]['message']
if message.get('function_call'):
function_name = message["function_call"]["name"]
arguments = json.loads(message["function_call"]["arguments"])
# 指定された関数を呼び出す（ここでは一つしかないので決め打ち）
function_response = self.get_weather(
city=arguments.get("city")
)
# 関数の結果をメッセージに追加して、もう一度OpenAIのAPIを呼び出す
messages.append({
"role": "function",
"name": function_name,
"content": function_response
})
response = openai.ChatCompletion.create(
model="gpt-3.5-turbo",
messages=messages
)
speak_output = response['choices'][0]['message']['content']
except Exception as e:
logger.error(f"OpenAI API request failed: {e}")
speak_output = "すみません、エラーが発生しました。しばらく時間をおいてからもう一度お試しください。"
# ChatGPTの回答をメッセージに追加
messages.append({
"role": "assistant",
"content": speak_output
})
# セッションにメッセージを保存
handler_input.attributes_manager.session_attributes["MESSAGES"] = messages
directive = ElicitSlotDirective(
slot_to_elicit="user_message",
updated_intent = Intent(
name = "ChatBotIntent",
confirmation_status = IntentConfirmationStatus.NONE,
slots ={
"user_message": Slot(name= "user_message", value = "", confirmation_status = SlotConfirmationStatus.NONE)
}
)
)
return (
handler_input.response_builder
.speak(speak_output)
.ask("なにか話しかけてください。")
.add_directive(directive)
.response
)
ベースとなったソースコードはこちらにあります。今回追加した箇所だけを抜き出して解説します。
なお、今回のサンプルコードには実用性はありません。そもそも天気予報なら、ChatGPTを介さずAlexaに直接聞いた方が早いですし、コード自体も荒削りです。あくまで、Function callingの仕組みを理解するために作ったものですので、ご了承ください。
解説
天気情報の取得
天気の情報はOpenWeatherのWeather APIを使って取得します。APIの呼び出しは、get_weatherという関数で行います。
def get_weather(self, city):
BASE_URL = "http://api.openweathermap.org/data/2.5/weather"
API_KEY = "your-api-key" # OpenWeatherのAPIキーを設定
response = requests.get(BASE_URL, params={"q": city, "appid": API_KEY})
weather_data = response.json()
# すべての温度をケルビンから摂氏に変換して上書き
weather_data["main"]["temp"] = weather_data["main"]["temp"] - 273.15
weather_data["main"]["feels_like"] = weather_data["main"]["feels_like"] - 273.15
weather_data["main"]["temp_min"] = weather_data["main"]["temp_min"] - 273.15
weather_data["main"]["temp_max"] = weather_data["main"]["temp_max"] - 273.15
return json.dumps(weather_data)
温度の単位がケルビンなので、摂氏に変換しています。オリジナルのAPIレスポンスの例は次のとおりです。
{
"coord": {
"lon": 139.6917,
"lat": 35.6895
},
"weather": [
{
"id": 801,
"main": "Clouds",
"description": "few clouds",
"icon": "02d"
}
],
"base": "stations",
"main": {
"temp": 306.46,
"feels_like": 311.68,
"temp_min": 304.54,
"temp_max": 308.03,
"pressure": 1007,
"humidity": 55
},
"visibility": 10000,
"wind": {
"speed": 5.66,
"deg": 180
},
"clouds": {
"all": 20
},
"dt": 1691109283,
"sys": {
"type": 2,
"id": 268105,
"country": "JP",
"sunrise": 1691092250,
"sunset": 1691142233
},
"timezone": 32400,
"id": 1850144,
"name": "Tokyo",
"cod": 200
}
Function calling用の関数定義
ChatGPTが、関数を呼ぶか自分で答えるか判断できるように、関数の情報を定義します。それがこちらです。
# Function calling用の関数定義
functions = [
{
"name": "get_weather",
"description": "天気情報を取得する",
"parameters": {
"type": "object",
"properties": {
"city": {
"type": "string",
"description": "都市名。例: Tokyo",
},
},
"required": ["city"],
},
}
]
descriptionに「天気情報を取得する」とあるので、ユーザーが天気のことを知りたがっていると判断できる場合には、この関数が呼ばれるわけです。
Function callingを有効にしたAPIの呼び出し
次の処理で、Function callingを有効にしたOpenAIのAPIの呼び出しを行っています。
# Function callingを有効にしてOpenAIのAPIを呼び出す
response = openai.ChatCompletion.create(
model="gpt-3.5-turbo",
messages=messages,
functions=functions,
function_call="auto"
)
function_callをautoにすることで、文脈に応じてChatGPTが関数を呼び出すかどうかを判断します。
関数の呼び出しと後処理
ChatGPTが関数を呼び出すべきだと判断した場合（今回はユーザーが「東京の天気は？」と尋ねた場合）、レスポンスのmessageには次のような値が返ります。
{
"role": "assistant",
"content": null,
"function_call": {
"name": "get_weather",
"arguments": {
"city": "Tokyo"
}
}
}
これを踏まえて、次のコードをご覧ください。
# ChatGPTがfunction_callを返したら、指定された関数を呼び出す
message = response['choices'][0]['message']
if message.get('function_call'):
function_name = message["function_call"]["name"]
arguments = json.loads(message["function_call"]["arguments"])
# 指定された関数を呼び出す（ここでは一つしかないので決め打ち）
function_response = self.get_weather(
city=arguments.get("city")
)
# 関数の結果をメッセージに追加して、もう一度OpenAIのAPIを呼び出す
messages.append({
"role": "function",
"name": function_name,
"content": function_response
})
response = openai.ChatCompletion.create(
model="gpt-3.5-turbo",
messages=messages
)
このif分岐は、ChatGPTがfunction_callを返した場合に実行されます。function_callの中には、関数名と引数が含まれています。ここでは、function_nameにはget_weather、argumentsには{"city": "Tokyo"}が入ります。
本来は、function_nameに応じて対応する関数を呼び出すようにすべきですが、今回はget_weatherのみなので、決め打ちで呼び出しています。
get_weather関数のレスポンスをmessagesに追加して、もう一度OpenAIのAPIを呼び出すことで、ChatGPTが関数の結果を考慮したメッセージを返すようになります。今回は横着して、ChatGPTにメッセージを作成してもらいましたが、get_weather関数の中でメッセージを組み立てて返すようにすると、回答のブレがなくなり応答も早くなります。
課題
現在のコードには次のような課題があります。実用性のないサンプルコードなので、課題を解決するモチベーションは低いのですが、興味があれば改良してみてください。
- OpenWeatherのAPIは、日本語の都市名に対応していないので、ChatGPTが日本語でJSONを生成するとエラーになる
- 現在の天気しかわからない
- APIキーがハードコーディングされている（Alexa Hostedの制約なので、自前のLambdaで動かす場合は環境変数に設定できる）
まとめと宣伝
今回は、Function callingを使って、会話中に別の機能を呼び出すというアイディアに挑戦してみました。この機能を使うことで、ChatGPTがリアルタイムに情報を調べて答えることができるようになりました。
Alexaスキルの作り方は、こちらの本の中で詳しく解説しています。ぜひご覧ください。
Discussion