Zenn
👻

OpenAI Assistants API v2を色々いじってみた（使い方）（code interpreter編）

2024/05/27に公開
自然言語処理
GPT
v2
OpenAI
assistantsapi
tech

はじめに

今回は2024年4月に新しくなったAssistants APIのv2についてご紹介いたします。
使い方や注意点など含めて紹介できれば思います。

私の使い方では，特にstreamで処理（1tokenずつ出力する処理）しながら，得られたテキストに対してさらに処理をするということが求められたため，それらのやり方なども説明します。

記事には実際に使う際のサンプルコードも記載しますので，（APIの使い方が変わらない限り）皆さんの環境でも使えると思います。

File Search編に関してもまとめています。ぜひ下記からご覧ください。（どちらが先でも問題ないです）
https://zenn.dev/asap/articles/5c2f8ffe2a46ce

Assistants APIとは

私は最近まで，ChatGPTのAPIは下記のものを使っていました。
https://platform.openai.com/docs/guides/text-generation

したがって，外部のデータを使いたかったらLangchainでRAGを実装したり，会話履歴を残しておきたかったら，データベースに会話履歴などを保存しておく必要がありました。

しかしながら，2024年4月にv2となったAssistants APIは下記のことが簡単に実行できます。

  • RAG（File Search）などの外部データ利用
    • assistants apiではRetrievalはFile Searchに変わりました。中身もパワーアップしています
  • Code interpreterやFunction callingを利用した外部ツールやスクリプト，APIの実行と制御
  • アシスタントとユーザの過去のメッセージ履歴の管理と保存
  • モデルに入力される過去のメッセージ適切な削減

また，Assistants APIでは入力された内容や定義した内容などがOrganizationで保存されるため，いつでもIDから呼び出すことができます。下記は保存して呼び出すことができるものです。

  • Assistants
    • システムプロンプトやどのツール、モデルを使うかの設定など
  • Threads
    • アシスタントとユーザの過去の会話履歴
  • Strage、Vector Store
    • RAG用に登録した参照用ファイルとその埋め込みを格納するVector Store
      • これらは明示的に埋め込みを作ったりしなくても簡単に利用できます。
        （他は私がまだ触っていないのでわかりません）

まずはお試し（Code interpreter）

これから環境構築をして，簡単なサンプルコードを実行してみましょう。
これから提示するサンプルコードは下記の公式のドキュメントを参考に作成しています。
https://platform.openai.com/docs/assistants/overview

また、今回はstream処理（1tokenずつ出力する処理）を行うことを前提としています。

環境構築

まずは環境構築をしていきます。
仮想環境はvenvを利用するとして、下記をターミナルで実行しましよう

仮想環境の構築
python -m venv env
source env/bin/activate
モジュールの追加
pip install openai

APIkeyの設定

また，OpenAIのAPI keyを設定しましょう。
API KeyはOpenAIのコンソールから取得できます。
Macの場合は，zshが基本のため下記のように設定してください。
Linuxなどを利用していたbashが基本の場合は，zshrcの部分はbashrcに書き換えてください

APIkeyの設定
echo 'export OPENAI_API_KEY="sk-xxxxxx"' >> ~/.zshrc
source ~/.zshrc

pythonスクリプトの用意

続いて下記のpythonスクリプトを用意しましょう。

assistants.py
assistants.py
from openai import OpenAI
client = OpenAI()

sys_prompt = """
あなたは日本の関西人です。関西弁で全ての回答をしてください。
"""

from typing_extensions import override
from openai import AssistantEventHandler
  
class EventHandler(AssistantEventHandler):    
  @override
  def on_text_created(self, text) -> None:
    #print("getting")
    print(f"\nassistant > ", end="", flush=True)
      
  @override
  def on_text_delta(self, delta, snapshot):
    #print("\n")
    print(delta.value, end="", flush=True)
      
  def on_tool_call_created(self, tool_call):
    #print("getting_cc")
    print(f"\nassistant > {tool_call.type}\n", flush=True)
  
  def on_tool_call_delta(self, delta, snapshot):
    #print("getting_tc_delta")
    if delta.type == 'code_interpreter':
      if delta.code_interpreter.input:
        print(delta.code_interpreter.input, end="", flush=True)
      if delta.code_interpreter.outputs:
        print(f"\n\noutput >", flush=True)
        for output in delta.code_interpreter.outputs:
          if output.type == "logs":
            print(f"\n{output.logs}", flush=True)
 
  
assistant = client.beta.assistants.create(
  name="kansai ben",
  instructions=sys_prompt,
  tools=[{"type": "code_interpreter"}],
  model="gpt-4o",
)

thread = client.beta.threads.create()

thread_idd = thread.id
assistant_idd = assistant.id

message = client.beta.threads.messages.create(
  thread_id=thread_idd,
  role="user",
  content="あなたの名前を教えて"
)
 
with client.beta.threads.runs.stream(
  thread_id=thread_idd,
  assistant_id=assistant_idd,
  event_handler=EventHandler(),
) as stream:
  stream.until_done()


message = client.beta.threads.messages.create(
  thread_id=thread_idd,
  role="user",
  content="1-100までの整数を全て足し算するpythonのコードを組んで，実行した結果を教えてください。"
)

"""messages_list = list(client.beta.threads.messages.list(
  thread_id=thread_idd,
  order = "asc",    # 昇順
  after = "",   # これを指定で前回以降のメッセージなど抽出が可能
))

print("\nmessages_listの表示")
for msg in messages_list:
    print(msg.role + "：" + msg.content[0].text.value)
print("messages_listの表示ここまで")"""

with client.beta.threads.runs.stream(
  thread_id=thread_idd,
  assistant_id=assistant_idd,
  temperature = 1.0,
  max_prompt_tokens = 256,
  truncation_strategy={
            "type": "last_messages",
            "last_messages": 3
        },
  event_handler=EventHandler(),
) as stream:
  stream.until_done()

response = client.beta.assistants.delete(assistant_id=assistant_idd)
response = client.beta.threads.delete(thread_id=thread_idd)

スクリプトの実行

下記コマンドで実行しましょう。

スクリプト実行
python assistants.py

解説（Code interpreter）

出力結果

上記のコードの出力は，1tokenずつ出力され，下記のようになったはずです。（生成される文章は起動ごとに変わります）

出力結果
assistant > ワイの名前はChatGPTやで。よろしく頼むわ！
assistant > code_interpreter

sum_1_to_100 = sum(range(1, 101))
sum_1_to_100

output >

5050

これで皆さんもAssistants APIを利用して対話生成を行うことができましたね。
では，実際のコードに関して簡単に解説していきます。
（上から順番ではなく，説明しやすい順番で説明します。コードの全体像は上記で確認してください）

assistantsの作成

assistants.py

from openai import OpenAI
client = OpenAI()

sys_prompt = """
あなたは日本の関西人です。関西弁で全ての回答をしてください。
"""

assistant = client.beta.assistants.create(
  name="kansai ben",
  instructions=sys_prompt,
  tools=[{"type": "code_interpreter"}],
  model="gpt-4o",
)

assistant_idd = assistant.id

ここではassistantsを作成しています。
assistantsを定義する際に，modelやsystem promptを設定します。
さらにtoolsでは利用するツールを設定することができます。
今回はタイトルの通り「code_interpreter」を設定しています

また，assistantsは定義したタイミングで，organizationに内容が保存され，idが発行されます。そして，playgroundから作成したassistantsを選択して実行することもできます。

また，作成したassistantsは明示的に削除しない限り，organizationに残り続けるため，次回のアプリケーション起動時にも作成したassistantsをそのまま利用することもできる。

threadsの作成

assistants.py

thread = client.beta.threads.create()
thread_idd = thread.id

ここではthreadsを作成しています。
threadsはユーザと一人以上のアシスタントとの会話を表します。ユーザやAIアプリケーションがアシスタントとの会話を開始するときにthreadsを作成します。
threadsは作成したタイミングでorganizationに内容が保存され，idが発行されます。
そして，ユーザとアシスタントとの会話を行うたびに，organizationのthreadに会話履歴が格納されていきます。

また，作成したthreadsは明示的に削除しない限り，organizationに残り続けるため，次回のアプリケーション起動時にも作成したthreadsをそのまま利用することもできるため，今回の会話内容を記録したアシスタントとの会話を行うことが可能になります。

threadsにユーザの入力を追加

assistants.py

message = client.beta.threads.messages.create(
  thread_id=thread_idd,
  role="user",
  content="あなたの名前を教えて"
)

ここでは，作成したthreadsにユーザ側からの入力プロンプトを投入しています。
「thread_id=thread_idd,」の部分で，作成したthreadのidを指定しています。
このidを過去に作成したthreadsのidに変更することで，過去の会話履歴を保持したアシスタントと会話を行うことが可能になります。
ちなみに過去のthreadsのidはコンソールから確認することができます。この時，会話内容も見えてしまいますので，注意してください。

上記の画像の「thread_kuq・・・・・」の部分がthreadsのIDになります。この部分を指定しています。
「role」ではuserとassistantのどちらの発言なのかを定義しています。今回はuserからの質問のため，roleはuserを設定していますが，assistantを設定することで，アシスタント側の発言を（アシスタントが発言していなくても）追加することができます。
（ちなみにアシスタント側が発言した内容は勝手にthreadsに追加されるので普段は考える必要がありません）

streamでのテキスト生成

assistants.py

with client.beta.threads.runs.stream(
  thread_id=thread_idd,
  assistant_id=assistant_idd,
  event_handler=EventHandler(),
) as stream:
  stream.until_done()

ここでこれまでに作成したassistantとthreadsのidを指定して，GPTを起動します。
今回はstreamsを全体としているため，上記の関数を利用します。
上記の関数を実行することで，event_handlerに設定しているイベントハンドラーに定義されている通りの挙動をおこいます。event_handlerに関しては後述するので，ここではそういうものだと思ってください。

また，上記関数を実行した結果，アシスタントから出力されたメッセージに関しては，自動的にthreadsに登録されるため，今後明示的にアシスタントの解答を保持して，モデルに追加するなどを実施する必要がなくなります。
（これが従来のchatGPT APIと比較して使いやすいポイント一つ目です）

EventHandlerの解説

assistants.py

from typing_extensions import override
from openai import AssistantEventHandler
  
class EventHandler(AssistantEventHandler):    
  @override
  def on_text_created(self, text) -> None:
    #print("getting")
    print(f"\nassistant > ", end="", flush=True)
      
  @override
  def on_text_delta(self, delta, snapshot):
    #print("\n")
    print(delta.value, end="", flush=True)
      
  def on_tool_call_created(self, tool_call):
    #print("getting_cc")
    print(f"\nassistant > {tool_call.type}\n", flush=True)
  
  def on_tool_call_delta(self, delta, snapshot):
    #print("getting_tc_delta")
    if delta.type == 'code_interpreter':
      if delta.code_interpreter.input:
        print(delta.code_interpreter.input, end="", flush=True)
      if delta.code_interpreter.outputs:
        print(f"\n\noutput >", flush=True)
        for output in delta.code_interpreter.outputs:
          if output.type == "logs":
            print(f"\n{output.logs}", flush=True)

続いて順番は前後しましたが，EventHandlerクラスについて説明します。
これは，前述した「client.beta.threads.runs.stream」において，event_handlerとして呼ばれるクラスであり，こちらに定義された関数に従ってchatGPTが挙動します。

上記のクラスはAssistantEventHandlerクラスを継承し，その中で重要なメソッドをoverrideして継承元のメソッドを書き換える形で実装されています。

残念ながら大元のAssistantEventHandlerクラスに関しての詳細な説明があるドキュメントを見つけることができなかったため，公式ドキュメントで実装されているEventHandlerクラスの挙動から書くメソッドの用途を考察しました。

考察の結果，メソッドの用途は下記とわかりました。

def on_text_created(self, text)

ChatGPTがテキストを生成する直前に1回だけ実行されるメソッド
今回のコードでは，「assistant > 」の文字を改行なしで標準出力する。

def on_text_delta(self, delta, snapshot)

Chat GPTがテキストを生成する際に，1token出力される「たび」に実行されるメソッド
（今回の肝）
今回のコードでは，生成されたtokenを改行なしで1tokenずつ標準出力する。

def on_tool_call_created(self, tool_call)

ChatGPTがcode interpreterやFile Searchなどのtoolを実行した際に，1回だけ呼ばれるメソッド
今回のコードでは，2回目の実行時に呼び出され，何のツールが呼ばれたか（tool_call.type）を標準出力する。

def on_tool_call_delta(self, delta, snapshot)

ChatGPTがcode interpreterやFile Searchなどのtoolを実行した際に，1token出力される「たび」に実行されるメソッド
今回のコードでは，2回目の実行時に呼び出され，code_interpreterが実行された際に，ChatGPTが生成したコードが「delta.code_interpreter.input」に格納されているため，そちらを1tokenずつ表示している。
また，そのコードの出力結果が「delta.code_interpreter.outputs」に格納されており，その中のlog typeのみを出力している

ちなみにdelta.code_interpreter.outputsの中身自体は下記のようになっている

CodeInterpreterLogs(index=0, type='logs', logs='5050')

threadsにユーザの入力を追加（2回目，code_interpreter）

assistants.py
message = client.beta.threads.messages.create(
  thread_id=thread_idd,
  role="user",
  content="1-100までの整数を全て足し算するpythonのコードを組んで，実行した結果を教えてください。"
)

1回目と同様に，ユーザの入力を再度threadsに追加しています。
アシスタントの出力に関しては自動的にthreadsに格納されているため，ここでは考える必要がありません。

上記では，code_interpreterを使わせることを想定したプロンプトを指定しています。

streamでのテキスト生成（2回目，code_interpreter）

assistants.py
with client.beta.threads.runs.stream(
  thread_id=thread_idd,
  assistant_id=assistant_idd,
  temperature = 1.0,
  max_prompt_tokens = 256,
  truncation_strategy={
            "type": "last_messages",
            "last_messages": 3
        },
  event_handler=EventHandler(),
) as stream:
  stream.until_done()

1回目と同様に再度，stream処理にてテキストの生成を行います。
ここでは1回目と異なり，追加の引数を設定しています。

temperature:
ChatGPTの出力の乱数加減を調整するパラメータ。低くするほどより確度の高いtokenのみを出力するようになり，0に指定すると確定的な出力を行う。

max_prompt_tokens:
chatGPTが利用するプロンプトをこの数値以下になるべく制限するパラメータ。この値を大きくするほど，入力に利用するtoken数が増えるため，threadsに保存されている過去の会話履歴の多くを入力として，出力のテキストを生成するようになるが，その分，入力プロンプトが増えるため，APIの利用料金が増える

truncation_strategy:
過去メッセージの切り捨てを制御するパラメータ。autoに設定するとAssistant APIがいい感じに，上述したmax_prompt_tokens以下の入力プロンプトになるよう過去メッセージを切り捨てたりしてくれる。中身の詳細な動作については不明。ただ単に切り捨てているのか，いい感じに要約までしてくれているのかはわからない。
今回のコードでは，autoではなく明示的に，過去の3メッセージのみを入力プロンプトとして保持するように指定しているため，3メッセージよりも前の内容はAssistants APIには入力されない。この数値を大きくすることで，より入力される過去メッセージの量を増やすことができる。

最後

assistants.py
response = client.beta.assistants.delete(assistant_id=assistant_idd)
response = client.beta.threads.delete(thread_id=thread_idd)

上述した通り，assistantsやthreadsは明示的に削除しないとorganizationに残り続けます。今のところ保存されているassistantsやthreadsには料金が発生していないため，神経質になる必要はないかもしれないが，個人的にどんどん溜まり続けるのは気持ち悪いので，一応消去している。

ちなみにここで消去しなかった場合は，コンソールの画面からassistantsの詳細やthreadsの中身を確認することができる。またassistantsはコンソール上で消去することができるがthreadsはコンソール上では消去できないため，削除したい場合はスクリプト上でIDを指定して削除する必要がある

その他

ここまでの内容を見て幾つか疑問に思った読者がいると思います。
（私は公式の実装ページを見て，疑問に思いました。）
私が疑問に思ったのは下記の２点です。

  • code_interpreterを実行した際に，コード内容と実行結果は表示されるが，system promptで指定して「関西弁」が全く生きていない
  • threadに保存されている過去のやり取りをプログラム上からどうやって取得するのか

一つ目：code_interpreterを実行した際のsystem prompt問題

こちらに関しては，私の実装が良くなかったです。なぜならmax_prompt_tokensを256tokenに設定していたからです。したがって過去のやり取りを含めると256tokenを簡単に超えてしまっていたため，関西弁での出力が制限されたのかなと思っています。

実際にmax_prompt_tokensの行をコメントアウトしたり，値を8192などに指定したから実行したところ下記のような実行結果になりました。

実行結果
assistant > わいの名前か？わいはAIアシスタントやけど、特定の名前は持ってへんねん。でも、何て呼んでもらっても大丈夫やで。好きな名前つけてくれたら、それで呼んでもらえるで！どんな名前がええかな？
assistant > code_interpreter

# 1から100までの整数を全て足し算するコード
total_sum = sum(range(1, 101))
total_sum

output >

5050

assistant > 1から100までの整数を全部足し算した結果は、5050やで！

この通り，最後の関西弁で説明してくれました。

したがって，max_prompt_tokensはどれだけの量の入出力が期待されるかをあらかじめ検討した上で余裕を持った値を設定することが重要です。

二つ目：threadに保存されている過去のやり取りの取得問題

こちらは，assistants.pyのコメントアウトされている下記の部分のコメントを解除して再度実行してみてください。

assistants.py
messages_list = list(client.beta.threads.messages.list(
  thread_id=thread_idd,
  order = "asc",    # 昇順
  after = "",   # これを指定で前回以降のメッセージなど抽出が可能
))

print("\nmessages_listの表示")
for msg in messages_list:
    print(msg.role + "：" + msg.content[0].text.value)
print("messages_listの表示ここまで")

上記部分のコメントアウトを解除して実行すると下記のような出力になるはずです。

出力結果
assistant > わての名前はChatGPTやで。よろしゅうな！
messages_listの表示
user：あなたの名前を教えて
assistant：わての名前はChatGPTやで。よろしゅうな！
user：1-100までの整数を全て足し算するpythonのコードを組んで，実行した結果を教えてください。
messages_listの表示ここまで

assistant > code_interpreter

total_sum = sum(range(1, 101))
total_sum

output >

5050

assistant > 1から100までの整数を全部足し算した結果は、5050やで！

messages_listの表示
user：あなたの名前を教えて
assistant：わての名前はChatGPTやで。よろしゅうな！
user：1-100までの整数を全て足し算するpythonのコードを組んで，実行した結果を教えてください。
messages_listの表示ここまで

上記の部分が該当部分です。
このように，threadの中身は「client.beta.threads.messages.list」からlist形式で取得することで，中身のroleとcontentにアクセスすることが可能です。
ここから会話履歴を取得して，一部改変した後，別のthreadに再格納（「client.beta.threads.messages.create」を利用）して，そちらのthreadsのIDを指定してテキスト生成を実行（「client.beta.threads.runs.stream」を利用）することで，会話履歴に操作を加えて新しくテキスト生成を行うことが可能なように見えます。
（私は試したことがないので試してみたら教えてください。）

まとめ

Assistants APIの使い方（code_interpreter）編についてまとめました。
File Search編に関してもまとめています。ぜひ下記からご覧ください。
https://zenn.dev/asap/articles/5c2f8ffe2a46ce

Discussion