すべての実運用エージェントは、最終的に設計されていない状況に直面します。問題は、それが失敗するかどうかではなく、実際に被害を与える前にそれを検知する仕組みを組み込んでいるかどうかです。
起こしたくないインシデント
エージェントはタスクを実行します。入力に少し差異がある――重複レコード、データの端のケース、あいまいな指示。自信はぎりぎり。エージェントはそれでも前進します。
結果: 不適切な顧客へ送られた一括メール。データベースのレコードが上書きされる。請求が二重に処理される。
今、インシデント対応モードとなり、利害関係者に「完全自律」AIシステムには停止して確認する手段がない理由を説明します。
Human-in-the-loop(HITL)設計は、実運用エージェントには必須です。デモと、実際に信頼できるものを区別する要因です。
5段階の介入レベル
すべての人間の監視が等しく優れているわけではありません。HITL設計で最も大きな誤りの一つは、それを二値として扱うことです――エージェントがすべてを要求するか、何も要求しないか、どちらかに振れてしまうと目的を失います。
正しい抽象化: 5段階のスペクトラム。
class HITLLevel(Enum):
FULL_AUTO = 0 # Act without approval
NOTIFY_ONLY = 1 # Act + notify after
SOFT_APPROVAL = 2 # Wait with timeout (silent consent)
HARD_APPROVAL = 3 # Block until explicit approval
HUMAN_TAKEOVER = 4 # Hand off completely
各レベルの使いどころ:
| Level | 使用条件 |
|---|---|
| FULL_AUTO | 不可逆で低コスト、信頼度 > 0.85 |
| NOTIFY_ONLY | 人間の認識は必要だが制御は不要 |
| SOFT_APPROVAL | 人間が承認する可能性が高く、可視性を望む;タイムアウトは同意とみなす |
| HARD_APPROVAL | 不可逆、金融、PII、規制対象の領域 |
| HUMAN_TAKEOVER | 複数の失敗、あいまいな状況、エージェントの信頼度 < 0.5 |
肝要な洞察: ほとんどのアクションには HARD_APPROVAL は不要。厳格なゲートを過度に使用すると自律性を奪います。過少な使用はインシデントの原因となります。このキャリブレーションを正しく取るのが技術の技です。
信頼度認識によるエスカレーション
以下は、発生前に80%のインシデントを捕捉するパターンです。実行前にエージェント自身の信頼度を評価させます。
CONFIDENCE_PROMPT = """Before proceeding with this task, assess your confidence level.
Task: {task}
Planned Action: {planned_action}
Evaluate:
1. How clear is the task specification? (ambiguous vs. explicit)
2. Are there edge cases you're uncertain about?
3. Do you have all information needed, or are you making assumptions?
4. What’s the consequence if you’re wrong?
Respond with:
CONFIDENCE_SCORE: [0.0-1.0]
RATIONALE: [one sentence]
UNCERTAINTIES: [comma-separated list]
RECOMMENDATION: [PROCEED | CLARIFY | ESCALATE]"""
これは安価で高速なモデル(いわゆる「俳句レベル」)を用いたメタ認知の前処理で、実際の行動へ移る前に自己評価を行います。コストは低く、エッジケースの検出率は驚くほど高いです。
信頼度を HITL レベルへマッピング:
def confidence_to_hitl_level(score: float, recommendation: str) -> HITLLevel:
if recommendation == "ESCALATE" or score < 0.5:
return HITLLevel.HUMAN_TAKEOVER
elif score < 0.65:
return HITLLevel.HARD_APPROVAL
elif score < 0.80:
return HITLLevel.SOFT_APPROVAL
else:
return HITLLevel.FULL_AUTO
承認ゲートパターン
中核となるインフラストラクチャ: 4つの非自動レベルを一貫した動作で扱う承認ゲート。
class ApprovalGate:
def __init__(self, notifier, storage,
soft_approval_timeout_s = 300, # 5 min
hard_approval_timeout_s = 86400): # 24 hours
self.notifier = notifier
self.storage = storage
self.soft_timeout = soft_approval_timeout_s
self.hard_timeout = hard_approval_timeout_s
self._pending: dict[str, asyncio.Future] = {}
Note the asymmetry: soft approval timeout means approved (human had the chance to object). Hard approval timeout means escalate (you can't assume consent for high-stakes actions).
Async Flows: Don't Block Your Server
The most common HITL mistake in web services: blocking an HTTP connection waiting for human input.
‚ùå Wrong:
[HTTP Request] ‚Üí [Agent starts] ‚Üí [Waits 2 hours for approval] ‚Üí [Connection times out] ‚Üí üí•
‚úÖ Right:
[HTTP Request] ‚Üí [Agent starts] ‚Üí [Saves state + task_id] ‚Üí [Returns 202 Accepted]
‚Üì
[Human reviews] ‚Üí [POST /approve with task_id] ‚Üí [Agent resumes] ‚Üí [Sends result]
The implementation: break approval flows into two HTTP request lifecycle. Store pending task state in Redis. Return a task_id immediately. Provide a polling endpoint and a webhook endpoint for approval responses.
@app.post(\"/tasks/{task_id}/start\")
async def start_task(task_id: str, input: TaskInput):
# タスクを開始し、状態を保存し、task_id を返す
# 承認が必要な場合 status = \"pending_approval\"
return {\"task_id\": task_id, \"status\": \"pending_approval\"}
@app.post(\"/tasks/approve\")
async def approve_task(webhook: ApprovalWebhook):
# ヒューマンがトリガーするエンドポイント
# 実行または停止されたタスクを再開
result = await orchestrator.resume_after_approval(
task_id=webhook.task_id,
approved= webhook.approved,
reviewer_id= webhook.reviewer_id,
)
return result
@app.get(\"/tasks/{task_id}/status\")
async def task_status(task_id: str):
return await state_store.get(task_id)
Progressive Autonomy: Trust as a Ratchet
エージェントは特定の HITL レベルに永久にとどまるべきではありません。信頼は、示された信頼性によって築かれます。
@dataclass
class AutonomyProfile:
agent_id: str
current_level: HITLLevel = HITLLevel.SOFT_APPROVAL
consecutive_successes: int = 0
promote_after_successes: int = 10 # Conservative
demote_after_failures: int = 2 # Fast demotion
failure_window_hours: int = 24
def record_outcome(self, success: bool):
if success:
self.consecutive_successes += 1
if self.consecutive_successes >= self.promote_after_successes:
# Promote to less oversight
new_value = max(0, self.current_level.value - 1)
self.current_level = HITLLevel(new_value)
self.consecutive_successes = 0
else:
self.consecutive_successes = 0
self.recent_failures += 1
if self.recent_failures >= self.demote_after_failures:
# Demote to more oversight immediately
new_value = min(4, self.current_level.value + 1)
self.current_level = HITLLevel(new_value)
実用的な効果: 新しいエージェントは SOFT_APPROVAL から開始します。連続して 10 回の成功の後、NOTIFY_ONLY へ昇格します。さらに 20 回で、そのアクションタイプの FULL_AUTO になります。24 時間で 2 回の失敗があると、直ちに SOFT_APPROVAL に戻ります。
ラチェット原理: 昇進は遅い(成功数10)、降格は速い(失敗数2)。この非対称性は現実を反映している—信頼はゆっくりと築かれ、壊れるのは早い。
人間による円滑な引き継ぎ
HUMAN_TAKEOVERが発動した場合、エージェントをただ停止させないでください。人間が継続するために必要なすべてを提供してください。
...コードブロックは省略されました...
LLM生成のハンドオフパッケージは、人間がログを読まずに文脈を理解できるようにします。状況を理解するのにかかる30秒は、フォレンジック調査の30分よりも有益です。
HITL監査の追跡記録
規制産業、エンタープライズ顧客、そして事後インシデントのレビューのためには、完全な記録が必要です。
...コードブロックは省略されました...
スキーマのヒント: 承認リクエストから解決までの latency_ms を含めてください。この指標は、通知パイプラインが機能しているかどうか、レビュアーの応答がどれだけ迅速かを示します。SLA設計には、どちらも重要です。
HITL意思決定マトリクス(クイックリファレンス)
アクションは不可逆ですか?
YES 金融、PII、規制対象ですか? HARD_APPROVAL は常に
NO 自信度 > 0.75?
YES SOFT_APPROVAL
NO HARD_APPROVAL
NO コスト > $100? HARD_APPROVAL
No 自信度 > 0.85? FULL_AUTO / NOTIFY_ONLY
No SOFT_APPROVAL
過去24時間で複数回の失敗ですか? HUMAN_TAKEOVER は上記に関係なく
本番環境での見え方
うまく設計された HITL システムは、物事が正しく動作しているときにはほとんど見えません。アクションは流れ、人間には時折通知が届き、監査ログは背景で静かに蓄積されます。
システムは、問題が起こる場合、あるいはほとんど起こらない場合に価値を示します。境界信頼度のアクションはソフト承認へと振り分けられます。人間はそれを見てエッジケースを認識し、却下します。エージェントは拒否を記録し、文脈を調整し、別のアプローチを試みます。インシデントは発生しません。ポストモーテムもありません。
それが目標です:エージェントを閉じ込めるのではなく、状況がその確信を超えたときに信頼できるフォールバックを提供すること。
さらに読む
この投稿はアーキテクチャのパターンを扱っています。完全な実装―― 完全な ApprovalGate クラス、Slack 通知、FastAPI を使用した非同期 AsyncHITLOrchestrator、完全な ProgressiveAutonomyManager、LLM 生成パッケージを使用した GracefulTakeoverHandler、HITLAuditLogger、マルチエージェントの権限委譲、そして 35 点のリリース前チェックリスト――は、Machina Market(MAC-016, 0.016 ETH)にリファレンスパックとしてまとめました。
パターンのいずれかについての質問ですか? コメントで具体的な点を掘り下げるのは喜んで対応します。
タグ: #ai #python #agents #architecture #safety