この記事はレコチョク Advent Calendar 2023 の 6 日目の記事となります。

はじめに

株式会社レコチョクでバックエンドエンジニアをしている新卒2年目の小林です。
現在は、PythonによるAPIの開発やSolidityによるBlockchainのスマートコントラクトの開発を行っています。
趣味はゲームやアニメで、最近ハマっている曲はTVアニメ「ぼっち・ざ・ろっく!」作中バンドである結束バンドの「ギターと孤独と蒼い惑星」です。
今回は、2023/11/7 の OpenAI DevDay で発表された Assistants API を利用して簡単なチャットボットを作成する方法をまとめました。

Assistants API とは

ChatGPTの新機能の1つで、Assistants APIを利用することで自作のアプリケーションにAIアシスタントを構築することができます。
AIアシスタントは、さまざまなモデルやツールを用いて、ユーザーのリクエストに適切な応答を提供します。
β版（2023/11/22時点）では以下のツールが提供されています。

Functions
- ユーザーのリクエストに応じて定義済みの関数を実行し、回答を生成することができる
Code Interpreter
- GPT内でプログラミングコードを実行できる
Retrieval
- ユーザーがアップロードしたファイルからデータを取得し、それに関する回答を生成することができる

今回は主にRetrievalを利用して簡単なチャットボット形式の英単語一問一答を作成していきます。

開発準備

Pythonの仮想環境作成・起動

$ python --version
Python 3.12.0

$ python -m venv env
$source ./env/bin/activate

FastAPIとUvicorn、OpenAIのインストール

$ pip install fastapi uvicorn openai

$ pip list
Package           Version
---------------   ------------
fastapi           0.104.1
openai            1.3.3
uvicorn           0.24.0.post1

アプリケーションファイル作成

chat.pyファイルを作成し、以下のようにhealth_check関数を作成します。
```
from fastapi import FastAPI

app = FastAPI()

@app.get("/")
def health_check():
  return {"status": "success"}
```

アプリケーションの実行テスト

$ uvicorn chat:app --reload
INFO:     Will watch for changes in these directories: ['/path/chat']
INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
INFO:     Started reloader process [26279] using StatReload
INFO:     Started server process [26281]
INFO:     Waiting for application startup.
INFO:     Application startup complete.

上記が完了後 http://127.0.0.1:8000 にアクセス。
{"status": "success"} が出力されていれば開発準備完了。

OpenAI クライアント作成

import openai

api_key = "xxxxxxxxxx"
client = openai.OpenAI(api_key=api_key)

api_key は OpenAI の APIキーとなります。
OpenAIアカウント作成後、こちらから作成・取得することができます。
セキュリティ上のリスクを避けるため、APIキーは公開されることのないように保管してください。

Assistants API作成

以下のコードで、Assistants API に参照させるファイルのアップデート ~ Assistants API 作成を実行します。

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

class CreateAssistantResponse(BaseModel):
    assistant_id: str
    file_id: str

@app.post("/create_assistant", response_model=CreateAssistantResponse)
def create_assistant():
    try:
        # アシスタントに使用するファイルの登録
        with open("./en_word.pdf", "rb") as file:
            file_response = client.files.create(file=file, purpose="assistants")

        # アシスタントの作成
        assistant = client.beta.assistants.create(
            instructions="添付されたファイルを読み込み、そこからランダムで英単語を一問一答形式で出題してください。出題するのは単語で、単語の意味を問う形式です。正解の場合は「正解！」と返答し、不正解の場合は「不正解！」と正答をあわせて返答してください。また、正否を問わず返答後に次の問題を出題してください。これらのやり取りはすべて日本語で行ってください。",
            description="英単語を一問一答形式で出題するアシスタントです",
            name="英単語一問一答",
            tools=[{"type": "retrieval"}],
            model="gpt-4-1106-preview",
            file_ids=[file_response.id],
        )

        return CreateAssistantResponse(
            assistant_id=assistant.id, file_id=file_response.id
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

ファイルのアップロード

Assistants APIにファイルを読み込ませる場合は次のようにファイルをアップロードする必要があります。
今回の一問一答ではen_word.pdfに記載されている単語の中から出題してもらいます。

purpose="assistants" と渡すことでAssistants APIが利用するファイルとして設定されます。

with open("./en_word.pdf", "rb") as file:
    file_response = client.files.create(file=file, purpose="assistants")

Assistants API 作成

Assistants API の基本的な情報と上記でアップロードした際のResponseに含まれる id をもとにAssistants APIを作成します。

assistant = client.beta.assistants.create(
    instructions="添付されたファイルを読み込み、そこからランダムで英単語を一問一答形式で出題してください。出題するのは単語で、単語の意味を問う形式です。正解の場合は「正解！」と返答し、不正解の場合は「不正解！」と正答をあわせて返答してください。また、正否を問わず返答後に次の問題を出題してください。これらのやり取りはすべて日本語で行ってください。",
    description="英単語を一問一答形式で出題するアシスタントです",
    name="英単語一問一答",
    tools=[{"type": "retrieval"}],
    model="gpt-4-1106-preview",
    file_ids=[file_response.id],
)

client.beta.assistants.create() の引数は以下のとおりです。

instructions：Assistants APIに与える指示
description：Assistants APIに関する説明
tools：利用するツールを設定。上述の通りRetrievalを使用
model：利用するモデルを設定。今回は2023/11/22現在で最新のmodelである gpt-4-1106-preview を使用
file_ids：ファイルを利用する際に必要な、アップロードされたファイルのID

assistant_idとfile_idはAssistants APIを実際に呼び出す際に必要となるため、この関数のレスポンスに含められています。

Assistants APIが作成できているか確認

以下の関数を作成し、上記で作成したAssistants API を取得します。

def get_assistant(assistant_id: str):
    try:
        return client.beta.assistants.retrieve(assistant_id=assistant_id)
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

今回は以下のような結果になります。

{
    "id": "asst_l5dfU3smDsB4Lv2xXCK0c8LG",
    "created_at": 1700475087,
    "description": "英単語を一問一答形式で出題するアシスタントです",
    "file_ids": [
        "file-EQuyKRFCzhOx6im1oQv3GOpZ"
    ],
    "instructions": "添付されたファイルを読み込み、そこからランダムで英単語を一問一答形式で出題してください。出題するのは単語で、単語の意味を問う形式です。正解の場合は「正解！」と返答し、不正解の場合は「不正解！」と正答をあわせて返答してください。また、正否を問わず返答後に次の問題を出題してください。これらのやり取りはすべて日本語で行ってください。",
    "metadata": {
    },
    "model": "gpt-4-1106-preview",
    "name": "英単語一問一答",
    "object": "assistant",
    "tools": [
        {
            "type": "retrieval"
        }
    ]
}

Assistants APIにRequestを送る

Assistants APIが作成されていることを確認した後、以下の関数で Assistants API にRequestを送ります。

import time

from pydantic import BaseModel
from typing import Optional

class ChatRequest(BaseModel):
    assistant_id: str
    message: str
    file_id: str
    thread_id: Optional[str] = None

class ChatResponse(BaseModel):
    thread_id: str
    message: str

@app.post("/chat", response_model=ChatResponse)
def post_chat(request: ChatRequest):
    try:
        # スレッドの作成
        thread_id = request.thread_id
        if not thread_id:
            thread = client.beta.threads.create()
            thread_id = thread.id

        # スレッドにメッセージを追加
        client.beta.threads.messages.create(
            thread_id=thread_id,
            role="user",
            content=request.message,
            file_ids=[request.file_id],
        )

        # アシスタントの実行
        run = client.beta.threads.runs.create(
            thread_id=thread_id,
            assistant_id=request.assistant_id,
        )

        # 実行結果の取得
        while True:
            response = client.beta.threads.runs.retrieve(
                thread_id=thread_id,
                run_id=run.id,
            )
            if response.status == "completed":
                break
            time.sleep(10)

        # スレッドのメッセージを取得
        messages = client.beta.threads.messages.list(thread_id=thread_id)
        # アシスタントのメッセージのみを抽出
        assistant_messages = [msg for msg in messages.data if msg.role == "assistant"]
        # 最新のメッセージを取得
        latest_message = max(assistant_messages, key=lambda msg: msg.created_at)
        res_message = latest_message.content[0].text.value

        return ChatResponse(thread_id=thread_id, message=res_message)
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

スレッドの作成

ここで作成されるスレッドは、Persistent Threadsという機能により会話の履歴情報を保持することができます。
そのため、ChatCompletion APIのように会話の履歴をこちらで保存する必要がありません。
以前のやり取りを引き継ぎたい場合はthread_idを指定し、新規でやり取りを始めたい場合はここでスレッドを作成します。

thread_id = request.thread_id
if not thread_id:
    thread = client.beta.threads.create()
    thread_id = thread.id

スレッドにメッセージを追加

上記で作成したスレッドにメッセージを追加します。
このとき、role に user を設定することで Assistants API へのRequestとしてメッセージが追加されます。
contentがユーザーからの質問文で、返答の生成にファイルを指定する場合はfile_idsにリストで渡します。

client.beta.threads.messages.create(
    thread_id=thread_id,
    role="user",
    content=request.message,
    file_ids=[request.file_id],
)

Assistants API を実行

Assistants API 側が上記のRequestを読み取り、検索やツールを駆使して最適な返答を生成します。
この返答は、roleが assistant としてスレッドに追加されます。
この処理は非同期で実行されます。

run = client.beta.threads.runs.create(
    thread_id=thread_id,
    assistant_id=request.assistant_id,
)

実行結果の取得

上記で作成された Assistants API の返答を取得します。
runは非同期で実行されるため、client.beta.threads.runs.retrieve()のレスポンスに含まれるstatusが completed となるまでこの処理を繰り返し実行する必要があります。
time.sleep(10) の時間は任意です。

while True:
    response = client.beta.threads.runs.retrieve(
        thread_id=thread_id,
        run_id=run.id,
    )
    if response.status == "completed":
        break
    time.sleep(10)

メッセージの取得

スレッドに追加されたメッセージのリストを取得します。
client.beta.threads.messages.list() では同じスレッドに追加された以前までのメッセージが全て含まれるため、role が assistant である最新のメッセージを取得します。

# スレッドのメッセージを取得
messages = client.beta.threads.messages.list(thread_id=thread_id)
# アシスタントのメッセージのみを抽出
assistant_messages = [msg for msg in messages.data if msg.role == "assistant"]
# 最新のメッセージを取得
latest_message = max(assistant_messages, key=lambda msg: msg.created_at)
res_message = latest_message.content[0].text.value

使ってみた

それでは実際に上記のコードを実行してみます。
簡単なチャット形式になるようにChatGPTに以下のようなフロント部分を作成してもらいました。

では実際に一問一答の出題を依頼します。

このように、client.beta.assistants.create() でAssistant APIを作成した際に設定したinstructionsとfileに基づいて英単語を問い続けるチャットボットを作成することができました。

まとめ

最後までお読みいただきありがとうございました。
私の Assistants API の学習も兼ねてこの記事を作成しましたが、とても簡単にチャットボットが作成できた事に驚いています。
2023/11/22時点でAssistants API はまだβ版ということで、今後さらに使いやすく高性能になることを考えるとますます楽しみになります。

明日のレコチョク Advent Calendar 2023 は7日目 pytestのmockerについてです。
お楽しみに！

参考

この記事はレコチョクのエンジニアブログの記事を転載したものとなります。