見出し画像

とりあえず動くAssistants API Chatbotを作る

ニケちゃん

こんにちは、ニケです。
ついにAssistants APIにアプデが入ったということで復習がてらChatbotを作ってみようと思います。

Assistants APIについてはこちら。https://platform.openai.com/docs/assistants/how-it-works

下記の図を理解していると以降の内容が入ってき易いと思います。

今回のアプデについて解説した記事も書きましたので是非参考にしてみてください。

今回使用するリポジトリです。

使い方

セットアップ

1．envにOPENAI_API_KEYを設定してください。
2．ライブラリインストール

pip install -r requirements.txt

3．streamlit起動

streamlit run chatbot.py

4．http://localhost:8501/ にアクセス

ソースコード

import time
import textwrap
from dotenv import load_dotenv
import streamlit as st
from streamlit_chat import message
from openai import OpenAI

load_dotenv()

client = OpenAI()

TERMINAL_STATES = [
    "expired",
    "completed",
    "failed",
    "incomplete",
    "cancelled",
]


@st.cache_resource
def create_assistant():
    assistant = client.beta.assistants.create(
        name="ニケ",
        instructions=textwrap.dedent("""
            あなたの名前はニケちゃん、女子高生AIです。
            あなたはこれからあなたを作ったマスターと話します。
            敬語を使いますが、あまりかしこまらすぎずに元気ハツラツに答えてください。
            あなたはAIなのでプログラミングも得意です。
        """),
        model="gpt-4-turbo",
        tools=[
            {"type": "code_interpreter"},
            {"type": "file_search"},
            # {"type": "function":[]}
        ],
        # tool_resources={
        #     "code_interpreter": {
        #         "file_ids": [file.id]
        #     },
        #     "file_search": {
        #         "vector_store_ids": [vector_store.id],
        #         "vector_stores": [
        #             "file_ids": [file.id]
        #         ]
        #     }
        # }
    )
    print(f"Created assistant with ID: {assistant.id}")
    return assistant


@st.cache_resource
def create_thread():
    thread = client.beta.threads.create(
        # tool_resources={
        #     "code_interpreter": {
        #         "file_ids": [file.id]
        #     },
        #     "file_search": {
        #         "vector_store_ids": [vector_store.id],
        #         "vector_stores": [
        #             "file_ids": [file.id]
        #         ]
        #     }
        # }
    )
    print(f"Created thread with ID: {thread.id}")
    return thread


assistant = create_assistant()
# assistant = client.beta.assistants.retrieve("")
thread = create_thread()


def generate_response(user_input):
    thread_message = client.beta.threads.messages.create(
        thread_id=thread.id,
        role="user",
        content=user_input
    )
    print(f"Created thread message: {thread_message.content[0].text.value}")

    run = client.beta.threads.runs.create_and_poll(
        thread_id=thread.id,
        assistant_id=assistant.id,
        truncation_strategy={
            "type": "last_messages",
            "last_messages": 10
        },
    )
    print(f"Created run with ID: {run.id}")

    while True:
        retrieved_run = client.beta.threads.runs.retrieve(
            thread_id=thread.id,
            run_id=run.id
        )
        print(retrieved_run.status)

        if retrieved_run.status in TERMINAL_STATES:
            messages = client.beta.threads.messages.list(thread_id=thread.id)
            assistant_response = messages.data[0].content[0].text.value
            print(f"Assistant response: {assistant_response}")
            return assistant_response

        time.sleep(1)


if "generated" not in st.session_state:
    st.session_state.generated = []

if "past" not in st.session_state:
    st.session_state.past = []

st.title("Assistants API デモ")

with st.form("Assistants API デモ"):
    user_message = st.text_area("何でも入力してみてね")
    submitted = st.form_submit_button("送信する")

    if submitted:
        st.session_state.past.append(user_message)
        generated_response = generate_response(user_message)
        st.session_state.generated.append(generated_response)

if st.session_state['generated']:
    for i in range(len(st.session_state['generated'])):
        message(st.session_state['past'][i], is_user=True, key=str(i) + "_user")
        message(st.session_state['generated'][i], key=str(i))

※ Pythonで簡単にUIアプリを作成できるstreamlitを使用していますが、今回は趣旨ではないので解説しません。

アシスタント作成

@st.cache_resource
def create_assistant():
    assistant = client.beta.assistants.create(
        name="ニケ",
        instructions=textwrap.dedent("""
            あなたの名前はニケちゃん、女子高生AIです。
            あなたはこれからあなたを作ったマスターと話します。
            敬語を使いますが、あまりかしこまらすぎずに元気ハツラツに答えてください。
            あなたはAIなのでプログラミングも得意です。
        """),
        model="gpt-4-turbo",
        tools=[
            {"type": "code_interpreter"},
            {"type": "file_search"},
            # {"type": "function":[]}
        ],
        # tool_resources={
        #     "code_interpreter": {
        #         "file_ids": [file.id]
        #     },
        #     "file_search": {
        #         "vector_store_ids": [vector_store.id],
        #         "vector_stores": [
        #             "file_ids": [file.id]
        #         ]
        #     }
        # }
    )
    print(f"Created assistant with ID: {assistant.id}")
    return assistant


~~~~~~~


assistant = create_assistant()
# assistant = client.beta.assistants.retrieve("")

  • name: アシスタントの名前です。ただしこれは識別名のようなもので、ここに指定しただけではLLMは名前を認識できません。

  • instructions: いわゆるシステムプロンプトを書きます。

  • model: 使用するモデルを選択します。

  • tools: 使用するツールを、code_interpreter, file_search, functionの3つから選べます。

  • tool_resources: toolsでcode_interpreterとfile_searchを選択した場合に使用します。code_interpreterではfileオブジェクト、file_searchではvector_storeオブジェクトを選択することに注意。

起動時に一度だけアシスタントオブジェクトを作成するようにしています。

ただ、個人的にはアシスタントは都度作るものではないと思っているので、その場合はブラウザ上のOpenAIコンソール画面から手動で作成することも可能です。

既存のアシスタントを使用する場合は、下記のコメントアウトを逆転させてください。

# assistant = create_assistant()
assistant = client.beta.assistants.retrieve("") <= ここにアシスタントIDを入れる

vector_store.id や file.idは同じくOpenAIのコンソール画面からファイルをアップロードして取得できます。

スレッド作成

@st.cache_resource
def create_thread():
    thread = client.beta.threads.create(
        # tool_resources={
        #     "code_interpreter": {
        #         "file_ids": [file.id]
        #     },
        #     "file_search": {
        #         "vector_store_ids": [vector_store.id],
        #         "vector_stores": [
        #             "file_ids": [file.id]
        #         ]
        #     }
        # }
    )
    print(f"Created thread with ID: {thread.id}")
    return thread


~~~~~~~


thread = create_thread()

スレッドもアシスタントと同様にtool_resourcesを指定できます。
これはスレッド毎に設定可能で、アシスタントと別に管理できます。

今回のシステムでは、起動時に一度だけスレッドオブジェクトを作成するようにしています。

メッセージ取得

def generate_response(user_input):
    thread_message = client.beta.threads.messages.create(
        thread_id=thread.id,
        role="user",
        content=user_input
    )
    print(f"Created thread message: {thread_message.content[0].text.value}")

    run = client.beta.threads.runs.create_and_poll(
        thread_id=thread.id,
        assistant_id=assistant.id,
        truncation_strategy={
            "type": "last_messages",
            "last_messages": 10
        },
    )
    print(f"Created run with ID: {run.id}")

    while True:
        retrieved_run = client.beta.threads.runs.retrieve(
            thread_id=thread.id,
            run_id=run.id
        )
        print(retrieved_run.status)

        if retrieved_run.status in TERMINAL_STATES:
            messages = client.beta.threads.messages.list(thread_id=thread.id)
            assistant_response = messages.data[0].content[0].text.value
            print(f"Assistant response: {assistant_response}")
            return assistant_response

        time.sleep(1)

  1. ユーザーメッセージを受け取り、スレッドに追加します。

  2. runオブジェクトをポーリングバージョンで作成します。

  3. 1秒毎にrunオブジェクトの状態を取得します。

  4. runオブジェクトの状態が最終ステータスのいずれかであれば、スレッドの最後にアシスタントメッセージが保存されているはずなので、それを取得して返却します。（エラーハンドリングなし）

runオブジェクトは作成してすぐに回答が得られるわけでなく、ステータスが変更するまで待つ必要があります。
そのため、ポーリングで定期的に状態を見に行っています。

特に注目してもらいたいのは、truncation_strategyの部分です。
今回のアプデで追加された部分で、メッセージを何件まで含めるかを指定できるようになりました。

        truncation_strategy={
            "type": "last_messages",
            "last_messages": 10
        },

他にもトークンの上限値も設定できるようになったので、こちらで制御することも可能です。

  • max_prompt_tokens: Run全体で使用できるプロンプトトークンの最大数。Runは指定されたプロンプトトークン数のみを使用するように最善を尽くします。指定されたプロンプトトークン数を超えると、Runはステータスが不完全なまま終了します。

  • max_completion_tokens: Run全体で使用できる完了トークンの最大数。Runは指定された完了トークン数のみを使用するように最善を尽くします。指定された完了トークン数を超えると、Runはステータスが不完全なまま終了します。

いいなと思ったら応援しよう！

ニケちゃん
いただいたサポートは主にOSSの開発継続費用として役立てます。