OpenAI Assistants API入門：AIアシスタント開発の決定版ガイド

OpenAI Assistants APIの基本フロー：Assistant作成からThread・Run実行まで

Assistants APIを使ってステートフルな会話エージェントを構築する基本フローを実装します。Assistant定義 → Thread作成 → Message追加 → Run実行 → 結果取得という一連のステップを確認しましょう。

from openai import OpenAI
import time

client = OpenAI()

# ステップ1: Assistantの作成（ツール付き）
assistant = client.beta.assistants.create(
    name="Data Analysis Helper",
    instructions="""あなたはデータ分析の専門家です。
ユーザーの質問に対して、分かりやすく丁寧に回答してください。
必要に応じてCode Interpreterを使い、計算やグラフ生成を行ってください。
回答は日本語でお願いします。""",
    model="gpt-4o",
    tools=[{"type": "code_interpreter"}]
)
print(f"Assistant作成完了: {assistant.id}")

# ステップ2: Threadの作成（会話セッション）
thread = client.beta.threads.create()
print(f"Thread作成完了: {thread.id}")

# ステップ3: Messageの追加
message = client.beta.threads.messages.create(
    thread_id=thread.id,
    role="user",
    content="1から100までの素数を列挙し、その個数と合計を教えてください。"
)
print(f"Message追加完了: {message.id}")

# ステップ4: Runの実行
run = client.beta.threads.runs.create(
    thread_id=thread.id,
    assistant_id=assistant.id
)
print(f"Run開始: {run.id} (status={run.status})")

# ステップ5: Run完了をポーリングで待機
while run.status in ["queued", "in_progress"]:
    time.sleep(1)
    run = client.beta.threads.runs.retrieve(
        thread_id=thread.id,
        run_id=run.id
    )
    print(f"  status: {run.status}")

if run.status == "completed":
    # ステップ6: Assistantの応答メッセージを取得
    messages = client.beta.threads.messages.list(thread_id=thread.id)
    for msg in reversed(messages.data):
        role = msg.role
        for block in msg.content:
            if block.type == "text":
                print(f"\n[{role}] {block.text.value}")
elif run.status == "failed":
    print(f"Run失敗: {run.last_error}")

# 後続メッセージの追加（同一Thread内で会話継続）
follow_up = client.beta.threads.messages.create(
    thread_id=thread.id,
    role="user",
    content="その素数をヒストグラムで可視化してください。"
)
run2 = client.beta.threads.runs.create(
    thread_id=thread.id,
    assistant_id=assistant.id
)
while run2.status in ["queued", "in_progress"]:
    time.sleep(1)
    run2 = client.beta.threads.runs.retrieve(
        thread_id=thread.id, run_id=run2.id
    )
print(f"\nFollow-up完了: status={run2.status}")

# クリーンアップ（不要になったAssistantを削除）
client.beta.assistants.delete(assistant.id)
print("Assistant削除完了")

Assistants APIのポイントは3つあります。第一に、Threadが会話状態を保持するため、開発者側でメッセージ履歴を管理する必要がなく、follow-upメッセージを追加するだけで文脈を維持した会話が継続できます。第二に、Code Interpreterツールを有効にすることで、Assistantが自律的にPythonコードを実行して計算やグラフ生成を行えるため、データ分析ユースケースに強力です。第三に、Runは非同期で実行されるため、ポーリングまたはストリーミングで完了を検知する設計が必要です。本番環境ではclient.beta.threads.runs.create_and_poll()を使うとポーリングコードを簡略化できます。