shigel/langchain-function-calling

1# 必要なモジュールをインポート2import gradio as gr3import os4import sys5import json6import csv7import dotenv8import openai9from langchain.chat_models import ChatOpenAI10from langchain.agents import initialize_agent, Tool11from langchain.schema import (12    AIMessage,13    AgentAction,14    HumanMessage,15    FunctionMessage16)17from langchain.chat_models import ChatOpenAI18from langchain.agents import AgentType19 20# .envファイルから環境変数をロード21dotenv.load_dotenv(".env")22 23# OpenAIキーをosモジュールで取得24openai.api_key = os.environ.get("OPENAI_API_KEY")25 26# 民間伝承を取得する関数27def fetch_folklore(location):28    folklore_lookup = {}29    # CSVファイルからデータを読み取り、地点をキー、伝承を値とする辞書を作成30    with open('folklore.csv', 'r') as f:31        reader = csv.DictReader(f)32        folklore_lookup = {row['location']: row['folklore'] for row in reader}33        type_lookup = {row['type']: row['folklore'] for row in reader}34 35    # 指定された地点の伝承などを返す。存在しない場合は不明を返す。36    folklore = folklore_lookup.get((location), f"その地域の伝承は不明です。")37    type = type_lookup.get((location), f"その地域の伝承は不明です。")38    print("type:", type)39    return folklore40 41def serialize_agent_action(obj):42    if isinstance(obj, AgentAction):43        return { "tool": obj.tool, "tool_input": obj.tool_input, "log": obj.log}44    if isinstance(obj, _FunctionsAgentAction):45        return { "tool": obj.tool, "tool_input": obj.tool_input, "log": obj.log, "message_log": obj.message_log}46    if isinstance(obj, AIMessage):47        return { "content": obj.content, "additional_kwargs": obj.additional_kwargs, "example": obj.example}48    raise TypeError(f"Type {type(obj)} not serializable")49 50# LangChainエージェントからレスポンスを取得する関数51def get_response_from_lang_chain_agent(query_text):52    # ChatOpenAIを使用して言語モデルを初期化53    language_model = ChatOpenAI(model_name='gpt-3.5-turbo-0613')54    tools = [55        # 民間伝承を取得するToolを作成56        Tool(57            name="Folklore",58            func=fetch_folklore,59            description="伝承を知りたい施設や地名を入力。例: 箱根",60        )61    ]62    # エージェントを初期化してから応答を取得63    agent = initialize_agent(tools, language_model, agent="zero-shot-react-description",64                             verbose=True, return_intermediate_steps=True)65    response = agent({"input": query_text})66    print(type(response))67    response = json.dumps(response, default=serialize_agent_action, indent=2, ensure_ascii=False)68 69    return response70 71# Function Callingからレスポンスを取得する関数72def get_response_from_function_calling(query_text):73    function_definitions = [74        # 関数の定義を作成75        {76            "name": "fetch_folklore",77            "description": "伝承を調べる",78            "parameters": {79                "type": "object",80                "properties": {81                    "location": {82                        "description": "伝承を知りたい施設や地名。例: 箱根",83                    },84                },85                "required": ["location"],86            },87        }88    ]89    messages = [HumanMessage(content=query_text)]90    language_model = ChatOpenAI(model_name='gpt-4')91    # 言語モデルを使ってメッセージを予測92    message = language_model.predict_messages(93        messages, functions=function_definitions)94 95    if message.additional_kwargs:96        # 関数の名前と引数を取得97        function_name = message.additional_kwargs["function_call"]["name"]98        arguments = message.additional_kwargs["function_call"]["arguments"]99 100        # JSON 文字列を辞書に変換101        arguments = json.loads(arguments)102        location=arguments.get("location")103        # type=arguments.get("type")104 105        # 関数を実行してレスポンスを取得106        function_response = fetch_folklore(location=location)107        # 関数メッセージを作成108        function_message = FunctionMessage(109            name=function_name, content=function_response)110        # 関数のレスポンスをメッセージに追加して予測111        messages.append(function_message)112        second_response = language_model.predict_messages(113            messages=messages, functions=function_definitions)114        content = second_response.content115    else:116        content = message.content117    return content118 119# Function Call Agentからレスポンスを取得する関数120def get_response_from_function_calling_agent(query_text):121    language_model = ChatOpenAI(model_name='gpt-3.5-turbo-0613')122    tools = [123        # 民間伝承情報を提供するツールの追加124        Tool(125            name="Folklore",126            func=fetch_folklore,127            description="伝承を知りたい施設や地名を入力。例: 箱根"128        )129    ]130    # エージェントの初期化とレスポンスの取得131    agent = initialize_agent(tools, language_model, agent=AgentType.OPENAI_FUNCTIONS,132                             verbose=True, return_intermediate_steps=True)133    response = agent({"input": query_text})134    response = json.dumps(response, default=serialize_agent_action, indent=2, ensure_ascii=False)135    return response136 137# メインの実行部分138 139 140def main(query_text, function_name="all"):141 142    response1 = ""143    response2 = ""144    response3 = ""145    146    if function_name == "all" or function_name == "langchain":147        # LangChainエージェントからのレスポンス148        response1 = get_response_from_lang_chain_agent(query_text)149        print(response1)150 151    if function_name == "all" or function_name == "functioncalling":152        # Function Callingからのレスポンス153        response2 = get_response_from_function_calling(query_text)154        print(response2)155 156    if function_name == "all" or function_name == "functioncallingagent":157        # Function Callingエージェントからのレスポンス158        response3 = get_response_from_function_calling_agent(query_text)159        print(response3)160 161    return response1, response2, response3162 163 164# スクリプトが直接実行された場合にmain()を実行165if __name__ == "__main__":166    if len(sys.argv) == 2:167        query_text = sys.argv[1]168        main(query_text=query_text)169    elif len(sys.argv) > 2:170        query_text = sys.argv[1]171        function_name = sys.argv[2]172        main(query_text=query_text, function_name=function_name)173    else:174        import time175 176        # インプット例をクリックした時のコールバック関数177        def click_example(example):178            # クリックされたインプット例をテキストボックスに自動入力179            inputs.value = example180            time.sleep(0.1)  # テキストボックスに文字が表示されるまで待機181            # 自動入力後に実行ボタンをクリックして結果を表示182            execute_button.click()183 184        # gr.Interface()を使ってユーザーインターフェースを作成します185        # gr.Text()はテキスト入力ボックスを作成し、186        # gr.Textbox()は出力テキストを表示するためのテキストボックスを作成します。187        iface = gr.Interface(188            fn=main,189            examples=[190                ["葛飾区の伝承を教えてください。"],191                ["千代田区にはどんな伝承がありますか？"],192                ["江戸川区で有名な伝承？"],193            ],194            inputs=gr.Textbox(195                lines=5, placeholder="質問を入力してください"),196            outputs=[197                gr.Textbox(label="LangChain Agentのレスポンス"),198                gr.Textbox(label="Function Callingのレスポンス"),199                gr.Textbox(label="Function Calling Agentのレスポンス")200            ],201            title="日本各地の伝承AI (東京23区版)",202            description="最新のGPTモデルを使用し、LangChain, Function Calling, Function Calling + LangChain Agentの対話モデルのAIから回答を取得するシステムです。以下のインプット例をクリックすると入力欄に自動入力されます。",203            example_columns=3,204            example_callback=click_example205        )206 207        # インターフェースを起動します208        iface.launch()209