OpenAI Assistant API
ChatGPT とやりとりを行うAPIは複数ある．
代表的なものが，Chat Completion API と Assistant API になる．
その名の通りで，チャットのみを行うなら実装が容易な前者だが，Assistantをカスタマイズし拡張性を持たせようとすると専門的な実装を行う後者になる．
Assistant API が行える Assistant Tool の実装に手間取ったので，備忘録として実装を交えながら書き残す．
- 参考
Assistant Tool
Assistant API を使用して作成できる Assistant を拡張できる機能．
ファイルの中を検索させたり，Python を実行したり，外部サーバーと通信させたりできる．
File Search
いわゆるRAG(Retrieval-Augmented Generation)と呼ばれるファイル検索を行う機能．
ファイル検索は、独自の製品情報やユーザーから提供されたドキュメントなど、モデル外の知識でアシスタントを補強します。 OpenAIは自動的にドキュメントを解析してチャンクし、埋め込みを作成して保存し、ベクトル検索とキーワード検索の両方を使用して、ユーザーのクエリに答えるために関連するコンテンツを取得します。
対応ファイルフォーマットはプログラミングコードからPDF,tex,docx,pptxなどもいける．
基本的には文字情報が載っている系のものが多く，ファイルフォーマットも画像ファイルには対応していない．
多分PDFもフォント埋め込みを前提としている気がする．
Code Interpreter
プログラミングコードの生成・実行が可能になる機能
Code Interpreterは、アシスタントがサンドボックス化された実行環境でPythonコードを記述し、実行することを可能にします。 このツールは、多様なデータやフォーマットのファイルを処理し、データやグラフの画像を含むファイルを生成することができます。 Code Interpreterを使用すると、アシスタントはコードを繰り返し実行して、難しいコードや数学の問題を解くことができます。 アシスタントが実行に失敗するコードを記述した場合、コードの実行が成功するまで別のコードを実行して、このコードを繰り返し実行することができます。
対応ファイルフォーマットはFileSearchに加えて画像系(jpg,gif,png)がある上にzipまでいけるっぽい．
Function Calling
GPT が状況に応じて特定の外部のシステムとやりとりを行う機能（ユーザの入力（プロンプトの内容）から引数っぽいものを探索する機能付き）1
(e.g.) 天気取得APIを予め設定することで，天気の情報を持っていないため通常は天気を訊かれても答えられない ChatGPT に天気を返答させることができる．
ユーザの問いかけに答えるには設定したFunctionが有用であることを判断し，そのFunctionの実行のための引数の推測と Function の実行・結果の口語的設計などを一括して行う．
Chat Completions API (※Assistant API 以前のAPI)と同様に、Assistant API は Function Calling をサポートしています。
Function Calling では、 Assistants API に関数を記述し、呼び出す必要のある関数を引数とともにインテリジェントに返すことができます。
GUI (API PlayGround ではなく Web版) と API
fileのアップロード
ChatGPT を GUI で使っていると📎でなんでもアップロードするが，API的には，いくつか種類がある．
とりわけ躓いたのは，画像ファイルの取り扱い．
「この写真に描かれているものはなんですか？」というような，内部的には（おそらく）画像埋め込みモデルが動作するような message を送る のであればになるので，ChatGPT には message に付随させる必要がある．
よって，file の purpose は vision になる．
しかし，「この画像を文字起こししてください」「これらの画像を読み取ってください」「写真からCSVに変換してください」などの文字起こし系や，「回転させてください」「縮小してください」「背景を抜き取ってください」などの画像処理系のような 特定の専門的な動作系(OpenCVやTesseract)を使用するような message を送る のであれば，内部的に使用するのは（おそらく）Code Interpreter になるので，Assistant のAssistant Tool に Code Interpreter 部分に付随させる必要がある．
よって，file の purpose は assistants になる．
以上から，message を送られた ChatGPT が Python を動作させるか否かで，file の purpose を vision か assistant に調整するのみならず，file を付帯させる場所も assistant tool か message かで調整する必要がある．
実装
packageの準備
! pip install --upgrade pip
! pip install --upgrade openai
! pip freeze > requirements.txt
API準備
import os
import openai
from dotenv import load_dotenv
# .envファイルからAPIキーを読み込む
load_dotenv()
client = openai.OpenAI()
ファイル系処理
# ファイル名のリストを指定する
file_names = ['hoge.png','fuga.png']
# アップロード済みのファイルのリストを取得する
files = client.files.list()
uploaded_files = {f.filename: f.id for f in files.data}
# ファイルIDのリストを格納するための配列を初期化
file_ids = []
# 各ファイル名について処理を行う
for file_name in file_names:
# ファイルパスを指定する
image_dir = './../hoge/fuga'
filepath = os.path.join(image_dir, file_name)
# 同名ファイルのIDを取得
file_id = uploaded_files.get(file_name)
if file_id:
# 同名ファイルがある場合、そのファイルを取得
file = client.files.retrieve(file_id)
print('retrieve')
else:
# ファイルをアップロードする
file = client.files.create(
file=open(filepath, 'rb'),
purpose='assistants'
)
file_id = file.id
print('create')
# ファイルIDを配列に追加
file_ids.append(file_id)
# 最終的なファイルIDの配列を出力
print('File IDs:', file_ids)
Assistant作成
# list
assistants_list = client.beta.assistants.list(
order='desc',
# limit='20',
)
# print([asst.name for asst in assistants_list.data])
# 作成したい名前
new_name = 'test_fig-fit'
# 同じ名前のものが存在するかチェック
existing_assistant = next((assistant for assistant in assistants_list.data if assistant.name == new_name), None)
if existing_assistant:
# 同名のAssistantが存在する場合、そのAssistantを取得
assistant = client.beta.assistants.retrieve(existing_assistant.id)
print('retrieve')
else:
# 同名のAssistantが存在しない場合、新規作成
assistant = client.beta.assistants.create(
# instructions='You are a personal math tutor. When asked a question, write and run Python code to answer the question.',
name=new_name,
# tools=[{'type': 'code_interpreter'}],
model='gpt-4o',
# file_ids=[file.id for file in files],
)
print('create')
print(assistant)
thread関連
# Thread作成
thread = client.beta.threads.create()
print(thread)
thread_messages = client.beta.threads.messages.list(thread.id)
thread_messages.data
messageを削除したいとき
# deleted_message = client.beta.threads.messages.delete(
# message_id='msg_xxxxxxxx',
# thread_id=thread.id,
# )
message関連
theme = ''
instruction='''
あなたはMBTI診断で仲介者（INFP）のような性格の人です。仲介者（INFP）は控えめ、または静かそうに見えるかもしれませんが、心の中は情熱であふれ、生き生きとしている人です。独創的かつ想像力豊かなので、色々な空想をしながら、さまざまな会話やストーリを作り上げることが好きなタイプです。繊細な気質の持ち主として知られていて、音楽、芸術、自然、そして周りの人に対して、深く感情的に反応する人です。
'''
instruction = instruction.replace('$theme', theme)
print(instruction)
# Message作成
# thread_message = client.beta.threads.messages.create(
# thread.id,
# role='user',
# content=[
# {'type': 'text', 'text': instruction},
# {'type': 'image_file', 'image_file':{'file_id': file_ids[0]}},
# ],
# )
# 初期のコンテンツに指示文を追加
content = [{'type': 'text', 'text': instruction}]
# 最大で10回までのループにする
for file_id in file_ids[:10]:
content.append({'type': 'image_file', 'image_file': {'file_id': file_id}})
# contentリストを使ってメッセージを作成
thread_message = client.beta.threads.messages.create(
thread.id,
role='user',
content=content,
)
print(thread_message)
Run関連
import time
# Run を作成する
run = client.beta.threads.runs.create(
thread_id=thread.id,
assistant_id=assistant.id,
)
# Run のステータスが`queued`もしくは`in_progress`の状態では待機
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)
print(run.status)
# Thread 上の message のリストを降順で取得する
messages = client.beta.threads.messages.list(
thread_id=thread.id,
order='desc'
# order='asc'
)
# Assistant によって追加された 内容を取得する
assistant_messages = []
for thread_message in messages.data:
role = thread_message.role
if role == 'assistant':
for content in thread_message.content:
content = content.text.value
assistant_messages.append({'role': role, 'content': content})
break
出力関連
messages.data
assistant_messages
print(theme)
print(file_names)
print(thread.id)
print(assistant_messages[0]['content'])
import os
import json
from datetime import datetime
# # Example variables
# theme = 'Example Theme'
# file_names = ['file1.txt', 'file2.txt']
# thread = type('Thread', (object,), {'id': 12345})()
# assistant_messages = [{'content': 'This is a message from the assistant'}]
# Create the logs directory if it doesn't exist
log_dir = './logs'
os.makedirs(log_dir, exist_ok=True)
# Prepare the log data
log_data = {
'theme': theme,
'file_names': file_names,
'thread_id': thread.id,
'assistant_message': assistant_messages[0]['content']
}
# Get the current timestamp and create a filename
current_time = datetime.now().strftime('%Y%m%d_%H%M%S')
log_file_path = os.path.join(log_dir, f'{current_time}.json')
# Write the log data to a JSON file
with open(log_file_path, 'w', encoding='utf-8') as log_file:
json.dump(log_data, log_file, ensure_ascii=False, indent=4)
print(f'Log saved to {log_file_path}')