OpenAIの新機能「Function calling」を使ってReActエージェントを自作してみる。

本記事の対象と目指すところ

前回、function callingの概念やでできること・できないことをまとめました。
今回は実装面について触れていこうかと思います。そのため、本記事では基本的な部分の詳細や概念には触れません。

本記事では、以下を実装した際の概要やポイントをまとめます。

• function callingの基本的な使い方
• ユーザの入力に応じて複数の関数を実行していく単純なエージェント機能を実装
• エージェントにReasong and Actingを追加 (思考→行動→観察を繰り返し、ユーザの目的を達成する)

達成したいこと

「 複数の関数を使いながらユーザの入力に対して回答を出力する 」
「 関数の実行結果を基に次のアクションを判断する 」

今回用意する関数は2つです。

• search_file()
  • 社内情報を検索することを想定した関数
• search_web()
  • Web検索により外部情報を取得することを想定した関数

これらの関数をユーザの入力に応じていい感じに使ってもらうことを目指します。その次のステップとして、各関数の実行結果を基に次のアクションを判断するような機構を実装してみます。
想定する流れは以下の図の通りです。

※function callingの実際の流れとは異なる部分があります。

各関数の中身にはこだわっておらず、文字列をreturnするだけの簡単な関数です。以下のような値を返すように設定しています。

関数名	returnする値
search_file()	ChatGPTは社内利用禁止です。
search_web()	ChatGPTの利用を禁止している会社があります。社内独自のChatGPTのようなシステムを構築する動きが増えています。

そのため、「 ChatGPTって社内で使っていいの？他社動向も教えて 」と聞いた場合は、「 社内では利用禁止。他社では禁止しているところもある中で、社内独自のシステムを構築している動きがある 」といった回答が得られれば期待通りの動作ということになります。

【基礎編】function callingの使い方

まずは基本的な使い方ということで、search_file()を必要なタイミングで呼出せるようにします。

function callingの実装例

function callingの実装

import os
import json
import openai
from dotenv import load_dotenv

# APIキーを環境変数から取得する
load_dotenv()
openai.api_key = os.getenv("OPENAI_API_KEY")

# function callingを実行可能な関数
def function_calling(functions_metadata:list, messages:list, user_input:str) -> dict:
    if user_input is not None:
        # ユーザの入力をmessagesに追加する
        messages.append({"role": "user", "content": user_input})

    # messagesを基に推論実行し、返答を取得する
    resposne = openai.ChatCompletion.create(
        model = "gpt-3.5-turbo-0613",
        messages = messages,
        functions=functions_metadata,
        function_call="auto"
    )

    return resposne

# 社内ファイル検索用の関数
def search_file(query:str, source:str) -> str:
    result = {
        "source": source,
        "query": query,
        "result": "ChatGPTは社内利用禁止です。",
    }
    
    return json.dumps(result)

# 社内ファイル検索用の関数のメタデータ
search_file_metadata = {
    "name": "search_file",
    "description": "社内情報を取得するための関数。社内ファイルやデータベースを検索可能。",
    "parameters": {
        "type": "object",
        "properties": {
            "query": {
                "type": "string",
                "description": "ファイル検索のクエリ"
            },
            "source":{
                "type": "string",
                "enum": ["Fileserver, Database"]
            }
        },
        "required": ["query", "source"]
    }
}

# 関数のメタデータをリストに格納
functions_metadata = [
    search_file_metadata
]

# 関数名をキーにして、関数を呼び出せるようにしておく
functions_callable = {
    search_file_metadata["name"]: search_file
}

# システムのプロンプト
SYSTEM_PROMPT = """
あなたはユーザを助けるアシスタントです。
ユーザの入力に正しく回答を出力するために、ステップバイステップで慎重に考えることができます。
"""

# 文脈を保持するリスト
messages = [
    {"role":"system", "content": SYSTEM_PROMPT},
]

# ユーザの入力=ゴール
user_input = """
必要に応じてfunction_callを使用する。
ChatGPTって社内で使っていい？
"""

# 推論実行
res = function_calling(functions_metadata, messages, user_input)

# function_callがある場合は、関数を呼び出す
if res["choices"][0]["message"].get("function_call"):
    # functionを実行するために必要な情報を取得する
    func_name = res["choices"][0]["message"]["function_call"]["name"]
    func_parameters = json.loads(res["choices"][0]["message"]["function_call"]["arguments"])

    # functionを実行する
    func_result = functions_callable[func_name](**func_parameters)
    
    # 実行結果をmessageに追加する
    messages.append(
        {
            "role": "function", 
            "name": func_name, 
            "content": func_result
        }
    )

    # functionの実行結果を踏まえて再度推論実行する
    res = function_calling(functions_metadata, messages, None)

print(res["choices"][0]["message"]["content"])

実装の流れは以下の通りです。個人的なイメージとしては前半・後半に分かれています。

• 前半： 初回のOpenAI APIを呼び出し

  • ①function callingで使用したい関数を定義
  • ②関数のメタデータを定義 (いつ使うか、引数はなにかなど)
  • ③APIを叩く時に関数のメタデータを添える

• 後半: 初回OpenAI APIを呼び出しの結果を基にした処理

  • ④レスポンスに「 function_call 」が含まれるかチェック
  • ⑤含まれる場合は関数名と引数を抽出
  • ⑥抽出した関数名と引数を用いて関数を実行
  • ⑦実行結果を含めて、再度OpenAI APIを呼び出す

ユーザの入力に基づいて「どの関数をどのような引数で実行するか？」をAIさんに考えてもらう。
考えてもらった内容を踏まえて、ユーザ側で関数を実行する。といったイメージですね。

【応用編①】エージェント機能

続いて、search_file() , search_web()を両方利用可能な状態にします。
さきほどと異なる点は、複数の関数を使用する点です。

今回問いかける内容は、「 ChatGPTって社内で使っていいの？他社動向も教えて 」です。
期待する動作は以下の通りです。

• search_fileを実行して社内情報を取得
• search_webを実行して外部情報取得
• 取得した内容を踏まえてユーザの入力に対する回答を出力

現在の状態を認識させることで実装可能

今回の実装では以下のような方式としました。

初めに、ユーザの入力に基づいて実行すべき関数を考えさせます。

次に、ユーザ側で関数を実行し、いままでのやり取りに関数の実行結果を追加します。その情報を基に、再度OpenAI APIでリクエストを投げつけます。
この処理をループすることで、一つずつ必要な関数を実行しながら最終的な出力を生成できるという流れです。

詳しい実装例は以降で記載していますが、注目ポイントは⑤です。
「すでにsearch_file()は実行済みなんだな」ということを認識させないと永遠に同じ関数を実行しようとします。
以下のようにmessagesに実行結果と状態を追加してあげることで期待通りの動作が得やすくなります。

func_name = "実行した関数名"
func_result = "実行結果"

messages.append(
    {
        "role": "function", 
        "name": func_name, 
        "content": f"実行結果:{func_result}, 状態: {func_name}関数実行済み"
    }
)

実装例

Tipsに記載していますが、search_file(), search_web()をクラスとして扱えるようにしています。
Semantic KernelやLangchainのマネですね。

search_file()とsearch_web()の定義

search_file()とsearch_web()の定義

class SearchFile:
    # 社内ファイル検索用の関数のメタデータ
    metadata = {
        "name": "search_file",
        "description": "社内情報を取得するための関数。社内ファイルやデータベースを検索可能。",
        "parameters": {
            "type": "object",
            "properties": {
                "query": {
                    "type": "string",
                    "description": "ファイル検索のクエリ"
                },
                "source":{
                    "type": "string",
                    "enum": ["Fileserver, Database"]
                }
            },
            "required": ["query", "source"]
        }
    }
    
    # 社内ファイル検索用の関数
    def run(self, query:str, source:str) -> dict:
        result = {
            "source": source,
            "query": query,
            "result": "ChatGPTは社内利用禁止です。",
        }
        
        return result
    

class SearchWeb:
    # Web検索用の関数のメタデータ
    metadata = {
        "name": "search_web",
        "description": "Web検索を用いて情報を取得するための関数。外部情報を取得可能。",
        "parameters": {
            "type": "object",
            "properties": {
            "query": {
                "type": "string",
                "description": "検索クエリ"
            }
            },
            "required": ["query"]
        }
    }

    # Web検索用の関数
    def run(self, query:str) -> dict:
        result = {
            "query": query,
            "result": "ChatGPTの利用を禁止している会社があります。社内独自のChatGPTのようなシステムを構築する動きが増えています。",
        }

        return result

function callingメイン処理

import os
import json
import openai
from dotenv import load_dotenv

# APIキーを環境変数から取得する
load_dotenv()
openai.api_key = os.getenv("OPENAI_API_KEY")


# システムのプロンプト
SYSTEM_PROMPT = """
あなたはユーザを助けるアシスタントです。
ユーザの入力に正しく回答を出力するために、ステップバイステップで慎重に考えることができます。
"""

class FunctionCalling:
    def __init__(self):
        # function_callingで使用する関数をインスタンス化
        search_file = SearchFile()
        search_web = SearchWeb() 

        # 文脈を保持するリストを初期化
        self.messages = [
            {"role":"system", "content": SYSTEM_PROMPT},
        ]

        # 関数のメタデータをリストに格納
        self.functions_metadata = [
            search_file.metadata,
            search_web.metadata
        ]

        # 関数を呼び出すための辞書を作成
        self.functions_callable = {
            search_file.metadata["name"]: search_file.run,
            search_web.metadata["name"]: search_web.run
        }


    # function callingを実行する関数
    def function_calling(self, functions_metadata:list, messages:list, user_input:str) -> dict:
        # ユーザの入力をmessagesに追加する
        if user_input is not None:
            messages.append({"role": "user", "content": user_input})

        # messagesを基に推論実行し、返答を取得する
        resposne = openai.ChatCompletion.create(
            model = "gpt-3.5-turbo-0613",
            messages = self.messages,
            functions=functions_metadata,
            function_call="auto"
        )

        return resposne

    # function callingをループさせる関数
    def function_calling_loop(self, response:dict, user_input:str) -> dict:
        while response["choices"][0]["message"].get("function_call"):
            # functionを実行するために必要な情報を取得する
            func_name = response["choices"][0]["message"]["function_call"]["name"]
            func_parameters = json.loads(response["choices"][0]["message"]["function_call"]["arguments"])

            # functionを実行する
            func_result = self.functions_callable[func_name](**func_parameters)

            # 実行された関数の詳細を表示
            print(f"実行された関数: {func_name}\n引数: {func_parameters}\n実行結果: {func_result['result']}\n")

            # 実行結果をmessageに追加する
            self.messages.append(
                {
                    "role": "function", 
                    "name": func_name, 
                    "content": f"実行結果:{func_result}, 状態: {func_name}関数実行済み"
                }
            )

            # 再度function_callingを実行する
            response = self.function_calling(self.functions_metadata, self.messages, user_input)

        return response
    

    def main(self, user_input: str):
        # 初回の推論実行
        response = self.function_calling(self.functions_metadata, self.messages, user_input)

        # OpenAI APIのレスポンスを処理する
        if response["choices"][0]["message"].get("function_call"):
            result = self.function_calling_loop(response, user_input)
        
        # 最終的な返答
        return result



# function callingをインスタンス化
function_calling = FunctionCalling()

# ユーザの入力=ゴール
user_input = """
必要に応じてfunction_callを使用する。
ChatGPTって社内で使っていい？他社の動向も含めておしえて
"""

# function callingを実行
result= function_calling.main(user_input)

# 最終的な回答を表示
print(result["choices"][0]["message"]["content"])

実行結果

実装例に記載のプログラムを実行したところ、以下のような回答が得られました。期待通りの動作ですね。
ただ、実行可能な関数が増えたりした場合どうなるかを検証していないため、引き続き色々試していきたいと思います。

### ユーザの入力 ###
user_input = """
必要に応じてfunction_callを使用する。
ChatGPTって社内で使っていい？他社の動向も含めておしえて
"""

### function callingにより実行された関数 ###
実行された関数: search_file
引数: {'query': 'ChatGPT', 'source': 'Fileserver'}
実行結果: ChatGPTは社内利用禁止です。

実行された関数: search_web
引数: {'query': 'ChatGPTの利用状況他社'}
実行結果: ChatGPTの利用を禁止している会社があります。社内独自のChatGPTのようなシステムを構築する動きが増えています。

### 最終的な出力 ###
ChatGPTは社内利用が禁止されています。他社の動向に関しては、一部の会社がChatGPTの利用を禁止しているという情報があります。また、社内独自のChatGPTのようなシステムを構築する動きも増えています。ご参考までにお伝えいたします。

【応用編②】ReAct

続いて、ReAct的な機能を追加します。
今回の例ではファイル検索や外部情報を検索できるエージェントを想定しているため、以下のような機能を追加したいと思います。

• いい感じの情報が得られない場合、自動でクエリを変えて再検索
• あまりに見つからない場合は別の関数を使用

実装例

ソースコード全文

ReActエージェント

### function callingで使用する関数群の定義 ###
class SearchFile:
    # 社内ファイル検索用の関数のメタデータ
    metadata = {
        "name": "search_file",
        "description": "社内情報を取得するための関数。社内ファイルやデータベースを検索可能。",
        "parameters": {
            "type": "object",
            "properties": {
                "query": {
                    "type": "string",
                    "description": "ファイル検索のクエリ"
                },
                "source":{
                    "type": "string",
                    "enum": ["Fileserver, Database"]
                }
            },
            "required": ["query"]
        }
    }
    
    # 社内ファイル検索用の関数
    def run(self, query:str, source:str="Fileserver") -> dict:
        # わざと情報が見つからないようにする。
        query += "__aaaaa"
        source += "__aaaaa"

        # queryが条件を満たしている場合のみ検索結果を返す
        if query == "ChatGPT" and source == "Fileserver":
            result = {
                "parameters": {
                    "source": source,
                    "query": query
                },
                "result": "ChatGPTは社内利用禁止です。",
            }
        else:
            result = {
                "parameters": {
                    "source": source,
                    "query": query,
                },
                "result": "情報が見つかりませんでした。",
                "state": "error"
            }
        
        return result
    

class SearchWeb:
    # Web検索用の関数のメタデータ
    metadata = {
        "name": "search_web",
        "description": "Web検索を用いて情報を取得するための関数。外部情報を取得可能。",
        "parameters": {
            "type": "object",
            "properties": {
                "query": {
                    "type": "string",
                    "description": "検索クエリ"
                }
            },
            "required": ["query"]
        }
    }

    # Web検索用の関数
    def run(self, query:str) -> dict:
        result = {
            "parameters": {
                "query": query
            },
            "result": "ChatGPTの利用を禁止している会社があります。社内独自のChatGPTのようなシステムを構築する動きが増えています。",
        }

        return result


### メイン処理 ###
import os
import json
import openai
from dotenv import load_dotenv
import time

# APIキーを環境変数から取得する
load_dotenv()
openai.api_key = os.getenv("OPENAI_API_KEY")


# システムのプロンプト
SYSTEM_PROMPT = """
あなたはユーザを助けるアシスタントです。
ユーザの入力に正しく回答を出力するために、ステップバイステップで慎重に考えることができます。
"""

# ReAct用のプロンプト
REACT_PROMPT = """
# 指示
「実行した関数の詳細」に基づいて次のアクションを判断し、実行まで行って。
アクションの選択肢は以下の通りです。

# アクション
- 優先度1: parametersを変えて試す
- 優先度10: 何回も失敗する場合は別のfunctionで試す

# 実行した関数の詳細
name: 
{}

description: 
{}

# 条件
必要に応じてfunction_callを使用して。
"""

class FunctionCalling:
    def __init__(self):
        # function_callingで使用する関数をインスタンス化
        search_file = SearchFile()
        search_web = SearchWeb() 

        # 文脈を保持するリストを初期化
        self.messages = [
            {"role":"system", "content": SYSTEM_PROMPT},
        ]

        # 関数のメタデータをリストに格納
        self.functions_metadata = [
            search_file.metadata,
            search_web.metadata
        ]

        # 関数を呼び出すための辞書を作成
        self.functions_callable = {
            search_file.metadata["name"]: search_file.run,
            search_web.metadata["name"]: search_web.run
        }


    # function callingを実行する関数
    def function_calling(self, functions_metadata:list, messages:list, user_input:str) -> dict:
        # ユーザの入力をmessagesに追加する
        if user_input is not None:
            messages.append({"role": "user", "content": user_input})

        # messagesを基に推論実行し、返答を取得する
        resposne = openai.ChatCompletion.create(
            model = "gpt-3.5-turbo-0613",
            messages = self.messages,
            functions=functions_metadata,
            function_call="auto"
        )

        return resposne
    
    # Reasoning and Actiongを行う関数
    def reasoning_and_actioning(self, messages:list, function_name:str, function_result:dict) -> dict:
        # 過去の文脈と関数の実行結果から次に何をする必要があるか考える
        react_prompt = REACT_PROMPT.format(function_name, function_result)
        response = self.function_calling(self.functions_metadata, messages, react_prompt)

        return response

    # function callingをループさせる関数
    def function_calling_loop(self, response:dict, user_input:str) -> dict:
        while response["choices"][0]["message"].get("function_call"):
            # functionを実行するために必要な情報を取得する
            func_name = response["choices"][0]["message"]["function_call"]["name"]
            func_parameters = json.loads(response["choices"][0]["message"]["function_call"]["arguments"])

            # functionを実行する
            func_result = self.functions_callable[func_name](**func_parameters)

            # 実行された関数の詳細を表示
            print(f"実行された関数: {func_name}\n引数: {func_parameters}\n実行結果: {func_result['result']}\n")

            # 実行結果をmessageに追加する
            self.messages.append(
                {
                    "role": "function", 
                    "name": func_name, 
                    "content": f"実行結果:{func_result}"
                }
            )

            # 関数の実行結果の状態が十分かどうかを確認する
            response = self.reasoning_and_actioning(self.messages, func_name, func_result)
            if response["choices"][0]["message"].get("function_call"):
                time.sleep(15)
                # function_callがある場合はループを続ける
                continue
            else:
                break
            
        return response
    

    def main(self, user_input: str):
        # 初回の推論実行
        response = self.function_calling(self.functions_metadata, self.messages, user_input)

        # OpenAI APIのレスポンスを処理する
        if response["choices"][0]["message"].get("function_call"):
            result = self.function_calling_loop(response, user_input)
        
        # 最終的な返答
        return result



# function callingをインスタンス化
function_calling = FunctionCalling()

# ユーザの入力=ゴール
user_input = """
必要に応じてfunction_callを使用する。
ChatGPTって社内で使っていい？他社の動向も含めておしえて
"""

# function callingを実行
result= function_calling.main(user_input)

# 最終的な回答を表示
print(result["choices"][0]["message"]["content"])

ReAct追加前の動作イメージ

先ほどの「現在の状態を更新することで実装可能」で示した図を使ってイメージを書いてみます。
ReAct追加前では、たとえファイルの検索処理を行い何も情報が見つからなくても、「 実行済みだから次行こう 」と判断されてしまいます。見つからない場合はクエリを変えるなりして、もう少し粘ってほしいですよね。

ReAct追加後の動作イメージ

ReActっぽく、以下のような思考回路を設けます。
ファイル(情報)が見つからない場合は、単純に次にいくのではなく、一度立ち止まってなにをすればいいかを考えてもらうイメージです。

• Observation: 実行結果を観察
• Thought: search_file()の結果が十分でない場合、次に取るべきアクションを判断させる
• Act: アクションを実行する

実装のポイント

大きく変更したポイントは以下の通りです。

①関数実行時に期待通りの値が得られなかった場合、state:errorを追加してからreturnする

①変更後の社内情報を取得する関数

①変更後の社内情報を取得する関数

class SearchFile:
    # 社内ファイル検索用の関数のメタデータ
    metadata = {...}
    
    # 社内ファイル検索用の関数
    def run(self, query:str, source:str="Fileserver") -> dict:
        # わざと情報が見つからないようにする。
        query += "__aaaaa"
        source += "__aaaaa"

        # queryが条件を満たしている場合のみ検索結果を返す
        if query == "ChatGPT" and source == "Fileserver":
            result = {
                "parameters": {
                    "source": source,
                    "query": query
                },
                "result": "ChatGPTは社内利用禁止です。",
            }
        else:
            result = {
                "parameters": {
                    "source": source,
                    "query": query,
                },
                "result": "情報が見つかりませんでした。",
                "state": "error"  #「エラーですよ。」という状態を追加
            }
        
        return result

②ReAct用のプロンプトを挟む

プロンプト

②ReAct用のプロンプト

REACT_PROMPT = """
# 指示
「実行した関数の詳細」に基づいて次のアクションを判断し、実行まで行って。
アクションの選択肢は以下の通りです。

# アクション
- 優先度1: 別のparametersで試す
- 優先度10: 何回も失敗する場合は別のfunctionで試す

# 実行した関数の詳細
name: 
{}

description: 
{}

# 条件
必要に応じてfunction_callを使用して。
"""

今回は、アクションに優先度を付けています。
理由としては、「まずはクエリを変えて試してみて、それでもだめなら別の関数を実行してほしかったため」です。
社内情報×外部情報を答えさせるタスクなのに、社内情報検索が1回ダメだったから外部情報を検索しにいこうとなってしまうのは避けたいですよね。

具体的な実装としては、過去の文脈を踏まえて上記のプロンプトを実行するプロセスを挟んでいるだけです。詳細は実装例をご参照ください。

# Reasoning and Actiongを行う関数
def reasoning_and_actioning(self, messages:list, function_name:str, function_result:dict) -> dict:
    # 過去の文脈と関数の実行結果から次に何をする必要があるか考える
    react_prompt = REACT_PROMPT.format(function_name, function_result)
    response = self.function_calling(self.functions_metadata, messages, react_prompt)

    return response

実行結果

今回、search_file()はどのような引数で実行されても「情報が見つかりませんでした。」と返すようにしています。毎回実行が失敗するような意地悪な関数を用意しておいて挙動を確認するイメージです。
また、search_file()のパラメータは以下の2つです。

• query: 検索クエリ
• source: ファイルサーバから取得するか、データベースから取得するか

queryを変えながら試行錯誤してくれるのを期待していましたが、sourceを変えながら試していますね
関数のメタデータを定義する際に、パラメータにも優先度を付ければ改善できそうです。
ただ、何回か失敗した後、諦めて別の関数を実行しているのでそこは期待通りです

### function calling × ReActエージェントにより実行された関数 ###
実行された関数: search_file
引数: {'query': 'ChatGPTの社内利用可否'}
実行結果: 情報が見つかりませんでした。

実行された関数: search_file
引数: {'query': 'ChatGPTの社内利用可否'}
実行結果: 情報が見つかりませんでした。

実行された関数: search_file
引数: {'query': 'ChatGPTの社内利用可否', 'source': 'Fileserver'}
実行結果: 情報が見つかりませんでした。

実行された関数: search_file
引数: {'query': 'ChatGPTの社内利用可否', 'source': 'Fileserver, Database'}
実行結果: 情報が見つかりませんでした。

実行された関数: search_file
引数: {'query': 'ChatGPTの社内利用可否', 'source': 'Fileserver, Database'}
実行結果: 情報が見つかりませんでした。

実行された関数: search_web
引数: {'query': 'ChatGPTの社内利用可否'}
実行結果: ChatGPTの利用を禁止している会社があります。社内独自のChatGPTのようなシステムを構築する動きが増えています。


### 最終的な出力 ###
結果を見ると、`search_web`関数による検索の結果、ChatGPTの利用を禁止している会社があることがわかりました。また、社内独自のChatGPTのようなシステムを構築する動きも増えているようです。
次のアクションとして、この結果をユーザに伝えることができます。それでは、結果を回答として出力しましょう。
回答:
ChatGPTの利用を禁止している会社が存在します。また、社内独自のChatGPTのようなシステムを構築する動きも増えているようです。

その他・Tips

実行したい関数群をclassとして管理すると便利かも？

Semantic KernelやLangchainでは、スキル/ツールなどをclassで管理できます。
その方式を採用し、今回はfunction callingで呼出したい関数群をclassとして管理してみましたが、意外と取り回し良さそう？
思い付きで実装してみただけなので改善の余地は大いにありそうです。

Funciton callingで使用したい関数の定義 (class)

使用したい関数をクラス化

class SearchWeb:
    # Web検索用の関数のメタデータ
    metadata = {
        "name": "search_web",
        "description": "Web検索を用いて情報を取得するための関数。外部情報を取得可能。",
        "parameters": {
            "type": "object",
            "properties": {
            "query": {
                "type": "string",
                "description": "検索クエリ"
            }
            },
            "required": ["query"]
        }
    }

    # Web検索用の関数
    def run(self, query:str) -> dict:
        result = {
            "query": query,
            "result": "ChatGPTの利用を禁止している会社があります。社内独自のChatGPTのようなシステムを構築する動きが増えています。",
        }

        return result

Function calling側でクラスを利用

色々省略しています。

class FunctionCalling:
    def __init__(self):
        # function_callingで使用する関数をインスタンス化
        search_web = SearchWeb() 

        # 文脈を保持するリストを初期化
        self.messages = [
            {"role":"system", "content": SYSTEM_PROMPT},
        ]

        # 関数のメタデータをリストに格納
        self.functions_metadata = [
            search_file.metadata,
            search_web.metadata
        ]

        # 関数を呼び出すための辞書を作成
        self.functions_callable = {
            search_file.metadata["name"]: search_file.run,
            search_web.metadata["name"]: search_web.run
        }

    def function_calling():
        # messagesを基に推論実行し、返答を取得する
        resposne = openai.ChatCompletion.create(
            model = "gpt-3.5-turbo-0613",
            messages = self.messages,
            functions=functions_metadata,
            function_call="auto"
        )

        # functionを実行するために必要な情報を取得する
        func_name = response["choices"][0]["message"]["function_call"]["name"]
        func_parameters = json.loads(response["choices"][0]["message"]["function_call"]["arguments"])

        # functionを実行する
        func_result = self.functions_callable[func_name](**func_parameters)

functioin callingを優先させたいとき

ユーザの入力に「必要に応じてfunction_callを使用する。」という文言を追加すると優先してくれるようになりました。
ただ、attentionが強く働きすぎると永遠にループし始めます。この辺りは色々テクニックがありそうですね。（要調査）

# ユーザの入力=ゴール
user_input = """
必要に応じてfunction_callを使用する。
ChatGPTって社内で使っていい？他社の動向も含めておしえて
"""

# function callingを実行
result= function_calling.main(user_input)

レート制限ってどうやって回避するの？（未解決）

今回の実装はOpenAI APIへのリクエストを投げまくるループ処理となっています。
そのため、以下のようなエラーが発生するため注意が必要です。今回はtime.sleep()でその場しのぎをしています。

回避策は要調査。

Rate limit reached for default-gpt-3.5-turbo in organization org-asgsgrg8 on requests per min. Limit: 3 / min.

まとめ

function callingめちゃくちゃ便利な気がしてきた。
以前もなにかに書きましたが、Semantic KernelやLangchainでエージェント的な機能を実装しようと思うとプロンプトチューニングが必須な気がしているんですよね、、。
期待通りのJSON形式を出力してくれなかったり、JSON形式としては正しいけども引数の型が正しくなかったり、、。それをチューニングするのが結構辛いです。

Function callingでは、プロンプトチューニングせずとも、関数の処理内容とメタデータ（いつ使うか・引数など）を定義してリクエスト投げるだけでよく、精度もめちゃくちゃいいと思ってます。
そのため、「 ユーザの入力に応じて実行すべき関数を実行する 」という仕組みを実現するのが楽になったのかなあと。

また、「 Aをした後にBしてほしい 」といった入力に対して、複数の関数を実行する方法も今回実装できたため応用の幅が広いなと思っています。めちゃくちゃ主観ですが、動的に関数を実行する系は全部Function callingベースになる気がしています。