OpenAI APIを使い始めるときのリアルな落とし穴
以前、AuthenticationErrorのデバッグだけで半日潰したことがある。原因はAPIキーをコピーするときに余分なスペースが混入していただけだった。別の機会では、ローカルでは問題なく動いていたコードが、サーバーに上げたらリトライ処理がないせいで頻繁にタイムアウトしていた。OpenAI APIを使い始めるなら、この記事で自分が経験したハマりどころを事前に知っておいてほしい。
OpenAI APIはChat Completionsだけでなく、Embeddings、Image generation、Speech-to-textなど多様な機能を提供している。しかし実際のDevOpsチームでは、ユースケースの80%がChat Completionsによる自動化に集中している:コードレビュー、ドキュメント生成、エラーログの解析、レポートの要約といった作業だ。この記事ではその部分に絞って解説する。
事前に押さえておくべき基本概念
モデルと料金体系
OpenAIには複数のモデルがあり、それぞれ速度・コスト・品質のトレードオフが異なる:
- gpt-4o — 最も高性能、マルチモーダル(画像処理対応)、複雑なタスクに最適
- gpt-4o-mini — gpt-4oより約15倍安く、多くのシンプルなタスクで十分に機能する
- gpt-3.5-turbo — 旧世代で安価だが、gpt-4o-miniが完全にこの役割を引き継いでいると感じる
費用はトークンで計算される。トークンとはテキストの最小単位で(英語では約1 token ≈ 0.75単語;日本語は特殊文字が多いためトークンを多く消費する傾向がある)。input token(送信するプロンプト)とoutput token(返ってくるレスポンス)の2種類があり、料金が異なる。通常outputの方が高い。
メッセージの構造
Chat Completions APIは3種類のロールを持つ会話形式を使用する:
system— モデルの全体的な振る舞いを指示する(例:「あなたはシニアDevOpsエンジニアです」)user— ユーザー側またはコードからのメッセージassistant— モデルのレスポンス(会話履歴を維持する場合に使用)
実践的なステップ
ステップ1:インストールとAPIキーの取得
OpenAIの公式ライブラリをインストールする:
pip install openai
# またはuvを使う場合
uv add openai
APIキーはplatform.openai.com → API keys → Create new secret keyから取得する。重要な注意点:キーは作成時に一度しか表示されない。後から確認する方法はないため、必ず安全な場所に保存すること。
キーは環境変数に保存し、コードに直接書き込むことは絶対に避けること:
# .envファイル(.gitignoreに追加する)
OPENAI_API_KEY=sk-proj-xxxxxxxxxxxx
# またはシェルで直接exportする
export OPENAI_API_KEY=sk-proj-xxxxxxxxxxxx
ステップ2:基本的なAPI呼び出し
APIが正常に動作することを確認するための最小限のコード:
from openai import OpenAI
# クライアントは環境変数OPENAI_API_KEYを自動的に読み込む
client = OpenAI()
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=[
{"role": "system", "content": "あなたはDevOpsをサポートするアシスタントです。"},
{"role": "user", "content": "'docker system prune -af'コマンドが何をするか説明してください。"}
]
)
print(response.choices[0].message.content)
実行してみてoutputを確認しよう。AuthenticationErrorが発生した場合はAPIキーを確認する。echo $OPENAI_API_KEYで環境変数が設定されているか確認できる。
ステップ3:エラーハンドリングつき再利用可能な関数の作成
痛い経験から学んだ教訓:コードベース全体に20か所以上のAPI呼び出しが散在すると、リトライやロギングを追加するたびに全箇所を修正しなければならない。最初からwrapperを一度書いておくだけで、後々の頭痛を大幅に減らせる:
import time
import logging
from openai import OpenAI, RateLimitError, APITimeoutError, APIConnectionError
logging.basicConfig(level=logging.INFO)
log = logging.getLogger(__name__)
client = OpenAI(timeout=30.0) # タイムアウト30秒
def call_openai(
prompt: str,
system: str = "You are a helpful assistant.",
model: str = "gpt-4o-mini",
max_retries: int = 3
) -> str | None:
"""リトライロジックつきでOpenAI APIを呼び出す。"""
for attempt in range(max_retries):
try:
response = client.chat.completions.create(
model=model,
messages=[
{"role": "system", "content": system},
{"role": "user", "content": prompt}
],
temperature=0.7,
max_tokens=2000
)
return response.choices[0].message.content
except RateLimitError:
wait = 2 ** attempt # exponential backoff: 1s、2s、4s
log.warning(f"レート制限に達した。{attempt+1}/{max_retries}回目のリトライまで{wait}秒待機")
time.sleep(wait)
except (APITimeoutError, APIConnectionError) as e:
log.error(f"接続エラー: {e}、リトライ {attempt+1}/{max_retries}")
time.sleep(1)
except Exception as e:
log.error(f"予期しないエラー: {e}")
return None
log.error("リトライ上限に達した。スキップする。")
return None
# 使用例
result = call_openai(
prompt="このSQLを確認して、潜在的なN+1クエリを探してください: SELECT * FROM users WHERE id IN (...)",
system="あなたはシニアバックエンド開発者です。コードレビューは簡潔に、具体的な問題を指摘してください。"
)
if result:
print(result)
ステップ4:ストリーミングでリアルタイムにレスポンスを表示
レポートの作成やコード生成のような長い出力が必要なタスクでは、ストリーミングによってUXが大幅に向上する。完了を待つ代わりに、レスポンスが少しずつ表示されるようになる:
def call_openai_stream(prompt: str, model: str = "gpt-4o-mini") -> str:
"""レスポンスをストリーミングし、チャンクを順次出力してフルテキストを返す。"""
full_response = []
with client.chat.completions.create(
model=model,
messages=[{"role": "user", "content": prompt}],
stream=True
) as stream:
for chunk in stream:
delta = chunk.choices[0].delta.content
if delta:
print(delta, end="", flush=True)
full_response.append(delta)
print() # 末尾の改行
return "".join(full_response)
result = call_openai_stream("ディスク使用量を監視してアラートを送信するPythonスクリプトを書いてください。")
ステップ5:構造化されたJSONアウトプットの処理
テキストではなく構造化データとしてアウトプットをパースする必要がある場合は、response_formatを使用する:
import json
def analyze_log_entry(log_line: str) -> dict | None:
"""ログの1行を解析して構造化されたdictを返す。"""
try:
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=[
{
"role": "system",
"content": "ログエントリをJSONにパースする。返却形式: {severity, service, message, is_error}"
},
{"role": "user", "content": log_line}
],
response_format={"type": "json_object"},
temperature=0 # JSONアウトプットには決定論的な出力が必要
)
return json.loads(response.choices[0].message.content)
except Exception as e:
log.error(f"パースに失敗: {e}")
return None
# テスト
result = analyze_log_entry(
"2024-01-15 14:32:01 ERROR payment-service Connection timeout to database after 30s"
)
print(result)
# 出力: {'severity': 'ERROR', 'service': 'payment-service', 'message': 'Connection timeout...', 'is_error': True}
トークン使用量とコストのトラッキング
トークンのトラッキングは最初は見落とされがちだ。しかしスケールアップすると、コストが思った以上に急増する。500トークンのプロンプトで1日1万リクエストを処理するパイプラインは、インプットだけで月$20〜50を消費することもある。予期せぬ請求を避けるために、自分は最初からusageをログに記録するようにしている:
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": prompt}]
)
usage = response.usage
log.info(
f"トークン — 入力: {usage.prompt_tokens}, "
f"出力: {usage.completion_tokens}, "
f"合計: {usage.total_tokens}"
)
# gpt-4o-miniの料金: 入力~$0.15/1M、出力~$0.60/1M(platform.openai.comで最新料金を確認)
input_cost = usage.prompt_tokens * 0.15 / 1_000_000
output_cost = usage.completion_tokens * 0.60 / 1_000_000
log.info(f"推定コスト: ${input_cost + output_cost:.6f}")
まとめ
このAPIは技術的に難しくない。よく見落とされるのは4つのポイントだ:適切なモデルの選択、明確なsystem promptの記述、正しいエラーハンドリング、そして最初からコストをトラッキングすること——月末に予想外の請求書が届いてから気づくのでは遅すぎる。
自分のチームでは自動化タスクの約95%をgpt-4o-miniで動かしている。gpt-4oに比べて約15倍安く、速度も明らかに速い。画像処理や複雑な推論が必要なタスクだけ上位モデルに切り替えている。
さらに深く学びたいなら、次のステップとして:Function Calling(モデルが自分のツールを呼び出せるようにする)、Embeddings API(セマンティック検索の構築)、Assistants API(メモリを持つエージェントの作成)を調べるといい。それぞれがまったく異なるユースケースの扉を開いてくれる。

