運用に移したらジョブが二重に動いた、承認が止まって全体が滞った、エラー時にどの処理を巻き戻すべきか分からない――こうしたつまずきは、現場でワークフローを「ただ動かす」段階でよく起きます。本記事では、RAGやモデル推論、リトレーニング、監視をつなぐ「実行時オーケストレーション」に必要な要素を、Pythonで実装できる手順を中心に整理します。読み進めることで、自社の一つの業務フローを実際に動かせる状態を目指します。
なぜCI/CDや単純な監視だけでは足りないのか
CI/CDはデプロイやテストの自動化には強い一方で、実行時の複雑な条件分岐や人的承認、非同期リトライ、復旧（compensation）などを柔軟に扱う作りにはなっていません。現場で求められる要件を整理すると次の通りです。
- スケジュール（定期実行）とイベントトリガー（外部アラートやファイル到着）の両方を扱うこと
- 条件分岐と分岐後の合流（fork/join）ができること
- 再試行・指数バックオフ・タイムアウトなどの耐障害性
- 人的承認ポイントの挿入と監査ログの保存
- 可観測性（ログ／メトリクス／トレース）と状態の可視化
- 状態不整合や二重実行を防ぐための排他制御やidempotency
アーキテクチャ概観
まずは高レベルの構成を押さえます。図は記事用のイメージで、実際は自組織の要件に合わせて削る／足すを検討してください。
| コンポーネント | 役割 |
|---|---|
| トリガー | 時間（cron相当）、イベント（S3アップロード、Webhook）、監視アラートからジョブを起動 |
| ワークフローエンジン / スケジューラ | ジョブの依存関係、再試行、状態管理を担う（Airflow/Prefect/Dagster/自作） |
| 状態保持層 | ジョブの状態、ロック、承認履歴をDBやキューで保存（Postgres, Redis, SQS等） |
| モデル推論・RAGパイプライン | データ取り込み、埋め込み、モデル呼び出し、結果格納 |
| 人的承認 | Slack/Teams通知と承認API、簡易UI（Flask/FastAPI）で承認フローを実装 |
| 監視・可観測性 | ログ（structured log）、メトリクス（Prometheus）、分散トレース（OpenTelemetry） |
ツール選定ガイド（実務向け比較）
小規模〜中小企業でPython中心に運用する前提で、主要選択肢を比較します。
| ツール | 長所 | 短所 | 推奨ケース |
|---|---|---|---|
| cron / supervisor | 導入が最も簡単。OSレベルで動く。 | 複雑な依存や再試行管理ができない。状態可視化が弱い。 | 単純な定期バッチや簡易ジョブの運用に最適。 |
| Airflow | 成熟度高、可視化豊富、スケジュール重視のワークフローに強い。 | 運用コストと学習コストが比較的高い。軽量には向かない。 | 複数のETLジョブやデータパイプラインを本格的に管理する場合。 |
| Prefect | Pythonフレンドリー、ローカルからクラウドまで柔軟。再試行/状態管理が扱いやすい。 | クラウド版の商用機能があるため、大規模化時にコスト検討が必要。 | 中小チームが比較的短時間でワークフローを構築するのに適する。 |
| Dagster | 型を意識したパイプライン設計、ローカルでのテストがしやすい。 | 概念を理解する必要があり学習コストがやや高い。 | データプロダクト化を視野に入れた運用に向く。 |
| サーバレス（Step Functions 等） | スケールと可用性に優れる。マネージドで運用負荷低め。 | クラウドロックイン、細かいロジックのテストがやや面倒。 | クラウド中心の組織で稼働済のインフラを活かす場合。 |
おすすめ：小規模でPython中心のチームならまずはPrefectや軽量な自作オーケストレータ（DBで状態管理）でプロトタイプを作り、信頼性が必要ならAirflowかサーバレスに移行するのが現実的です。
実装パターン（手順ベース）
1) idempotency（冪等性）の実装
キーは「同じ入力で同じ処理が複数回走っても副作用が一度だけ起きる」ことです。実装例は次の方針。
- ジョブ開始時に一意の実行IDを発行し、DBに状態レコード（status: started/finished/failed、updated_at）を作る。
- 処理はすべてそのIDに紐づけて書き込み、コミットが完了したらstatusをfinishedに更新する。
- 再実行時は既存のfinishedを検出して処理をスキップする。
| 手順 | 擬似コード（説明） |
|---|---|
| 1. 実行ID発行 | create run record (run_id, status=’started’, payload_hash) |
| 2. 既存確認 | if record.status == ‘finished’: return |
| 3. 処理 | do work, write outputs atomically |
| 4. 完了更新 | update record.status = ‘finished’ |
2) 再試行と指数バックオフ
- 短時間で解消する外部依存（ネットワーク、APIレート）には再試行を入れる。
- 試行回数は上限を決め、指数バックオフ（base * 2^n）にジッタを入れる。
3) 補償処理（compensating actions）とトランザクション性
外部システムへの副作用がある場合、部分的に失敗したら補償アクションで整合性を回復できる設計にします。例：外部DBに書き込んだが通知に失敗した場合は通知失敗を再試行、可能であれば書き込みを元に戻す補償処理を用意する。
4) ロック / 排他制御
並列実行を防ぐため、DBで楽観ロックやRedisのSETNX（名前付きロック）を使います。ロックにTTLを付けてデッドロックを避けることが重要です。
人的承認ワークフローの作り方
実務では自動処理と人の判断をつなぐポイントが必要になります。基本的な流れと実装のヒントを示します。
- ワークフロー中に承認ポイントを置き、状態を”pending_approval”に遷移させる。
- Slack/Teamsに通知を投げ、承認用のURL（短いトークン付き）を送る。
- 承認APIは簡易なFastAPI/Flaskアプリで実装し、承認結果をDBに保存する。
- 承認が一定時間来なければタイムアウト処理を実行（自動ロールバック、あるいは代替承認者へのエスカレーション）。
| 要素 | 実装のポイント |
|---|---|
| 通知 | Slack Incoming Webhook / Block Kitで承認ボタンを送る。ボタンは承認APIのエンドポイントを呼ぶ。 |
| 承認API | 受け取ったトークンでDBのrunレコードを更新し、監査ログを追加。処理を進めるワーカーに通知。 |
| タイムアウト | ワークフローエンジン側でタイムアウト監視を行い、期限切れ時の代替処理を実行。 |
オーケストレーションと既存ワークフローの接続
Manage AIシリーズの既存回とどう繋ぐか具体例です。
- 第93回（RAG）：RAGのingestion完了をイベントで拾って、ワークフローを起動（ファイル到着やDBイベント）。
- 第94回（CI/CD）：モデルの新バージョンがデプロイされたら、リグレッション用のジョブをワークフローでキックして評価を自動化。
- 第103回（リトレーニング）：リトレーニング完了後、検査→承認→本番入替という流れをワークフローで組む。
ポイントは「どのタイミングで人的確認を入れるか」を設計テンプレートにすることです。例：リトレーニング→自動評価→閾値を超えたら自動で適用、それ以外は承認待ち、というルール化。
テストと本番移行のチェックリスト
| フェーズ | チェック項目 |
|---|---|
| ローカル | ドライランでステート遷移を検証。DBロールバック／補償処理を手動で試す。 |
| ステージング | 実際の外部APIやモデルをモックせずに通す。監視アラートの発砲をテスト。 |
| フェイルオーバー | ワーカー停止・DB断などの障害シナリオを実施。復旧時間を計測。 |
| 本番移行 | SOPに従い段階的リリース。最初は低トラフィックのジョブから稼働。 |
よくある失敗事例と回避策
- 状態不整合：原因は部分的なコミット失敗。対策はトランザクションか補償処理の明確化。
- 二重実行：ロックやidempotencyキーで防止。ジョブ発行側も重複防止策を導入。
- 人的承認の滞留：承認期限とエスカレーション経路を設け、期限切れ時の自動処理を用意。
- 監視の死角：ログ、メトリクス、トレースを必ず揃え、ラベル付けでフィルタ可能にする。
可観測性フックの具体例：各ステップで構造化ログ（run_id, step, status, duration）、メトリクス（success_count, failure_count, queue_latency）、分散トレース（request_id）を出すこと。
まずこの1つを動かす — 最小構成サンプル（段階的手順）
以下は最短でひとつのワークフローを動かすための最小構成です。
- インフラ準備：Postgres（状態保存）、Redis（ロック）、ワーカーを動かす実行環境（VMやコンテナ）を用意する。
- ワークフロー実装：Prefectや軽量のPythonスクリプトで以下のフローを作る。
ステップ 説明 1 ジョブ起動（schedule or webhook）→ run_id作成 2 RAG ingestion呼び出し → 成功なら次へ 3 モデル推論（外部API呼び出し）→ 成功で結果保存 4 承認ポイント（optional）→ Slack通知、承認で続行 - 観測基盤：各ステップでログを出力し、メトリクスを（PrometheusやGrafana）で見る。まずはログにrun_idを付けるだけでも可視化効果が高いです。
- テスト：ローカルでドライラン→ステージングで外部依存をそのまま走らせる→本番へ段階的にリリース。
まとめ
実行時のオーケストレーションは、単にジョブを定期的に動かすだけではなく、再試行、状態管理、人的承認、そして観測性を含めて設計することが重要です。小規模なチームならまずはPrefectや軽量な自作仕組みでプロトタイプを作り、idempotency・ロック・補償処理・承認の基本パターンを1つのワークフローで実践してみてください。本記事の最小構成サンプルに沿って一つずつ確認すれば、自社の業務フローを安定して回せる第一歩になります。必要なら次回は具体的なPrefect例やFastAPIでの承認APIのコードを載せて説明します。