Human-in-the-Loop 承認ゲート
リスクの高いエージェントアクションを一時停止し、要求して承認するパターンを学びます。
「Human-in-the-Loop 承認ゲート」はCoddyKit上の無料AI Agentsレッスンです。 これはレッスン4/4です。 下記で完全なレッスンを無料で読むことができます。その後、ブラウザ内の組み込みコードエディタと24時間対応のAIチューターでハンズオン演習できます。 これはAI Agents学習パスの一部であり、ウェブとCoddyKitアプリ全体で進捗が同期されます。 AI Agentsコースには全4レッスンが含まれています。
エージェントに人間の承認が必要な場合
契約書の送信、本番データの削除、予算の使用、公開告知の送信など、重大な結果を伴うアクションについては、エージェントが自律的に実行してはいけません。実行する前に一時停止して人間の承認を求める必要があります。
これはHuman-in-the-Loop（HITL）パターンです。
重大な結果を伴うアクションの定義
人間の承認が必要なアクションを特定する分類関数を定義します。しきい値は、リスク許容度に応じてデプロイごとに調整できます。
HIGH_STAKES_ACTIONS = {
'send_email',
'delete_records',
'publish_content',
'transfer_funds',
'modify_production_config',
'export_all_data',
'send_push_notification_to_all'
}
HIGH_STAKES_THRESHOLDS = {
'transfer_funds': 1000, # USD — require approval above this
'delete_records': 10, # rows
'send_email': 50, # recipients
'push_notification': 1000 # users
}
def requires_approval(action: str, parameters: dict) -> bool:
if action not in HIGH_STAKES_ACTIONS:
return False
threshold = HIGH_STAKES_THRESHOLDS.get(action)
if threshold is None:
return True # all instances require approval
# Check parameter against threshold
amount = parameters.get('amount') or parameters.get('count') or 0
return float(amount) >= threshold
if __name__ == '__main__':
print('Small transfer:', requires_approval('transfer_funds', {'amount': 200}))
print('Large transfer:', requires_approval('transfer_funds', {'amount': 5000}))
print('Publish content:', requires_approval('publish_content', {}))
承認リクエストの作成
承認が必要になると、エージェントは承認リクエストのレコードを作成し、実行を一時停止します。リクエストには、実行される内容の説明、パラメーター、期限を含めます。
import uuid, time
approval_requests: dict[str, dict] = {} # approval_id -> request
def create_approval_request(agent_id: str, user_id: str, action: str,
parameters: dict, timeout_minutes: int = 30) -> str:
approval_id = str(uuid.uuid4())
approval_requests[approval_id] = {
'approval_id': approval_id,
'agent_id': agent_id,
'user_id': user_id,
'action': action,
'parameters': parameters,
'status': 'pending', # pending / approved / rejected / timed_out
'created_at': time.time(),
'expires_at': time.time() + timeout_minutes * 60,
'decided_by': None,
'decided_at': None
}
return approval_id
if __name__ == '__main__':
approval_id = create_approval_request(
'agent-1', 'user-42', 'delete_records', {'count': 50}
)
print('Created approval request:', approval_id)
print('Status:', approval_requests[approval_id]['status'])
Slackによる通知
承認リクエストをSlack経由で承認者に送信します。エージェントが実行しようとしている内容の概要、承認または拒否を行うリンク、タイムアウトの期限を含めてください。
import requests
SLACK_WEBHOOK = 'https://hooks.slack.com/services/YOUR/WEBHOOK/URL'
APPROVAL_BASE_URL = 'https://your-agent-dashboard.com/approvals'
def notify_approver_slack(approval_id: str, approver_slack_id: str):
req = approval_requests[approval_id]
import time as t
from datetime import datetime
expires = datetime.fromtimestamp(req['expires_at']).strftime('%H:%M UTC')
payload = {
'text': f'<@{approver_slack_id}> Agent approval required',
'attachments': [{
'color': '#ff9900',
'fields': [
{'title': 'Action', 'value': req['action'], 'short': True},
{'title': 'Requested by','value': req['user_id'], 'short': True},
{'title': 'Parameters', 'value': str(req['parameters'])[:200]},
{'title': 'Expires', 'value': expires, 'short': True}
],
'actions': [
{'type': 'button', 'text': 'Approve',
'url': f'{APPROVAL_BASE_URL}/{approval_id}/approve'},
{'type': 'button', 'text': 'Reject',
'url': f'{APPROVAL_BASE_URL}/{approval_id}/reject'}
]
}]
}
requests.post(SLACK_WEBHOOK, json=payload)メールによる通知
バックアップ用のチャネルとして、またはSlackを利用していない組織の主要チャネルとして、明確な承認・拒否リンクを記載した承認リクエストをメールで送信します。
import smtplib
from email.mime.text import MIMEText
SMTP_HOST = 'smtp.yourcompany.com'
SMTP_PORT = 587
SMTP_USER = 'agent-noreply@yourcompany.com'
SMTP_PASS = 'YOUR_SMTP_PASSWORD'
def notify_approver_email(approval_id: str, approver_email: str):
req = approval_requests[approval_id]
body = (
f'An AI agent is requesting approval for:\n\n'
f'Action: {req["action"]}\n'
f'Parameters: {req["parameters"]}\n\n'
f'Approve: {APPROVAL_BASE_URL}/{approval_id}/approve\n'
f'Reject: {APPROVAL_BASE_URL}/{approval_id}/reject\n\n'
f'This request expires in 30 minutes.'
)
msg = MIMEText(body)
msg['Subject'] = f'Agent Approval Required: {req["action"]}'
msg['From'] = SMTP_USER
msg['To'] = approver_email
with smtplib.SMTP(SMTP_HOST, SMTP_PORT) as server:
server.starttls()
server.login(SMTP_USER, SMTP_PASS)
server.send_message(msg)承認判断のポーリング
通知を送信した後、エージェントは判断を待ちます。短いスリープ間隔を設定したポーリングループを使用してください。ステータスが「pending」から変わるか、期限を過ぎたらポーリングを停止します。
import time
def wait_for_approval(approval_id: str, poll_interval: float = 5.0) -> dict:
while True:
req = approval_requests.get(approval_id)
if not req:
return {'decision': 'error', 'reason': 'Approval request not found'}
if req['status'] == 'approved':
return {'decision': 'approved', 'decided_by': req['decided_by']}
if req['status'] == 'rejected':
return {'decision': 'rejected', 'decided_by': req['decided_by']}
if time.time() > req['expires_at']:
req['status'] = 'timed_out'
return {'decision': 'timed_out', 'reason': 'No decision within deadline'}
time.sleep(poll_interval)承認判断の記録
承認者がApproveまたはRejectをクリックしたら、判断した人物と時刻を記録します。これにより、リクエスト作成 → 通知 → 判断 → 実行（またはキャンセル）という完全な承認監査証跡が作成されます。
def record_decision(approval_id: str, decision: str,
decided_by: str) -> dict:
req = approval_requests.get(approval_id)
if not req:
return {'error': 'Approval request not found'}
if req['status'] != 'pending':
return {'error': f'Request already in state: {req["status"]}'}
if time.time() > req['expires_at']:
req['status'] = 'timed_out'
return {'error': 'Request has expired'}
req['status'] = decision # 'approved' or 'rejected'
req['decided_by'] = decided_by
req['decided_at'] = time.time()
return {'ok': True, 'decision': decision}タイムアウト処理：自動キャンセル
タイムアウト期間内に判断が届かなかった場合、エージェントはアクションを自動的にキャンセルし、元のリクエスト担当者に通知します。これにより、応答しない承認者によってアクションが永久にブロックされるのを防ぎます。
def handle_timeout(approval_id: str, agent_session: dict) -> str:
req = approval_requests.get(approval_id, {})
action = req.get('action', 'unknown')
user_id = req.get('user_id', 'unknown')
# Log the timeout
import logging
logging.warning(
'Approval timeout: action=%s user=%s approval_id=%s',
action, user_id, approval_id
)
# Tell the user
timeout_message = (
f'The "{action}" action was automatically cancelled because '
f'no approver responded within the 30-minute window. '
f'Please request again or contact your administrator.'
)
return timeout_message承認ゲートの完全なオーケストレーション
すべての手順を組み合わせます。承認が必要か確認し、リクエストを作成して通知を送り、待機した後、判断に基づいて実行またはキャンセルします。
def execute_with_approval_gate(agent_id: str, user_id: str, action: str,
parameters: dict, approver_email: str,
execute_fn) -> dict:
# Step 1: Check if approval needed
if not requires_approval(action, parameters):
result = execute_fn(action, parameters)
return {'approved': True, 'auto': True, 'result': result}
# Step 2: Create approval request
approval_id = create_approval_request(agent_id, user_id, action, parameters)
# Step 3: Notify approver
notify_approver_email(approval_id, approver_email)
print(f'Approval requested: {approval_id}. Waiting...')
# Step 4: Wait for decision
decision = wait_for_approval(approval_id)
# Step 5: Act on decision
if decision['decision'] == 'approved':
result = execute_fn(action, parameters)
return {'approved': True, 'decided_by': decision['decided_by'], 'result': result}
elif decision['decision'] == 'rejected':
return {'approved': False, 'reason': 'Rejected by approver'}
else:
msg = handle_timeout(approval_id, {})
return {'approved': False, 'reason': msg}承認監査証跡
承認ライフサイクルのすべてのイベントを監査ログに記録する必要があります。対象は、リクエストの作成、通知の送信、判断、アクションの実行またはキャンセルです。この証跡はSOC 2への準拠に必要です。
import logging, json, time
approval_logger = logging.getLogger('agent.approvals')
def audit_approval_event(event: str, approval_id: str, details: dict):
entry = {
'timestamp': time.time(),
'event': event,
'approval_id': approval_id,
**details
}
approval_logger.info(json.dumps(entry))
# Usage flow:
# audit_approval_event('request_created', approval_id, {'action': 'delete_records', 'user': 'u123'})
# audit_approval_event('notification_sent', approval_id, {'channel': 'email', 'approver': 'admin@co.com'})
# audit_approval_event('decision_received', approval_id, {'decision': 'approved', 'by': 'admin@co.com'})
# audit_approval_event('action_executed', approval_id, {'result': 'success'})
if __name__ == '__main__':
import sys
approval_logger.setLevel(logging.INFO)
approval_logger.addHandler(logging.StreamHandler(sys.stdout))
audit_approval_event('request_created', 'appr-1', {'action': 'delete_records', 'user': 'u123'})
主承認者が対応できない場合のエスカレーション
主承認者がタイムアウト期間の半分以内に応答しない場合は、二次承認者にエスカレーションします。これにより、1人の不在によってすべての承認が停止するのを防ぎます。
def escalate_if_needed(approval_id: str, secondary_email: str,
escalation_at_pct: float = 0.5):
req = approval_requests.get(approval_id)
if not req or req['status'] != 'pending':
return
total_window = req['expires_at'] - req['created_at']
elapsed = time.time() - req['created_at']
escalation_at = req['created_at'] + total_window * escalation_at_pct
if time.time() >= escalation_at and not req.get('escalated'):
req['escalated'] = True
notify_approver_email(approval_id, secondary_email)
print(f'Escalated approval {approval_id} to {secondary_email}')タイムアウトの期限を過ぎると保留中の承認リクエストはどうなりますか
タイムアウト処理は、HITL設計における重要な要素です。誤ったデフォルトを選択すると、安全性と使いやすさに異なる影響が生じます。
Human-in-the-Loop承認ゲートの総括
HITL承認ゲートは、重大な結果を伴うアクションの分類、一時停止状態の承認リクエストの作成、Slackまたはメールによる承認者への通知、判断のポーリング、タイムアウト時の自動キャンセル、監査証跡へのすべてのライフサイクルイベントの記録によって機能します。
二次承認者へのエスカレーションにより、承認者の不在による承認のデッドロックを防げます。
よくある質問
「Human-in-the-Loop 承認ゲート」レッスンは無料ですか？
はい。「Human-in-the-Loop 承認ゲート」の完全なテキストはこのウェブで無料で読めます。インタラクティブに演習し（組み込みコードエディタと24時間対応のAIチューター）、AI Agentsコースの残りをアンロックするには、CoddyKit PROにアップグレードしてください。 AI Agentsコースには全4レッスンが含まれています。
「Human-in-the-Loop 承認ゲート」で何を学びますか？
リスクの高いエージェントアクションを一時停止し、要求して承認するパターンを学びます。 ブラウザで直接実行するハンズオンコードでAI Agentsを演習し、24時間対応のAIチューターがレッスンを進める中での質問に答えます。
AI Agentsを始めるのに経験は必要ですか？
事前経験は必要ありません。CoddyKitのAI Agentsは初級者から上級者向けに構成されているため、ここから始めるか最初から始めて、自分のペースで進むことができます。 これはレッスン4/4です。
「Human-in-the-Loop 承認ゲート」レッスンにはどのくらい時間がかかりますか？
ほとんどのCoddyKitレッスンは約5～10分かかります。各レッスンはコンパクトでインタラクティブなので、着実に進歩し、ウェブとアプリ全体で正確に前回の場所から再開できます。
このAI Agentsレッスンでコードを書いて実行できますか？
はい。すべてのAI Agentsレッスンに組み込みコードエディタが含まれているため、ブラウザでリアルコードを書いて実行し、即座のAIフィードバックを取得できます。ローカル設定は不要です。
このコースのすべてのレッスン
- エージェントの改ざん不能なアクションログ
- エージェントアクションへのポリシー適用
- 規制コンプライアンス：GDPR と SOC2
- Human-in-the-Loop 承認ゲート