GPT-4oとStreamlitでOpenAI Assistants APIのCode Interpreterを検証した現状と課題

はじめに

OpenAIのAssistants APIをそのまま使用することで、自前でLangChainのエージェントなどを使用して同様の処理を実装する手間を省け、非常に便利です。ただ、現状（2024/05/18）ではまだβ版ということもあり、APIのインタフェースの改変も多く見られます。

Assitants APIを用いたcode-interpreterのUIをstreamlitで実装
においても、実装例が紹介されていますが、そのままでは動作しないこともあり、最新版での動作検証も兼ねてStreamlitでの実装例を紹介します。
また、本記事ではStreaming対応済みの実装を取り入れており、よりリアルタイムな対話が可能となっています。
扱っているモデルは2024/05/14に発表されたGPT-4oを用いています。

目次

1. はじめに
2. 実装例
   • app.py
   • openai_handler.py
3. デモ
   • 実行手順
   • ファイルの生成
   • ファイルの生成に関して
   • ファイルの読み取り
   • ファイルの読み取りに関して
4. まとめ

実装例

以下に、Streamlit（Python）を用いた実装例を示します。
Github Repositoryはこちらになります。

1. 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"])

2. 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

デモ

実行手順

1. 上記のリポジトリで、make runを実行
2. 下記の画面に遷移する
   

ファイルの生成

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を用いた実装の参考になれば幸いです。