Closed2023/07/26にクローズ7

OpenAIのFunction callingをやってみる

OpenAI

LangChain

LLM

functioncalling

kun432

周回遅れも甚だしいが、Function Callingを理解するために手を動かしてやってみる。LangChainでの使い方も見てみる予定。

Colaboratoryを使う。

kun432

Azure OpenAI Serviceでも使えるようになったのでAzureで。公式のドキュメントに従ってやってみる。

変数は適宜設定
- 今回はgpt-3.5-turboを使用。
- APIバージョンは現時点だと2023-07-01-previewでないとFunction Callingは動作しないので注意。

import os
import requests
import json
from pprint import pprint

import openai

openai.api_key = AZURE_OPENAI_KEY
openai.api_base = AZURE_OPENAI_ENDPOINT
openai.api_type = 'azure'
openai.api_version = AZURE_OPENAI_API_VERSION
deployment_name = AZURE_OPENAI_CHAT_DEPLOYMENT

messages= [
    {"role": "system", "content": "あなたは親切なAIアシスタントです。ユーザーの質問に日本語で答えます。"},
    {"role": "user", "content": "サンディエゴで、ビーチに面していて、朝食は無料で、月300ドル以下のホテルを教えて。"}
]

functions= [  
    {
        "name": "search_hotels",
        "description": "Retrieves hotels from the search index based on the parameters provided",
        "parameters": {
            "type": "object",
            "properties": {
                "location": {
                    "type": "string",
                    "description": "The location of the hotel (i.e. Seattle, WA)"
                },
                "max_price": {
                    "type": "number",
                    "description": "The maximum price for the hotel"
                },
                "features": {
                    "type": "string",
                    "description": "A comma separated list of features (i.e. beachfront, free wifi, etc.)"
                }
            },
            "required": ["location"],
        },
    }
]  

response = openai.ChatCompletion.create(
    engine=deployment_name,
    messages=messages,
    functions=functions,
    function_call="auto", 
)

pprint(response)

返ってくるレスポンスはこんな感じのJSONオブジェクト。

{
  "id": "chatcmpl-7g3wgMWue031LWXFCPyH2pEC2J4xk",
  "object": "chat.completion",
  "created": 1690259298,
  "model": "gpt-35-turbo",
  "prompt_annotations": [
    {
      "prompt_index": 0,
      "content_filter_results": {
        "hate": {
          "filtered": false,
          "severity": "safe"
        },
        "self_harm": {
          "filtered": false,
          "severity": "safe"
        },
        "sexual": {
          "filtered": false,
          "severity": "safe"
        },
        "violence": {
          "filtered": false,
          "severity": "safe"
        }
      }
    }
  ],
  "choices": [
    {
      "index": 0,
      "finish_reason": "function_call",
      "message": {
        "role": "assistant",
        "function_call": {
          "name": "search_hotels",
          "arguments": "{\n  \"location\": \"\u30b5\u30f3\u30c7\u30a3\u30a8\u30b4\",\n  \"max_price\": 300,\n  \"features\": \"\u30d3\u30fc\u30c1,\u7121\u6599\u671d\u98df\"\n}"
        }
      },
      "content_filter_results": {}
    }
  ],
  "usage": {
    "completion_tokens": 45,
    "prompt_tokens": 181,
    "total_tokens": 226
  }
}

ポイントはここ。（Unicodeエンコードされてる部分はデコードしてある）

(snip)
  "choices": [
    {
      "index": 0,
      "finish_reason": "function_call",
      "message": {
        "role": "assistant",
        "function_call": {
          "name": "search_hotels",
          "arguments": "{\n  \"location\": \"\サンディエゴ\",\n  \"max_price\": 300,\n  \"features\": \"ビーチ,無料朝食\"\n}"
        }
      },
(snip)

ユーザーの入力内容から予め定義された関数を使うべき、とLLMが判断した場合、function_callプロパティに呼び出すべき関数名と引数を返してくれると。なるほど。

ちなみに呼び出すべき関数がない場合はこうなる。（Unicodeエンコードされてる部分はデコードしてある）

(snip)
  "choices": [
    {
      "index": 0,
      "finish_reason": "stop",
      "message": {
        "role": "assistant",
        "content": "申し訳ありませんが、私は天気情報を提供できる機能を備えていません。天気予報に関する情報を調べるためには、インターネット上の天気予報サイトや天気予報アプリをご利用ください。"
      },
(snip)

普通にcontentで返される。

ドキュメントによると、function_callとcontentの両方が返ってくる場合があるらしい（補足的な説明とか）。なので、function_callが定義されているかでルーティングしてやればよいらしい。

で関数は実行してくれるというわけではなくて、自分で実行する。そしてその結果を再度LLMに渡すという流れになる。さっきのコードに続けて書く。実行される関数もダミーで追加してみた。

(snip)
first_response = response['choices'][0]['message']

from typing import List

def search_hotels(location: str, max_price: int, features: str) -> str:
    # 実際にやるならばここで検索APIに条件を渡す感じ。ここでは単純のため、全部を連結した文字列を返す。
    list_features = [f for f in features.split(',')]
    str_features = '/'.join(list_features)
    return f"場所: {location}、料金上限: {max_price} ドル、サービス: {str_features} のオススメホテルは、ホテル{location}です。"

if first_response.get("function_call"):
    function_name = first_response["function_call"]["name"]

    available_functions = {
            "search_hotels": search_hotels,
    }
    function_to_call = available_functions[function_name] 

    function_args = json.loads(first_response["function_call"]["arguments"])
    function_response = function_to_call(**function_args)

    messages.append(
        {
            "role": first_response["role"],
            "name": first_response["function_call"]["name"],
            "content": first_response["function_call"]["arguments"],
        }
    )
    messages.append(
        {
            "role": "function",
            "name": function_name,
            "content": function_response,
        }
    ) 

    second_response = openai.ChatCompletion.create(
            messages=messages,
            engine=deployment_name,
    )
    print(second_response["choices"][0]["message"]["content"])
else:
    print(first_response["choices"][0]["message"]["content"])

結果

サンディエゴで、ビーチに面していて、朝食は無料で、月300ドル以下のホテルをお探しですね。オススメのホテルは「ホテルサンディエゴ」です。価格帯や条件に合わせて、ビーチに近い立地で滞在することができます。ぜひ予約を検討してみてください。

messagesで会話履歴を保持しているようなのでpprintしてみる。

[{'content': 'あなたは親切なAIアシスタントです。ユーザーの質問に日本語で答えます。', 'role': 'system'},
 {'content': 'サンディエゴで、ビーチに面していて、朝食は無料で、月300ドル以下のホテルを教えて。', 'role': 'user'},
 {'content': '{\n'
             '  "location": "サンディエゴ",\n'
             '  "max_price": 300,\n'
             '  "features": "ビーチ,無料朝食"\n'
             '}',
  'name': 'search_hotels',
  'role': 'assistant'},
 {'content': '場所: サンディエゴ、料金上限: 300 ドル、サービス: ビーチ/無料朝食 のオススメホテルは、ホテルサンディエゴです。',
  'name': 'search_hotels',
  'role': 'function'}]

なるほど、以下も含めて会話履歴に残すということか。

assistantロールがsearch_hotels関数を選択したことが記録される
functionロールでsearch_hotels関数を実行したことが結果とともに記録される

kun432

Function callingを試す前は、関数定義しておけばLLM側でそれっぽく実行した結果を安定した出力で返してくれるもん、と思いこんでいたけど、全然違った。なんというかエージェントっぽいルーティングをしつつ、関数の出力結果をチェーンで繋いでいくようなイメージ。

で、それはそれとして、安定した出力で返してほしい、例えば、JSONとか、っていうニーズはある。以下の記事が参考になる。

ということで書いてみた。

import os
import requests
import json
from pprint import pprint
from typing import List
import random

import openai

openai.api_key = AZURE_OPENAI_KEY
openai.api_base = AZURE_OPENAI_ENDPOINT
openai.api_type = 'azure'
openai.api_version = AZURE_OPENAI_API_VERSION
deployment_name = AZURE_OPENAI_CHAT_DEPLOYMENT

def search_hotels(location: str, max_price: int, features: str) -> str:
    # 実際にやるならばここで検索APIに条件を渡す感じ。ここでは単純のため、全部を連結した文字列を返す。
    list_features = [f for f in features.split(',')]
    str_features = '/'.join(list_features)
    show_url = random.randrange(2)
    url_string = ", ホテル画像URL="
    if show_url == 1:
        print("URL provided")
        url_string += "https://www.example.com/sample.jpg"
    else:
        print("URL not provided")
        url_string += "画像なし"

    search_result = f"検索条件: 場所={location}, 料金={max_price}ドル, サービス={str_features}\n検索結果: ホテル名=ホテル{location}, ホテル画像URL={url_string}"
    print(f"検索ログ: {search_result}")
    return search_result

system_prompt = "あなたは親切なAIアシスタントです。ユーザーの質問に日本語で答えます。"
user_query = "サンディエゴで、ビーチに面していて、朝食は無料で、月300ドル以下のホテルを教えて。"

messages= [
    {"role": "system", "content": system_prompt},
    {"role": "user", "content": user_query}
]

functions= [  
    {
        "name": "search_hotels",
        "description": "Retrieves hotels from the search index based on the parameters provided",
        "parameters": {
            "type": "object",
            "properties": {
                "location": {
                    "type": "string",
                    "description": "The location of the hotel (i.e. Seattle, WA)"
                },
                "max_price": {
                    "type": "number",
                    "description": "The maximum price for the hotel"
                },
                "features": {
                    "type": "string",
                    "description": "A comma separated list of features (i.e. beachfront, free wifi, etc.)"
                }
            },
            "required": ["location"],
        },
    },
]  

static_function = [
    {
        "name": "hotel_info_json",
        "description": "handle the previous search result about the hotel as JSON",
        "parameters": {
            "type": "object",
            "properties": {
                "text": {
                    "type": "string",
                    "description": "text description and recommendation about the hotel based on the previous search result. do not include URL string or mention about URL."
                },
                "imageUrl": {
                    "type": "string",
                    "description": "URL string of the hotel based on the previous search result. must be blank if not provided in the previous search result."
                },
            },
            "required": ["text","imageUrl"],
        }
    }
]

print("1st API called to select function or not")
res = openai.ChatCompletion.create(
    engine=deployment_name,
    messages=messages,
    functions=functions,
    function_call="auto", 
)

first_response = res['choices'][0]['message']

if first_response.get("function_call"):
    print("function called")
    function_name = first_response["function_call"]["name"]

    available_functions = {
            "search_hotels": search_hotels,
    }
    function_to_call = available_functions[function_name] 

    function_args = json.loads(first_response["function_call"]["arguments"])
    function_response = function_to_call(**function_args)

    messages.append(
        {
            "role": first_response["role"],
            "name": first_response["function_call"]["name"],
            "content": first_response["function_call"]["arguments"],
        }
    )
    messages.append(
        {
            "role": "function",
            "name": function_name,
            "content": function_response,
        }
    ) 

    print("2nd API called with function for output")
    second_response = openai.ChatCompletion.create(
            messages=messages,
            engine=deployment_name,
            functions=static_function,
            function_call={"name": "hotel_info_json"}

    )
    answer = json.loads(second_response["choices"][0]["message"]["function_call"]["arguments"])
else:
    answer = {"text": first_response["content"]}

print(answer)

前提として、最終的な出力は常に以下のJSONフォーマットになるようにしてみた。

{
    "text": "<検索結果から生成した、ホテルの説明やオススメ内容等>"
    "imageUrl": "<検索結果に画像URLが含まれていた場合はここにURLを含める>"
}

流れとしてはこんな感じ。

1回目のAPIリクエストで、ユーザーの入力内容から、Function calling経由でsearch_hotels関数を使うべきかどうか、を推論させる。

(snip)
functions= [  
    {
        "name": "search_hotels",
        "description": "Retrieves hotels from the search index based on the parameters provided",
        "parameters": {
            "type": "object",
            "properties": {
                "location": {
                    "type": "string",
                    "description": "The location of the hotel (i.e. Seattle, WA)"
                },
                "max_price": {
                    "type": "number",
                    "description": "The maximum price for the hotel"
                },
                "features": {
                    "type": "string",
                    "description": "A comma separated list of features (i.e. beachfront, free wifi, etc.)"
                }
            },
            "required": ["location"],
        },
    },
] 
(snip)
print("1st API called to select function or not")
res = openai.ChatCompletion.create(
    engine=deployment_name,
    messages=messages,
    functions=functions,
    function_call="auto", 
)

first_response = res['choices'][0]['message']
(snip)

Function callingでsearch_hotels関数を使うべき、と判断したら、ユーザーの入力内容から各引数を生成し、実行する。

(snip)
if first_response.get("function_call"):
    print("function called")
    function_name = first_response["function_call"]["name"]

    available_functions = {
            "search_hotels": search_hotels,
    }
    function_to_call = available_functions[function_name] 

    function_args = json.loads(first_response["function_call"]["arguments"])
    function_response = function_to_call(**function_args)
(snip)

実際に実行されるsearch_hotels関数もすこし変えた。

def search_hotels(location: str, max_price: int, features: str) -> str:
    list_features = [f for f in features.split(',')]
    str_features = '/'.join(list_features)
    show_url = random.randrange(2)
    url_string = ", ホテル画像URL="
    if show_url == 1:
        print("URL provided")
        url_string += "https://www.example.com/sample.jpg"
    else:
        print("URL not provided")
        url_string += "画像なし"

    search_result = f"検索条件: 場所={location}, 料金={max_price}ドル, サービス={str_features}\n検索結果: ホテル名=ホテル{location}, ホテル画像URL={url_string}"
    print(f"検索ログ: {search_result}")
    return search_result

実際にはデータベース等から検索せずにダミーの文字列を返すだけというのは最初の例と同じだが、前回は一つの文字列としてマルっと出力していただけなのを多少データベースからの検索結果っぽいフォーマットに変えてみた。あと、最終的に期待するフォーマットには画像のURLもあるので、それもこの関数の中で返すように追加した。

で、画像のURLはランダムで生成するようにした。例えば既存のコンテンツをVector化してドキュメント検索させるような場合、画像URLなどの付加情報は本来ならメタデータで管理したほうがいい気がするんだけど、

膨大な既存のコンテンツをテキストとメタデータ（画像URL）にきちんと分けてVector DB化するのって大変じゃない？マルっとテキストに画像URLも含めておいて、検索結果を元に回答を生成する際にうまく分けてくれたら嬉しい
テキストと画像URLが必ずしもペアになるとは限らず、むしろテキストだけの情報のほうが圧倒的に多いはず。となると、画像があれば出力してくれるし、なければ出力しない、というふうにすればいい。

と考えた次第。

画像URLが得られない場合、最終出力するJSONから画像URLのキー（imageUrl）をそもそも含めない、というのも試してみたんだけど、どうも出力が安定せず架空のURLを生成したりするので、画像URL
がない場合でも{"imageUrl": "画像なし"}として含めるようにした。

で、この辺はもう一つのFunction calling用の関数の定義にも絡んでくるのでそれは後述。

2回目のAPIリクエストにコンテキストとして受け渡すために、ここまでの流れをmessagesに追加しておく。

    messages.append(
        {
            "role": first_response["role"],
            "name": first_response["function_call"]["name"],
            "content": first_response["function_call"]["arguments"],
        }
    )
    messages.append(
        {
            "role": "function",
            "name": function_name,
            "content": function_response,
        }
    )

で、2回目のAPIリクエスト。

(snip)
static_function = [
    {
        "name": "hotel_info_json",
        "description": "handle the previous search result about the hotel as JSON",
        "parameters": {
            "type": "object",
            "properties": {
                "text": {
                    "type": "string",
                    "description": "text description and recommendation about the hotel based on the previous search result. do not include URL string or mention about URL."
                },
                "imageUrl": {
                    "type": "string",
                    "description": "URL string of the hotel based on the previous search result. must be blank if not provided in the previous search result."
                },
            },
            "required": ["text","imageUrl"],
        }
    }
]
(snip)
    print("2nd API called with function for output")
    second_response = openai.ChatCompletion.create(
            messages=messages,
            engine=deployment_name,
            functions=static_function,
            function_call={"name": "hotel_info_json"}

    )
    answer = json.loads(second_response["choices"][0]["message"]["function_call"]["arguments"]) 
(snip)

最初の例では、そのままLLMに最終回答となるテキストを生成させていたけど、今回はフォーマット化された出力を常に返してほしいので、ここで再度Function callingを使う。

ポイントとしては、

openai.ChatCompletion.createでfunction_callプロパティに{"name": "<関数名>"}を定義すると、関数の選択を判断することなく、必ず指定された関数を使うことを提案するようになる。
今回は「検索結果をJSONで出力する」ための関数としてFunction calling用に定義。
ただし実際に実行する必要はない。なぜなら必要なのはFunction callingで生成される「引数部分」がJSONになってるのでこれを使えば良いだけ。なので実際に実行する関数の定義が不要なほんとのダミー関数となる。
Function calling用の関数定義における、descriptionやパラメータのpropertiesは、Functaion calling用のプロンプトとして使用される。よって必要な情報が必要なフォーマットで得られやすくするために工夫する必要がある
- 上の方で述べたけど、検索結果に画像URLがない場合の振る舞いを定義しておかないと、架空のURLが生成されたり、URLがあっても出力から抜け落ちたり、descriptionに「画像はありません」と出力されたり、と出力が不安定な感じになったので、それを避けるようにプロンプトを追加。
- また、"required"でプロパティが必須なのかオプションなのかを指定できるけど、オプションにするとこれもまた不安定な出力になる。プロンプトと同じで、ブレがあったり、自由度があったりすような定義だとより不安定になりやすい印象。
- ということで、出力のフォーマットは完全に固定、キーは常に存在するけど値がなければ空文字にする、ということにした。

あたり。この辺は結構トライ＆エラーな感じ。これをプロンプトだけで全部実現するのはいろいろ大変だけど、Function callingにおいてもいくらか楽にはなるもののプロンプトはやはり重要な感じです。

で前後してしまうけど、1回目のAPIリクエストでFunction callingが提案されなかった場合は普通にLLMで文字列として回答が生成されてしまうので、ここでも同じフォーマットでJSON出力するようにしておく。

(snip)
else:
    answer = {"text": first_response["content"], "imageUrl": ""}
(snip)

kun432

では実際に実行してみる。すこしデバッグ的な出力も含めているので実際に実行してみると処理の流れがわかると思う。

まず、普通にホテル検索を必要とする質問が来た場合。検索結果に画像URLがある場合・ない場合で分けてみてみる。

user_query = "サンディエゴで、ビーチに面していて、朝食は無料で、月300ドル以下のホテルを教えて。"

画像がある場合

{'text': 'サンディエゴでおすすめのホテルは、ホテルSan Diegoです。ビーチに面しており、朝食は無料です。価格は月300ドル以下です。', 'imageUrl': 'https://www.example.com/sample.jpg'}

画像がない場合。

{'text': 'ホテルサンディエゴは、サンディエゴのビーチに面しており、朝食は無料です。月額料金は300ドル以下です。', 'imageUrl': ''}

次にホテル検索を必要としない関係のない質問が来た場合。

user_query = "明日の天気は？"

{'text': '申し訳ありませんが、私は天気情報を取得する機能を持っていません。天気予報を確認するためには、天気予報のウェブサイトやアプリをご利用ください。', 'imageUrl': ''}

上は一応想定通りにはなったけど、繰り返し実行しているとこのフォーマットにならない場合も当然ある。よって、受け取った結果のチェックは必須かなという気はするし、いっそ自前でパースしたほうが安全かもという気はしている。

kun432

参考

npaka神

https://note.com/npaka/n/n917463f55b8a

AzureのFunction callingサンプル

出力フォーマットの固定はこちらも

RAGで使う場合のメリット。roleを明確に分けれるのはとても良いと思う。

また、embeddingで独自データを利用する手法では、これまでIn-Context Learningと呼ばれるように検索したテキストを最終的にはプロンプトに詰め込んでLLMに渡していました。今回のアップデートで、role:userによる発言ではなく、role:functionによる発言(?)として、messageのリストに入れることができるようになりました。ユーザーの発言とシステム上も区別できるという点から、こちらの方が無理矢理感が薄くていいですね。

Agentっぽいってところはそうだと思う

kun432

LangChainのAgentでツールを選択する際のAgent typeにFunction callingを選択できる

あとChainでも使える

kun432

Parallel Function Calling等DevDayでの変更内容はこちら

このスクラップは2023/07/26にクローズされました