深夜3時、Slackが鳴った
「エージェントが同じAPIを400回叩いている」——今年の春先、あるクライアント環境で受け取った通知の第一報だ。障害というほど派手ではない。ただ、社内の在庫同期ワークフローに載せたエージェントが、レスポンスの解釈に失敗して、同じ在庫更新エンドポイントを延々と再試行していた。コストは一晩で$47。金額は小さい。ただ、これが送金APIだったら、と考えると背筋が寒くなる。
前回はMCP で社内ツール群を束ねて複数システムを横断するエージェントの構築を書いた。実装レベルでは動くものが出来上がる。しかし、本番投入となると話がまるで変わる。「動く」と「本番で回せる」の間には、承認境界・ガードレール・監査ログという、地味だが決定的な設計が横たわっている。
前提: なぜ2026年に「暴走させない設計」が急に注目されているのか
2026年3月末、総務省と経済産業省の「AI事業者ガイドライン」改定でAIエージェントが正式な対象として明記された。特に、エージェントが外部に影響を与える操作を行う前に「人間の判断を必須とする仕組み(Human-in-the-Loop、以下HITL)」を組み込むことが望ましいとされたのが大きい。詳細な運用要件はまだ各社の解釈に委ねられている部分もあるが、少なくとも「フルオート前提のPoCをそのまま本番に持っていく」時代は終わったと見て良い。
正直に言えば、筆者もつい半年前までは「まずは動かしてから考える」派だった。だが、冒頭のインシデントを経て、順序は逆だと考えを改めた。ガードレールを先に、機能はその中で。この記事はその方針転換の記録でもある。
設計の骨格 — 4層で考える
本番運用のエージェントに必要な統制を、筆者は次の4層で整理している。
行動前: 影響度分類 → 承認境界の決定 (HITL)
行動中: 暴走防止ガードレール (ステップ上限・コスト上限・ループ検知)
行動後: 監査ログ・ロールバック
運用継続: 失敗検知・アラート・メトリクス
順に見ていく。コードは Python + LangGraph 想定で書くが、フレームワークに依存しない設計なので、自前実装や他 SDK にも移植できる。
層1: 影響度別の承認境界 (HITL)
エージェントの全アクションを人間承認にすると、そもそもエージェントを使う意味がない。かといってフルオートは怖い。現実解は、アクションを影響度で分類して段階的に扱うことだ。筆者が使っている分類は次の4段階。
Low: 分類・要約・読み取り → ログのみ、承認不要
Medium: 内部システム更新 (社内Wiki編集など) → サンプリング審査(例: 10%を事後レビュー)
High: 外部通信・アカウント変更 → 事前承認必須
Critical: 送金・権限変更・大量削除 → 二重承認 + ロールバック計画必須
ツールの登録時にこの impact レベルをメタデータとして持たせておくのが実装上のコツ。以下は最小例。
from dataclasses import dataclass
from enum import Enum
from typing import Callable, Any
class Impact(str, Enum):
LOW = "low"
MEDIUM = "medium"
HIGH = "high"
CRITICAL = "critical"
@dataclass
class Tool:
name: str
fn: Callable[..., Any]
impact: Impact
description: str
例: 在庫の読み取りは low、更新は high
inventory_read = Tool(
name="inventory_read",
fn=lambda sku: {"sku": sku, "qty": 42},
impact=Impact.LOW,
description="SKUの在庫数を取得",
)
inventory_update = Tool(
name="inventory_update",
fn=lambda sku, delta: {"sku": sku, "new_qty": 42 + delta},
impact=Impact.HIGH,
description="在庫数を増減",
)
そしてエージェントのツール呼び出しをラップして、impact に応じて承認フローを差し込む。
import uuid
import time
APPROVAL_QUEUE: dict[str, dict] = {} # 実運用ではRedis/DB
def request_approval(tool: Tool, args: dict, timeout_sec: int = 300) -> bool:
req_id = str(uuid.uuid4())
APPROVAL_QUEUE[req_id] = {
"tool": tool.name,
"args": args,
"impact": tool.impact.value,
"status": "pending",
"requested_at": time.time(),
}
# 実運用ではここで Slack/Teams 通知を飛ばす
notify_slack(req_id, tool, args)
deadline = time.time() + timeout_sec
while time.time() < deadline:
status = APPROVAL_QUEUE[req_id]["status"]
if status in ("approved", "rejected"):
return status == "approved"
time.sleep(2)
APPROVAL_QUEUE[req_id]["status"] = "timeout"
return False
def invoke_tool(tool: Tool, args: dict, run_id: str) -> Any:
audit_log(run_id, "tool_call_requested", tool.name, args, tool.impact.value)
if tool.impact in (Impact.HIGH, Impact.CRITICAL):
approved = request_approval(tool, args)
if not approved:
audit_log(run_id, "tool_call_rejected", tool.name, args)
raise PermissionError(f"{tool.name}: 承認が得られなかった")
if tool.impact == Impact.CRITICAL:
# 二重承認: 別ロールの承認をもう1件要求
approved2 = request_approval(tool, args, timeout_sec=600)
if not approved2:
raise PermissionError(f"{tool.name}: 二重承認の2件目が失敗")
result = tool.fn(**args)
audit_log(run_id, "tool_call_completed", tool.name, args, result=result)
return result
ここで一つ、実装してみて気づいた罠。承認待ちの間、エージェントの思考ループを止めるかどうかは意外と悩ましい。同期的にブロックすると LLM のコンテキストが古くなり、承認された時点で状況が変わっている可能性がある。筆者は「承認待ちに入ったら run 全体を suspend し、承認 webhook 側から再開する」という非同期モデルに落ち着いた。LangGraph の checkpointer と相性が良い。
層2: 暴走防止ガードレール
冒頭の「同じAPIを400回」事件は、まさにこの層が不在だったせいで起きた。実装すべきは3つ。
(a) ステップ上限。max_iterations は 5〜10 くらいから始めて、ワークフロー特性に応じて調整する。10 を超えるならタスク分割を疑うべき。
(b) コスト上限。これが一番効く。ある海外の運用レポートで見た数字だが、「API gateway 側で $10/day のハードキャップ」を入れるだけでインシデントの95%が止まる、というものがあった。金額は小さく見えるが、暴走はほぼ確実にこの閾値を超えるので検知線として機能する、という話だ。筆者の環境でもこれは体感と合っている。
(c) ループ検知。同じ tool + 同じ args が N 回連続したら止める。あるいは「N ステップ進捗なし」ヒューリスティック。
from collections import deque
class RunGuard:
def init(self, max_steps: int = 8, max_cost_usd: float = 2.0,
loop_window: int = 3):
self.max_steps = max_steps
self.max_cost_usd = max_cost_usd
self.loop_window = loop_window
self.steps = 0
self.cost_usd = 0.0
self.recent_calls: deque = deque(maxlen=loop_window)
def before_step(self, tool_name: str, args: dict):
self.steps += 1
if self.steps > self.max_steps:
raise RuntimeError(f"step_limit_exceeded: {self.steps}")
signature = (tool_name, tuple(sorted(args.items())))
self.recent_calls.append(signature)
if (len(self.recent_calls) == self.loop_window
and len(set(self.recent_calls)) == 1):
raise RuntimeError(f"loop_detected: {tool_name} x{self.loop_window}")
def add_cost(self, usd: float):
self.cost_usd += usd
if self.cost_usd > self.max_cost_usd:
raise RuntimeError(
f"cost_limit_exceeded: ${self.cost_usd:.2f} > ${self.max_cost_usd}")
ここで細かい注意。LLM の思考が同じ入力に対して同じ出力を返すとは限らないため、「引数の完全一致」だけでループ判定するのは弱い。API のレスポンスハッシュや、対象リソース ID の一致を見るとより堅い。ただ、まず素朴な実装で入れて、実データで誤検知/見逃しを見ながら調整するのが早い。
層3: 監査ログ — 「あとから再現できる」を最優先
監査ログで一番よくある失敗は、「LLM の出力だけ残してツール実行結果を残さない」パターンだ。これだとインシデント時に何が起きたか再現できない。最低限、以下は1つの run_id に紐付けて時系列で残す。
system prompt のバージョン識別子(ハッシュや git SHA)
ユーザー入力・retrieved context
LLM 呼び出しごとの request / response(トークン数・コスト込み)
tool 呼び出しの args と result
承認要求と承認者・タイムスタンプ
最終ステータス(success / rejected / cost_exceeded / loop_detected など)
実装は素朴で十分だが、追記専用テーブル(改ざん耐性)にしておくとインシデント対応で信用できる。
import json
import sqlite3
from datetime import datetime
DB = sqlite3.connect("agent_audit.db", check_same_thread=False)
DB.execute("""
CREATE TABLE IF NOT EXISTS agent_events (
id INTEGER PRIMARY KEY AUTOINCREMENT,
run_id TEXT NOT NULL,
ts TEXT NOT NULL,
event_type TEXT NOT NULL,
tool_name TEXT,
payload TEXT,
prompt_version TEXT
)""")
DB.commit()
def audit_log(run_id: str, event_type: str, tool_name: str = None,
args: dict = None, result: Any = None,
prompt_version: str = "v2026-07-01"):
payload = json.dumps({"args": args, "result": result}, ensure_ascii=False,
default=str)
DB.execute(
"INSERT INTO agent_events (run_id, ts, event_type, tool_name, payload, prompt_version) "
"VALUES (?, ?, ?, ?, ?, ?)",
(run_id, datetime.utcnow().isoformat(), event_type, tool_name,
payload, prompt_version)
)
DB.commit()
SQLite はプロトタイプ用。本番では Splunk / Datadog / Microsoft Sentinel のような SIEM に流す構成が定番。特に SOC を持つ組織なら、既存の監査ログ基盤に相乗りするのが最も現実的だ。
層4: ロールバック・失敗検知・アラート
Critical カテゴリのツールには、必ずロールバック手順を「実装として」持たせておく。手順書ではなく、コードとして。理由は単純で、深夜3時のインシデント時に手順書を見ながら復旧するのは無理があるからだ。
失敗検知は、監査ログのイベントタイプを見ればほぼ済む。step_limit_exceeded / loop_detected / cost_limit_exceeded / tool_call_rejected のカウントを 5 分ウィンドウで集計し、閾値を超えたら Slack で on-call に飛ばす。派手なダッシュボードより、この単純アラートの方が実運用では効く。
def notify_slack(req_id: str, tool: Tool, args: dict):
webhook = "YOUR_SLACK_WEBHOOK_URL"
text = (f"[{tool.impact.value.upper()}] 承認要求: {tool.name}\n"
f"args: {json.dumps(args, ensure_ascii=False)}\n"
f"承認: /approve {req_id} 拒否: /reject {req_id}")
# requests.post(webhook, json={"text": text})
print(text) # 動作確認用
動作確認 — インシデントを再現してみる
冒頭の「同じAPIを400回」を、ガードレール込みで再現してみる。inventory_update をわざとループさせる。
run_id = str(uuid.uuid4())
guard = RunGuard(max_steps=8, max_cost_usd=2.0, loop_window=3)
try:
for _ in range(20): # エージェントの暴走を模擬
guard.before_step("inventory_update", {"sku": "A-001", "delta": 1})
guard.add_cost(0.02) # LLM呼び出しコストを加算
# invoke_tool(inventory_update, {"sku": "A-001", "delta": 1}, run_id)
except RuntimeError as e:
audit_log(run_id, "guard_triggered", payload={"reason": str(e)})
print("stopped:", e)
出力例:
stopped: loop_detected: inventory_update x3
ステップが1桁のうちに loop_detected で止まる。$47 のインシデントは、この10行で防げていた。書きながら少し虚しくなるが、これが現実だ。
運用メトリクス — 最低限これだけ
ダッシュボードに載せるものは意外と少なくていい。筆者が定点観測しているのは次の6つ。
成功率 (finished_ok / total_runs)
承認待ち中央値時間(HITL のボトルネック検出)
承認却下率(却下率が高いツールは自動化の適性を再検討)
ガード発動率(step / cost / loop 別)
1 run あたりの平均コスト
ロールバック実行回数(0 に近いほど良いが、0 が続いたらロールバック経路自体が腐っていないか点検)
最後の「ロールバック 0 が続くと経路が腐る」は、実際に一度やらかしたので書いておく。半年間 rollback を呼ばずにいたら、認証トークンの更新に追従できておらず、いざ叩いたら 401 で返ってきた。それ以降、月一で dry-run を回すようにしている。
シリーズを振り返って
3回にわたって、社内システムを操作するエージェントを扱ってきた。初回は tool-use の基本、前回は MCP による複数システム横断、そして今回は本番運用の統制設計。「読む(S5)→操作する(S6-1,2)→安全に運用する(S6-3)」という流れで、社内 LLM/エージェント活用の一つの筋道は通せたのではないかと思う。
本シリーズはここで一区切り。エージェントは、フレームワークの華やかさに目を奪われがちだが、本番で回すために必要なのは、地味な承認境界と監査ログとキルスイッチだ。派手ではないが、これが無いと止まる。逆に、これがあれば、失敗しても復旧できる。エージェントを止めない最良の方法は、いつでも止められるようにしておくことだ、というのが3回書き終えた今の結論だ。
筆者は 5years+ で、韓国・日本の企業向けに AI/LLM の業務応用と本番運用設計を担当している。読者の環境でも役立つ部分があれば嬉しい。