Zenn
🐥

Azure上のAssistants APIでBing Searchを試す

2024/03/21に公開
AI
Azure
Microsoft
OpenAI
ChatGPT
tech

Assistants function calling with Bing Search

以下のサンプルを参考にBing Search APIs と function calling を使用して応答にBing Searchの情報を含めるアシスタントを作成します。この方法を使用することで回答に最新の情報を含めることできます。

https://github.com/Azure-Samples/azureai-samples/tree/main/scenarios/Assistants/function_calling

以下のサンプルを進める前にBing Searchのリソースを作成してください。
Bing Search resource

はじめに

必要パッケージ

%pip install requests openai

Parameters

パラメーター設定を行います。Bing Searchのリソースのキーとエンドポイントを設定します。
キーはAzure Portalから取得できます。

azure_endpoint = "https://<YOUR_RESOURCE_NAME>.openai.azure.com"
api_version = "2024-02-15-preview"
aoai_api_key = "<AOAI_RESOURCE_API_KEY>"
deployment_name = "<DEPLOYMENT_NAME>"
bing_search_subscription_key = "<BING_SEARCH_SUBSCRIPTION_KEY>"
bing_search_url = "https://api.bing.microsoft.com/v7.0/search"

Bing Search APIを呼び出す関数を定義

LLM での Bing 検索 API の使用要件等は以下から確認してください。
Bing Search APIs, with your LLM.

import json
import requests
import time

from openai import AzureOpenAI
from pathlib import Path
from typing import Optional

以下のPythonコードは、指定された検索クエリに対してBing検索を実行し、その結果をリストとして返す関数を定義しています。

  • 関数の定義:

    • search関数を定義します。この関数は、検索クエリを引数として受け取り、検索結果のリストを返します。

  • Bing Search APIのリクエストヘッダー:

    • Bing Search APIのリクエストヘッダーを定義します。このヘッダーには、Bing Search APIのサブスクリプションキーが含まれます。

  • リクエストパラメータ:

    • リクエストパラメータを定義します。これには、検索クエリとテキスト装飾のオプションが含まれます。

  • Bing Search APIへのGETリクエスト:

    • requests.get関数を使用してBing Search APIにGETリクエストを送信します。このリクエストには、Bing Search APIのURL、リクエストヘッダー、およびリクエストパラメータが含まれます。

  • リクエストの成功確認:

    • response.raise_for_status()を呼び出して、リクエストが成功したかどうかを確認します。これは、リクエストが失敗した場合に例外を発生させます。

  • レスポンスボディの解析:

    • response.json()を呼び出して、レスポンスボディをJSON形式で解析し、その結果をsearch_results変数に格納します。

  • 検索結果の処理:

    • 空のリストoutputを作成します。このリストは、検索結果を格納するためのものです。
    • search_results["webPages"]["value"]をループして、各検索結果を処理します。各検索結果について、そのタイトル（result["name"]）、リンク（result["url"]）、およびスニペット（result["snippet"]）を取得し、それらを辞書としてoutputリストに追加します。

  • 結果の返却:

    • json.dumps(output)を呼び出して、outputリストをJSON形式の文字列に変換し、それを返します。

この関数は、指定された検索クエリに対してBing検索を実行し、その結果をJSON形式の文字列として返すためのものです。

def search(query: str) -> list:
    """
    Perform a bing search against the given query

    @param query: Search query
    @return: List of search results

    """
    headers = {"Ocp-Apim-Subscription-Key": bing_search_subscription_key}
    params = {"q": query, "textDecorations": False}
    response = requests.get(bing_search_url, headers=headers, params=params)
    response.raise_for_status()
    search_results = response.json()

    output = []

    for result in search_results["webPages"]["value"]:
        output.append({"title": result["name"], "link": result["url"], "snippet": result["snippet"]})

    return json.dumps(output)
  • 試しにBing APIを使用してみます。
search("2024年のオリンピックはどこで開催されますか？")

以下のような出力がされるはずです。
'[{"title": "Paris 2024 Olympics - Latest News, Schedules & Results", "link": "・・・・

Assistantの作成

アシスタントに必要ないくつかの関数を事前に定義します。これらの関数はすべて最後のセルで使用します。
Web検索アシスタントを定義し、その機能に関する指示を与え、質問してみます。

def poll_run_till_completion(
    client: AzureOpenAI,
    thread_id: str,
    run_id: str,
    available_functions: dict,
    verbose: bool,
    max_steps: int = 10,
    wait: int = 3,
) -> None:
    """
    Poll a run until it is completed or failed or exceeds a certain number of iterations (MAX_STEPS)
    with a preset wait in between polls

    @param client: OpenAI client
    @param thread_id: Thread ID
    @param run_id: Run ID
    @param assistant_id: Assistant ID
    @param verbose: Print verbose output
    @param max_steps: Maximum number of steps to poll
    @param wait: Wait time in seconds between polls

    """

    if (client is None and thread_id is None) or run_id is None:
        print("Client, Thread ID and Run ID are required.")
        return
    try:
        cnt = 0
        while cnt < max_steps:
            run = client.beta.threads.runs.retrieve(thread_id=thread_id, run_id=run_id)
            if verbose:
                print("Poll {}: {}".format(cnt, run.status))
            cnt += 1
            if run.status == "requires_action":
                tool_responses = []
                if (
                    run.required_action.type == "submit_tool_outputs"
                    and run.required_action.submit_tool_outputs.tool_calls is not None
                ):
                    tool_calls = run.required_action.submit_tool_outputs.tool_calls

                    for call in tool_calls:
                        if call.type == "function":
                            if call.function.name not in available_functions:
                                raise Exception("Function requested by the model does not exist")
                            function_to_call = available_functions[call.function.name]
                            tool_response = function_to_call(**json.loads(call.function.arguments))
                            tool_responses.append({"tool_call_id": call.id, "output": tool_response})

                run = client.beta.threads.runs.submit_tool_outputs(
                    thread_id=thread_id, run_id=run.id, tool_outputs=tool_responses
                )
            if run.status == "failed":
                print("Run failed.")
                break
            if run.status == "completed":
                break
            time.sleep(wait)

    except Exception as e:
        print(e)

poll_run_till_completion関数は、指定されたOpenAIクライアント、スレッドID、実行IDを使って、ある実行が完了、失敗する、または指定された最大ステップ数(MAX_STEPS)を超えるまで状態を定期的にチェックするためのポーリングメカニズムです。ポーリングの間隔はwaitパラメータで設定します。

  • もしクライアント、スレッドID、実行IDのいずれかが指定されていない場合、関数はエラーメッセージを表示して終了します。
  • max_stepsの回数だけ次の処理をループします：
    • 実行のステータスを確認します。verboseがTrueの場合、現在のポーリング回数とステータスを表示します。
    • ステータスがrequires_actionの場合、指定されたアクション（必要なツールの出力を送信するなど）を実行します。
    • ステータスがfailedの場合、エラーメッセージを表示してループを終了します。
    • ステータスがcompletedの場合、ループを終了します。
    • 上記のいずれにも該当しない場合、指定されたwait時間だけ待機します。
    • 何らかの例外が発生した場合、その例外を表示します。

この関数は、長時間実行される処理や非同期タスクの完了を効率的に管理するために使用されます。
Runのライフサイクルは以下が参考になります。

https://platform.openai.com/docs/assistants/how-it-works/runs-and-run-steps

def create_message(
    client: AzureOpenAI,
    thread_id: str,
    role: str = "",
    content: str = "",
    file_ids: Optional[list] = None,
    metadata: Optional[dict] = None,
    message_id: Optional[str] = None,
) -> any:
    """
    Create a message in a thread using the client.

    @param client: OpenAI client
    @param thread_id: Thread ID
    @param role: Message role (user or assistant)
    @param content: Message content
    @param file_ids: Message file IDs
    @param metadata: Message metadata
    @param message_id: Message ID
    @return: Message object

    """
    if metadata is None:
        metadata = {}
    if file_ids is None:
        file_ids = []

    if client is None:
        print("Client parameter is required.")
        return None

    if thread_id is None:
        print("Thread ID is required.")
        return None

    try:
        if message_id is not None:
            return client.beta.threads.messages.retrieve(thread_id=thread_id, message_id=message_id)

        if file_ids is not None and len(file_ids) > 0 and metadata is not None and len(metadata) > 0:
            return client.beta.threads.messages.create(
                thread_id=thread_id, role=role, content=content, file_ids=file_ids, metadata=metadata
            )

        if file_ids is not None and len(file_ids) > 0:
            return client.beta.threads.messages.create(
                thread_id=thread_id, role=role, content=content, file_ids=file_ids
            )

        if metadata is not None and len(metadata) > 0:
            return client.beta.threads.messages.create(
                thread_id=thread_id, role=role, content=content, metadata=metadata
            )

        return client.beta.threads.messages.create(thread_id=thread_id, role=role, content=content)

    except Exception as e:
        print(e)
        return None

Pythonコードは、AzureのOpenAIクライアントを使用してスレッド内にメッセージを作成する関数create_messageを定義しています。

  • 関数の定義:

    • create_message関数は、OpenAIクライアント、スレッドID、メッセージの役割（ユーザーまたはアシスタント）、メッセージの内容、メッセージのファイルID、メッセージのメタデータ、メッセージIDを引数として受け取ります。

  • 初期設定:

    • メタデータとファイルIDがNoneの場合、それぞれを空の辞書と空のリストに設定します。

  • 前提条件の確認:

    • クライアントとスレッドIDが指定されていることを確認します。これらが指定されていない場合、エラーメッセージを表示して関数を終了します。

  • メッセージの処理:

    • tryブロックを開始します。このブロック内で、メッセージIDが指定されている場合、client.beta.threads.messages.retrieveメソッドを使用してメッセージを取得します。
    • ファイルIDとメタデータが指定されている場合、client.beta.threads.messages.createメソッドを使用して新しいメッセージを作成します。このメソッドは、スレッドID、役割、内容、ファイルID、メタデータを引数として受け取ります。
    • ファイルIDのみが指定されている場合、メタデータのみが指定されている場合、どちらも指定されていない場合について、同様にclient.beta.threads.messages.createメソッドを使用して新しいメッセージを作成します。

  • 例外処理:

    • 例外が発生した場合、その例外を表示してNoneを返します。

この関数は、OpenAIのスレッド内に新しいメッセージを作成するためのものです。

def retrieve_and_print_messages(
    client: AzureOpenAI, thread_id: str, verbose: bool, out_dir: Optional[str] = None
) -> any:
    """
    Retrieve a list of messages in a thread and print it out with the query and response

    @param client: OpenAI client
    @param thread_id: Thread ID
    @param verbose: Print verbose output
    @param out_dir: Output directory to save images
    @return: Messages object

    """

    if client is None and thread_id is None:
        print("Client and Thread ID are required.")
        return None
    try:
        messages = client.beta.threads.messages.list(thread_id=thread_id)
        display_role = {"user": "User query", "assistant": "Assistant response"}

        prev_role = None

        if verbose:
            print("\n\nCONVERSATION:")
        for md in reversed(messages.data):
            if prev_role == "assistant" and md.role == "user" and verbose:
                print("------ \n")

            for mc in md.content:
                # Check if valid text field is present in the mc object
                if mc.type == "text":
                    txt_val = mc.text.value
                # Check if valid image field is present in the mc object
                elif mc.type == "image_file":
                    image_data = client.files.content(mc.image_file.file_id)
                    if out_dir is not None:
                        out_dir_path = Path(out_dir)
                        if out_dir_path.exists():
                            image_path = out_dir_path / (mc.image_file.file_id + ".png")
                            with image_path.open("wb") as f:
                                f.write(image_data.read())

                if verbose:
                    if prev_role == md.role:
                        print(txt_val)
                    else:
                        print("{}:\n{}".format(display_role[md.role], txt_val))
            prev_role = md.role
        return messages
    except Exception as e:
        print(e)
        return None

retrieve_and_print_messages関数は、指定されたスレッドIDに関連するメッセージのリストを取得し、それらを表示する役割を果たします。

  • 前提条件の確認:

    • clientまたはthread_idがNoneの場合、エラーメッセージを表示してNoneを返します。

  • スレッドのメッセージリストの取得:

    • スレッドのメッセージリストを取得します。

  • メッセージの表示設定:

    • メッセージの役割（ユーザーまたはアシスタント）に対応する表示名を設定します。
    • verboseがTrueの場合、"CONVERSATION:"と表示します。

  • メッセージの処理:

    • 取得したメッセージを逆順に処理します。
    • 前のメッセージがアシスタントからのもので、現在のメッセージがユーザーからのものである場合、verboseがTrueの場合は"------"と表示します。
    • メッセージの内容を処理します。内容がテキストの場合、そのテキストを取得します。内容が画像の場合、その画像のデータを取得し、指定された出力ディレクトリに保存します。
    • verboseがTrueの場合、メッセージの役割と内容（テキスト）を表示します。前のメッセージと現在のメッセージの役割が同じ場合は、内容だけを表示します。

  • 例外の処理:

    • 何らかの例外が発生した場合、その例外を表示してNoneを返します。

  • パラメータ:

    • client: OpenAIのクライアント
    • thread_id: スレッドID
    • verbose: 詳細表示の有無
    • output_dir: 画像の保存先ディレクトリ

この関数は、上記のパラメータを使用して、スレッドのメッセージを取得し、それらを表示します。

name = "BingSearchAssistant"
instructions = """ユーザーが質問に答えるのを助けるように設計されたアシスタントです。

あなたはBing Searchを使ってWEBに問い合わせることができます。質問が最新の情報を必要とする場合や、WEBデータの恩恵を受ける可能性がある場合は、いつでもBing Searchを呼び出す必要があります。回答は日本語で答えてください。
"""

message = {"role": "user", "content": "富士山の高さは何mですか？"}


tools = [
    {
        "type": "function",
        "function": {
            "name": "search_bing",
            "description": "WEBから最新の情報を得るためにBingを検索する。",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "The search query",
                    }
                },
                "required": ["query"],
            },
        },
    }
]

available_functions = {"search_bing": search}
verbose_output = True

client = AzureOpenAI(api_key=aoai_api_key, api_version=api_version, azure_endpoint=azure_endpoint)

assistant = client.beta.assistants.create(
    name=name, description="", instructions=instructions, tools=tools, model=deployment_name
)

thread = client.beta.threads.create()
create_message(client, thread.id, message["role"], message["content"])


run = client.beta.threads.runs.create(thread_id=thread.id, assistant_id=assistant.id, instructions=instructions)
poll_run_till_completion(
    client=client, thread_id=thread.id, run_id=run.id, available_functions=available_functions, verbose=verbose_output
)
messages = retrieve_and_print_messages(client=client, thread_id=thread.id, verbose=verbose_output)

出力結果
Poll 0: queued　
Poll 1: in_progress　
Poll 2: requires_action　
Poll 3: in_progress　
Poll 4: completed

CONVERSATION:
User query:
富士山の高さは何mですか？
Assistant response:
富士山の高さは3,776mです。

Bing APIを活用して回答を返してくれています。

コードの内容は、OpenAIのAzure APIを使用して、Bing Searchを活用した質問応答アシスタントを作成し、実行するものです。

  • アシスタントの定義:

    • アシスタントの名前と指示を定義します。このアシスタントは、最新の情報を必要とする質問やウェブデータが役立つ質問に対して、Bing Searchを呼び出すように設計されています。

  • ユーザーからのメッセージ:

    • ユーザーが「富士山の高さは何mですか？」という質問をしています。

  • 使用可能なツールの定義:

    • Bingを検索して最新の情報を取得する関数search_bingをツールとして定義しています。

  • AzureOpenAIクライアントの作成:

    • AzureOpenAIクライアントを作成し、そのクライアントを使用してアシスタントを作成します。アシスタントの作成には、定義した名前、指示、ツール、およびモデル名が必要です。

  • スレッドの作成とメッセージの追加:

    • アシスタントが作成されたら、スレッドを作成し、ユーザーからのメッセージをそのスレッドに追加します。

  • スレッドの実行:

    • アシスタントがユーザーからのメッセージに対してアクションを実行します。この例では、アシスタントはBingを検索して富士山の高さを調べます。

  • メッセージの取得と出力:

    • スレッドからメッセージを取得し、それらを出力します。これにより、アシスタントがユーザーの質問にどのように応答したかを確認することができます。
    • Pollからrunライフサイクルのどこを実行しているか確認します。

Azure Portal上からも作成したアシスタントが確認できます。

また、以下のように実行することで、assistant、messages、run等の詳細を確認できます。

import json

def show_json(obj):
    display(json.loads(obj.model_dump_json()))

show_json(assistant)

サンプル例:

show_json(messages)

show_json(run)

まとめ

  • Assistants APIを活用し、Bing Searchのデータを取り込んで、ユーザーの質問にリアルタイムで回答する方法について解説しました。
  • 本記事では、Bing Search APIと機能呼び出しを組み合わせたサンプルコードを用いて、最新情報を応答に含めるプロセスを紹介。
  • この技術を駆使することで、ユーザーに最新の情報を提供するアシスタントアプリを開発できます。

Discussion