はじめに — うまく説明できないエラーで時間を取られていませんか？
AIを業務に組み込むと、予期せぬエラーが発生しがちです。外部APIの一時障害、入力データの不整合、モデルの推論失敗、ファイルI/Oの問題……どれも現場では頻出です。しかし重要なのは“何が起きたかを説明できて、対応できること”です。本記事では、実務で使える例外設計とハンドリング手順を、コード例・ログ設計・テスト・運用チェックリストとともに示します。読了後には実装に移れるレベルを目指します。
問題提起：現場でよくある失敗パターン
| カテゴリ | 症状 | 原因（例） |
|---|---|---|
| 外部API | タイムアウト、502/503 | ネットワーク/レート制限/一時的障害 |
| データ不整合 | パース失敗、スキーマ違反 | 入力CSVの欠損・型違い |
| モデル推論 | 推論例外、メモリ不足 | モデルの入力前提違反、リソース不足 |
| ファイルI/O | 読み書き失敗 | 権限、ディスク容量、同時アクセス |
例外設計の基本
実務向けには、標準例外を乱用せず、業務ドメインに応じたカスタム例外を追加することを推奨します。命名規約は「場所＋原因＋Action」で分かりやすくします（例：CsvParseError、ApiTimeoutError、ModelInferenceError）。
下記は基本の階層例です。
| 基底 | 目的 | 例 |
|---|---|---|
| AppError(Exception) | アプリケーション全体で共通の基底 | ログやアラートの一括ハンドリング用 |
| TransientError(AppError) | 再試行で回復が期待できる一時的エラー | ApiTimeoutError、ServiceUnavailableError |
| PermanentError(AppError) | 再試行しても意味がない恒久的エラー | InvalidInputError、DataSchemaError |
| ResourceError(AppError) | 外部リソースに関わる問題 | FileIOError、DatabaseError |
振る舞いマップ（例外ごとの実務アクション）
次の表は例外種別と現場で望ましい振る舞いのマッピングです。実運用ではこれをルール化してワークフローに落とし込みます。
| 例外種別 | 現場アクション | ログ/メトリクス |
|---|---|---|
| TransientError（外部APIタイムアウト等） | 自動再試行（指数バックオフ）、失敗後はアラート | retry_count、last_error_type |
| PermanentError（入力データ不備等） | 処理をスキップしてユーザー通知/エラーレポート | bad_record_count、validation_errors |
| ResourceError（ディスク/DB権限） | 即時停止＋オンコールへ通知、想定外ならロールバック | resource_status、failure_rate |
| ModelError（推論失敗） | 再試行＋軽量なフォールバック（簡易モデル）または手動介入 | model_failures、latency |
実装ガイド
カスタム例外（dataclassベース）
from dataclasses import dataclass
class AppError(Exception):
"""アプリ全体の基底例外"""
pass
@dataclass
class TransientError(AppError):
message: str
retry_after: float = 0.0
@dataclass
class PermanentError(AppError):
message: str
field: str = ""
例外にコンテキストを持たせる（traceback保持）
raise from を使い、元の例外を保持しておくと原因追跡が容易になります。
try:
result = external_api.call(payload)
except TimeoutError as e:
raise TransientError("API timeout", retry_after=2.0) from e
例外をラップしてドメインに翻訳するパターン
外部ライブラリの例外はそのまま流すのではなく、ドメイン例外へ翻訳します。
def fetch_with_wrap():
try:
return third_party.fetch()
except third_party.HttpError as e:
# 外部のHTTPエラーを自ドメインのTransientErrorに変換
raise TransientError("third_party http error", retry_after=5.0) from e
ロギング＆可観測性連携
標準loggingで構造化ログを出し、SentryやPrometheusに必要な情報を渡します。ポイントは「何が」「どこで」「どの程度」の情報を一貫して出すことです。
構造化ログの例
import logging
logger = logging.getLogger(__name__)
def handle_error(exc, context):
logger.error("processing_error", extra={
"error_type": type(exc).__name__,
"message": getattr(exc, 'message', str(exc)),
"context": context,
})
メトリクスの置き場所
Prometheus等には次のようなメトリクスを送ります：
- error_count{type=…}
- retry_count
- processing_latency_seconds
| ログ/メトリクス | 用途 | 閾値の目安 |
|---|---|---|
| error_rate (5m) | 全体の健全性 | >1%で要調査 |
| transient_error_rate | 外部依存の劣化 | 急増は外部障害の兆候 |
テストとCI
例外処理はユニットテストとCIでの回帰チェックが必須です。外部依存はモックで一貫して疑似障害を作るとよいです。
ユニットテストの例（pytest）
def test_api_timeout(monkeypatch):
def fake_call(_):
raise TimeoutError("timeout")
monkeypatch.setattr('external_api.call', fake_call)
with pytest.raises(TransientError) as excinfo:
fetch_with_wrap()
assert excinfo.value.retry_after == 5.0
CIでの検証項目（例）
- 例外階層のドキュメントが最新か
- 主要な例外でのユニットテスト通過
- ログに必要なフィールドが出力されているかのスナップショット
デプロイ／運用チェックリスト
| 項目 | 確認内容 |
|---|---|
| 例外リスト | 主要例外と期待される振る舞い（retry/skip/alert）がドキュメント化されている |
| アラートテスト | オンコールに通知されることをテスト済み |
| ログ/メトリクス | 構造化ログと主要メトリクスが収集されている |
| ロールバック手順 | 障害時のロールバック手順が用意されている |
小さなハンズオン：CSV一括推論パイプライン
以下は簡易サンプルです。CSVを読み、1行ずつ推論して結果を書き出す。APIタイムアウトは再試行、データ不備はスキップしてレポートします。
import csv
import time
import logging
logger = logging.getLogger(__name__)
class CsvParseError(PermanentError):
pass
class ApiTimeoutError(TransientError):
pass
def predict_with_retry(payload, max_retry=3):
for i in range(max_retry):
try:
return external_api.predict(payload)
except TimeoutError as e:
if i == max_retry - 1:
raise ApiTimeoutError("api timeout", retry_after=2.0) from e
time.sleep(2 ** i)
def process_csv(in_path, out_path):
bad_rows = []
with open(in_path) as inf, open(out_path, 'w', newline='') as outf:
reader = csv.DictReader(inf)
writer = csv.DictWriter(outf, fieldnames=reader.fieldnames + ['prediction'])
writer.writeheader()
for row in reader:
try:
if not row.get('text'):
raise CsvParseError('missing text', field='text')
pred = predict_with_retry(row['text'])
row['prediction'] = pred
writer.writerow(row)
except PermanentError as e:
logger.warning('skip_row', extra={'reason': e.message, 'row': row})
bad_rows.append(row)
except TransientError as e:
logger.error('transient_failure', extra={'reason': e.message})
raise
return bad_rows
読後アクション（まず取り組む3つ）
- 自分のワークフローで発生しうる主要例外を一覧化する。
- 各例外に対して期待される振る舞い（retry/skip/alert）を書き出す。
- 1種類の例外（例：APIタイムアウト）を実装してユニットテスト＆デプロイする。
まとめ
実務で使えるエラーハンドリングは、例外を分類し「期待される振る舞い」を明文化することが出発点です。カスタム例外の設計、例外のラップ、構造化ログとメトリクス連携、そしてユニットテストとCIでの回帰検証をセットにすることで、何が起きたか説明でき、対応できる運用が作れます。本記事の表やチェックリストを自分のワークフローに当てはめ、まず一つの例外を実装・検証してみてください。
関連：第124回（再試行）、第126回（API連携）、第123回（可観測性）も合わせて参照すると、今回の設計を運用に落とし込みやすくなります。