運用を始めると「何を記録すれば良いか」「どこまで自動化できるか」で迷うことが多いです。本記事は、監査ログ（予測の起点・入力・モデルバージョン・出力）を確実に残し、SLO/SLIを定義して日次・週次チェックをPythonで自動化する実務手順をまとめます。現場でそのまま試せるスキーマ例、チェックリスト、簡易ランブック、サンプルコードと運用上の注意点を提供します。
目次
- 概要と要件定義
- 監査ログ設計：JSONLスキーマ例と必須フィールド
- SLO/SLIの定義と閾値設計
- Pythonで作る日次／週次自動チェック（サンプルスクリプト）
- ランブックと対応フロー
- CI/スケジューラへの組み込みとレポート配信
- 運用でよくある落とし穴と改善ループ
- まとめと次につなげる企画案
① 概要と要件定義
まず最初に決めるのは「どのイベントを残すか」と「保持期間・マスキング方針」です。実務では記録不足で原因追跡ができないことがよくあるため、後から必要になりそうな情報は最小限でも残す方針が安全です。以下のポイントを決めてから実装を始めてください。
- 記録するイベント：リクエスト受信、前処理、モデル呼び出し、応答、エラー、モデルデプロイ／ロールバック
- 保持期間：法規や契約に従う。一般的には30日～1年を検討（個人情報含む場合は短縮）
- マスキング：個人情報（PII）は保存しないかハッシュ化／部分マスクする
- アクセス制御：監査ログは読み取り権限を限定
- 改ざん防止：ハッシュや署名で改ざん検知できる仕組みを検討
② 監査ログ設計：JSONLスキーマ例と必須フィールド
実務では行指向（JSONL）が扱いやすく、ストリーム処理や検索にも適しています。まずは必須フィールドを定義し、運用に必要なメタデータを含めます。
JSONLスキーマ（例）
{"timestamp": "2026-08-31T09:12:34.123Z", "request_id": "req_0001", "user_id_hash": "sha256:...", "model": "predictor", "model_version": "v1.2.3", "input_summary": "token_count=45, features_ok=true", "input_hash": "sha256:...", "prediction": "approve", "confidence": 0.87, "latency_ms": 210, "error": null, "provenance": {"pipeline_version": "p1.0"}, "signature": "hmac:..."}上記の1行が1レコード（JSONL）です。運用ではgzipで圧縮して保管することもあります。
| フィールド | 説明 | 必須 |
|---|---|---|
| timestamp | UTCタイムスタンプ（ISO8601） | 必須 |
| request_id | リクエストを一意に識別するID | 必須 |
| user_id_hash | 個人を特定しないハッシュ／マスク済みID | 必須（PIIは非保存が原則） |
| model / model_version | 使用したモデルとバージョン | 必須 |
| input_hash / input_summary | 入力のハッシュと簡易要約（全文保存しない場合） | 必須 |
| prediction / confidence | 出力と確信度 | 必須 |
| latency_ms | 推論所要時間（ミリ秒） | 推奨 |
| error | エラー情報（発生時） | 推奨 |
| provenance | 処理パイプラインや前処理のバージョン | 推奨 |
| signature | ログ改ざん検知用の署名（省略可） | オプション |
③ SLO/SLIの定義と閾値設計
SLO（Service Level Objective）は運用目標、SLI（Service Level Indicator）はそれを測る指標です。現実的な閾値を決めるためには、基線となるメトリクスを数週間観察することが重要です。
| 指標（SLI） | 式（例） | SLO目標例 |
|---|---|---|
| 応答時間（P95 latency） | 95パーセンタイルの遅延（ms） | P95 < 500ms（可用性：99%） |
| エラー率 | エラー数 / 総リクエスト数 | < 1%（月間） |
| データ品質（欠損率） | 欠損あるいは不整合レコードの割合 | < 0.5%（週次） |
SLOは技術的に達成可能で、かつビジネスに意味のある水準で設定します。最初は緩めに設定して運用で引き締めるのが現実的です。
SLOを計算するサンプル関数（Python）
def calc_error_rate(records):
total = len(records)
errors = sum(1 for r in records if r.get("error"))
return errors / total if total else 0.0
def calc_p95_latency(latencies):
if not latencies:
return None
latencies = sorted(latencies)
import math
idx = math.ceil(0.95 * len(latencies)) - 1
return latencies[idx]④ Pythonで作る日次／週次自動チェック（サンプルスクリプト付き）
ここでは、JSONLログを読み込み、SLOを計算してCSVと簡易HTMLレポートを出力し、Slackへ送る軽量な流れを示します。標準ライブラリ中心で書き、必要に応じてpandas導入を検討します。
ポイントで使う標準要素
- logging と RotatingFileHandler（ログローテーション）
- json / csv モジュール
- datetime, timezone（UTC管理）
- pathlib（ファイル操作）
- 例外処理と簡易リトライ
サンプル：日次チェックの簡易スクリプト
import logging
from logging.handlers import RotatingFileHandler
import json
from pathlib import Path
from datetime import datetime, timezone
import csv
import requests
# ログ設定
logger = logging.getLogger("audit_checker")
handler = RotatingFileHandler("logs/audit_checker.log", maxBytes=5*1024*1024, backupCount=3)
formatter = logging.Formatter("%(asctime)s %(levelname)s %(message)s")
handler.setFormatter(formatter)
logger.addHandler(handler)
logger.setLevel(logging.INFO)
DATA_DIR = Path("/var/logs/audit")
REPORT_DIR = Path("/var/reports/audit")
REPORT_DIR.mkdir(parents=True, exist_ok=True)
# JSONLを読み込むユーティリティ
def read_jsonl(path):
with path.open("r", encoding="utf-8") as f:
for line in f:
try:
yield json.loads(line)
except json.JSONDecodeError:
logger.warning(f"invalid json: {path} : {line[:100]}")
# 集計処理
def aggregate_for_period(paths):
records = []
latencies = []
for p in paths:
for r in read_jsonl(p):
records.append(r)
if r.get("latency_ms") is not None:
latencies.append(r["latency_ms"])
return records, latencies
# SLO計算（先ほどの関数を活用）
def calc_error_rate(records):
total = len(records)
errors = sum(1 for r in records if r.get("error"))
return errors / total if total else 0.0
def calc_p95_latency(latencies):
if not latencies:
return None
latencies = sorted(latencies)
import math
idx = math.ceil(0.95 * len(latencies)) - 1
return latencies[idx]
# レポート出力
def write_csv_report(path, summary):
with path.open("w", newline="", encoding="utf-8") as f:
writer = csv.writer(f)
writer.writerow(["metric", "value"])
for k, v in summary.items():
writer.writerow([k, v])
# Slackへ送る（軽量な例）
def post_slack(webhook_url, text):
try:
requests.post(webhook_url, json={"text": text}, timeout=5)
except Exception as e:
logger.warning(f"slack post failed: {e}")
# メイン処理（例：前日分）
def run_daily_check(date_str, slack_webhook=None):
# 日付文字列 -> ファイルを選ぶルールは運用で決める
day_file = DATA_DIR / f"audit_{date_str}.jsonl"
records, latencies = aggregate_for_period([day_file])
error_rate = calc_error_rate(records)
p95 = calc_p95_latency(latencies)
summary = {"date": date_str, "total_requests": len(records), "error_rate": error_rate, "p95_latency_ms": p95}
csv_path = REPORT_DIR / f"summary_{date_str}.csv"
write_csv_report(csv_path, summary)
logger.info(f"daily check done: {summary}")
if slack_webhook:
post_slack(slack_webhook, f"Daily audit report {date_str}: {summary}")
# 実行例
if __name__ == '__main__':
today = datetime.now(timezone.utc).strftime('%Y-%m-%d')
run_daily_check(today, slack_webhook=None)上記は最小限の例です。実運用ではログのローテーション、古いログの削除、圧縮、取得失敗時のリトライなどを追加してください。
⑤ ランブックと対応フロー（トリアージ、エスカレーション、復旧手順）
平時から復旧までの手順を文章化しておくと、担当が不在でも対応が回ります。以下は簡易ランブックの例です。
| アラート種別 | 優先度 | 初動対応（担当） | エスカレーション先 |
|---|---|---|---|
| SLO違反（エラー率急上昇） | 高 | ログ確認、直近のデプロイ確認（オンコール） | 開発リード／インフラ担当 |
| モデルパフォーマンス劣化（急低下） | 中 | サンプルの入力と出力を抽出して比較（運用担当） | MLエンジニア |
| ログ書き込み失敗 | 中 | ディスク容量・権限を確認（運用担当） | インフラ担当 |
トリアージのポイント
- まずは影響範囲（ユーザー数、バッチ処理かリアルタイムか）を評価する
- 短時間で復旧可能かを判断し、ロールバックや一時遮断を検討する
- 根本原因調査は別タスクにして、まずサービス維持を優先する
⑥ CI/スケジューラへの組み込みとレポート配信
定期チェックはcronやGCP Cloud Scheduler、GitHub Actionsなどに組み込みます。成果物はCSV/HTMLレポートとして保存し、Slackやメールで配信します。レポートは短く要点を示す形式が受け入れられやすいです。
軽量なSlack送信例（requests使用）は前述の post_slack を参照してください。メール送信はSMTPライブラリ、あるいは外部サービスのAPI利用を検討します。
⑦ 運用でよくある落とし穴と改善ループ
- 大量ログのコスト増：すべて保存せずサンプリングや条件付き保存（エラー時のみ詳細を保存）を設計する
- プライバシー：PIIは生データで保存しない。ハッシュ化や部分マスキングの具体例をポリシー化する
- 改ざん対策：ログに単純なHMACを付与して改ざん検知の第一歩とする（秘密鍵は別管理）
- SLOの現実的妥協：初期は広めに設定し、運用で引き締める。過度なSLOは誤検知や過剰対応を招く
- 担当1人でも回せる仕組み：手順を簡潔に、復旧手順を優先度付きで定義する
運用チェックリスト（そのまま使える表）
| 項目 | チェック内容 | 推奨頻度 | 責任者 | 備考 |
|---|---|---|---|---|
| ログ保持 | 保存ポリシーが適用されているか（古いログは削除/アーカイブ） | 月次 | 運用担当 | 法規遵守を確認 |
| アクセス制御 | ログストアのアクセス権を確認 | 月次 | セキュリティ担当 | 最小権限の原則 |
| マスキング | PIIがマスク／ハッシュされているか | 週次 | 運用担当 | 抜き打ち検査を推奨 |
| 異常検知閾値 | SLO/SLIの閾値が現状に合っているか | 四半期 | SRE/運用 | 基線を見て調整 |
| 通知先 | Slack/メールの通知先が最新か | 週次 | 運用担当 | オンコール表と同期 |
| 抜き打ちテスト | テストリクエストでログ整合性を確認 | 月次 | 運用担当 | 結果は記録して改善へ |
まとめ
監査ログ設計、SLO/SLI定義、日次／週次の自動チェック、そしてランブック化は運用の基本です。本記事では現場で使えるJSONLスキーマ例、SLO計算、Pythonを使った自動チェックの骨子、実務的な注意点を示しました。まずは最小限のログを確実に取り、SLOを設定して自動化することを優先してください。運用は作って終わりではなく、定期的な見直し（改善ループ）が重要です。
次につなげる企画案
- ポストモーテムと改善ループの作り方（テンプレート付き）
- 監査ログを検索・分析する軽量な検索基盤構築（ELK代替）
- 監査ログからの自動改善サイクル（フィードバックループ）実装例
この記事は「AIとPythonの実務」シリーズの一部です。実務での導入や具体的なスクリプト改良の相談があれば、次回以降でより深掘りします。