執筆日
2025/02/28
概要
以前、Function Callingについての記事を書いていましたがこの時は並列関数呼び出しに対応しておらず、一度のリクエスト実行で一つの関数しか実行してくれませんでした（一つの関数を複数回呼び出しは出来た）。が、いつの間にか対応して出来るようになっていました。API的には2023年12月には出来たよとMS Learnに書いてあるけど……？
使用方法
AzureOpenAIのクライアントのchat.completions.createメソッドで今まで通りtool_choice = "auto"を指定していればいいだけです。AzureOpenAIでは使えなかったparallel_tool_callsオプションはデフォルトでTrueになっているため、Falseにすることで今まで通り一つの関数のみが呼び出されるようにできます。
コード例
functions.py
import datetime, json
import pytz
def get_current_time(location: str):
locations = {
"Tokyo": "Asia/Tokyo",
"New York": "America/New_York",
"London": "Europe/London"
}
if location not in locations:
return "指定された都市名が不正です"
tz = pytz.timezone(locations[location])
now = datetime.datetime.now(tz)
result = {
"role": "function",
"name": "get_current_time",
"content": f"{location}の現在時刻は{now}です"
}
return result
def get_current_weather(location: str):
locations = {
"Tokyo": "晴れ",
"New York": "曇り",
"London": "雨"
}
if location not in locations:
return "指定された都市名が不正です"
result = {
"role": "function",
"name": "get_current_weather",
"content": f"{location}の天気は{locations[location]}です"
}
return result
function_map = {
"get_current_time": get_current_time,
"get_current_weather": get_current_weather,
}
tools = [
{
"type": "function",
"function": {
"name": "get_current_time",
"description": "指定された場所の現在時刻を取得します",
"parameters": {
"type": "object",
"properties": {
"location": {
"type": "string",
"description": "取得したい都市名 (使用可能な都市名: Tokyo, New York, London)",
},
},
"required": ["location"],
},
}
},
{
"type": "function",
"function": {
"name": "get_current_weather",
"description": "指定された場所の現在の天気を取得します",
"parameters": {
"type": "object",
"properties": {
"location": {
"type": "string",
"description": "取得したい都市名 (使用可能な都市名: Tokyo, New York, London)",
},
},
"required": ["location"],
},
}
},
]
def get_tools_result(res_message):
func_results = []
for tool_call in res_message.tool_calls:
tool_name = tool_call.function.name
tool_arguments = json.loads(tool_call.function.arguments)
result = function_map[tool_name](**tool_arguments)
result["tool_call_id"] = tool_call.id
func_results.append(result)
return func_results
call.py
import os
from dotenv import load_dotenv
from openai import AzureOpenAI
from functions import tools, get_tools_result
load_dotenv(".env")
OPENAI_ENDPOINT = os.getenv("OPENAI_ENDPOINT")
OPENAI_KEY = os.getenv("OPENAI_KEY")
OPENAI_API_VERSION = os.getenv("OPENAI_API_VERSION")
OPENAI_DEPLOYMENT_NAME = os.getenv("OPENAI_DEPLOYMENT_NAME")
def get_gpt_response(
client: AzureOpenAI,
system_message: str = "",
user_message: str = "",
tools: list = None,
tool_choice: str = "none", # "auto" or "none" or {"type": "function", "function": {"name": "function_name"}}
recall_chat: bool = False
):
messages = []
messages.append({
"role": "system",
"content": [{"type": "text", "text": system_message}]
})
messages.append({
"role": "user",
"content": [{"type": "text", "text": user_message}]
})
try:
response = client.chat.completions.create(
model = OPENAI_DEPLOYMENT_NAME,
messages = messages,
tools = tools,
tool_choice = tool_choice,
)
except Exception as e:
return str(e), None
res_message = response.choices[0].message
if res_message.tool_calls:
func_results = get_tools_result(res_message)
if recall_chat:
messages.extend(func_results)
response = client.chat.completions.create(
model = OPENAI_DEPLOYMENT_NAME,
messages = messages,
)
content = response.choices[0].message.content
else:
content = "\n".join([func_result["content"] for func_result in func_results])
else:
print("No tool calls found")
content = res_message.content
tokens = {"input": response.usage.prompt_tokens, "output": response.usage.completion_tokens}
return content, tokens
def ask_with_calling(client, system_message, user_message):
response, tokens = get_gpt_response(
client,
system_message,
user_message,
tools = tools_1,
tool_choice = "auto")
print(response)
print(f"input: {len(system_message) + len(user_message):5} characters, {tokens["input"]:5} tokens")
print(f"output: {len(response):5} characters, {tokens["output"]:5} tokens")
if __name__ == "__main__":
client = AzureOpenAI(
azure_endpoint = OPENAI_ENDPOINT,
api_key = OPENAI_KEY,
api_version = OPENAI_API_VERSION
)
system_message = "テンサイクンという名前の汎用的なアシスタントとしてふるまってください"
ask_with_calling(client, system_message, "東京の時間と天気を教えて下さい")
$ python call.py
>> Tokyoの現在時刻は2025-02-28 23:17:34.854961+09:00です
>> Tokyoの天気は晴れです
>> input: 49 characters, 320 tokens
>> output: 59 characters, 46 tokens
おわり
なんで前まで出来なかったんだろう。そしていつになったらfunction callingはpreviewじゃなくなるんだろう。
マルチエージェントシステムを作る際のオーケストレータとして使うのがユースケースかなと思っています。Azureであれば、
- 並列関数呼び出しで複数関数の実行パラメータを取得
- Azure Durable Functionsで関数の並列実行
- 実行結果を集約してGPTで回答生成
みたいなロジックを組むのが良さそうな気がしています。
参考
Discussion