はじめに — まずはその「つまずき」に寄り添います
AIプロトタイプやスクリプトは動くけれど、定期実行すると止まってしまう、ログが読めない、障害時に誰が対応するか決まっていない──そんな現場の悩みに応えます。本記事は、AIモデルの出力をPythonで定期実行するワークフローを、スケジューリングから監視、障害対応、コスト管理まで「実務で回す」ための手順とテンプレートを提供します。まずは1週間の稼働試験を目標に読み進めてください。
導入シナリオ（例）
想定例：毎朝8時に外部APIでデータを取得し、AIで解析してCSVを生成、Slackへ結果を通知する処理。要件は「確実に毎日実行」「失敗検知と自動再試行」「コストを抑える」などです。
アーキテクチャの選定基準
- 可用性：停止の許容時間はどれくらいか
- 運用負荷：ローカルで運用できるか、クラウドに任せるか
- 監視のしやすさ：ログ・メトリクスをどう集めるか
- セキュリティ：APIキーや機密情報の管理方法
- コスト：実行頻度とAPI利用量に基づく見積もり
スケジューリング比較
| 方式 | 向き・特徴 | 運用負荷 | 推奨場面 |
|---|---|---|---|
| ローカル/サーバーcron | シンプル、OS依存、単純なジョブに最適 | 中（ホスト管理必要） | 少数の低頻度ジョブ |
| APScheduler | Python内蔵で柔軟、複雑なスケジュール向け | 中（プロセス監視必要） | アプリ内スケジューラや動的ジョブ |
| Cloud Scheduler（GCP/AWS同等） | マネージド、可用性高、スケール容易 | 低（クラウド運用が主体） | 業務での安定稼働が必要なケース |
簡単な例（cron用シェルとAPSchedulerのサンプル）
cron用（crontabに追加）:
0 8 * * * /usr/bin/python3 /opt/app/daily_job.py >> /var/log/daily_job.log 2>&1APScheduler（アプリ内部）:
from apscheduler.schedulers.blocking import BlockingScheduler
sched = BlockingScheduler()
@sched.scheduled_job('cron', hour=8, minute=0)
def daily_job():
run_main()
if __name__ == '__main__':
sched.start()実行時のログ／メトリクス設計
ログとメトリクスは分けて考えます。ログは詳細トラブルシュート用、メトリクスは運用判断用です。
| 項目 | 内容（例） | 用途 |
|---|---|---|
| 成功率 | 成功数／合計実行数（1h,1d） | 障害検知、SLI |
| 処理時間 | 平均・p95 | 性能劣化検知 |
| API呼び出し数 | 外部APIのコール回数と課金見込み | コスト監視 |
| エラーログ（構造化） | {“job”:”daily_job”,”error”:”タイムアウト”,”trace”:”…”} | 自動解析と検索 |
ログ出力の簡単なPython例（構造化ログ）:
import json, time
def log_event(level, msg, **kwargs):
entry = {"ts": time.time(), "level": level, "msg": msg}
entry.update(kwargs)
print(json.dumps(entry))障害検知と自動再試行
再試行戦略は失敗のタイプごとに分けます。ネットワーク等の一時的失敗は指数バックオフで再試行、永続的なエラーは即通知して手動対応へ。
簡易リトライデコレータ（説明付き）:
import time
from functools import wraps
def retry(max_attempts=3, base_delay=2):
def deco(f):
@wraps(f)
def wrapped(*a, **kw):
for i in range(1, max_attempts+1):
try:
return f(*a, **kw)
except Exception as e:
if i == max_attempts:
raise
time.sleep(base_delay * (2 ** (i-1)))
return wrapped
return decoアラートとエスカレーション
- 閾値例：成功率90%未満でSlack通知、60%未満でメール＋オンコール（担当者呼び出し）
- 通知テンプレート（Slack）: {“title”:”daily_job 失敗”,”env”:”prod”,”last_error”:”タイムアウト”}
- Runbookテンプレ（簡易）: 障害発生→ログ確認→再試行→影響範囲報告→ロールバック案実行
コストとAPI利用管理
APIの呼び出し数は定期的にレポートして上限を設けます。APIキーは環境変数かシークレットマネージャーで保護してください。予算超過の自動停止やレート制限エラーを想定した設計を入れます。
デプロイとロールバック（簡易手順）
- ローカルで単体テスト→コンテナ化（Dockerfile）
- GitHub Actionsでビルド→ステージングへデプロイ→自動テスト
- 本番へロールアウト（段階的）→問題あればタグを戻して再デプロイ
運用チェックリストとテンプレ
| 段階 | チェック項目 |
|---|---|
| 導入前 | 要件確認、実行頻度、API課金見積、シークレット管理方針 |
| 導入直後（1週間） | ログ量確認、成功率確認（目標95%）、監視アラート動作確認 |
| 定期点検（週/月） | エラートレンド確認、コスト確認、Runbookの更新 |
簡易Runbook雛形（抜粋）:
- 発生時：まず最新の実行ログを取得（/var/log/…）
- 再現：ステージングで同条件を再現し原因切り分け
- 対応：再起動→再試行→ロールバックの順で対応
- 報告：影響範囲、暫定対応、恒久対応予定をまとめて共有
付録：すぐ使えるテンプレ（チェックリスト抜粋）
| 項目 | 完了 |
|---|---|
| シークレットを環境変数へ移行 | [ ] |
| 監視：成功率アラート設定 | [ ] |
| 1週間の稼働試験を開始（ログ監視） | [ ] |
まずの一歩（具体アクション）
- 今週中にcronかCloudSchedulerで1ジョブを週次→1週間稼働させる
- 成功率・処理時間のメトリクスを収集し、初回の閾値を設定する
- Runbookを1ページで作り、担当者へ共有する
まとめ
AI×Pythonの業務運用は「動かす」だけでなく、「監視する」「障害時に誰が何をするか」を定めることが成功の鍵です。本記事はスケジュールの選び方、ログとメトリクスの設計、再試行とアラートの具体例、デプロイ／ロールバック手順、そして運用チェックリストを提示しました。まずは1週間の稼働試験を行い、ログと成功率を見ながら閾値や再試行方針を調整してください。
次回（シリーズ）: 運用改善の自動化とABテスト — 実施後のフィードバックループを作る方法を扱います。