AI API 開発

Web インターフェースで AI チャットツールを使うのは便利ですが、AI の能力を実際に自分の製品、ワークフロー、自動化スクリプトに組み込むには、API が必要です。

API(Application Programming Interface)はアプリケーション間の通信の橋渡しです。API を通じて、自分のコードから直接 AI モデルにリクエストを送信し、応答を取得できます。手動で Web ページを開いてコピー&ペーストする必要はありません。

次のようなシナリオを想像してみてください:

  • EC サイトが各商品の説明文を自動生成する。

  • ノートアプリがユーザーが入力した長文を自動的に要約する。

  • カスタマーサポートシステムがユーザーの問い合わせを自動的に分類して返答する。

  • これらの機能はすべて AI API で実現できます。

このモジュールの目標は、「AI でチャットできるだけ」から「AI を自分のコードで働かせることができる」へと変わることです。


API 基本概念

コードを書く前に、いくつかの核心的な概念を理解しましょう。

API とは

API とは、あらかじめ取り決められた対話方法のことです。特定の形式でリクエストを送信し、相手は特定の形式で結果を返します。

レストランでの注文に例えてみましょう:

  • レストランに入り、メニュー(API ドキュメント)を受け取り、何を注文できるか、どう注文するか(リクエスト形式)が分かります。

  • ウェイターに「宮保鶏丁(クンパオチーディン)、辛さ控えめで」と伝えます(API リクエストを送信)。

  • 厨房で料理が作られ、あなたのところに運ばれてきます(API レスポンスを返す)。

この過程で、あなたは厨房に入る必要も、料理の作り方を知る必要もありません。これが API の価値です:複雑な内部処理をカプセル化し、シンプルなインターフェースだけを公開する。

REST API の基本原理

REST(Representational State Transfer)は、現在最も広く使われている API 設計スタイルです。

簡単に言うと、HTTP プロトコルを使って通信する方法です:

HTTP メソッド用途AI API での例
POSTリソースの作成チャットリクエストの送信、画像の生成
GETリソースの取得モデル一覧の照会、請求情報の取得
DELETEリソースの削除会話履歴の削除

AI API の大半は POST リクエストです。なぜなら「AI に処理してほしい内容を送信する」からです。

JSON 形式入門

JSON(JavaScript Object Notation)は、API 通信で最もよく使われるデータ形式です。

JSON の構文はとてもシンプルで、キーと値のペアの組み合わせです。

例

{
    "model": "gpt-4",
    "messages": [
        {
            "role": "user",
            "content": "こんにちは、サンプル について紹介してください"
        }
    ],
    "temperature": 0.7,
    "max_tokens": 1000
}

これは典型的な AI API リクエストボディです。

いくつかの重要なルール:

  • オブジェクトは{ }で囲み、配列は[ ]で囲みます。

  • 文字列はダブルクォートを使用し、"シングルクォートは使用できません。

  • キーと値のペアはコロンで:区切り、キーは文字列でなければなりません。

  • 複数のキーと値のペアはカンマで,区切ります。

JSON は Python の辞書とよく似ていますが、構文に違いがあるため、コードを書くときは注意が必要です。


開発環境の準備

良い仕事をするには、まず道具を整えなければなりません。

Python のインストールと設定

Python は AI API を呼び出す際に最もよく使われる言語で、エコシステムが整っており、ライブラリも豊富です。

まず、お使いのコンピューターに Python がインストールされているか確認してください:

# 检查 Python 版本(Windows、macOS、Linux 通用)
python --version
# 或者
python3 --version

Python 3.9 以降を推奨します。

まだインストールしていない場合は、python.orgにアクセスして最新の安定版をダウンロードしてください。

pip パッケージ管理

pip は Python のパッケージマネージャーで、サードパーティ製ライブラリのインストールに使用します。

# 检查 pip 版本
pip --version
# 或者
pip3 --version

# 升级 pip 到最新版本
pip install --upgrade pip

# 配置国内镜像源(可选,提升下载速度)
pip config set global.index-url https://pypi.tuna.tsinghua.edu.cn/simple

コードエディタの選択

AI API のコードを書く際は、以下のエディタのいずれかを推奨します:

エディタ特徴おすすめの人
VS Code無料、プラグインが豊富、軽量ほとんどの開発者
CursorAI アシスタント内蔵、コードを直接生成可能AI にコード作成を手伝ってもらいたい人
PyCharm機能が最も充実、Python サポートが最高プロフェッショナルな Python 開発者

本チュートリアルのサンプルコードは、どのエディタでも実行できます。


OpenAI API 入門

OpenAI API は最も定番で、エコシステムが最も整った AI API です。多くの他のベンダーもその形式と互換性があります。

登録と API Key の取得

API Key はあなたの通行証であり、身元を証明し、利用量を記録します。取得手順は次のとおりです:

  • 1. アクセスplatform.openai.comアカウントを登録

  • 2. API Keys ページに入る

  • 3. "Create new secret key" をクリック

  • 4. この Key をコピーして保存します(一度しか表示されません!)

重要:API Key を公開のコードリポジトリにコミットしてはいけません。一度漏洩すると、他人があなたの利用枠を使って費用を発生させる可能性があります。

最初の API 呼び出し

まず OpenAI の公式 SDK をインストールします:

pip install openai

それでは、最初の API 呼び出しプログラムを書いてみましょう:

例

# ファイルパス: first_api_call.py
# 最初の OpenAI API 呼び出しの例

from openai import OpenAI

# クライアントの初期化
# 注意:API Key は環境変数から読み取ることを推奨します。コードにハードコードしないでください。
# ここではデモのために直接書いていますが、実際のプロジェクトでは環境変数を使用してください。
client = OpenAI(
    api_key="your-api-key-here"  # あなたの API Key に置き換えてください
)

# チャットリクエストを送信
response = client.chat.completions.create(
    model="gpt-3.5-turbo",  # モデルを選択
    messages=[
        {
            "role": "user",  # ロール:user はユーザーを表す
            "content": "サンプル を一言で紹介してください"  # ユーザーが入力した内容
        }
    ]
)

# 完全なレスポンスを出力(何が返ってきたか確認)
print("完全なレスポンス:")
print(response)
print("\n" + "="*50 + "\n")

# AI の応答を抽出
ai_reply = response.choices[0].message.content
print("AI の応答:")
print(ai_reply)

OpenAI の api_key がない場合は、中国国内の代替サービスも利用できます。中国国内の多くのモデルは OpenAI と互換性があります。例えば DeepSeek です。アクセスしてhttps://platform.deepseek.com/api_keysapi_key を申請し、以下のコードで自分のものと置き換えてください。api_keyとmodelDeepSeek がサポートするモデルは deepseek-v4-flash、deepseek-v4-pro(思考モード)です。API を使用して大規模モデルを呼び出すことができます:

例

from openai import OpenAI

# クライアントの初期化
# 注意:API Key は環境変数から読み取ることを推奨します。コードにハードコーディングしないでください。
# ここではデモの都合上直接書いていますが、実際のプロジェクトでは環境変数を使用してください。
client = OpenAI(
    api_key="sk-xxxx",  # api_key を設定
    base_url="https://api.deepseek.com")  # リクエストURLを設定


# チャットリクエストを送信
response = client.chat.completions.create(
    model="deepseek-v4-flash",  # モデルを選択
    messages=[
        {
            "role": "user",  # ロール:user はユーザーを表す
            "content": "サンプル 初心者向けチュートリアルを一言で紹介してください"  # ユーザーの入力内容
        }
    ]
)

# 完全なレスポンスを出力(何が返ってきたか確認)
print("完全なレスポンス:")
print(response)
print("\n" + "="*50 + "\n")

# AIの返信を抽出
ai_reply = response.choices[0].message.content
print("AIの返信:")
print(ai_reply)

このプログラムを実行します:

python first_api_call.py

すべて正常なら、次のような出力が表示されます:

完整响应:
ChatCompletion(
  id='chatcmpl-xxx',
  choices=[Choice(finish_reason='stop', index=0, message=ChatMessage(content='サンプル是一个专注于提供编程技术教程的学习平台...', role='assistant'))],
  model='gpt-3.5-turbo',
  usage=CompletionUsage(completion_tokens=30, prompt_tokens=18, total_tokens=48)
)

==================================================

AI 回复:
サンプル是一个专注于提供编程技术教程的学习平台,涵盖多种编程语言和技术领域。

おめでとうございます!AI API の呼び出しに成功しました。

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

OpenAI API には調整可能なパラメータが多数あります。以下は最も一般的なものです:

パラメータ名型必須かどうか説明デフォルト値
modelstringである使用するモデルID(例:gpt-3.5-turbo、gpt-4)无
messagesarrayである会話履歴のメッセージリスト无
temperaturenumber否サンプリング温度。0〜2の間。高いほどランダム、低いほど決定的1.0
max_tokensinteger否生成する最大トークン数無制限
top_pnumber否核サンプリングパラメータ。0〜1の間1.0
stopstring/array否停止シーケンス。これらの文字に遭遇すると生成を停止null

temperature について重点的に説明します。これは最もよく使われる調整パラメータです:

例

# ファイルパス:temperature_demo.py
# temperature パラメータが出力に与える影響をデモ

from openai import OpenAI

client = OpenAI(api_key="your-api-key-here")

def ask_ai(temperature_value: float, question: str) -> str:
    """temperature を指定して質問"""
    response = client.chat.completions.create(
        model="gpt-3.5-turbo",
        messages=[{"role": "user", "content": question}],
        temperature=temperature_value,
        max_tokens=200
    )
    return response.choices[0].message.content

# 同じ質問を、異なる temperature で3回問う
question = "サンプル のスローガンを考えてください"

print("--- temperature=0(最も決定的、毎回ほぼ同じ結果)---")
print(ask_ai(0.0, question))
print("\n--- temperature=0.7(バランスが良く、日常使用におすすめ)---")
print(ask_ai(0.7, question))
print("\n--- temperature=1.8(最もランダム、毎回結果が大きく異なる)---")
print(ask_ai(1.8, question))

経験則:

  • temperature=0:事実回答、コード生成、正確な結果が必要な場面

  • temperature=0.7:日常会話、一般的なタスク

  • temperature>1.0:クリエイティブライティング、ブレインストーミング


Anthropic Claude API

Claude は Anthropic 社の大規模言語モデルで、安全性と長文処理能力で知られています。

OpenAI API との違いと共通点

どちらもテキスト生成ができますが、設計思想にいくつかの違いがあります:

比較項目OpenAI APIClaude API
メッセージ構造system、user、assistant が交互に来るsystem は別に設定し、user、assistant が交互に来る
コンテキスト長16k〜128k まで様々Claude 3 シリーズは 200k をサポート
パラメータ設計temperature、top_p などtemperature、top_p など、概念は類似

Claude Messages API の構造

まず Claude SDK をインストールします:

pip install anthropic

Claude API の基本的な呼び出し:

例

# ファイルパス:claude_first_call.py
# Claude API 基本呼び出しの例

import anthropic

# クライアントを初期化
client = anthropic.Anthropic(
    api_key="your-api-key-here"  # あなたの Claude API Key に置き換える
)

# メッセージリクエストを送信
response = client.messages.create(
    model="claude-3-sonnet-20240229",  # Claude モデルを選択
    max_tokens=1024,  # Claude では max_tokens の設定が必須
    messages=[
        {
            "role": "user",
            "content": "サンプル 初心者向けチュートリアルを3文で紹介してください"
        }
    ]
)

# レスポンスを出力
print("完全なレスポンス:")
print(response)
print("\n" + "="*50 + "\n")

# 返信を抽出
print("Claude の返信:")
print(response.content[0].text)

同様に、中国国内互換の大規模モデル(例:DeepSeek)に置き換えて、指定しますapi_key 、 base_url、modelこの3つのパラメータ:

例

import anthropic

# クライアントを初期化
client = anthropic.Anthropic(
    api_key="sk-xxx",  # あなたの API Key に置き換える
    base_url="https://api.deepseek.com/anthropic"  # Claude API のベースURL
)

# メッセージリクエストを送信
response = client.messages.create(
    model="deepseek-v4-flash",  # Claude モデルを選択
    max_tokens=1024,  # Claude では max_tokens の設定が必須
    messages=[
        {
            "role": "user",
            "content": "サンプル 初心者向けチュートリアルを3文で紹介してください"
        }
    ]
)

# レスポンスを出力
print("完全なレスポンス:")
print(response)
print("\n" + "="*50 + "\n")

# 返信を抽出
print("Claude の返信:")
print(response.content[0].thinking)

System Prompt の設定

Claude API では、System Prompt は独立したパラメータであり、messages 配列の中には置きません:

例

# ファイルパス:claude_system_prompt.py
# Claude System Prompt 使用例

import anthropic

client = anthropic.Anthropic(api_key="your-api-key-here")

response = client.messages.create(
    model="claude-3-sonnet-20240229",
    max_tokens=1024,
    # System Prompt はここで設定
    system="あなたはプロの技術ドキュメント編集者です。回答は簡潔かつ正確にし、リスト形式を多用してください。",
    messages=[
        {
            "role": "user",
            "content": "Python を学ぶ際に押さえるべき中核的な知識は何ですか?"
        }
    ]
)

print(response.content[0].text)

マルチターン対話管理

単発の会話は簡単ですが、実際に役立つアプリケーションは「コンテキストを記憶する」必要があります。

会話履歴の管理方法

考え方はとても簡単です:過去の会話をすべて保存し、毎回のリクエストで AI に一緒に送信します。

例

# ファイルパス:multi_turn_chat.py
# マルチターン会話管理の例

from openai import OpenAI

client = OpenAI(api_key="your-api-key-here")

class ChatBot:
    """シンプルなマルチターン対話ボット"""

    def __init__(self, system_prompt: str = "あなたは人助けが好きなアシスタントです"):
        self.messages = []  # リストを使って会話履歴を保存
        # 最初に system prompt を追加
        self.messages.append({
            "role": "system",
            "content": system_prompt
        })

    def chat(self, user_input: str) -> str:
        """メッセージを送信して返信を取得"""
        # ユーザーメッセージを履歴に追加
        self.messages.append({
            "role": "user",
            "content": user_input
        })

        # API を呼び出す
        response = client.chat.completions.create(
            model="gpt-3.5-turbo",
            messages=self.messages
        )

        # AI の返信を取得
        ai_reply = response.choices[0].message.content

        # AI の返信も履歴に追加
        self.messages.append({
            "role": "assistant",
            "content": ai_reply
        })

        return ai_reply

    def get_history(self) -> list:
        """会話履歴を取得"""
        return self.messages


# 使用例
if __name__ == "__main__":
    bot = ChatBot(system_prompt=あなたはプログラミングアシスタントで、サンプル と Python に関する質問に専門のに答えます)

    print("=== マルチターン会話デモ ===")

    # 1ターン目
    r1 = bot.chat("Python とは何ですか?")
    print("ユーザー:Python とは何ですか?")
    print("AI:", r1)
    print()

    # 2ターン目(AI は前のターンの話題を覚えている)
    r2 = bot.chat("その特徴は何ですか?")
    print("ユーザー:その特徴は何ですか?")
    print("AI:", r2)
    print()

    # 3ターン目
    r3 = bot.chat("サンプル には Python のチュートリアルはありますか?")
    print("ユーザー:サンプル には Python のチュートリアルはありますか?")
    print("AI:", r3)

重要なポイント:会話履歴は順序付きリストであり、user-assistant-user-assistant の順番を厳密に守ります。毎回のリクエストで完全な履歴を送信する必要があり、AI がコンテキストを理解できます。履歴はトークンを消費するため、定期的なクリーンアップが必要です。

コンテキストウィンドウ超過の処理

各モデルにはコンテキストウィンドウの制限があります。例えば gpt-3.5-turbo は 16k、gpt-4 は 8k または 32k です。

会話履歴が長すぎると、エラーが発生します。

いくつかの一般的な対処戦略:

例

# ファイルパス:context_management.py
# コンテキストウィンドウ管理戦略の例

from openai import OpenAI

client = OpenAI(api_key="your-api-key-here")


class ChatBotWithContextLimit:
    """コンテキスト制限付きの対話ボット"""

    def __init__(self, system_prompt: str, max_messages: int = 10):
        self.system_prompt = system_prompt
        self.max_messages = max_messages  # 最大で何件のメッセージを保持するか
        self.messages = []

    def _trim_messages(self):
        """メッセージリストをトリミングし、最近のメッセージを保持"""
        # 常に system prompt を保持し、次に最新の max_messages 件のみを保持
        if len(self.messages) > self.max_messages:
            # 最新の N 件を保持
            self.messages = self.messages[-self.max_messages:]

    def chat(self, user_input: str) -> str:
        # ユーザーメッセージを追加
        self.messages.append({
            "role": "user",
            "content": user_input
        })

        # 履歴をトリミング
        self._trim_messages()

        # 完全なリクエストを構築(system prompt + 履歴)
        full_messages = [
            {"role": "system", "content": self.system_prompt}
        ] + self.messages

        try:
            response = client.chat.completions.create(
                model="gpt-3.5-turbo",
                messages=full_messages
            )
            ai_reply = response.choices[0].message.content

            self.messages.append({
                "role": "assistant",
                "content": ai_reply
            })

            return ai_reply

        except Exception as e:
            # それでもエラーになる場合は、より積極的なトリミングが必要かもしれません
            return f"エラーが発生しました:{str(e)}"


# 戦略2:AIで履歴会話を要約する
def summarize_conversation(messages: list) -> str:
    """AIで履歴会話を要約し、元の履歴の代わりにする"""
    history_text = "\n".join([
        f"{m['role']}: {m['content']}"
        for m in messages
    ])

    prompt = f"""以下の会話を簡潔な要約にまとめ、重要な情報を保持してください:

{history_text}

要約:"""


    response = client.chat.completions.create(
        model="gpt-3.5-turbo",
        messages=[{"role": "user", "content": prompt}],
        temperature=0
    )

    return response.choices[0].message.content


# 要約戦略を使用する
print("--- 会話履歴の要約例 ---")
sample_history = [
    {"role": "user", "content": "私の名前は何?"},
    {"role": "assistant", "content": "すみません、あなたの名前は知りません。教えていただけますか?"},
    {"role": "user", "content": "私はシャオミンです。サンプルでPythonを学んでいます。"},
    {"role": "assistant", "content": "こんにちは、シャオミン!お会いできて嬉しいです。サンプルはとても良い学習プラットフォームです。"},
    {"role": "user", "content": "リストを学びたいのですが、教えてもらえますか?"},
    {"role": "assistant", "content": "もちろんです!Pythonのリストは..."},
]

summary = summarize_conversation(sample_history)
print("会話の要約:", summary)

コンテキスト管理の方法をまとめると:

戦略長所短所適用シーン
直近のN件のみ保持シンプルで高速重要な情報を失う可能性がある短い会話や雑談シーン
AIで履歴を要約意味情報を保持APIの追加呼び出しが必要長い会話、重要な情報が多い
スライディングウィンドウシンプルさと完全性のバランス実装はやや複雑一般的なシーンで汎用的

ストリーミング出力

ウェブ上でAIとチャットすると、文字がひとつずつ飛び出すのが見えます。これがストリーミング出力です。

ストリーミング出力とは

  • 通常モード:AIが完全な返答を生成してから一括で返すため、数秒から数十秒待つことになります。

  • ストリーミングモード:AIがトークンを生成するたびに送信するため、ユーザー体験が向上します。

技術的には、ストリーミング出力はSSE(Server-Sent Events)プロトコルを使用し、サーバーがデータを継続的にプッシュします。

タイプライター効果の実装

例

# ファイルパス:streaming_demo.py
# ストリーミング出力のデモ(タイプライター効果)

from openai import OpenAI

client = OpenAI(api_key="your-api-key-here")

print("=== ストリーミング出力デモ ===")
print("質問:サンプルについての紹介を書いてください\n")
print("回答:", end="", flush=True)

# 重要なパラメータ:stream=True
response = client.chat.completions.create(
    model="gpt-3.5-turbo",
    messages=[
        {"role": "user", "content": "サンプル(初心者向けチュートリアル)についての紹介を書いてください"}
    ],
    stream=True  # ストリーミング出力を有効にする
)

collected_reply = []

for chunk in response:
    # 現在のchunkの内容を抽出する
    if chunk.choices[0].delta.content:
        content = chunk.choices[0].delta.content
        print(content, end="", flush=True)  # リアルタイムに出力し、改行しない
        collected_reply.append(content)

print()  # 最後に改行を出力する
print("\n" + "="*50)
print("完全な回答:", "".join(collected_reply))

このプログラムを実行すると、文字がひとつずつ表示されるのがわかります。

Claudeのストリーミング出力の使い方も同様です:

例

# ファイルパス:claude_streaming.py
# Claudeのストリーミング出力例

import anthropic

client = anthropic.Anthropic(api_key="your-api-key-here")

print("=== Claude ストリーミング出力 ===")
print("質問:プログラミングを学ぶには?\n")
print("回答:", end="", flush=True)

with client.messages.stream(
    model="claude-3-sonnet-20240229",
    max_tokens=1024,
    messages=[{"role": "user", "content": "プログラミングを体系的に学ぶには?"}]
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)

print()

ユーザー体験のアドバイス:「完全な結果を待つ必要がある」というシーンでない限り、ストリーミング出力を優先して使用しましょう。ユーザーに進捗を見せると、体験が大幅に向上します。


API 料金の計算と最適化

AI APIは使用量に応じて課金されます。使うほど、費用がかかります。

Token 課金の仕組み

AI APIは文字数や会話回数ではなく、Tokenに基づいて課金されます。Token,TokenTokenはAIがテキストを処理する基本単位です。

1トークンはおおよそ次の数に相当します:

  • 英語:0.75語(または4文字)

  • 中国語:漢字1〜2文字

例を挙げます:

  • "Hello, 世界" — 約3〜4トークン

  • "サンプル很棒" — 約6〜8トークン

課金方法:

Prompt tokens(你发给 AI 的) + Completion tokens(AI 返回给你的)= 总 tokens

モデルによって価格が異なります。例えば、gpt-3.5-turboは$0.0015/1k prompt tokens、$0.002/1k completion tokensです。

API コストを抑える方法

例

# ファイルパス:cost_tracking.py
# API費用の追跡と制御の例

from openai import OpenAI

client = OpenAI(api_key="your-api-key-here")

# モデル価格表(例、実際の価格は公式ドキュメントを参照してください)
# 単位:米ドル / 1k tokens
MODEL_PRICES = {
    "gpt-3.5-turbo": {
        "prompt": 0.0015,
        "completion": 0.002
    },
    "gpt-4": {
        "prompt": 0.03,
        "completion": 0.06
    }
}


def chat_with_cost_tracking(model: str, messages: list):
    """費用統計付きチャット"""
    response = client.chat.completions.create(
        model=model,
        messages=messages
    )

    # レスポンスからトークン使用量を取得する
    usage = response.usage
    prompt_tokens = usage.prompt_tokens
    completion_tokens = usage.completion_tokens
    total_tokens = usage.total_tokens

    # 費用を計算する
    price = MODEL_PRICES.get(model, {"prompt": 0, "completion": 0})
    prompt_cost = (prompt_tokens / 1000) * price["prompt"]
    completion_cost = (completion_tokens / 1000) * price["completion"]
    total_cost = prompt_cost + completion_cost

    return {
        "reply": response.choices[0].message.content,
        "usage": {
            "prompt_tokens": prompt_tokens,
            "completion_tokens": completion_tokens,
            "total_tokens": total_tokens
        },
        "cost": {
            "prompt_cost": prompt_cost,
            "completion_cost": completion_cost,
            "total_cost": total_cost
        }
    }


# 使用例
if __name__ == "__main__":
    result = chat_with_cost_tracking(
        model="gpt-3.5-turbo",
        messages=[{"role": "user", "content": "サンプルを紹介してください"}]
    )

    print("AIの応答:", result["reply"])
    print()
    print("使用統計:")
    print(f"  Prompt tokens: {result['usage']['prompt_tokens']}")
    print(f"  Completion tokens: {result['usage']['completion_tokens']}")
    print(f"  Total tokens: {result['usage']['total_tokens']}")
    print()
    print("費用統計(米ドル):")
    print(f"  Prompt cost: ${result['cost']['prompt_cost']:.6f}")
    print(f"  Completion cost: ${result['cost']['completion_cost']:.6f}")
    print(f"  Total cost: ${result['cost']['total_cost']:.6f}")

コスト最適化の実用的なテクニック:

テクニック方法節約率
適切なモデルを選ぶ簡単なタスクにはgpt-3.5-turboを使い、gpt-4は使わない90%+
max_tokensを制限する適切な上限を設定し、AIが話しすぎないようにする場合による
コンテキストを削減する必要な履歴会話のみを保持する50%+
プロンプトは簡潔にsystem promptは長く書きすぎない場合による
結果のキャッシュ同じ質問にはキャッシュ結果を直接返す場合による

キャッシュ戦略

重複する質問には、キャッシュ結果を使うと多くの費用を節約できます:

例

# ファイルパス:response_cache.py
# シンプルなレスポンスキャッシュの実装

import hashlib
import json
from openai import OpenAI

client = OpenAI(api_key="your-api-key-here")


class SimpleCache:
    """シンプルなメモリキャッシュ"""

    def __init__(self):
        self.cache = {}

    def _make_key(self, model: str, messages: list) -> str:
        """リクエスト内容からキャッシュキーを生成する"""
        # 重要な情報をシリアライズしてハッシュを取得する
        key_data = {
            "model": model,
            "messages": messages
        }
        key_str = json.dumps(key_data, sort_keys=True)
        return hashlib.md5(key_str.encode()).hexdigest()

    def get(self, model: str, messages: list):
        """キャッシュを取得する"""
        key = self._make_key(model, messages)
        return self.cache.get(key)

    def set(self, model: str, messages: list, value: str):
        """キャッシュを設定する"""
        key = self._make_key(model, messages)
        self.cache[key] = value


# キャッシュ付きチャット関数
cache = SimpleCache()

def chat_with_cache(model: str, messages: list):
    # まずキャッシュを確認する
    cached = cache.get(model, messages)
    if cached:
        print("(キャッシュにヒットしたため、直接返します)")
        return cached

    # キャッシュミスの場合、APIを呼び出す
    response = client.chat.completions.create(
        model=model,
        messages=messages
    )
    reply = response.choices[0].message.content

    # キャッシュに保存する
    cache.set(model, messages, reply)

    return reply


# テスト
if __name__ == "__main__":
    messages = [{"role": "user", "content": "Pythonとはどんなウェブサイトですか?"}]

    print("1回目の呼び出し...")
    r1 = chat_with_cache("gpt-3.5-turbo", messages)
    print(r1)
    print()

    print("2回目の呼び出し(同じ質問)...")
    r2 = chat_with_cache("gpt-3.5-turbo", messages)
    print(r2)

エラーハンドリングとリトライ機構

ネットワークリクエストは必ずエラーが発生します。APIも例外ではありません。堅牢なコードはさまざまな例外状況に対処する必要があります。

よくあるエラーの種類

エラーの種類HTTPステータスコード原因対処方法
認証失敗401APIキーの誤りまたは期限切れAPIキーを確認する
クォータ不足402アカウントの残高がないチャージするかアカウントを変更する
制限超過429リクエストが速すぎるか、クォータを使い切ったしばらく待って再試行する
サーバーエラー5xxAIサーバー側に問題があるリトライ

指数バックオフによるリトライ

一時的なエラー(429、5xxなど)が発生した場合、最もよく使われる戦略は指数バックオフリトライです:

  • 1回目の失敗 → 1秒待つ → リトライ

  • 2回目の失敗 → 2秒待つ → リトライ

  • 3回目の失敗 → 4秒待つ → リトライ

  • 4回目の失敗 → 8秒待つ → リトライ

毎回の待ち時間を2倍にすることで、サービスが回復する機会を与えます。

例

# ファイルパス:error_handling.py
# エラー処理と指数バックオフリトライ

import time
from openai import OpenAI, APIError, RateLimitError, APIConnectionError

client = OpenAI(api_key="your-api-key-here")


def chat_with_retry(
    model: str,
    messages: list,
    max_retries: int = 3,
    initial_wait: float = 1.0
):
    """リトライ機構付きチャット関数"""

    retries = 0

    while retries <= max_retries:
        try:
            response = client.chat.completions.create(
                model=model,
                messages=messages
            )
            return {
                "success": True,
                "reply": response.choices[0].message.content
            }

        except RateLimitError as e:
            # レート制限エラー:しばらく待って再試行
            retries += 1
            if retries > max_retries:
                return {
                    "success": False,
                    "error": f"リトライ回数を使い果たしました:{str(e)}"
                }
            wait_time = initial_wait * (2 ** (retries - 1))
            print(f"レート制限が発生しました。{wait_time}秒待ってから再試行します...")
            time.sleep(wait_time)

        except APIConnectionError as e:
            # ネットワーク接続エラー:リトライ
            retries += 1
            if retries > max_retries:
                return {
                    "success": False,
                    "error": f"ネットワーク接続に失敗しました:{str(e)}"
                }
            wait_time = initial_wait * (2 ** (retries - 1))
            print(f"ネットワークエラーが発生しました。{wait_time}秒待ってから再試行します...")
            time.sleep(wait_time)

        except APIError as e:
            # APIエラー:状況に応じてリトライ
            if e.status_code and 500 <= e.status_code < 600:
                retries += 1
                if retries > max_retries:
                    return {
                        "success": False,
                        "error": f"サーバーエラー:{str(e)}"
                    }
                wait_time = initial_wait * (2 ** (retries - 1))
                print(f"サーバーエラーが発生しました。{wait_time}秒待ってから再試行します...")
                time.sleep(wait_time)
            else:
                # 5xx以外のエラーはリトライしない
                return {
                    "success": False,
                    "error": f"API エラー:{str(e)}"
                }

        except Exception as e:
            # その他のエラーは再試行しない
            return {
                "success": False,
                "error": f"未知のエラー:{str(e)}"
            }


# 使用例
if __name__ == "__main__":
    result = chat_with_retry(
        model="gpt-3.5-turbo",
        messages=[{"role": "user", "content": "こんにちは、サンプル"}]
    )

    if result["success"]:
        print("成功:", result["reply"])
    else:
        print("失敗:", result["error"])

既製のライブラリを使用してリトライロジックを簡素化することもできます。たとえば tenacity は、次のコマンドでインストールします:

pip install tenacity

例

# ファイルパス:tenacity_demo.py
# tenacity ライブラリを使用してリトライロジックを簡素化

from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
from openai import OpenAI, RateLimitError, APIError

client = OpenAI(api_key="your-api-key-here")


@retry(
    stop=stop_after_attempt(3),  # 最大3回再試行
    wait=wait_exponential(multiplier=1, min=1, max=10),  # 指数バックオフ:1, 2, 4... 最大10秒
    retry=retry_if_exception_type((RateLimitError, APIError))  # 特定の例外のみ再試行
)
def chat_with_tenacity(model: str, messages: list):
    """tenacity デコレータでリトライを実装"""
    response = client.chat.completions.create(
        model=model,
        messages=messages
    )
    return response.choices[0].message.content


# 使用
try:
    reply = chat_with_tenacity(
        "gpt-3.5-turbo",
        [{"role": "user", "content": "サンプル を紹介してください"}]
    )
    print(reply)
except Exception as e:
    print(f"最終失敗:{e}")

実践プロジェクト:コマンドライン AI チャットツール

これまで学んだ知識を統合して、完全なコマンドラインチャットツールを作りましょう。

例

# ファイルパス:サンプル_chatbot.py
# サンプル AI チャットボット - 完全なコマンドラインツール

import sys
import json
import os
from datetime import datetime
from openai import OpenAI


class サンプルChatBot:
    """完全な機能を備えたコマンドラインチャットボット"""

    def __init__(self, config_file: str = "config.json"):
        # 設定を読み込む
        self.config = self._load_config(config_file)
        self.client = OpenAI(api_key=self.config["api_key"])

        # 会話状態
        self.messages = []
        self.total_tokens = 0
        self.total_cost = 0.0

        # system prompt を初期化
        self._init_system_prompt()

    def _load_config(self, config_file: str) -> dict:
        """設定ファイルを読み込む"""
        default_config = {
            "api_key": "your-api-key-here",
            "model": "gpt-3.5-turbo",
            "system_prompt": "あなたは サンプル が技術サポートを提供する、親切な AI アシスタントです。",
            "max_history": 20,
            "stream": True,
            "temperature": 0.7
        }

        if os.path.exists(config_file):
            with open(config_file, "r", encoding="utf-8") as f:
                user_config = json.load(f)
                default_config.update(user_config)

        return default_config

    def _save_config(self, config_file: str = "config.json"):
        """設定ファイルを保存する"""
        with open(config_file, "w", encoding="utf-8") as f:
            json.dump(self.config, f, indent=2, ensure_ascii=False)

    def _init_system_prompt(self):
        """system prompt を初期化する"""
        self.messages = [
            {
                "role": "system",
                "content": self.config["system_prompt"]
            }
        ]

    def _trim_history(self):
        """会話履歴をトリミングする"""
        # system prompt + 直近の N 件の会話を保持
        max_messages = self.config["max_history"]
        if len(self.messages) > max_messages + 1:
            self.messages = [self.messages[0]] + self.messages[-max_messages:]

    def chat(self, user_input: str) -> str:
        """メッセージを送信して返信を取得する"""
        # ユーザーメッセージを追加
        self.messages.append({
            "role": "user",
            "content": user_input
        })

        # 履歴をトリミング
        self._trim_history()

        if self.config["stream"]:
            return self._chat_stream()
        else:
            return self._chat_normal()

    def _chat_normal(self) -> str:
        """通常モード(非ストリーミング)"""
        response = self.client.chat.completions.create(
            model=self.config["model"],
            messages=self.messages,
            temperature=self.config["temperature"]
        )

        ai_reply = response.choices[0].message.content

        # 使用量を集計
        if hasattr(response, "usage"):
            self.total_tokens += response.usage.total_tokens
            # 費用を簡単に見積もる(gpt-3.5-turbo の価格)
            self.total_cost += (response.usage.total_tokens / 1000) * 0.002

        self.messages.append({
            "role": "assistant",
            "content": ai_reply
        })

        return ai_reply

    def _chat_stream(self) -> str:
        """ストリーミングモード"""
        response = self.client.chat.completions.create(
            model=self.config["model"],
            messages=self.messages,
            temperature=self.config["temperature"],
            stream=True
        )

        collected_chunks = []
        print("AI:", end="", flush=True)

        for chunk in response:
            if chunk.choices[0].delta.content:
                content = chunk.choices[0].delta.content
                print(content, end="", flush=True)
                collected_chunks.append(content)

        print()

        ai_reply = "".join(collected_chunks)

        self.messages.append({
            "role": "assistant",
            "content": ai_reply
        })

        # ストリーミングモードの token 統計は追加処理が必要ですが、ここでは簡略化しています
        # 実際のプロジェクトでは usage API を呼び出すか推定できます
        self.total_tokens += len(ai_reply) // 2  # おおよその見積もり

        return ai_reply

    def save_conversation(self, filename: str = None):
        """会話履歴を保存する"""
        if not filename:
            timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
            filename = f"conversation_{timestamp}.json"

        data = {
            "model": self.config["model"],
            "total_tokens": self.total_tokens,
            "estimated_cost": self.total_cost,
            "messages": self.messages,
            "saved_at": datetime.now().isoformat()
        }

        with open(filename, "w", encoding="utf-8") as f:
            json.dump(data, f, indent=2, ensure_ascii=False)

        return filename

    def clear_history(self):
        """会話履歴をクリアする"""
        self._init_system_prompt()
        self.total_tokens = 0
        self.total_cost = 0.0

    def show_stats(self):
        """統計情報を表示する"""
        print("\n" + "="*40)
        print("会話統計")
        print("="*40)
        print(f"モデル:{self.config['model']}")
        print(f"メッセージ数:{len(self.messages) - 1}")
        print(f"合計 tokens:{self.total_tokens}")
        print(f"推定費用:${self.total_cost:.4f}")
        print("="*40 + "\n")


def print_help():
    """ヘルプ情報を表示する"""
    print("\n" + "="*40)
    print("サンプル AI チャットツール - コマンド一覧")
    print("="*40)
    print("/help - ヘルプを表示")
    print("/clear - 会話履歴をクリア")
    print("/stats - 統計情報を表示")
    print("/save - 会話履歴を保存")
    print("/model - モデルを切り替え")
    print("/temp - temperature を調整")
    print("/system - システムプロンプトを変更")
    print("/quit - プログラムを終了")
    print("="*40 + "\n")


def main():
    print("="*40)
    print(" サンプル AI チャットツール")
    print("="*40)
    print("入力 /help でコマンドを表示、文字を直接入力してチャットを開始\n")

    bot = サンプルChatBot()

    while True:
        try:
            user_input = input("あなた:").strip()

            if not user_input:
                continue

            # コマンドを処理
            if user_input.startswith("/"):
                cmd = user_input.lower()

                if cmd in ["/quit", "/exit", "/q"]:
                    print("さようなら!")
                    break

                elif cmd == "/help":
                    print_help()

                elif cmd == "/clear":
                    bot.clear_history()
                    print("会話履歴はクリアされました")

                elif cmd == "/stats":
                    bot.show_stats()

                elif cmd == "/save":
                    filename = bot.save_conversation()
                    print(f"会話は保存されました:{filename}")

                elif cmd.startswith("/model "):
                    model = user_input[7:].strip()
                    bot.config["model"] = model
                    bot._save_config()
                    print(f"モデルは切り替わりました:{model}")

                elif cmd.startswith("/temp "):
                    try:
                        temp = float(user_input[6:].strip())
                        if 0 <= temp <= 2:
                            bot.config["temperature"] = temp
                            bot._save_config()
                            print(f"Temperature は設定されました:{temp}")
                        else:
                            print("Temperature は 0-2 の間である必要があります")
                    except ValueError:
                        print("有効な数字を入力してください")

                elif cmd.startswith("/system "):
                    system_prompt = user_input[8:].strip()
                    bot.config["system_prompt"] = system_prompt
                    bot.clear_history()
                    bot._save_config()
                    print("System prompt は更新され、会話はリセットされました")

                else:
                    print("不明なコマンドです。/help でヘルプを表示してください")

            else:
                # 通常チャット
                if not bot.config["stream"]:
                    print("AI:", end="")

                bot.chat(user_input)
                print()

        except KeyboardInterrupt:
            print("\n入力 /quit で終了するか、チャットを続けます)

        except Exception as e:
            print(f"エラーが発生しました:{e}")


if __name__ == "__main__":
    main()

設定ファイルを作成します:

{
    "api_key": "your-api-key-here",
    "model": "gpt-3.5-turbo",
    "system_prompt": "你是一个乐于助人的 AI 助手,由 サンプル 提供技术支持。",
    "max_history": 20,
    "stream": true,
    "temperature": 0.7
}

これでこのチャットツールを実行できます:

python サンプル_chatbot.py

次のように表示されます:

========================================
  サンプル AI 聊天工具
========================================
输入 /help 查看命令,直接输入文字开始聊天

你:你好
AI:你好!很高兴见到你。我是由 サンプル 提供技术支持的 AI 助手,有什么可以帮你的吗?

你:/stats

========================================
对话统计
========================================
模型:gpt-3.5-turbo
消息数:2
总 tokens:50
估算费用:$0.0001
========================================

你:/quit
再见!
その他の拡張