はじめに
タイトルの通りOpenAI APIのfunction_callingとStreamを同時使用する場合、引数がStreamで渡されます。この現象に日本語で解説した情報があまりなさそうだったのでまとめていこうと思います。
現象
いきなりですがコードを見てもらうのが一番手っと早く理解できそうなので、こちらをご覧ください！
response = openai.chat.completions.create(
model="gpt-4-1106-preview",
messages=messages,
functions=[
{
"name": "get_current_weather",
"description": "Get the current weather in a given location",
"parameters": {
"type": "object",
"properties": {
"location": {
"type": "string",
"description": "The city and state, e.g. San Francisco, CA",
},
"unit": {"type": "string", "enum": ["celsius", "fahrenheit"]},
},
"required": ["location"],
},
}
],
function_call="auto",
stream=True,
)
for chunk in response:
function_call = chunk.choices[0].delta.function_call
print(function_call)
上記コードを実行した出力は以下になります。
ChoiceDeltaFunctionCall(arguments='', name='get_current_weather')
ChoiceDeltaFunctionCall(arguments='{"', name=None)
ChoiceDeltaFunctionCall(arguments='location', name=None)
ChoiceDeltaFunctionCall(arguments='":"', name=None)
ChoiceDeltaFunctionCall(arguments='Mat', name=None)
ChoiceDeltaFunctionCall(arguments='sum', name=None)
ChoiceDeltaFunctionCall(arguments='oto', name=None)
ChoiceDeltaFunctionCall(arguments=',', name=None)
ChoiceDeltaFunctionCall(arguments=' JP', name=None)
ChoiceDeltaFunctionCall(arguments='","', name=None)
ChoiceDeltaFunctionCall(arguments='unit', name=None)
ChoiceDeltaFunctionCall(arguments='":"', name=None)
ChoiceDeltaFunctionCall(arguments='c', name=None)
ChoiceDeltaFunctionCall(arguments='elsius', name=None)
ChoiceDeltaFunctionCall(arguments='"}', name=None)
見ての通り、引数がChunk毎に分割して送られてきます。
解決策
response = openai.chat.completions.create(
model="gpt-4-1106-preview",
messages=messages,
functions=[
{
"name": "get_current_weather",
"description": "Get the current weather in a given location",
"parameters": {
"type": "object",
"properties": {
"location": {
"type": "string",
"description": "The city and state, e.g. San Francisco, CA",
},
"unit": {"type": "string", "enum": ["celsius", "fahrenheit"]},
},
"required": ["location"],
},
}
],
function_call="auto",
stream=True,
)
function_info = {
"name": None,
"arguments": "",
}
for chunk in response:
function_call = chunk.choices[0].delta.function_call
if function_call:
if function_call.name is not None and function_call.name != "":
function_info["name"] = function_call.name
if function_call.arguments is not None and function_call.arguments != "":
function_info["arguments"] += function_call.arguments
continue
# chunkの吐き出しが終わると"finish_reason"が渡される
if chunk.choices[0].finish_reason == "function_call":
function_to_execute = globals()[function_info["name"]]
arguments = json.loads(function_info["arguments"])
# 関数実行
こんな感じで愚直ですが、引数を貯めておきましょう。
終わりに
function_callingとStreamを同時に使う場合は一手間ですが必要になるので、参考になれば幸いです。
ここから実プロダクトになると、「functionがcallされなかった時の処理」や「戻り値を元にOpenAI APIに再回答を求める場合」等まだまだ考慮が必要になってきそうですが、そちらは私も納得のいくコードが書けていないのでまたの機会に