2023/11/11に公開された OpenAI Cookbook｜「Python SDKを使ったAssistants APIの概要」を、一部補足を入れながら翻訳しましたので掲載いたします。
参照元: openai-cookbook/examples/Assistants_API_overview_python.ipynb at main · openai/openai-cookbook (github.com)

Python SDKを使ったアシスタントAPIの概要

OpenAIが提供する新しいアシスタントAPIは、チャット補完APIを発展させたもので、アシスタントのような体験を簡単に作成するためのものです。開発者は、Code InterpreterやRetrievalなどの強力なツールにアクセスできます。

チャット補完APIとアシスタントAPIの比較

チャット補完APIの基本は`Messages`で、`Model`（`gpt-3.5-turbo`、`gpt-4`など）を使って`Completion`を行います。これは軽量で強力ですが、基本的にステートレスなので、会話の状態、ツールの定義、リトリーバルドキュメント、コード実行などを手動で管理する必要があります。

アシスタントAPIの基本は以下の通りです。

• `Assistants`：基本モデル、指示、ツール、（コンテキスト）ドキュメントをカプセル化します。

• `Threads`：会話の状態を表します。

• `Runs`：テキスト応答とマルチステップツールの使用を含む`Thread`上での`Assistant`の実行を行います。

これらを使って、強力でステートフルな体験を作成する方法を見ていきましょう。

セットアップ

Python SDK

!pip install --upgrade openai

【Windowsの場合】
pip install --upgrade openai

バージョンが最新であることを確認するために以下を実行します。

!pip show openai | grep Version

【Windowsの場合】
pip show openai | findstr Version

Version: 1.2.3以上で表示されていればOKです。

Pretty Printing Helper

import json

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

アシスタントAPIを使った完全な例

アシスタント

アシスタントAPIを使い始める最も簡単な方法は、アシスタントプレイグラウンドを通じてです。

Math Tutorを作成しましょう。ドキュメントにあるように、新しいアシスタントを作成します。

作成されたアシスタントはアシスタントダッシュボードで見ることができます。

また、APIを直接使ってアシスタントを作成することもできます。

from openai import OpenAI

client = OpenAI()

assistant = client.beta.assistants.create(
    name="Math Tutor",
    instructions="You are a personal math tutor. Answer questions briefly, in a sentence or less.",
    model="gpt-4-1106-preview",
)
show_json(assistant)

アシスタントを作成したら、そのIDを追跡することが重要です。これにより、スレッドとランを通じてアシスタントを参照します。

次に、新しいスレッドを作成し、メッセージを追加します。これは会話の状態を保持するので、毎回完全なメッセージ履歴を再送する必要がなくなります。

スレッド

新しいスレッドを作成します。

thread = client.beta.threads.create()
show_json(thread)

そしてスレッドにメッセージを追加します。

message = client.beta.threads.messages.create(
    thread_id=thread.id,
    role="user",
    content="I need to solve the equation `3x + 11 = 14`. Can you help me?",
)
show_json(message)

ラン

スレッドを作成したことに注意してください。これは先ほど作成したアシスタントとは関連付けられていません！スレッドはアシスタントから独立して存在します。

アシスタントからスレッドに対する補完を得るためには、ランを作成する必要があります。ランを作成することで、アシスタントに対してスレッド内のメッセージを確認し、行動を取るよう指示します。

アシスタントにユーザーに応答するようにさせるために、ランを作成します。先ほど言及したように、アシスタントとスレッドの両方を指定する必要があります。

run = client.beta.threads.runs.create(
    thread_id=thread.id,
    assistant_id=assistant.id,
)
show_json(run)

チャット補完APIで補完を作成するのとは異なり、ランの作成は非同期操作です。すぐにランのメタデータを返し、`status`が最初に`queued`に設定されます。アシスタントが操作を実行する（ツールの使用やメッセージの追加など）につれて`status`が更新されます。

アシスタントが処理を完了したかどうかを知るために、ループでランをポーリングすることができます。（ストリーミングサポートは近日中に提供予定！）ここでは`queued`または`in_progress`ステータスのみを確認していますが、実際にはランがさまざまなステータス変更を経る可能性があります。これらはユーザーに表示することができます。（これらはステップと呼ばれ、後で説明します。）

import time

def wait_on_run(run, thread):
    while run.status == "queued" or run.status == "in_progress":
        run = client.beta.threads.runs.retrieve(
            thread_id=thread.id,
            run_id=run.id,
        )
        time.sleep(0.5)
    return run

そして、ランが完了したら、スレッド内のメッセージをリストして、アシスタントが追加した内容を確認します。

メッセージ

ランが完了したので、スレッド内のメッセージをリストしてアシスタントが何を追加したかを見てみましょう。

messages = client.beta.threads.messages.list(thread_id=thread.id)
show_json(messages)

ご覧の通り、メッセージは逆時系列順に並んでいます。これは、最新の結果が常に最初の`page`にあるようにするためです（結果はページネーションされる可能性があります）。この点を注意してください。これはチャット補完APIのメッセージ順序とは逆です。

アシスタントに結果をさらに説明してもらいましょう。

# 新しいメッセージをスレッドに追加します。
message = client.beta.threads.messages.create(
    thread_id=thread.id, role="user", content="Could you explain this to me?"
)

# ランを実行します。
run = client.beta.threads.runs.create(
    thread_id=thread.id,
    assistant_id=assistant.id,
)

# 完了を待ちます。
wait_on_run(run, thread)

# 最後のユーザーメッセージの後に追加されたすべてのメッセージを取得します。
messages = client.beta.threads.messages.list(
    thread_id=thread.id, order="asc", after=message.id
)
show_json(messages)

これは単純な例では多くのステップに感じられるかもしれませんが、実際には非常に強力な機能をコードをほとんど変更せずにアシスタントに追加することができます。

例

すべてをまとめてみましょう。以下は、作成したアシスタントを使用するために必要なすべてのコードです。

Math TutorのIDを`MATH_ASSISTANT_ID`に保存し、2つの関数を定義しました。

• `submit_message`: スレッドにメッセージを作成し、新しいランを開始（そして返す）関数

• `get_response`: スレッド内のメッセージのリストを返す関数

また、`create_thread_and_run`関数を定義しました。これは再利用できます（APIの`client.beta.threads.create_and_run`関数とほぼ同じです）。最後に、モックユーザーリクエストを新しいスレッドに送信できます。

これらのAPI呼び出しは非同期操作であり、`asyncio`のような非同期ライブラリを使用せずに実際に非同期動作をコードで得ることができます。

from openai import OpenAI

MATH_ASSISTANT_ID = assistant.id  # またはハードコードされたID "asst-..."

client = OpenAI()

def submit_message(assistant_id, thread, user_message):
    client.beta.threads.messages.create(
        thread_id=thread.id, role="user", content=user_message
    )
    return client.beta.threads.runs.create(
        thread_id=thread.id,
        assistant_id=assistant_id,
    )

def get_response(thread):
    return client.beta.threads.messages.list(thread_id=thread.id, order="asc")

def create_thread_and_run(user_input):
    thread = client.beta.threads.create()
    run = submit_message(MATH_ASSISTANT_ID, thread, user_input)
    return thread, run

# 並行したユーザーリクエストをエミュレートします。
thread1, run1 = create_thread_and_run(
    "I need to solve the equation `3x + 11 = 14`. Can you help me?"
)
thread2, run2 = create_thread_and_run("Could you explain linear algebra to me?")
thread3, run3 = create_thread_and_run("I don't like math. What can I do?")

# すべてのランが実行中です...

すべてのランが動作している間、各ランを待ち、応答を取得できます。

import time

# Pretty printing ヘルパー関数
def pretty_print(messages):
    print("# Messages")
    for m in messages:
        print(f"{m.role}: {m.content[0].text.value}")
    print()

# ループで待機する関数
def wait_on_run(run, thread):
    while run.status == "queued" or run.status == "in_progress":
        run = client.beta.threads.runs.retrieve(
            thread_id=thread.id,
            run_id=run.id,
        )
        time.sleep(0.5)
    return run

# Run 1 を待ちます。
run1 = wait_on_run(run1, thread1)
pretty_print(get_response(thread1))

# Run 2 を待ちます。
run2 = wait_on_run(run2, thread2)
pretty_print(get_response(thread2))

# Run 3 を待ちます。
run3 = wait_on_run(run3, thread3)
pretty_print(get_response(thread3))

# スレッド 3 でアシスタントに感謝のメッセージを送ります :)
run4 = submit_message(MATH_ASSISTANT_ID, thread3, "Thank you!")
run4 = wait_on_run(run4, thread3)
pretty_print(get_response(thread3))

これで完成です！

このコードは実際には数学アシスタントに特有のものではなく、アシスタントIDを変更するだけで新しいアシスタントを作成するために使用できます。これがアシスタントAPIの力です。

ツール

アシスタントAPIの重要な機能は、Code Interpreter、Retrieval、カスタムFunctionsなどのツールをアシスタントに装備することです。それぞれ見ていきましょう。

Code Interpreter

Math TutorにCode Interpreterツールを装備してみましょう。これはダッシュボードから...

...またはAPIを使用して、アシスタントIDを使って行うことができます。

assistant = client.beta.assistants.update(
    MATH_ASSISTANT_ID,
    tools=[{"type": "code_interpreter"}],
)
show_json(assistant)

次に、アシスタントに新しいツールを使ってもらいましょう。

thread, run = create_thread_and_run(
    "Generate the first 20 Fibonacci numbers with code."
)
run = wait_on_run(run, thread)
pretty_print(get_response(thread))

これで完了です！アシスタントは背後でCode Interpreterを使用し、最終的な応答を与えました。

ステップ

ランは1つ以上のステップで構成されています。ランと同様に、各ステップにはクエリ可能な`status`があります。これは、ユーザーに進行状況（たとえば、アシスタントがコードを書いている間や情報を取得している間のスピナー）を表示するのに役立ちます。

run_steps = client.beta.threads.runs.steps.list(
    thread_id=thread.id, run_id=run.id, order="asc"
)

それぞれのステップの`step_details`を見てみましょう。

for step in run_steps.data:
    step_details = step.step_details
    print(json.dumps(show_json(step_details), indent=4))

ランには以下のような`step_details`があります：

1. `tool_calls`（複数形、複数のツールが単一のステップで使用される可能性があるため）

2. `message_creation`

最初のステップは、`code_interpreter`を使用する`tool_calls`です。これには以下が含まれます：

• `input`：ツールが呼び出される前に生成されたPythonコード

• `output`：Code Interpreterの実行結果

2番目のステップは`message_creation`で、スレッドに追加されたメッセージを含んでいます。

Retrieval

アシスタントAPIのもう1つの強力なツールはRetrievalです。これはファイルをアップロードして、アシスタントが質問に答ええる際の知識ベースとして使用する機能です。ダッシュボードまたはAPIから有効にすることができ、使用するファイルをアップロードできます。

# ファイルをアップロード
file = client.files.create(
    file=open(
        "data/language_models_are_unsupervised_multitask_learners.pdf",
        "rb",
    ),
    purpose="assistants",
)
# アシスタントを更新
assistant = client.beta.assistants.update(
    MATH_ASSISTANT_ID,
    tools=[{"type": "code_interpreter"}, {"type": "retrieval"}],
    file_ids=[file.id],
)
show_json(assistant)

thread, run = create_thread_and_run(
    "What are some cool math concepts behind this ML paper pdf? Explain in two sentences."
)
run = wait_on_run(run, thread)
pretty_print(get_response(thread))

Functions

アシスタントに対してカスタムFunctionsを指定することができます。ラン中にアシスタントは指定された関数を呼び出すことを示すことができ、呼び出し側は関数を実行し、その出力をアシスタントに提供する責任があります。

Math Tutorに`display_quiz()`という関数を定義する例を見てみましょう。この関数は`title`と`question`の配列を取り、クイズを表示し、ユーザーから各質問の入力を取得します。

# ユーザーのモックレスポンスを取得する関数
def get_mock_response_from_user_multiple_choice():
    return "a"

def get_mock_response_from_user_free_response():
    return "I don't know."

def display_quiz(title, questions):
    print("Quiz:", title)
    print()
    responses = []

    for q in questions:
        print(q["question_text"])
        response = ""

        # 複数選択の場合、オプションを表示
        if q["question_type"] == "MULTIPLE_CHOICE":
            for i, choice in enumerate(q["choices"]):
                print(f"{i}. {choice}")
            response = get_mock_response_from_user_multiple_choice()

        # それ以外の場合、レスポンスを取得
        elif q["question_type"] == "FREE_RESPONSE":
            response = get_mock_response_from_user_free_response()

        responses.append(response)
        print()

    return responses

アシスタントでこの関数を呼び出せるように、JSON形式で関数のインターフェースを定義します。

function_json = {
    "name": "display_quiz",
    "description": "Displays a quiz to the student, and returns the student's response. A single quiz can have multiple questions.",
    "parameters": {
        "type": "object",
        "properties": {
            "title": {"type": "string"},
            "questions": {
                "type": "array",
                "description": "An array of questions, each with a title and potentially options (if multiple choice).",
                "items": {
                    "type": "object",
                    "properties": {
                        "question_text": {"type": "string"},
                        "question_type": {
                            "type": "string",
                            "enum": ["MULTIPLE_CHOICE", "FREE_RESPONSE"],
                        },
                        "choices": {"type": "array", "items": {"type": "string"}},
                    },
                    "required": ["question_text"],
                },
            },
        },
        "required": ["title", "questions"],
    },
}

ダッシュボードまたはAPIを使用してアシスタントを更新します。

assistant = client.beta.assistants.update(
    MATH_ASSISTANT_ID,
    tools=[
        {"type": "code_interpreter"},
        {"type": "retrieval"},
        {"type": "function", "function": function_json},
    ],
)
show_json(assistant)

クイズを求めてリクエストを行います。

thread, run = create_thread_and_run(
    "Make a quiz with 2 questions: One open ended, one multiple choice. Then, give me feedback for the responses."
)
run = wait_on_run(run, thread)
run.status

ランの`status`を確認すると`requires_action`となっています。これはツールが実行を待っており、アシスタントに出力を提出する必要があることを示しています。

# 単一のツールコールを抽出
tool_call = run.required_action.submit_tool_outputs.tool_calls[0]
name = tool_call.function.name
arguments = json.loads(tool_call.function.arguments)

print("Function Name:", name)
print("Function Arguments:")
arguments

実際に`display_quiz`関数をアシスタントから提供された引数で呼び出してみましょう。

responses = display_quiz(arguments["title"], arguments["questions"])
print("Responses:", responses)

レスポンスを取得したので、アシスタントに提出しましょう。`tool_call`のIDが必要で、これは先ほど解析した`tool_call`から取得できます。

run = client.beta.threads.runs.submit_tool_outputs(
    thread_id=thread.id,
    run_id=run.id,
    tool_outputs=[
        {
            "tool_call_id": tool_call.id,
            "output": json.dumps(responses),
        }
    ],
)
show_json(run)

ランが再び完了するのを待ち、スレッドをチェックしましょう。

run = wait_on_run(run, thread)
pretty_print(get_response(thread))

おめでとうございます 🎉

結論

このノートブックでは多くのことをカバーしました。疲れたら自分を褒めてください！このノートブックを通じて、Code Interpreter、Retrieval、Functionsなどのツールを使った強力でステートフルな体験を構築するためのしっかりとした基盤を築くことができたはずです。

省略したセクションもいくつかありますので、さらに詳しく学びたい方は以下のリソースを確認してください：

• Annotations: ファイルの引用の解析

• Files: スレッドスコープとアシスタントスコープのファイル

• Parallel Function Calls: 単一のステップで複数のツールを呼び出し

• マルチアシスタントスレッドラン: 複数のアシスタントからのメッセージを持つ単一のスレッド

• ストリーミング: 近日提供予定！

これで素晴らしいものを作りましょう！

追伸

何故か最後にこのリンクが張られていたので貼っておきます💦