やりたいこと
MVCからMVAというMSのイベントで提唱されていたことを理解するために、Functions Callingを理解する必要があったので、ついでにAssistants APIのFunction Callingを試しました。
公式ドキュメント
Function Callingとは
こちらの記事を読めば、使い方と概念が理解出来ます。
Assistants APIで試す（サンプルコード）
渋谷の防災情報を教えてくれるボットを作ったとして、Function CallingでAPIを振り分けられていそうかの確認を簡単にしてみます。
import os
import time
import json
from openai import AzureOpenAI
client = AzureOpenAI(
api_key= AZURE_OPENAI_KEY,
api_version="2024-02-15-preview",
azure_endpoint = AZURE_OPENAI_ENDPOINT
)
toolsという配列を用意します。使いたいfunctionを複数定義します。
functions = [{
"type": "function",
"function": {
"name": "getCurrentWeather",
"description": "Get the weather in location",
"parameters": {
"type": "object",
"properties": {
"location": {"type": "string", "description": "東京都渋谷区"},
"unit": {"type": "string", "enum": ["℃"]}
},
"required": ["location"]
}
}}, {
"type": "function",
"function": {
"name": "getEarthquakeInformation",
"description": "都市の地震情報を取得する",
"parameters": {
"type": "object",
"properties": {
"location": {"type": "string", "description": "東京都渋谷区"},
},
"required": ["location", "timestamp"]
}
}}, {
"type": "function",
"function": {
"name": "getTransportationInformation",
"description": "都市の交通情報を取得する",
"parameters": {
"type": "object",
"properties": {
"station": {"type": "string", "description": "渋谷駅"},
},
"required": ["station"]
}
} }, {
"type": "function",
"function": {
"name": "getShelterInformation",
"description": "都市の避難所情報を取得する",
"parameters": {
"type": "object",
"properties": {
"location": {"type": "string", "description": "東京都渋谷区"},
"latitude": {"type":"string", "description": "ユーザーの緯度" },
"longitude": {"type":"string", "description": "ユーザーの経度" },
},
"required": ["location"]
}
}
}]
適当に関数を用意します。上記のtoolsで定義した関数の引数などがないと、関数実行時に怒られます。
def getCurrentWeather(location: str):
return {"temperature":23, "unit": "℃"}
def getEarthquakeInformation(location: str):
return "震度5弱の地震だったよ"
def getTransportationInformation(station: str):
return "銀座線が5分遅延"
def getShelterInformation(location: str):
pass
available_functions = {"getCurrentWeather": getCurrentWeather,
"getEarthquakeInformation": getEarthquakeInformation,
"getTransportationInformation": getTransportationInformation,
"getShelterInformation": getShelterInformation}
Assistants APIのクライアント、Threadを作成
assistant = client.beta.assistants.create(
instructions="あなたは渋谷区の防災課の職員です。災害発生時に適切なAPIを選択してください。",
model="モデル名",
tools=functions,
)
thread = client.beta.threads.create()
Assistants APIのレスポンスを出力するようにします。
※function callingを使った場合、requires_actionというAssistants側が待ちのステータスになるので、requires_actionのステータスになったら、submit_tool_outputsを実行する必要があります。
# アシスタントが回答のメッセージを返すまで待つ関数
def wait_for_assistant_response(thread_id, run_id):
while True:
time.sleep(5)
# 実行ステータスの取得
run = client.beta.threads.runs.retrieve(
thread_id=thread_id,
run_id=run_id
)
status = run.status
if status in ["completed", "cancelled", "expired", "failed", "requires_action"]:
print(status)
if run.status == "requires_action":
tool_responses = []
if (
run.required_action.type == "submit_tool_outputs"
and run.required_action.submit_tool_outputs.tool_calls is not None
):
tool_calls = run.required_action.submit_tool_outputs.tool_calls
for call in tool_calls:
if call.type == "function":
if call.function.name not in available_functions:
raise Exception("Function requested by the model does not exist")
print(call.function.name)
print(call.function.arguments)
function_to_call = available_functions[call.function.name]
tool_response = function_to_call(**json.loads(call.function.arguments))
tool_responses.append({"tool_call_id": call.id, "output": tool_response})
run = client.beta.threads.runs.submit_tool_outputs(
thread_id=thread_id, run_id=run.id, tool_outputs=tool_responses
)
break
# スレッドのメッセージを確認する関数
def print_thread_messages(thread_id):
msgs = client.beta.threads.messages.list(thread_id=thread_id)
for m in msgs:
# assert m.content[0].type == "text"
print({"role": m.role, "message": m.content[0].text.value})
試しに、震度を聞いてみると
message = client.beta.threads.messages.create(
thread_id=thread.id,
role="user",
content="今地震あったけど、震度どのくらい？"
)
run = client.beta.threads.runs.create(
thread_id=thread.id,
assistant_id=assistant.id
)
# アシスタントが回答のメッセージを返すまで待つ
wait_for_assistant_response(thread.id, run.id)
# スレッドのメッセージを確認
print_thread_messages(thread.id)
地震の情報を取得する「getEarthquakeInformation」が実行されているみたいです。
getEarthquakeInformation
{"location":"æ±äº¬é½æ¸è°·åº"}
{'role': 'user', 'message': '今地震あったけど、震度どのくらい？'}
同じように、電車について聞いたら、交通情報を聞いてみたら、それっぽいAPIを実行してくれたみたいです。
getTransportationInformation
{"station":"æ¸è°·é§
"}
{'role': 'user', 'message': '今地震あったけど、電車動いてる？'}
最後に定義していない、現在時間を聞くと下記にようになりました。
completed
{'role': 'assistant', 'message': '申し訳ありませんが、私は現在の時刻を提供する機能を持っていません。お使いのデバイスやインターネット上で現在時刻を確認してください。'}
{'role': 'user', 'message': '今何時？'}
{'role': 'assistant', 'message': '地震の影響で東京メトロ銀座線は現在5分遅れで運行しています。その他の線に関しては、情報がないため正常運行している可能性がありますが、最新の交通情報を確認することをお勧めします。'}
{'role': 'user', 'message': '今地震あったけど、電車動いてる？'}
{'role': 'assistant', 'message': '渋谷区で発生した地震の震度は5弱でした。'}
{'role': 'user', 'message': '今地震あったけど、震度どのくらい？'}
適切なAPI選択とAPI実行結果から回答を返してくれていそうです。
まとめ
MVA（Model-View-AI）はFunction Callingを上手く使うということかと思いましたが、
確かに考えとしてはわかりやすいし、コントローラーにロジックを人が書かなくなるのかもなーと思いました。
SLMは必要かと思いますが。
あと、今回のサンプルだとAssistants APIのFunction Callingを使う意味があまりなさそうなので、
ユースケースは考えたいなーと。
Discussion