前回の記事では、エラーの切り分けと再試行を扱いました。最後に「ログへ残す項目」を並べて終わっています。今回はそれを実際に動く形にします。
APIを1回試している段階なら、画面に出たものを見れば十分です。しかし夜間に100店舗を処理して朝こうなっていたとき、画面はもう残っていません。
朝、結果が「96店舗 成功、4店舗 失敗」だったとします。どの4店舗が、何時に、どのエラーで失敗したのか。 それを後から言えるかどうかが、ログがあるかどうかです。
先に、この記事の結論を3つ書いておきます。
- ログに書ける項目を、許可リストで縛ります。 「APIキーを書かない」をルールではなく仕組みにします
- 料金は、ログを書く時点で計算して入れます。 単価は改定されるので、後から当時のログを今の単価で計算し直すことはできません
- 速さは平均ではなく p95 で見ます。 平均は、遅い数件を見えなくします
この記事を読み終えると、次のことができるようになります。
print()とログの違いを説明できる- API呼び出しごとに記録すべき項目を決められる
- 成功時とエラー時の request ID を取得できる
- 接続エラーでログ処理自体が落ちる書き方を避けられる
- 処理時間・トークン数・概算料金をログへ残せる
- JSONL形式で、日付ごとのファイルへ追記できる
- APIキーや本文が構造的にログへ入らない設計にできる
- 成功率・p95・日別コストを1本のスクリプトで出せる
- 「昨日と今日で何が変わったか」をログから言える
先に、この記事で出てくる用語を整理します
ログの話は、用語が分からないだけで急に難しく感じます。この記事で使う言葉を先にまとめておきます。いま全部覚える必要はありません。 本文中でも初出のたびに説明しますので、分からなくなったらここへ戻ってきてください。
| 用語 | この記事での意味 |
|---|---|
| ログ | 後から調べるための記録。画面表示と違って残ります |
| 構造化ログ | 文章ではなく、項目を決めて記録したログ。集計できます |
| JSONL | 1行に1件のJSONを書く形式。JSON Lines の略。末尾に追記できます |
| レコード | ログの1行。ここではAPI呼び出し1回ぶん |
| request ID | リクエスト1件ごとにOpenAIが振る識別子。問い合わせや調査に使います |
| レイテンシ | 応答時間。この記事ではミリ秒(ms)で記録します |
| p95 | 速い順に並べて95%目の値。「悪いほうの5%を除いた最悪値」の目安 |
| 許可リスト | 「これだけ通す」と決めた一覧。逆は禁止リスト(「これは通さない」) |
| マスク | 記録するときに一部を伏せること。山田*** のような形 |
| 保存期間 | ログを何日残すか。決めないと増え続けます |
| UTC | 世界共通の時刻。日本時間より9時間前です |
| 推論トークン | モデルが答える前に内部で考えた分。画面に出ませんが課金されます |
print では足りなくなる場所
学習のあいだは、これで十分です。
print(response.output_text)
足りなくなるのは、人が画面を見ていない時間帯に動かし始めたときです。
いちばん下の行が見落とされがちです。ログは、書いた本人以外が読むものです。 だから項目を揃えます。そして、だから見せてはいけないものを入れてはいけません。
まず「書いてよい項目」を決める
ふつうの順番は「何を記録したいか」から考えることですが、この記事では逆から始めます。先に、書いてよい項目の一覧を決めます。
許可リストにする理由
「APIキーはログに書かない」「個人情報は書かない」というのは、そのとおりです。しかし、これは覚えておく系のルールです。覚えておく系のルールは、忙しいときに破られます。
- デバッグのため、いったんプロンプト全文も出しておこう
- 原因が分かった
- その1行を消し忘れる
- 半年後、顧客名の入ったログが18万行たまっている
そこで、禁止リスト(これは書かない)ではなく許可リスト(これだけ書く)にします。書いてよい項目を明示的に列挙し、それ以外が渡されたら書き込みを拒否します。
この記事で使う項目
許可する項目 = {
# いつ・何を
"timestamp", # 実行時刻(UTC)
"job", # 処理名。nightly_report など
"model", # モデル名
"analysis_id", # アプリ側で振るID(後述)
# 結果
"request_id", # OpenAI が振るID
"status", # success / error
"attempts", # 何回目の試行で確定したか
"latency_ms", # 応答時間
# エラーのとき
"error_type", # 例外クラス名
"status_code", # HTTPステータス
"error_code", # error.code
# 使用量とお金
"input_tokens",
"output_tokens",
"reasoning_tokens",
"total_tokens",
"cost_usd",
}
16項目です。 ここに prompt も output_text も api_key もありません。入れていないので、書けません。
| 項目 | 何のために残すか |
|---|---|
timestamp |
いつ起きたか。日別集計の軸にもなります |
job |
夜間バッチと画面からの問い合わせを分けるため。混ざると集計が意味を失います |
request_id |
個別調査とサポートへの問い合わせ |
attempts |
再試行が増えていないかの監視。 前回の記事で作った試行回数です |
latency_ms |
遅くなっていないか |
error_code |
同じ429でも「待てば直る」か「直らない」かの区別 |
reasoning_tokens |
料金が増えた原因の切り分け。 出力が伸びたのか、推論が伸びたのか |
cost_usd |
日別・月別の費用。その時点の単価で計算した値 |
JSONL で日付ごとに書く
JSONL とは
JSONL(JSON Lines)は、1行に1件のJSONを書く形式です。JSON配列と違って、末尾に1行足すだけで追記できます。
{"timestamp":"2026-09-18T02:00:00+00:00","status":"success","latency_ms":3140}
{"timestamp":"2026-09-18T02:00:07+00:00","status":"error","error_code":"server_is_overloaded"}
{"timestamp":"2026-09-18T02:00:14+00:00","status":"success","latency_ms":2980}
JSON配列([...])だと、追記のたびに末尾の ] を消して書き足して閉じ直す必要があります。JSONLならその手間がありません。人もそのまま読めます。
書き込む関数
import json
from datetime import date
from pathlib import Path
LOGDIR = Path("logs")
LOGDIR.mkdir(exist_ok=True)
def ログを書く(record: dict) -> None:
はみ出し = set(record) - 許可する項目
if はみ出し:
raise ValueError(f"ログに書けない項目です: {sorted(はみ出し)}")
path = LOGDIR / f"api_{date.today():%Y%m%d}.jsonl"
with path.open("a", encoding="utf-8") as f:
f.write(json.dumps(record, ensure_ascii=False) + "\n")
短いですが、3つのことをしています。
- 許可リストの外の項目があれば、その場で例外を出します。 静かに捨てるのではなく止めます。捨てると、書けていないことに気づけません
- ファイル名に日付を入れています。
api_20260918.jsonlのようになります "a"は追記モードです。すでにある内容を消さずに末尾へ足します
日付で分ける理由
1つのファイルに書き続けると、行数が増え続けます。1日1,000件なら、365日で365,000行です。読み込みも遅くなりますし、「90日より古いものを消す」がやりにくくなります。日付でファイルが分かれていれば、古いファイルを消すだけです。
from datetime import date, timedelta
保存日数 = 90
期限 = date.today() - timedelta(days=保存日数)
for path in LOGDIR.glob("api_*.jsonl"):
日付 = path.stem.removeprefix("api_") # "20260918"
if 日付 < f"{期限:%Y%m%d}":
path.unlink()
print("削除:", path.name)
日付が YYYYMMDD の8桁なので、文字列のまま比較できます。日付型に変換しなくても大小が正しく並びます。
ensure_ascii=Falseを付けているのは、日本語をそのまま書くためです。付けないと店舗のような形になり、人が読めなくなります。
request ID を取る
成功したとき
response = client.responses.create(
model="gpt-5.6-luna",
input="先週の売上を2文以内で要約してください。",
)
print(response._request_id) # req_abc123...
先頭にアンダースコアが付いていますが、これは公開されたプロパティです。SDKのドキュメントにも記載があります。ふつうアンダースコアは「内部向け」の印なので、ここだけ例外だと覚えてください。
_request_idが付くのはトップレベルのレスポンスだけです。response.usage._request_idと書くとAttributeErrorになります。中の入れ子のオブジェクトには付いていません。
エラーになったとき
try:
response = client.responses.create(...)
except openai.APIStatusError as exc:
print(exc.request_id) # req_abc123...
print(exc.status_code) # 429
print(exc.code) # rate_limit_exceeded
APIStatusError は4xx・5xxの例外すべての親クラスなので、これ1つで401でも429でも503でも取れます。
接続エラーでは、属性そのものが存在しない
ここが、ログ処理でいちばん踏みやすい落とし穴です。
ネットワークにつながらなかった場合、リクエストはOpenAIに届いていません。IDも振られていません。それだけなら None が返ればよさそうですが、SDKの APIConnectionError には request_id という属性自体がありません。
>>> exc = openai.APIConnectionError(request=req) >>> exc.request_id AttributeError: 'APIConnectionError' object has no attribute 'request_id' >>> exc.status_code AttributeError: 'APIConnectionError' object has no attribute 'status_code'
つまり、次のように「まとめて捕まえて楽をしよう」と書くと壊れます。
except openai.APIError as exc: # 全部まとめて捕まえる
ログを書く({
"request_id": exc.request_id, # ← 接続エラーだとここで AttributeError
"status_code": exc.status_code, # ← ここも
...
})
しかもこれはネットワークが切れたときにしか起きません。平常時のテストでは絶対に見つからず、障害のときだけログ処理そのものが落ちます。本来のエラーも、ログも、両方失われます。
対処は単純で、例外の種類ごとに except を分けます。
| 例外 | request_id |
status_code |
code |
|---|---|---|---|
APIStatusError(4xx・5xx) |
あり | あり | あり |
APIConnectionError |
属性なし | 属性なし | None |
APITimeoutError |
属性なし | 属性なし | None |
APITimeoutErrorはAPIConnectionErrorのサブクラスです。except openai.APIConnectionErrorと書けばタイムアウトもそこで捕まります。
料金は、書くときに計算する
トークン数だけ残して、料金は後から集計する——という設計にしたくなります。しかし、これには問題があります。
単価は改定されます。 3か月前のログを今の単価で計算し直しても、当時いくらかかったかは出ません。
そこで、ログを書く時点の単価で計算した金額を、そのまま cost_usd として残します。前回までの記事で作った関数がそのまま使えます。
IN_USD_PER_1M = 0.20
CACHED_USD_PER_1M = 0.02
OUT_USD_PER_1M = 1.20
def 概算料金(usage) -> float:
cached = usage.input_tokens_details.cached_tokens or 0
通常入力 = usage.input_tokens - cached
return round(通常入力 / 1_000_000 * IN_USD_PER_1M
+ cached / 1_000_000 * CACHED_USD_PER_1M
+ usage.output_tokens / 1_000_000 * OUT_USD_PER_1M, 8)
round(..., 8) としているのは、1回あたりが小数点以下6桁の世界だからです。4桁で丸めると、全部 0.0015 になって差が消えます。
単価を変えたときは、ログにも単価を残すか、変更日をどこかに記録しておくと後から追えます。
cost_usdだけだと「なぜこの日から単価が違うのか」が分からなくなります。
1行が何を表しているのかを、はっきりさせる
ここは設計の分かれ目なので、先に決めておきます。
前回の記事で確認したとおり、SDKは1回の create() の中で、既定で2回まで自動的に再試行します。つまりこういうことが起きます。
こちらから見れば create() を1回呼んだだけですが、その内側では「送信 → 503 → 0.5秒待つ → 送信 → 503 → 1.0秒待つ → 送信 → 成功」が起きています。ログは1行の「成功」でも、HTTPリクエストは3本、1.5秒の待機つきです。
この記事では、ログの1行を「業務としての1回の呼び出し」と決めます。上の例なら成功1行です。そのうえで、次の点をはっきりさせておきます。
上の図の下3行が、この設計で気をつける点です。とくに latency_ms は呼び出し全体の時間であって、APIの素の速さではありません。「昨日より遅い」を見るには十分ですが、「APIが何ミリ秒で返したか」の値としては使えません。
「APIの素の速さが知りたい」「失敗した試行のIDも全部欲しい」という場合は、max_retries=0 にして自前ループの各回を1行ずつ記録する設計になります。ただし前回の記事のとおり、それをすると Retry-After への対応を自分で書き直すことになります。監視のために可観測性を上げるか、実装をSDKに任せるかのトレードオフです。
まずはこの記事の形(1呼び出し=1行)で始めてください。 ほとんどの用途では、これで「昨日より遅い」「失敗が増えた」までは分かります。足りないと分かってから細かくすれば十分です。
ログを組み込んだ呼び出し関数
前回の記事で作った再試行ラッパーに、ログを足します。
import time
from datetime import datetime, timezone
import openai
from openai import OpenAI
MODEL = "gpt-5.6-luna"
client = OpenAI(timeout=30.0) # max_retries は既定の 2 のまま
待っても直らない = {
"credit_balance_exhausted",
"organization_spend_limit_exceeded",
"project_spend_limit_exceeded",
"organization_usage_limit_exceeded",
}
def 呼ぶ(入力, job, 制限秒=60.0):
開始 = time.perf_counter()
いま = datetime.now(timezone.utc).isoformat(timespec="seconds")
期限 = time.monotonic() + 制限秒
待ち, 試行 = 2.0, 0
while True:
試行 += 1
try:
r = client.responses.create(model=MODEL, input=入力)
u = r.usage
ログを書く({
"timestamp": いま, "job": job, "model": MODEL,
"request_id": r._request_id,
"status": "success",
"attempts": 試行,
"latency_ms": round((time.perf_counter() - 開始) * 1000),
"input_tokens": u.input_tokens,
"output_tokens": u.output_tokens,
"reasoning_tokens": u.output_tokens_details.reasoning_tokens or 0,
"total_tokens": u.total_tokens,
"cost_usd": 概算料金(u),
})
return r
except openai.APIStatusError as exc:
直前 = exc
情報 = {"request_id": exc.request_id,
"status_code": exc.status_code,
"error_code": exc.code}
諦める = (exc.code in 待っても直らない
or not isinstance(exc, (openai.RateLimitError,
openai.InternalServerError,
openai.ConflictError)))
except openai.APIConnectionError as exc:
直前 = exc
# 接続エラーには request_id も status_code も「属性が無い」
情報 = {"request_id": None, "status_code": None, "error_code": None}
諦める = False
残り = 期限 - time.monotonic()
if 諦める or 残り <= 待ち:
ログを書く({
"timestamp": いま, "job": job, "model": MODEL,
"status": "error",
"attempts": 試行,
"error_type": type(直前).__name__,
"latency_ms": round((time.perf_counter() - 開始) * 1000),
**情報,
})
raise 直前
time.sleep(待ち)
待ち = min(待ち * 2, 30.0)
- ログを書くのは2箇所だけです。 成功して返すところと、諦めるところ。途中の再試行では書きません。1呼び出し=1行を守るためです
情報という辞書に、例外の種類ごとの差を閉じ込めています。 こうすると、ログを書く側は「接続エラーかどうか」を気にしなくて済みますいまは呼び出しの開始時刻です。 再試行で1分かかっても、ログの時刻は開始時刻のままになります。「いつ始まった処理か」で並べたいからですjobを引数で受け取っています。 夜間バッチと画面からの問い合わせが同じファイルに混ざっても、後で分けられます
ログの書き込みが失敗したらどうするか。 ディスクが一杯だったり権限が無かったりすると、
ログを書く()自体が例外を出します。エラー処理の中でそれが起きると、本来のAPIエラーが上書きされて消えます。本番では、ログの書き込みをtryで包んで、失敗しても本来の例外を優先する形にしてください。
できあがるログ
2日ぶんの夜間バッチ(1日100店舗)を流すと、こうなります。
logs/ api_20260917.jsonl 100行 api_20260918.jsonl 100行
中身は1行ずつこうなっています(読みやすいよう改行しています。実際は1行です)。
{"timestamp": "2026-09-18T02:00:00+00:00",
"job": "nightly_report",
"model": "gpt-5.6-luna",
"request_id": "req_403915782",
"status": "success",
"attempts": 1,
"latency_ms": 3140,
"input_tokens": 431,
"output_tokens": 1338,
"reasoning_tokens": 1174,
"total_tokens": 1769,
"cost_usd": 0.00169166}
失敗した行はこうなります。
{"timestamp": "2026-09-18T02:03:30+00:00",
"job": "nightly_report",
"model": "gpt-5.6-luna",
"status": "error",
"attempts": 4,
"error_type": "InternalServerError",
"latency_ms": 18420,
"request_id": "req_915330268",
"status_code": 503,
"error_code": "server_is_overloaded"}
成功の行と失敗の行で、項目が違います。 JSONLは1行ごとに独立しているので、これで問題ありません。集計するときは、無い項目を前提にしないように書きます。
集計する
下書きの段階でよくあるのが、必要になるたびに value_counts() や mean() を手で打つことです。毎朝見るものなら、1本のスクリプトにしておきます。
読み込みと集計
import json
import statistics
from collections import Counter
from pathlib import Path
def 読み込む(ディレクトリ="logs"):
記録 = []
for path in sorted(Path(ディレクトリ).glob("api_*.jsonl")):
with path.open(encoding="utf-8") as f:
記録 += [json.loads(行) for 行 in f if 行.strip()]
return 記録
def 分位(値の並び, q):
xs = sorted(値の並び)
return xs[min(int(len(xs) * q), len(xs) - 1)]
def レポート(記録):
成功 = [r for r in 記録 if r["status"] == "success"]
失敗 = [r for r in 記録 if r["status"] == "error"]
print(f"件数 {len(記録)} (成功 {len(成功)} / 失敗 {len(失敗)})")
print(f"成功率 {len(成功) / len(記録):.1%}")
if 成功:
lat = [r["latency_ms"] for r in 成功]
print(f"応答時間 中央値 {分位(lat, .5):,}ms"
f" / p95 {分位(lat, .95):,}ms"
f" / 最大 {max(lat):,}ms")
print(f"コスト 合計 ${sum(r['cost_usd'] for r in 成功):.4f}"
f" / 1件あたり ${statistics.mean(r['cost_usd'] for r in 成功):.6f}")
推論 = sum(r["reasoning_tokens"] for r in 成功)
出力 = sum(r["output_tokens"] for r in 成功)
print(f"推論割合 出力トークンの {推論 / 出力:.0%}")
再試行 = [r for r in 記録 if r["attempts"] > 1]
print(f"再試行 {len(再試行)} 件({len(再試行) / len(記録):.0%})")
if 失敗:
print("失敗の内訳")
内訳 = Counter((r["error_type"], r.get("error_code")) for r in 失敗)
for (種類, コード), 件数 in 内訳.most_common():
print(f" {件数:3d} 件 {種類}" + (f" / {コード}" if コード else ""))
実行するとこうなります。
件数 200 (成功 193 / 失敗 7)
成功率 96.5%
応答時間 中央値 3,140ms / p95 5,400ms / 最大 5,905ms
コスト 合計 $0.3535 / 1件あたり $0.001832
推論割合 出力トークンの 89%
再試行 13 件(6%)
失敗の内訳
7 件 InternalServerError / server_is_overloaded
平均ではなく p95 を見る
応答時間を平均で見てはいけません。 平均は、遅い数件を薄めてしまいます。
p95 は「速い順に並べて95%目の値」です。100件なら95番目。「20回に1回はこれくらい待たされる」 という読み方をします。画面で人が待っている処理では、平均よりこちらが体感に近くなります。
def 分位(値の並び, q):
xs = sorted(値の並び)
return xs[min(int(len(xs) * q), len(xs) - 1)]
並べ替えて位置を取るだけです。min(..., len(xs) - 1) は、q=1.0 のときに範囲外にならないようにするためのものです。件数が少ないうちは目安程度に見てください。 10件でp95を出しても、実質は最大値です。
日別に並べる
ここからが本題です。1日ぶんを見ても「これが普通なのか」は分かりません。 並べて初めて分かります。
def 日別(記録):
日ごと = {}
for r in 記録:
日ごと.setdefault(r["timestamp"][:10], []).append(r)
見出し = ("日付", "件数", "成功率", "p95", "推論/出力", "コスト")
print(f"{見出し[0]:12s}{見出し[1]:>6s}{見出し[2]:>9s}"
f"{見出し[3]:>10s}{見出し[4]:>10s}{見出し[5]:>11s}")
for 日, rs in sorted(日ごと.items()):
成功 = [r for r in rs if r["status"] == "success"]
p95 = 分位([r["latency_ms"] for r in 成功], .95) if 成功 else 0
推論率 = (sum(r["reasoning_tokens"] for r in 成功)
/ sum(r["output_tokens"] for r in 成功)) if 成功 else 0
コスト = sum(r.get("cost_usd", 0) for r in 成功)
print(f"{日:12s}{len(rs):>6d}"
f"{len(成功)/len(rs):>9.1%}{p95:>9,}ms"
f"{推論率:>10.0%}{コスト:>10.4f}$")
r["timestamp"][:10] は、"2026-09-18T02:00:00+00:00" の先頭10文字を取って "2026-09-18" にしています。ISO形式の日時は先頭から日付なので、文字列を切るだけで日付が取れます。
ログから「何が変わったか」を言う
先ほどのスクリプトを2日ぶんに当てると、こうなります。
日付 件数 成功率 p95 推論/出力 コスト 2026-09-17 100 97.0% 3,433ms 86% 0.1531$ 2026-09-18 100 96.0% 5,641ms 90% 0.2004$
この2行から、次のことが言えます。
| 気づくこと | ログのどの項目から |
|---|---|
| 応答が1.6倍遅くなった | latency_ms の p95(3,433 → 5,641ms) |
| コストが1.3倍になった | cost_usd の合計($0.1531 → $0.2004) |
| 原因は出力が伸びたこと | reasoning_tokens の割合(86% → 90%) |
| 成功率はほぼ変わっていない | status(97.0% → 96.0%) |
3行目が、この記事でいちばん言いたいところです。
遅くなって高くなったとき、原因の候補はいくつもあります。APIが混んでいた、ネットワークが遅い、リクエストが増えた、モデルが変わった——。しかし reasoning_tokens が残っていれば、「モデルが前より長く考えるようになった。だから遅くて高い」 と1つに絞れます。遅さとコストが同じ原因から来ていることも分かります。
そして、月30日に広げるとこうなります。
9/17基準なら $0.1531 × 30 = $4.59、9/18基準なら $0.2004 × 30 = $6.01 です。1日では$0.05の差ですが、月では$1.42です。ログを取っていなければ、この変化には請求が来るまで気づきません。
秘密情報が入っていないことを確かめる
許可リストを入れたので、仕組みとしては入らないはずです。実際に確かめます。
許可リストが働くかを試す
危ない = [
{"status": "success", "api_key": "sk-xxxxxxxxxxxxxxxx"},
{"status": "success", "prompt": "顧客名:山田太郎 売上:281,020円"},
{"status": "success", "output_text": "1週間の売上合計は281,020円で…"},
{"status": "success", "Authorization": "Bearer sk-xxxxxxxxxxxxxxxx"},
{"status": "success", "input_tokens": 420},
]
for r in 危ない:
try:
ログを書く(r)
print("書けた ", list(r))
except ValueError as exc:
print("拒否 ", exc)
結果です。
拒否 ログに書けない項目です: ['api_key'] 拒否 ログに書けない項目です: ['prompt'] 拒否 ログに書けない項目です: ['output_text'] 拒否 ログに書けない項目です: ['Authorization'] 書けた ['status', 'input_tokens']
4つとも書き込む前に止まっています。 5つ目だけが通りました。「気をつける」ではなく「通らない」状態になっていることが確認できました。
ここで使っているAPIキーの文字列は
sk-xxxxxxxxxxxxxxxxというダミーです。テストコードにも本物のキーは書かないでください。
できたログを検査する
念のため、書き上がったファイルも見ます。
import re
raw = "".join(p.read_text(encoding="utf-8")
for p in Path("logs").glob("api_*.jsonl"))
print("sk- で始まる文字列:", re.findall(r"sk-[A-Za-z0-9_\-]{8,}", raw) or "なし")
print("Bearer トークン :", re.findall(r"Bearer\s+\S+", raw) or "なし")
現れた項目 = sorted({k for r in 読み込む() for k in r})
print("許可リスト外の項目:", sorted(set(現れた項目) - 許可する項目) or "なし")
sk- で始まる文字列: なし Bearer トークン : なし 許可リスト外の項目: なし
この3行を、ログ設計を変えたときに毎回回してください。 項目を1つ足したときに、うっかり本文を入れてしまうのを防げます。
本文を追いたい場合は、IDでつなぐ
「どうしてもプロンプトと回答を確認したい」という場面はあります。そのときも、ログ本体には入れません。 代わりにIDでつなぎます。
analysis_id = "analysis_20260918_042" 呼ぶ(入力, job="nightly_report") # ログには analysis_id だけを入れる
ログ側にはIDだけが残ります。
{"analysis_id": "analysis_20260918_042", "request_id": "req_403915782", ...}
実データ(プロンプトと回答)は、アクセス権を絞った別の場所に置きます。こうすると次のようになります。
| APIログ | 実データの保管先 | |
|---|---|---|
| 中身 | メタデータだけ | プロンプト・回答 |
| 誰が見るか | 運用担当・開発者 | 限られた人だけ |
| 保存期間 | 90日 | もっと短く |
| つなぐ鍵 | analysis_id |
|
「調査のときだけ、権限のある人が突き合わせる」形になります。ログを広く共有しても、本文は漏れません。
この記事で扱っていないこと
複数のプロセスから同時に書く場合
今回の ログを書く() は、1つのプロセスから順番に書く前提です。複数のプロセスが同じファイルへ同時に追記すると、行が混ざることがあります。 並列で処理するようになったら、プロセスごとにファイルを分けるか、ログ専用のライブラリを使ってください。
Python標準の logging
Pythonには logging という標準のモジュールがあります。出力先の切り替えやレベル分けができて便利ですが、今回のように「決まった項目を1行のJSONで残す」だけなら、自前の関数のほうが短く、何が書かれるか読めば分かります。
規模が大きくなったら、次の順で移っていけます。
- JSONLファイル(今回)
- 出力先を切り替えたくなったら → Pythonの
logging - 複数サーバーになったら → クラウドのログサービス
- すぐ気づきたくなったら → 監視・アラート
最初からいちばん下を目指す必要はありません。 JSONLファイル1つで、この記事の集計はすべてできています。
まとめ
APIを繰り返し動かすなら、print() ではなく記録を残します。今回の形はこうです。
- API呼び出し
- 成功/失敗を判定
- 許可リストの項目だけを辞書にする
- 日付ごとのJSONLへ1行追記
| 覚えておくこと | |
|---|---|
| 何を書くか | 許可リストで決める。 「書かない」ルールではなく「書けない」仕組みにする |
| 料金 | ログを書く時点で計算して入れる。 単価は改定される |
| request ID | 成功は response._request_id、エラーは exc.request_id。接続エラーには属性が無い |
| 1行の意味 | 1回の呼び出し。 SDK内部の再試行は latency_ms に含まれ、見えない |
| 速さの見方 | 平均ではなく p95。 平均は遅い数件を薄める |
| 確かめ方 | 危ない辞書を渡して拒否されることを確認する。ファイルも正規表現で検査する |
そして、ログが効くのは調査のときだけではありません。2日並べるだけで「遅くなった」「高くなった」「原因は推論が伸びたこと」まで言えます。 請求が来てから気づくのとは、打てる手が変わります。
