2026年8月5日 – AIに仕事を任せたい人のためのPython塾 ― Manage AI ―

監視でエラーを検知しても、すべてを人手で対応するのは現実的ではありません。ですが自動で何でもやらせると二重実行や負荷増大といった新たな問題が生じます。本稿では「安全に自動復旧する」ための実務的パターン――再試行（retry）・バックオフ・サーキットブレーカー・冪等性（idempotency）――を、手順と最小実装例で整理します。第123回で可観測性を整えた後に続けて読むことを想定しています。

まず整理：何を自動化し、いつ人に切り替えるか

自動復旧でよく悩む点は「どこまで機械任せにするか」です。指針を簡潔に示します。

短時間の一時的な障害（ネットワーク断・一時的なタイムアウト）は自動再試行で対応する。
外部依存（サードパーティAPI）の継続障害は早めにサーキットを開いてエスカレーションする。
金銭取引や戻せない操作は人手を挟むか、厳密な冪等性を担保してから自動化する。

再試行パターンの実装手順（段階的）

同期／非同期のAPI呼び出しを例に、実装の手順を示します。

例外の分類：再試行可能なエラー（一時的なタイムアウト、503等）と不可逆的なエラー（400、認証失敗）を分ける。
再試行可能か判定する関数を作る。
バックオフ戦略（指数バックオフ＋ジッタ）を用意する。
デコレータ／contextmanagerで重複を避けつつ適用する。

再試行の判定例（表）

状況	再試行推奨度	理由
HTTP 500/502/503/504	高	一時的なサーバー側障害の可能性が高い
HTTP 429（レート制限）	条件付き	待ち時間やバックオフで回復するがスロット調整が必要
HTTP 400（リクエスト不正）	低	再試行しても同じエラーになる可能性が高い
ネットワークタイムアウト	高	一時的な接続不良として再試行価値あり

最小実装（同期デコレータ）

シンプルな再試行デコレータ例です。実務ではテスト・ロギング・メトリクスを必ず追加してください。

import time
import random
from functools import wraps

def retry(max_attempts=3, base_delay=0.5, max_delay=10, retry_if=lambda e: True):
    def deco(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            attempt = 0
            while True:
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    attempt += 1
                    if attempt >= max_attempts or not retry_if(e):
                        raise
                    # 指数バックオフ + ジッタ
                    delay = min(max_delay, base_delay * (2 ** (attempt - 1)))
                    delay = delay * (0.5 + random.random() / 2)
                    time.sleep(delay)
        return wrapper
    return deco

非同期（asyncio）向けの最小実装

import asyncio
import random
from functools import wraps

def async_retry(max_attempts=3, base_delay=0.5, max_delay=10, retry_if=lambda e: True):
    def deco(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            attempt = 0
            while True:
                try:
                    return await func(*args, **kwargs)
                except Exception as e:
                    attempt += 1
                    if attempt >= max_attempts or not retry_if(e):
                        raise
                    delay = min(max_delay, base_delay * (2 ** (attempt - 1)))
                    delay = delay * (0.5 + random.random() / 2)
                    await asyncio.sleep(delay)
        return wrapper
    return deco

ライブラリ利用の注意点（tenacity / backoff等）

便利だが内部でのスリープやタイマーをテスト時に高速化する方法を設計する（フェイクタイマー）。
ライブラリ固有の例外フィルタやキャンセル挙動を理解する。async環境でのキャンセル取り扱いは特に注意。
メトリクスやログをライブラリのフックに差し込んで可観測性を確保する。

冪等性（idempotency）の実務設計

再試行と組み合わせると冪等性は欠かせません。ここでは実務的な設計パターンを表でまとめます。

レイヤ	パターン	ポイント
クライアント	冪等キー生成（UUID、ハッシュ）	リクエストごとにサーバーが一意判定できるキーを付与する
API	受信時のキーチェック	キー存在時は既存レスポンスを返す／副作用を避ける
DB	upsert（INSERT … ON CONFLICT / REPLACE）	重複書き込みを原子的に処理する
キュー	deduplication（メッセージID、TTL付きの重複チェック）	コンシューマー側で処理済みのIDを保持して再実行を防ぐ

実務での具体的な注意点

冪等キーはクライアント側で生成して送る。サーバーで生成してしまうと再試行で同キーが使えない。
DBのupsertは競合回避に有効だが、論理的な重複判定（同一注文IDなど）もアプリ側で確認する。
ログに冪等キーを含め、監査や追跡を容易にする。

サーキットブレーカーとフォールバック

外部サービスが継続的に失敗する場合、サーキットブレーカーで早期にアクセスを止めると被害を小さくできます。ここでは簡易実装と設計上のポイントを示します。

状態遷移（簡潔表）

状態	意味	遷移条件
Closed	通常運転。失敗カウントを監視	失敗率が閾値を超えるとOpenへ
Open	呼び出しを遮断する（即エラーまたはフォールバック）	一定時間経過するとHalf-Openへ
Half-Open	試験的に一部リクエストを許可し回復を確認	数件成功ならClosedへ、失敗が続けば再びOpenへ

簡易サーキットブレーカー（概念実装）

import time

class SimpleCircuit:
    def __init__(self, fail_threshold=5, reset_timeout=30):
        self.fail_threshold = fail_threshold
        self.reset_timeout = reset_timeout
        self.fail_count = 0
        self.state = 'closed'
        self.opened_at = None

    def record_success(self):
        self.fail_count = 0
        if self.state != 'closed':
            self.state = 'closed'

    def record_failure(self):
        self.fail_count += 1
        if self.fail_count >= self.fail_threshold:
            self.state = 'open'
            self.opened_at = time.time()

    def allow(self):
        if self.state == 'open':
            if time.time() - self.opened_at > self.reset_timeout:
                self.state = 'half-open'
                return True
            return False
        return True

フォールバック戦略（運用判断）

キャッシュがある処理はキャッシュ応答に切り替える（整合性とTTLの検討を忘れずに）。
旧処理（軽量版レスポンス）でユーザーに最低限の機能を提供する。
フォールバックを返すときは明確なステータスと追跡可能なログを残す。

並列処理・非同期での注意点

同時実行が増えると再試行が同時に走ってスパイクを生む可能性があります。対策を整理します。

問題	実務的な対策
再試行が同時発生して負荷増大	ジッタを混ぜた指数バックオフ、セマフォやトークンバケットで同時実行数を制限
タスクキャンセルでリソースリーク	タイムアウトとfinallyでのクリーンアップ、asyncioのキャンセル例外を適切に扱う
重複処理の競合	DBの楽観ロックや一意キーによる排他、キュー側のdeduplication

観測とアラート連携（運用の実務）

再試行ロジックを入れたらメトリクスを増やして監視に組み込みます。例を示します。

メトリクス	説明	推奨アラート閾値（例）
再試行回数/分	短時間で増えると一時的障害の可能性	過去1h比で+300% または > 1000/分
サーキット状態	Open/半開の数を可視化	Openが一定数以上でエスカレーション
成功率（外部API）	外部依存の健全性を示す	成功率<95%でアラート

自動復旧のrunbook（まず試す手順）と、人手対応に切り替える判断フローを用意しておくと現場で迷いません（例：自動再試行→サーキットオープン→30分以内に回復しない→オンコールにエスカレーション）。

テストとCIへの組み込み

再試行やバックオフはテストしにくい点があるため、設計段階でテストフレンドリーにします。

タイマーは注入可能にして、ユニットテストではフェイクタイマーで高速化する。
モックやフェイクでエラー注入し、再試行回数・バックオフ挙動を検証する。
統合テストでは故障注入（chaos testing 的）を行い、サーキットやフォールバックが期待通り動くか確かめる。

作業用コードスニペットと次の一歩

ここまでの説明を踏まえた最小限の実務スニペットを集めました。現場で貼って試せるレベルを意識しています。

再試行デコレータ（上記）
簡易サーキットブレーカー（上記）
冪等キー保存の例（SQLiteを使った簡単なupsert）

-- SQLite の例（概念）
-- CREATE TABLE idempotency (key TEXT PRIMARY KEY, response TEXT, created_at INTEGER);
-- INSERT OR REPLACE INTO idempotency(key, response, created_at) VALUES(?, ?, strftime('%s','now'));

次の一歩としては、ジョブスケジューラとの連携、運用の自動化（復旧台本のコード化）、より高度なレート制御（トークンバケット）を検討してください。

まとめ

本稿では、監視の後段として「安全に自動復旧する」ための実務パターンを整理しました。ポイントを改めてまとめます。

再試行は例外の分類と指数バックオフ＋ジッタで実装し、テストしやすく設計する。
冪等性はクライアント発行のキー・DBのupsert・キューの重複排除で実務的に担保する。
サーキットブレーカーで外部障害の拡大を防ぎ、フォールバックと明確なログで運用の判断を助ける。
並列環境では同時実行制限やキャンセル処理を忘れずに。観測用メトリクスとrunbookを必ず用意する。

次回は本稿の実装を踏まえ、復旧台本（Runbook）の自動化とジョブスケジューラ連携について具体例を示します。Manage AI シリーズ「AIとPythonの実務」の流れで、監視から自動復旧、運用までつなげていきましょう。

月	火	水	木	金	土	日
					1	2
3	4	5	6	7	8	9
10	11	12	13	14	15	16
17	18	19	20	21	22	23
24	25	26	27	28	29	30
31