ChatGPT APIをPythonで連携させ、実務アプリケーションに組み込む方法

本記事では、ChatGPT APIをPythonで実装し、実際のビジネスアプリケーションに組み込むための具体的な手順を解説します。認証設定からエラーハンドリング、レート制限対策まで、すぐに本番環境で使える実装パターンを習得できます。

ChatGPT APIの基本セットアップ

ChatGPT APIを使い始めるには、まずOpenAIのアカウント登録とAPIキーの取得が必須です。以下の手順に従ってセットアップを進めてください。

APIキーの取得と環境変数の設定

OpenAI公式ページからAPIキーを生成したら、環境変数として設定します。以下のコマンドで、システムにAPIキーを登録してください。

# macOS / Linux の場合
export OPENAI_API_KEY='your-api-key-here'

# Windowsの場合（PowerShell）
$env:OPENAI_API_KEY='your-api-key-here'

# または .env ファイルに記述（推奨）
# .envファイルの内容
OPENAI_API_KEY=your-api-key-here
  

# 公式OpenAI Pythonライブラリのインストール
pip install openai python-dotenv

# バージョン確認（2024年以降の新APIに対応）
pip show openai
  

ChatGPT APIの基本的な使い方は、ユーザーの質問をメッセージ形式で送信し、AIの応答を受け取るパターンです。以下が最小限の実装例です。

import os
from openai import OpenAI
from dotenv import load_dotenv

# .envファイルから環境変数を読み込む
load_dotenv()

# OpenAIクライアントの初期化
client = OpenAI(api_key=os.getenv('OPENAI_API_KEY'))

def chat_with_gpt(user_message):
    """
    ChatGPT APIにメッセージを送信し、応答を取得する基本関数
    """
    response = client.chat.completions.create(
        model="gpt-4o-mini",  # または gpt-4, gpt-3.5-turbo など
        messages=[
            {"role": "system", "content": "あなたは親切なアシスタントです。"},
            {"role": "user", "content": user_message}
        ],
        temperature=0.7,  # 0-2の範囲で、高いほど創造的
        max_tokens=500     # 最大出力トークン数
    )
    
    # レスポンスからテキストを抽出
    return response.choices[0].message.content

# 使用例
if __name__ == "__main__":
    user_input = "Pythonでリスト内包表記を説明してください"
    result = chat_with_gpt(user_input)
    print(f"ユーザー: {user_input}")
    print(f"ChatGPT: {result}")
  

このコードを実行すると、ChatGPT APIに質問が送信され、応答が標準出力に表示されます。

複数ターンの会話履歴を維持する実装

より実用的なチャットボットを構築するには、会話履歴を保持し、文脈を維持する必要があります。以下は会話の履歴を管理する実装例です。

from openai import OpenAI
import os
from dotenv import load_dotenv

load_dotenv()
client = OpenAI(api_key=os.getenv('OPENAI_API_KEY'))

class ChatBot:
    def __init__(self, system_prompt=""):
        """
        ChatBotインスタンスを初期化
        system_prompt: AIの振る舞いを定義するシステムプロンプト
        """
        self.conversation_history = []
        if system_prompt:
            self.conversation_history.append({
                "role": "system",
                "content": system_prompt
            })
    
    def add_message(self, role, content):
        """
        会話履歴にメッセージを追加
        role: "user" または "assistant"
        """
        self.conversation_history.append({
            "role": role,
            "content": content
        })
    
    def get_response(self, user_message, temperature=0.7):
        """
        ユーザーメッセージに対する応答を生成
        """
        # ユーザーメッセージを履歴に追加
        self.add_message("user", user_message)
        
        try:
            response = client.chat.completions.create(
                model="gpt-4o-mini",
                messages=self.conversation_history,
                temperature=temperature,
                max_tokens=800
            )
            
            # AIの応答を履歴に追加
            assistant_message = response.choices[0].message.content
            self.add_message("assistant", assistant_message)
            
            return assistant_message
        
        except Exception as e:
            print(f"APIエラーが発生しました: {e}")
            return None
    
    def clear_history(self):
        """会話履歴をクリア（システムプロンプトは保持）"""
        self.conversation_history = self.conversation_history[:1]

# 使用例
if __name__ == "__main__":
    # カスタマーサポートボットの例
    bot = ChatBot(system_prompt="あなたはカスタマーサポートの専門家です。丁寧に対応してください。")
    
    # 複数ターンの会話
    questions = [
        "配送料金は無料ですか？",
        "返品期限は何日ですか？",
        "支払い方法は何がありますか？"
    ]
    
    for q in questions:
        print(f"ユーザー: {q}")
        answer = bot.get_response(q)
        print(f"ボット: {answer}\n")
  

エラーハンドリングとレート制限への対応

一般的なエラーパターンと解決策

ChatGPT APIを運用する際、いくつかの一般的なエラーが発生します。以下はそれらへの対応実装です。

from openai import OpenAI, APIError, RateLimitError, APIConnectionError
import os
from dotenv import load_dotenv
import time

load_dotenv()
client = OpenAI(api_key=os.getenv('OPENAI_API_KEY'))

def call_gpt_with_retry(message, max_retries=3, backoff_factor=2):
    """
    リトライ機能付きのAPI呼び出し
    max_retries: 最大リトライ回数
    backoff_factor: 指数バックオフの係数
    """
    for attempt in range(max_retries):
        try:
            response = client.chat.completions.create(
                model="gpt-4o-mini",
                messages=[{"role": "user", "content": message}],
                temperature=0.7,
                max_tokens=500
            )
            return response.choices[0].message.content
        
        except RateLimitError:
            # レート制限に達した場合
            wait_time = backoff_factor ** attempt
            print(f"レート制限に達しました。{wait_time}秒待機中...")
            time.sleep(wait_time)
        
        except APIConnectionError as e:
            # 接続エラー（ネットワーク問題）
            print(f"接続エラー: {e}")
            if attempt < max_retries - 1:
                time.sleep(backoff_factor ** attempt)
        
        except APIError as e:
            # その他のAPIエラー（認証失敗、無効なパラメータなど）
            print(f"APIエラー (ステータス {e.status_code}): {e.message}")
            if e.status_code == 401:
                print("APIキーが無効です。設定を確認してください。")
                return None
            elif e.status_code == 429:
                # レート制限（別の対処）
                wait_time = backoff_factor ** attempt
                print(f"{wait_time}秒待機します...")
                time.sleep(wait_time)
            else:
                return None
    
    print("最大リトライ回数に達しました。")
    return None

# 使用例
if __name__ == "__main__":
    result = call_gpt_with_retry("Pythonの非同期処理について説明してください")
    if result:
        print(result)
  

コスト管理とトークン数の監視

ChatGPT APIは使用量に応じた課金体系です。コスト管理のため、以下のようにトークン数を監視することが重要です。

from openai import OpenAI
import os
from dotenv import load_dotenv
import tiktoken  # トークン数カウント用ライブラリ

# インストール: pip install tiktoken

load_dotenv()
client = OpenAI(api_key=os.getenv('OPENAI_API_KEY'))

def count_tokens(text, model="gpt-4o-mini"):
    """
    テキストのトークン数をカウント
    """
    encoding = tiktoken.encoding_for_model(model)
    tokens = encoding.encode(text)
    return len(tokens)

def call_gpt_with_cost_tracking(user_message, model="gpt-4o-mini"):
    """
    API呼び出しと同時にコストを追跡
    """
    # 入力トークン数をカウント
    input_tokens = count_tokens(user_message, model)
    print(f"入力トークン数: {input_tokens}")
    
    response = client.chat.completions.create(
        model=model,
        messages=[{"role": "user", "content": user_message}],
        temperature=0.7,
        max_tokens=500
    )
    
    # レスポンスから使用トークン情報を取得
    total_tokens = response.usage.total_tokens
    completion_tokens = response.usage.completion_tokens
    
    print(f"出力トークン数: {completion_tokens}")
    print(f"合計トークン数: {total_tokens}")
    
    # モデルごとの料金（2024年の参考値）
    # gpt-4o-mini: $0.15/100万入力トークン, $0.60/100万出力トークン
    if model == "gpt-4o-mini":
        input_cost = (response.usage.prompt_tokens / 1_000_000) * 0.15
        output_cost = (response.usage.completion_tokens / 1_000_000) * 0.60
        total_cost = input_cost + output_cost
        print(f"推定コスト: ${total_cost:.6f}")
    
    return response.choices[0].message.content

# 使用例
if __name__ == "__main__":
    result = call_gpt_with_cost_tracking("機械学習とディープラーニングの違いを説明してください")
    print(f"\n回答:\n{result}")
  

実務アプリケーションでよくあるユースケース

テキスト分類・感情分析の例

カスタマーレビュー、フィードバック、問い合わせメールなどを自動分類する場合、以下のような実装が有効です。

from openai import OpenAI
import json
import os
from dotenv import load_dotenv

load_dotenv()
client = OpenAI(api_key=os.getenv('OPENAI_API_KEY'))

def classify_customer_feedback(feedback_text):
    """
    顧客フィードバックをカテゴリと感情で分類
    """
    prompt = f"""以下の顧客フィードバックを分析してください。
JSON形式で以下の情報を返してください：
- category: "製品品質", "配送", "カスタマーサービス", "その他"
- sentiment: "positive", "neutral", "negative"
- confidence: 0-1の信頼度スコア

フィードバック:
{feedback_text}

JSON形式で返してください（日本語はそのまま）:"""
    
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}],
        temperature=0.3,  # 分類タスクなので低めに設定
        max_tokens=200
    )
    
    response_text = response.choices[0].message.content
    
    # JSON部分を抽出してパース
    try:
        # レスポンスからJSON部分を抽出
        json_start = response_text.find('{')
        json_end = response_text.rfind('}') + 1
        json_str = response_text[json_start:json_end]
        result = json.loads(json_str)
        return result
    except json.JSONDecodeError:
        print(f"JSON解析エラー: {response_text}")
        return None

# 使用例
if __name__ == "__main__":
    feedbacks = [
        "商品は素晴らしいですが、配送に1ヶ月かかりました",
        "最悪です。不良品が届きました",
        "サポートチームが親切に対応してくれました"
    ]
    
    for feedback in feedbacks:
        result = classify_customer_feedback(feedback)
        print(f"フィードバック: {feedback}")