はじめに
OpenAIのAssistants APIをそのまま使用することで、自前でLangChainのエージェントなどを使用して同様の処理を実装する手間を省け、非常に便利です。ただ、現状（2024/05/18）ではまだβ版ということもあり、APIのインタフェースの改変も多く見られます。
Assitants APIを用いたcode-interpreterのUIをstreamlitで実装 においても、実装例が紹介されていますが、そのままでは動作しないこともあり、最新版での動作検証も兼ねてStreamlitでの実装例を紹介します。
また、本記事ではStreaming対応済みの実装を取り入れており、よりリアルタイムな対話が可能となっています。扱っているモデルは2024/05/14に発表されたGPT-4oを用いています。
実装例
以下に、Streamlit（Python）を用いた実装例を示します。
Github Repositoryはこちらになります。
app.py
import streamlit as st from openai import OpenAI import openai_handler st.title("Assistant API Code Interpreter") client = OpenAI() with st.form("form", clear_on_submit=False): user_question = st.text_area("文章を入力") file = [st.file_uploader("ファイルをアップロード", accept_multiple_files=False)] or None submitted = st.form_submit_button("送信") if submitted: st.session_state["thread"], st.session_state["stream"] = openai_handler.submit_message( user_question, file ) openai_handler.wait_on_stream(st.session_state["stream"], st.session_state["thread"])
openai_handler.py
import time from os.path import dirname, join from typing import ( Any, Iterable, Literal, Optional, Tuple, TypedDict, ) import streamlit as st from dotenv import load_dotenv from openai import AssistantEventHandler, OpenAI from openai.lib.streaming._assistants import AssistantStreamManager from openai.pagination import SyncCursorPage from openai.types.beta.thread import Thread from openai.types.beta.thread_create_params import ( Message as CreateMessage, ) from openai.types.beta.thread_create_params import ( MessageAttachment, ) from openai.types.beta.threads import ( Message, Run, TextContentBlock, ) from openai.types.beta.threads.image_file import ImageFile from openai.types.beta.threads.text import Text from streamlit.runtime.uploaded_file_manager import ( UploadedFile, ) from typing_extensions import override class EventHandler(AssistantEventHandler): @override def on_text_created(self, text: Text) -> None: print("\nassistant > ", end="", flush=True) @override def on_text_delta(self, delta: Any, snapshot: Any) -> None: print(delta.value, end="", flush=True) @override def on_image_file_done(self, image_file: ImageFile) -> None: print("on_image_file_done image_file id:", image_file.file_id) st.image(get_file(image_file.file_id)) @override def on_end(self) -> None: print("on_end") if "thread" in st.session_state: thread = st.session_state["thread"] pretty_print(get_response(thread)) def on_tool_call_created(self, tool_call: Any) -> None: print(f"\nassistant > {tool_call.type}\n", flush=True) def on_tool_call_delta(self, delta: Any, snapshot: Any) -> None: if delta.type == "code_interpreter": if delta.code_interpreter.input: print(delta.code_interpreter.input, end="", flush=True) if delta.code_interpreter.outputs: print("\n\noutput >", flush=True) for output in delta.code_interpreter.outputs: if output.type == "logs": print(f"\n{output.logs}", flush=True) dotenv_path = join(dirname(__file__), ".env.local") load_dotenv(dotenv_path) client = OpenAI() # IF: https://platform.openai.com/docs/assistants/how-it-works/creating-assistants assistant = client.beta.assistants.create( name="汎用アシスタント", instructions="あなたは汎用的なアシスタントです。質問には簡潔かつ正確に答えてください。", tools=[{"type": "code_interpreter"}], model="gpt-4o", ) ASSISTANT_ID = assistant.id global_messages: list[Any] = [] class CustomMessage(TypedDict): role: Literal["user", "assistant"] content: str attachments: Optional[Iterable[MessageAttachment]] metadata: Optional[Any] # If no file is uploaded, the 'files' variable is assigned a list containing a single 'None' value. def submit_message( user_message: str, files: Optional[list[Optional[UploadedFile]]] = None, assistant_id: str = ASSISTANT_ID, ) -> Tuple[Thread, AssistantStreamManager[EventHandler]]: print("assistant_id:", assistant_id) print("user_message:", user_message) print("files:", files) with st.chat_message("user"): st.write(user_message) if files is None: files = [None] file_ids = submit_file(files) if files[0] is not None else [] messages: list[CustomMessage] = [ {"role": "user", "content": user_message, "attachments": None, "metadata": None} ] if len(file_ids) > 0: messages[0]["attachments"] = [ { "file_id": file_ids[0], "tools": [{"type": "code_interpreter"}], } ] # IFは修正される可能性があるため、下のURLを確認する # https://platform.openai.com/docs/assistants/how-it-works/managing-threads-and-messages _messages: Iterable[CreateMessage] = [CreateMessage(**msg) for msg in messages] thread = client.beta.threads.create(messages=_messages) print("thread_id:", thread.id) stream = client.beta.threads.runs.stream( thread_id=thread.id, assistant_id=assistant.id, instructions="ユーザーのメッセージと同じ言語で回答してください。回答を生成する際はユーザーへの確認は不要です。", event_handler=EventHandler(), ) return thread, stream def submit_file(files: list[Optional[UploadedFile]]) -> list[str]: if files: ids = [] for file in files: if file is not None: # IF: https://platform.openai.com/docs/assistants/how-it-works/creating-assistants _file = client.files.create( file=file.read(), purpose="assistants", ) ids.append(_file.id) return ids else: return [] def get_response(thread: Thread) -> SyncCursorPage[Message]: return client.beta.threads.messages.list(thread_id=thread.id, order="asc") def pretty_print(messages: SyncCursorPage[Message]) -> None: for m in messages: print("role:", m.role) print("content:", m.content) if m.role == "assistant": for content in m.content: cont_dict = content.model_dump() if cont_dict.get("text") is not None and isinstance(content, TextContentBlock): message_content = content.text annotations = message_content.annotations files = [] for ( index, annotation, ) in enumerate(annotations): message_content.value = message_content.value.replace( annotation.text, f" [{index}]", ) if file_path := getattr( annotation, "file_path", None, ): files.append( ( file_path.file_id, annotation.text.split("/")[-1], ) ) for file in files: st.download_button( f"{file[1]} : ダウンロード", get_file(file[0]), file_name=file[1], ) def wait_on_run(run: Run, thread: Thread) -> Run: while run.status == "queued" or run.status == "in_progress": print("wait_on_run", run.id, thread.id) run = client.beta.threads.runs.retrieve( thread_id=thread.id, run_id=run.id, ) print("run.status:", run.status) time.sleep(0.5) return run def wait_on_stream(stream: AssistantStreamManager[EventHandler], thread: Thread) -> None: with st.chat_message("assistant"): with stream as s: st.write_stream(s.text_deltas) s.until_done() def get_file(file_id: str) -> bytes: retrieve_file = client.files.with_raw_response.content(file_id) content: bytes = retrieve_file.content return content
デモ
実行手順
- 上記のリポジトリで、'make run' を実行します。
- 下記の画面に遷移します。
ファイルの生成
1. PDFの生成
指示文：「任意のPDF資料を作成してください。マーケティング関連のビジネス企画書でお願いします。」
・1回目生成失敗
・補足：他にも稀に文字化けしたPDFが生成されてしまうことがある。（日本語の文字エンコードを正しく指定できていない。）
・2回目：生成されたPDFファイル
＜結果＞
生成に失敗する場合もあり不安定に感じる。生成されたファイル自体には特に問題はない。
2. Wordの生成
指示文：「任意のワード資料を作成してください。マーケティング関連のビジネス企画書でお願いします。」
・生成されたワードファイル
＜結果＞
文字がメインでフォーマットも特に綺麗に整理されてはいないが、ワードファイル自体は問題なく生成されている。
3.Excelの生成
指示文：「任意のエクセル資料を作成してください。マーケティング関連の帳票でお願いします。」
・生成されたエクセルファイル
＜結果＞
サンプルデータとして特に問題のないエクセルファイルが生成されました。
4.PowerPointの生成
指示文：「任意のパワーポイント資料を作成してください。マーケティング関連のビジネスプレゼンテーションでお願いします。」
・生成されたパワーポイントファイル
＜結果＞
文字がメインでフォーマットも特に綺麗に整理されてはいないが、パワーポイント自体は問題なく生成されました。
5.CSVの生成
指示文：「任意のCSV資料を作成してください。マーケティング関連の帳票でお願いします。」
・生成されたCSVファイル
＜結果＞
・サンプルデータとして特に問題のないCSVファイルが生成されました。
6.PNG画像の生成
指示文：「任意のPNG画像を作成してください。マーケティング関連のビジネスに関するものでお願いします。」
・生成されたPNG画像
＜結果＞
テキストが記載されたシンプルな画像。文字が若干はみ出している点が気になる。
ファイルの生成に関して
全体的にPDF、ワード、エクセル、パワーポイント、CSV、PNG画像全て生成に成功しています。ただ、PDFのみ一度生成に失敗しており、若干不安定なところは目立ちました（日本語の文字コードを正しく指定できず文字化けすることもありました。）。また、PNG画像は生成に成功したと言ってもテキストが画像化されているだけで、DALLE等で生成される写真のような画像は取得できませんでした。
結論としてテキストベースで資料を作成したり、ダミーのデータを作成するにはCode Interpreterは問題なく使えると思います。
ファイルの読み取り
上記で生成したファイルを元に読み取りのタスクを行なってみます。
1.PDFの読み取り
指示文：「こちらのファイルの内容を全て抽出して説明してください。」
＜結果＞
初め文字化けが指摘されたが二度目のトライで抽出に成功しました。
2.Wordの読み取り
指示文：「こちらのファイルの内容を全て抽出して説明してください。」
＜結果＞
ファイルの中身は抽出できているが、中身を抽出するまでのプロセスが若干冗長に感じられる。
3.エクセルの読み取り
指示文：「こちらのファイルのデータはどういった内容が含まれているかを説明し、グラフで示してください。」
＜結果＞
注意点として、エクセルの中身に日本語が混ざっているとグラフの生成に失敗します。上記で生成したエクセルには日本語が入っていたので英語に直したもので読み取りを行っています。また、エクセルであることを判定するまでの工程が冗長に思われます。グラフ表示は問題ありませんでした。(修正後エクセルファイル)
4.PowerPointの読み取り
指示文：「こちらのファイルの`全スライド`の内容を全て抽出して説明してください。」
＜結果＞
問題なく各スライドの中身を抽出できている。ただ、スライドという指示分を足さないとzipを解凍して、ディクレトリを探索しながらそれぞれのxmlファイルを詳しく解説し始めてしまう点がある。
5.CSVの読み取り
こちらのファイルのデータはどういった内容が含まれているかを説明し、グラフで示してください。
＜結果＞
最初エクセルで解析しようとするが失敗し、その後CSVであることを特定できている。また、上記のエクセルと同様に日本語が混ざっていると抽出に失敗するため、上記でCSVとして生成したものは英語に直したもので読み取りを行っています。（修正後CSVファイル）
6.PNG画像の読み取り
指示文：「こちらのファイルのデータはどういった内容が含まれているかを説明してください。」
＜結果＞
敢えてファイルのデータと指定しましたが、画像データであることを特定して中身の説明まで上手くできています。
ファイルの読み取りに関して
全体的にどのファイルのデータも若干プロンプトの調整が必要な箇所はありますが中身をうまく抽出できています。ただ、CSVやExcelでグラフの可視化をする際に日本語が混ざっていると表示に失敗し、英語に変換する必要があるのは残念でした。ただ以前よりも抽出の精度は上がっているので今後の改善に期待です。
まとめ
この記事では、OpenAIのAssistants APIを用いたCode Interpreterの現状と課題について、Streamlitを使用して検証しました。以下が主要なポイントです。
1.実装例
Streamlitを使用して、OpenAI Assistants APIを統合したCode Interpreterの実装方法を紹介しました。
2.ファイル生成のデモ
PDF、ワード、エクセル、パワーポイント、CSV、PNG画像の生成タスクを実施し、全体的に成功しました。ただし、PDFの生成に関しては一度失敗することがあり、特に日本語の文字エンコードの問題が見られました。
3.ファイル読み取りのデモ
生成したファイルを読み取り、その内容を抽出するタスクを実施しました。各形式のファイルについて適切に内容を抽出できましたが、CSVやExcelでのグラフの可視化には、日本語が含まれていると表示に失敗する場合がありました。
4.課題と改善点
- PDFの生成において日本語の文字エンコードに問題がある点。
- CSVやExcelのグラフ可視化において、日本語が混ざると表示に失敗する点。
- ファイルの読み取りにおいてプロンプトの調整が必要な場合がある点。
全体として、OpenAIのAssistants APIは多様な形式のファイル生成と読み取りにおいて強力なツールであり、今後の改善によりさらに高い精度と安定性が期待されます。この記事が、OpenAI Assistants APIを用いた実装の参考になれば幸いです。
記事執筆者
梅本 誠也
パーソルキャリア株式会社
テクノロジー本部 デジタルテクノロジー統括部 デジタルソリューション部 クライアントエンジニアグループ
韓国で5年間正規留学し、その間に業務委託で機械学習とデータエンジニアリング方面の開発を経験。新卒でアプリケーションエンジニアとしてフロントエンド、バックエンド、インフラを幅広く経験。パーソルキャリア入社後はデータエンジニアとして、社内のデータ分析基盤の構築と運用保守を担当。一方で、生成系AIを用いたアプリケーション開発にも携わっている。
