こんにちは、おうどんです🍜
AI Agentに複数のツールを持たせると、調査だけでなく、メール送信・チケット更新・ファイル変更まで進められます。
便利なんですが、ここで避けて通れないのが「途中で落ちたので再実行したら、同じ処理をもう一度やった」です。
ログ取得が2回なら、まあ多いな。。。で済むかもしれません。メール送信や課金が2回だと、急に笑えなくなります。
この記事では、AI Agentを“賢いチャット”ではなく、失敗と再実行がある分散ワークフローとして設計します。ポイントは次の3つです。
- 副作用のある処理を冪等にする
- 会話履歴とは別に、実行状態を保存する
- 危険な操作は、具体的な引数を見せて人間に承認してもらう
まず整理：メモリとチェックポイントは別物
AI Agentには、会話履歴を保持する「メモリ」や「セッション」があります。OpenAI Agents SDKのSessionsも、実行前に履歴を読み、実行後に新しい項目を保存する仕組みです。
ただし、会話に「メールを送信しました」と残っていることと、送信APIが一度だけ実行されたことは同じではありません。
- 会話メモリ: モデルが文脈を思い出すための情報
- 実行台帳: どの操作を、どの引数で、いつ実行し、結果がどうなったかを判定する情報
- 業務データ: 注文・チケット・メールなど、実際に変更される対象
この3つを分けます。全部をプロンプトへ詰めれば解決……とはなりません。プロンプトは説明書であって、トランザクションログではないんですよね。
処理を状態遷移として設計する
実行フローは、最低でも次の状態を持たせます。
PLANNED
↓
WAITING_APPROVAL ── rejected → REJECTED
↓ approved
RUNNING
├─ success → SUCCEEDED
├─ retryable error → RETRY_WAIT → RUNNING
└─ unknown result → NEEDS_RECONCILIATION
大事なのは UNKNOWN 相当の状態です。外部APIがタイムアウトしたとき、「相手側では成功したが、応答だけ届かなかった」可能性があります。ここで無条件に再実行すると、二重処理が生まれます。
実行台帳と冪等キーを用意する
PostgreSQLなら、たとえば次のような台帳を作れます。
CREATE TABLE agent_action_log (
idempotency_key text PRIMARY KEY,
run_id text NOT NULL,
tool_name text NOT NULL,
args_hash text NOT NULL,
status text NOT NULL
CHECK (status IN (
'PLANNED', 'WAITING_APPROVAL', 'RUNNING',
'SUCCEEDED', 'REJECTED', 'RETRY_WAIT',
'NEEDS_RECONCILIATION'
)),
external_ref text,
result_json jsonb,
created_at timestamptz NOT NULL DEFAULT now(),
updated_at timestamptz NOT NULL DEFAULT now()
);
冪等キーは、単なる乱数ではなく「同じ業務操作なら同じ値になる」ように作ります。
import hashlib
import json
def canonical_json(value: dict) -> str:
return json.dumps(
value,
ensure_ascii=False,
sort_keys=True,
separators=(",", ":"),
)
def make_idempotency_key(
workflow_name: str,
business_key: str,
tool_name: str,
args: dict,
) -> str:
material = "|".join([
workflow_name,
business_key,
tool_name,
canonical_json(args),
])
return hashlib.sha256(material.encode("utf-8")).hexdigest()
たとえば「問い合わせ123へ回答メールを送る」なら、business_key は問い合わせIDにできます。再実行時に同じキーの SUCCEEDED があれば、ツールを呼ばず保存済みの結果を返します。
ただし、ここに罠があります。
外部API実行 → 成功
↓
台帳更新の直前に停止
自分のDBだけ冪等にしても、この隙間は消えません。送信先APIが冪等キーを受け取れるなら、同じキーを下流にも渡すのが第一候補です。対応していなければ、送信結果の照会API・外部参照ID・定期突合を使い、結果不明を NEEDS_RECONCILIATION に隔離します。
“exactly once”と唱えるより、at-least-once実行＋重複排除＋突合へ分解した方が、実務では扱いやすいです。
リトライ可否はツール単位で決める
一律に「3回リトライ」は危険です。ざっくり次のように分けます。
| 操作 | 例 | 自動リトライ |
|---|---|---|
| 読み取り | 検索、一覧取得 | 一時エラーなら可 |
| 冪等な更新 | UPSERT、冪等キー付きAPI | 条件付きで可 |
| 非冪等な更新 | メール送信、決済、削除 | 状態照会なしの盲目的リトライは不可 |
HTTP 429や一時的な5xxは指数バックオフの対象にできます。一方、入力不正・権限不足・業務ルール違反は、待っても直らないので即時停止です。
人間承認は「操作名」だけでなく引数まで固定する
MCPの2026-07-28仕様では、ツール呼び出しを拒否できる人間をループに含め、公開ツールや呼び出しを明確に表示し、操作前に確認を出すことが推奨されています。
承認画面には、少なくとも次を出します。
- ツール名
- 対象（宛先、ファイル、注文IDなど）
- 変更内容の要約
- 正規化した引数のハッシュ
- 期限と承認者
- 失敗時の再実行方針
承認後にモデルが引数を変えられると、別の操作を承認したことになります。承認対象の args_hash と実行直前のハッシュを比較し、違えば再承認に戻します。
OpenAI Agents SDKのHuman-in-the-loop機能では、承認が必要なツール呼び出しで実行を中断し、RunState を保存して承認後に再開できます。引数が不正で承認ルールを安全に評価できない場合は、手動承認へ倒す設計も明記されています。
実装時の確認ポイント
1. 障害注入テスト
正常系だけでなく、次の位置で意図的に停止させます。
- ツール呼び出し前
- 外部API成功後、台帳更新前
- 台帳更新後、モデルへ結果を返す前
- 承認後、実行前
再開後に同じ副作用が増えていないか、台帳と外部システムの両方で確認します。
2. 監視項目
run_id、idempotency_key、tool_name、args_hash、承認者、試行回数、外部参照IDを同じトレースへ載せます。本文や機密情報を丸ごとログへ出さず、検索に必要な識別子と監査情報を分けるのも大切です。
3. 評価対象
「最終回答が自然か」だけでなく、次を自動テストします。
- 同一入力の再送で副作用が1回に保たれるか
- 拒否した操作が実行されないか
- 引数変更時に再承認されるか
- 結果不明が成功扱いされないか
- 最大試行回数を超えて無限ループしないか
OpenAI Agents SDKには、ツール実行・リトライ・セッション動作など、アプリ側が管理するオーケストレーションをモデル呼び出しなしで検証するテスト機能もあります。外部APIそのものは、結合テスト環境で別に確認します。
まとめ
AI Agentの再実行で守るべきなのは、モデルの“記憶”ではなく、副作用の整合性です。
- 会話メモリと実行台帳を分ける
- 業務操作から冪等キーを作る
- 下流APIにも同じキーを渡す
- 結果不明を成功・失敗へ勝手に寄せない
- 危険な操作は具体的な引数で承認する
- 障害注入で、再開時の二重処理を検証する
AIに仕事を任せるほど、従来のバッチ設計・トランザクション設計・運用監視が効いてきます。
未来っぽい見た目をしていますが、足元を支えるのは地味で強い設計です。こういう地味さ、あとで効くんですよね。
参考資料
- OpenAI Agents SDK: Human-in-the-loop
- OpenAI Agents SDK: Sessions
- OpenAI Agents SDK: Testing
- Model Context Protocol 2026-07-28: Tools
注意: 掲載したDDL・コードは設計例です。実環境では、利用するDB、外部APIの冪等性仕様、トランザクション境界、個人情報・機密情報の取り扱いを確認し、検証環境でテストしてください。
コメント