はじめに

OpenAI Codexは、ChatGPTのWebインターフェースやCodex CLIから利用できるだけでなく、Responses APIを通じてプログラマティックに呼び出すことも可能です。APIを使用することで、開発ワークフローの自動化、CI/CDパイプラインへの統合、カスタムツールの構築など、より高度なユースケースを実現できます。

本記事では、Codex APIの基礎となるcodex-mini-latestモデルを使った自動化開発の方法を解説します。基本的なAPIリクエストの構造から、ストリーミングレスポンスの処理、そして本番環境で必須となるエラーハンドリングとリトライ戦略までを実践的に学びます。

codex-mini-latestモデルとは

codex-mini-latestは、Codex CLI向けに最適化されたo4-miniベースの推論モデルです。API経由でコード生成や分析タスクを実行する際に利用できます。

モデルの特徴

項目 値
コンテキストウィンドウ 200,000トークン
最大出力トークン 100,000トークン
知識カットオフ 2024年6月1日
入力料金 $1.50 / 1Mトークン
キャッシュ入力料金 $0.375 / 1Mトークン
出力料金 $6.00 / 1Mトークン

対応エンドポイント

codex-mini-latestは以下のエンドポイントで利用できます。

エンドポイント 対応状況
Responses API (v1/responses) 対応
Chat Completions (v1/chat/completions) 対応
Realtime API (v1/realtime) 対応
Assistants API (v1/assistants) 対応
Batch API (v1/batch) 対応

本記事では、OpenAIが推奨する最新のResponses APIを使用した実装を紹介します。

前提条件

Codex APIを利用するには、以下の準備が必要です。

要件 詳細
OpenAIアカウント API利用のためのアカウント登録
APIキー プロジェクト設定で生成したAPIキー
課金設定 従量課金のためのクレジット設定
Python環境 Python 3.10以降（本記事のサンプル実行用）

APIキーの取得

OpenAI Platformにログインし、プロジェクトのAPIキーを生成します。

1
2

# APIキーを環境変数に設定
export OPENAI_API_KEY="sk-proj-..."

Python SDKのインストール

OpenAIの公式Pythonライブラリをインストールします。

1

pip install openai

基本的なAPIリクエストの構造

Responses APIの概要

Responses APIは、OpenAIが提供する最新のモデル応答生成インターフェースです。テキストと画像の入力をサポートし、ファイル検索やWeb検索などのビルトインツールとの連携も可能です。

sequenceDiagram
    participant Client as クライアント
    participant API as Responses API
    participant Model as codex-mini-latest
    
    Client->>API: POST /v1/responses
    API->>Model: リクエスト転送
    Model-->>API: レスポンス生成
    API-->>Client: Response オブジェクト

最初のAPIリクエスト

Pythonを使用した基本的なリクエストの例を示します。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

from openai import OpenAI

client = OpenAI()

response = client.responses.create(
    model="codex-mini-latest",
    input="Pythonで素数判定を行う関数を作成してください。"
)

print(response.output_text)

このコードを実行すると、Codexがコードを生成して返します。

リクエストパラメータの詳細

Responses APIでは、様々なパラメータを指定してリクエストをカスタマイズできます。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

from openai import OpenAI

client = OpenAI()

response = client.responses.create(
    model="codex-mini-latest",
    input=[
        {
            "role": "user",
            "content": "TypeScriptで配列の重複を除去するユーティリティ関数を作成してください。"
        }
    ],
    instructions="あなたは熟練したソフトウェアエンジニアです。コードは必ずコメント付きで記述してください。",
    temperature=0.2,
    max_output_tokens=4096,
    store=True
)

# レスポンスの詳細を確認
print(f"Response ID: {response.id}")
print(f"Status: {response.status}")
print(f"Model: {response.model}")
print(f"Output: {response.output_text}")

主要なパラメータ

パラメータ 説明 デフォルト値
model 使用するモデルID 必須
input 入力テキストまたはメッセージ配列 必須
instructions システムメッセージ（開発者指示） なし
temperature 出力のランダム性（0-2） 1
max_output_tokens 最大出力トークン数 モデル依存
store レスポンスを保存するか true
stream ストリーミングを有効にするか false

レスポンスオブジェクトの構造

APIから返されるResponseオブジェクトには、生成結果に関する詳細情報が含まれています。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

from openai import OpenAI

client = OpenAI()

response = client.responses.create(
    model="codex-mini-latest",
    input="FizzBuzz問題をPythonで解いてください。"
)

# レスポンスオブジェクトの構造を確認
print(f"ID: {response.id}")
print(f"Object: {response.object}")
print(f"Created at: {response.created_at}")
print(f"Status: {response.status}")
print(f"Completed at: {response.completed_at}")

# 出力内容
for item in response.output:
    if item.type == "message":
        for content in item.content:
            if content.type == "output_text":
                print(f"Text: {content.text}")

# トークン使用量
print(f"Input tokens: {response.usage.input_tokens}")
print(f"Output tokens: {response.usage.output_tokens}")
print(f"Total tokens: {response.usage.total_tokens}")

ステータスの種類

レスポンスには以下のステータスが設定されます。

ステータス 説明
completed 正常に完了
failed エラーで失敗
in_progress 処理中
cancelled キャンセル済み
queued キュー待ち
incomplete 不完全な終了

マルチターン会話の実装

Responses APIでは、previous_response_idを使用して会話の状態を維持できます。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

from openai import OpenAI

client = OpenAI()

# 最初のリクエスト
response1 = client.responses.create(
    model="codex-mini-latest",
    input="Pythonでシンプルなクラスベースのカウンターを作成してください。"
)

print("=== 初回レスポンス ===")
print(response1.output_text)

# フォローアップリクエスト
response2 = client.responses.create(
    model="codex-mini-latest",
    input="このカウンターにリセット機能を追加してください。",
    previous_response_id=response1.id
)

print("\n=== フォローアップレスポンス ===")
print(response2.output_text)

# さらに続ける
response3 = client.responses.create(
    model="codex-mini-latest",
    input="最大値を設定できるようにしてください。最大値に達したらエラーを発生させます。",
    previous_response_id=response2.id
)

print("\n=== 追加機能レスポンス ===")
print(response3.output_text)

この方法により、前回の会話コンテキストを保持しながら、段階的にコードを改善していくことができます。

ストリーミングレスポンスの処理

長い応答を生成する場合、ストリーミングを使用することで、生成されたテキストをリアルタイムで受信できます。

基本的なストリーミング実装

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

from openai import OpenAI

client = OpenAI()

stream = client.responses.create(
    model="codex-mini-latest",
    input="Express.jsを使用したREST APIサーバーの完全な実装例を作成してください。",
    stream=True
)

# ストリーミングイベントを処理
for event in stream:
    print(event)

テキストデルタの抽出

実用的なアプリケーションでは、テキストの差分（デルタ）を抽出してリアルタイム表示します。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22

from openai import OpenAI

client = OpenAI()

stream = client.responses.create(
    model="codex-mini-latest",
    input="ReactでTodoアプリを作成してください。useState, useEffectを使用します。",
    stream=True
)

full_text = ""
for event in stream:
    # テキストデルタイベントを処理
    if event.type == "response.output_text.delta":
        delta_text = event.delta
        full_text += delta_text
        print(delta_text, end="", flush=True)
    
    # 完了イベント
    elif event.type == "response.completed":
        print("\n\n=== 生成完了 ===")
        print(f"Total tokens: {event.response.usage.total_tokens}")

ストリーミングイベントの種類

ストリーミング中に発生する主要なイベントは以下の通りです。

イベント 説明
response.created レスポンス生成開始
response.in_progress 処理中
response.output_text.delta テキストの差分出力
response.output_text.done テキスト出力完了
response.completed 全体の処理完了
error エラー発生

非同期ストリーミング

asyncioを使用した非同期実装も可能です。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

import asyncio
from openai import AsyncOpenAI

async def stream_response():
    client = AsyncOpenAI()
    
    stream = await client.responses.create(
        model="codex-mini-latest",
        input="Node.jsでファイルアップロード機能を実装してください。",
        stream=True
    )
    
    async for event in stream:
        if event.type == "response.output_text.delta":
            print(event.delta, end="", flush=True)
    
    print("\n完了")

# 実行
asyncio.run(stream_response())

エラーハンドリング

本番環境でAPIを使用する際は、適切なエラーハンドリングが必須です。

エラーの種類

OpenAI APIで発生する主なエラーは以下の通りです。

エラーコード 原因 対処法
401 認証エラー APIキーを確認
429 レート制限超過 リクエスト頻度を下げる
500 サーバーエラー リトライを実行
503 サービス過負荷 時間をおいてリトライ

基本的なエラーハンドリング

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45

import openai
from openai import OpenAI

client = OpenAI()

def generate_code(prompt: str) -> str | None:
    try:
        response = client.responses.create(
            model="codex-mini-latest",
            input=prompt
        )
        return response.output_text
    
    except openai.AuthenticationError as e:
        print(f"認証エラー: APIキーを確認してください - {e}")
        return None
    
    except openai.RateLimitError as e:
        print(f"レート制限エラー: リクエスト頻度を下げてください - {e}")
        return None
    
    except openai.APIConnectionError as e:
        print(f"接続エラー: ネットワーク設定を確認してください - {e}")
        return None
    
    except openai.APITimeoutError as e:
        print(f"タイムアウトエラー: リトライを検討してください - {e}")
        return None
    
    except openai.BadRequestError as e:
        print(f"リクエストエラー: パラメータを確認してください - {e}")
        return None
    
    except openai.InternalServerError as e:
        print(f"サーバーエラー: 時間をおいてリトライしてください - {e}")
        return None
    
    except openai.APIError as e:
        print(f"APIエラー: {e}")
        return None

# 使用例
result = generate_code("Pythonでバブルソートを実装してください。")
if result:
    print(result)

リトライ戦略

一時的なエラーに対しては、適切なリトライ戦略を実装することで、システムの信頼性を向上させます。

指数バックオフの実装

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56

import time
import random
import openai
from openai import OpenAI

def exponential_backoff_retry(
    func,
    max_retries: int = 5,
    base_delay: float = 1.0,
    max_delay: float = 60.0,
    jitter: bool = True
):
    """指数バックオフを使用したリトライ関数"""
    retries = 0
    delay = base_delay
    
    while retries < max_retries:
        try:
            return func()
        
        except (openai.RateLimitError, openai.APITimeoutError, 
                openai.InternalServerError) as e:
            retries += 1
            
            if retries >= max_retries:
                raise e
            
            # ジッターを追加してサンダリングハード問題を回避
            if jitter:
                delay = min(delay * 2, max_delay) + random.uniform(0, 1)
            else:
                delay = min(delay * 2, max_delay)
            
            print(f"リトライ {retries}/{max_retries}: {delay:.2f}秒後に再試行...")
            time.sleep(delay)
        
        except openai.APIError as e:
            # リトライ不可能なエラーは即座に例外を発生
            raise e
    
    return None

# 使用例
client = OpenAI()

def make_request():
    return client.responses.create(
        model="codex-mini-latest",
        input="クイックソートをPythonで実装してください。"
    )

try:
    response = exponential_backoff_retry(make_request)
    print(response.output_text)
except openai.APIError as e:
    print(f"リトライ後もエラー: {e}")

デコレータパターンによるリトライ

再利用可能なリトライロジックをデコレータとして実装します。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56

import time
import random
import functools
import openai

def retry_with_backoff(
    max_retries: int = 5,
    base_delay: float = 1.0,
    max_delay: float = 60.0,
    retryable_exceptions: tuple = (
        openai.RateLimitError,
        openai.APITimeoutError,
        openai.InternalServerError
    )
):
    """リトライ機能を提供するデコレータ"""
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            delay = base_delay
            last_exception = None
            
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except retryable_exceptions as e:
                    last_exception = e
                    if attempt < max_retries - 1:
                        sleep_time = delay + random.uniform(0, 1)
                        print(f"試行 {attempt + 1} 失敗: {sleep_time:.2f}秒後にリトライ")
                        time.sleep(sleep_time)
                        delay = min(delay * 2, max_delay)
            
            raise last_exception
        return wrapper
    return decorator

# デコレータの使用例
from openai import OpenAI

client = OpenAI()

@retry_with_backoff(max_retries=3)
def generate_with_retry(prompt: str) -> str:
    response = client.responses.create(
        model="codex-mini-latest",
        input=prompt
    )
    return response.output_text

# 実行
try:
    result = generate_with_retry("マージソートをPythonで実装してください。")
    print(result)
except openai.APIError as e:
    print(f"最終的にエラー: {e}")

実践的なコード生成の自動化

ここでは、Codex APIを使用した実践的なコード生成自動化の例を紹介します。

コードレビュー自動化

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42

from openai import OpenAI

client = OpenAI()

def review_code(code: str, language: str = "Python") -> str:
    """コードをレビューして改善点を提案する"""
    prompt = f"""以下の{language}コードをレビューしてください。
    
\`\`\`{language.lower()}
{code}
\`\`\`

以下の観点でレビューを行い、改善点を具体的に指摘してください：
1. バグや潜在的な問題
2. コードの可読性
3. パフォーマンスの改善点
4. ベストプラクティスへの準拠
5. 改善版のコード（必要な場合）
"""
    
    response = client.responses.create(
        model="codex-mini-latest",
        input=prompt,
        instructions="あなたは経験豊富なシニアエンジニアです。建設的なフィードバックを提供してください。",
        temperature=0.3
    )
    
    return response.output_text

# 使用例
sample_code = """
def find_duplicates(lst):
    duplicates = []
    for i in range(len(lst)):
        for j in range(i + 1, len(lst)):
            if lst[i] == lst[j] and lst[i] not in duplicates:
                duplicates.append(lst[i])
    return duplicates
"""

review_result = review_code(sample_code)
print(review_result)

テストコード自動生成

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40

from openai import OpenAI

client = OpenAI()

def generate_tests(code: str, framework: str = "pytest") -> str:
    """コードに対するテストを自動生成する"""
    prompt = f"""以下のPythonコードに対するテストを{framework}を使用して作成してください。

\`\`\`python
{code}
\`\`\`

要件：
- 正常系のテストケース
- エッジケースのテスト
- 異常系のテスト（エラーハンドリング）
- 境界値テスト
"""
    
    response = client.responses.create(
        model="codex-mini-latest",
        input=prompt,
        instructions="テストは網羅的で、実行可能なコードを出力してください。",
        temperature=0.2
    )
    
    return response.output_text

# 使用例
target_code = """
def calculate_discount(price: float, discount_percent: float) -> float:
    if price < 0:
        raise ValueError("Price cannot be negative")
    if discount_percent < 0 or discount_percent > 100:
        raise ValueError("Discount must be between 0 and 100")
    return price * (1 - discount_percent / 100)
"""

tests = generate_tests(target_code)
print(tests)

バッチ処理による大量タスクの実行

複数のコード生成タスクを効率的に処理する方法です。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40

import asyncio
from openai import AsyncOpenAI

async def generate_batch(prompts: list[str]) -> list[str]:
    """複数のプロンプトを並列処理する"""
    client = AsyncOpenAI()
    
    async def single_request(prompt: str) -> str:
        response = await client.responses.create(
            model="codex-mini-latest",
            input=prompt,
            temperature=0.3
        )
        return response.output_text
    
    # 並列実行
    tasks = [single_request(prompt) for prompt in prompts]
    results = await asyncio.gather(*tasks, return_exceptions=True)
    
    return results

# 使用例
async def main():
    prompts = [
        "Pythonでスタックを実装してください。",
        "Pythonでキューを実装してください。",
        "Pythonでリンクリストを実装してください。",
        "Pythonで二分探索木を実装してください。"
    ]
    
    results = await generate_batch(prompts)
    
    for i, result in enumerate(results):
        print(f"\n=== タスク {i + 1} ===")
        if isinstance(result, Exception):
            print(f"エラー: {result}")
        else:
            print(result[:500] + "..." if len(result) > 500 else result)

asyncio.run(main())

コスト管理のベストプラクティス

APIを本番環境で使用する際は、コスト管理が重要です。

トークン使用量の監視

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60

from openai import OpenAI
from dataclasses import dataclass

@dataclass
class UsageTracker:
    total_input_tokens: int = 0
    total_output_tokens: int = 0
    total_requests: int = 0
    
    # codex-mini-latestの料金（2026年1月時点）
    INPUT_COST_PER_1M = 1.50  # ドル
    OUTPUT_COST_PER_1M = 6.00  # ドル
    
    def add_usage(self, input_tokens: int, output_tokens: int):
        self.total_input_tokens += input_tokens
        self.total_output_tokens += output_tokens
        self.total_requests += 1
    
    def get_estimated_cost(self) -> float:
        input_cost = (self.total_input_tokens / 1_000_000) * self.INPUT_COST_PER_1M
        output_cost = (self.total_output_tokens / 1_000_000) * self.OUTPUT_COST_PER_1M
        return input_cost + output_cost
    
    def report(self):
        print(f"総リクエスト数: {self.total_requests}")
        print(f"総入力トークン: {self.total_input_tokens:,}")
        print(f"総出力トークン: {self.total_output_tokens:,}")
        print(f"推定コスト: ${self.get_estimated_cost():.4f}")

# 使用例
client = OpenAI()
tracker = UsageTracker()

def tracked_request(prompt: str) -> str:
    response = client.responses.create(
        model="codex-mini-latest",
        input=prompt
    )
    
    tracker.add_usage(
        response.usage.input_tokens,
        response.usage.output_tokens
    )
    
    return response.output_text

# 複数リクエストを実行
prompts = [
    "Hello Worldをpythonで出力してください。",
    "FizzBuzzをPythonで実装してください。",
    "バイナリサーチをPythonで実装してください。"
]

for prompt in prompts:
    result = tracked_request(prompt)
    print(f"生成完了: {len(result)} 文字")

# レポート出力
print("\n=== 使用量レポート ===")
tracker.report()

プロンプトキャッシュの活用

同じプロンプトプレフィックスを繰り返し使用する場合、キャッシュを活用してコストを削減できます。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

from openai import OpenAI

client = OpenAI()

# prompt_cache_keyを設定して類似リクエストのキャッシュヒット率を向上
response = client.responses.create(
    model="codex-mini-latest",
    input="Pythonでユーティリティ関数を作成してください。",
    prompt_cache_key="user_123_utility_functions",
    prompt_cache_retention="24h"  # 24時間キャッシュを保持
)

# キャッシュされたトークンを確認
cached = response.usage.input_tokens_details.cached_tokens
print(f"キャッシュされたトークン: {cached}")

まとめ

本記事では、Codex APIのcodex-mini-latestモデルを使用した自動化開発の基礎を解説しました。

学んだ内容

  • Responses APIの基本的なリクエスト構造とパラメータ
  • マルチターン会話による段階的なコード改善
  • ストリーミングレスポンスによるリアルタイム出力
  • 本番環境で必須のエラーハンドリング手法
  • 指数バックオフを使用したリトライ戦略
  • コスト管理とトークン使用量の監視

次のステップ

Codex APIの基礎を習得したら、次は以下のトピックに進むことをお勧めします。

  • カスタム開発ワークフローの構築: Issue自動トリアージ、PRレビュー自動化などの実装
  • CI/CDパイプラインへの統合: GitHub ActionsやJenkinsとの連携
  • Function Callingの活用: 外部システムとの連携によるエージェント機能の拡張

Codex APIを活用することで、開発プロセスの自動化を実現し、より創造的な作業に時間を使えるようになります。

参考リンク