ChatGPT の API を業務システムに組み込む案件が一気に増えました。
プロトタイプは半日で動きます。しかし本番に載せると別の問題が出てきます。
実際に詰まった順に記録します。
1. 応答が遅い（そして時々もっと遅い）
数秒で返ることもあれば、30秒近くかかることもあります。
同期処理でリクエストの中から呼ぶと、Gunicorn のワーカーが専有され、
同時アクセスで簡単に詰まります。
非同期に逃がすのが基本です。
# 画面はすぐ返す。処理はキューへ
task = GenerationTask.objects.create(prompt=prompt, status="queued")
enqueue(task.id)
return JsonResponse({"task_id": task.id})
画面側はポーリングかストリーミングで受け取ります。
「待たせない」のではなく「待っていることが見える」ようにするのが実務的な解です。
2. 失敗する前提で組む
レート制限、タイムアウト、一時的な障害。必ず起きます。
指数バックオフでリトライしますが、無限にリトライしないことが重要です。
for attempt in range(3):
try:
return call_api(prompt)
except RateLimitError:
time.sleep(2 ** attempt) # 1s, 2s, 4s
except_final:
mark_failed(task) # 諦めてユーザーに伝える
失敗を握りつぶして「生成中」のまま放置するのが最悪の体験です。
3. 出力が毎回違う
同じ入力でも文面が変わります。これが業務システムでは厄介で、
後続処理がパースに失敗します。
出力形式を固定したい場合は、プロンプトで指定するだけでは不十分でした。
必ず検証層を挟みます。
try:
data = json.loads(response)
validate(data) # 期待するキーがあるか
except (json.JSONDecodeError, ValidationError):
# 1回だけ「形式を直して」と投げ直す
response = call_api(FIX_FORMAT_PROMPT + response)
4. コストが読めない
トークン単価は安く見えますが、長い文脈を毎回投げると一気に膨らみます。
最初から使用量を記録しておくと、後で効きます。
class ApiUsage(models.Model):
prompt_tokens = models.IntegerField()
completion_tokens = models.IntegerField()
model = models.CharField(max_length=40)
created_at = models.DateTimeField(auto_now_add=True)
「機能ごとの1リクエスト単価」が見えるようになると、
どこを短くすべきかが判断できます。
5. 入れてはいけない情報が入る
いちばん重要な論点です。ユーザーが入力欄に個人情報を書くことは普通にあります。
送信前にマスクするか、そもそも送らない設計にします。
text = re.sub(r"[\w.+-]+@[\w-]+\.[\w.]+", "[EMAIL]", text)
text = re.sub(r"\d{3}-?\d{4}-?\d{4}", "[PHONE]", text)
加えて、何を送ったかのログを残すこと。
後から「その情報は外部に出したのか」と聞かれたとき、答えられる必要があります。
まとめ
- 同期で呼ばない。キューに逃がして進捗を見せる
- リトライは有限回。失敗したことをユーザーに伝える
- 出力は必ず検証する。形式は信用しない
- トークン使用量を最初から記録する
- 送信前のマスクと送信ログを必ず持つ
デモと本番の差は、ほとんどがこの5点でした。