Freigabestufen mit Human-in-the-Loop
Muster zum Pausieren, Anfordern und Freigeben für kritische Agentenaktionen.
Freigabestufen mit Human-in-the-Loop ist eine kostenlose AI Agents-Lektion auf CoddyKit. Dies ist Lektion 4 von 4. Du kannst die komplette Lektion unten kostenlos lesen – dann übst du sie direkt im Browser mit einem integrierten Code-Editor und einem KI-Tutor rund um die Uhr. Sie ist Teil des AI Agents-Lernpfads, und dein Fortschritt wird über Web und CoddyKit-App synchronisiert. Der AI Agents-Kurs umfasst insgesamt 4 Lektionen.
Wann Agenten eine menschliche Genehmigung benötigen
Bei Aktionen mit hohem Risiko — etwa dem Senden eines Vertrags, dem Löschen von Produktionsdaten, dem Ausgeben eines Budgets oder dem Versenden einer öffentlichen Ankündigung — sollte ein Agent nicht autonom handeln. Er sollte pausieren und eine menschliche Genehmigung anfordern, bevor er fortfährt.
Dies ist das Human-in-the-Loop-Muster (HITL).
Aktionen mit hohem Risiko definieren
Definieren Sie eine Klassifizierungsfunktion, die ermittelt, für welche Aktionen eine menschliche Genehmigung erforderlich ist. Der Schwellenwert kann je nach Bereitstellung auf Grundlage der Risikotoleranz angepasst werden.
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', {}))
Eine Genehmigungsanfrage erstellen
Wenn eine Genehmigung erforderlich ist, erstellt der Agent einen Datensatz für die Genehmigungsanfrage und unterbricht die Ausführung. Die Anfrage enthält eine Beschreibung des Vorgangs, die Parameter und eine Frist.
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'])
Über Slack benachrichtigen
Senden Sie die Genehmigungsanfrage über Slack an die genehmigende Person. Fügen Sie eine Zusammenfassung der geplanten Aktion des Agenten, einen Link zum Genehmigen oder Ablehnen sowie die Frist für den Timeout hinzu.
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)Über E-Mail benachrichtigen
Senden Sie Genehmigungsanfragen als zusätzlichen oder primären Kanal für Organisationen ohne Slack per E-Mail mit eindeutig gekennzeichneten Links zum Genehmigen und Ablehnen.
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)Die Genehmigungsentscheidung abfragen
Nach dem Senden der Benachrichtigung wartet der Agent auf die Entscheidung. Verwenden Sie eine Polling-Schleife mit einem kurzen Warteintervall. Beenden Sie das Polling, sobald sich der Status von 'pending' ändert oder die Frist abläuft.
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)Die Genehmigungsentscheidung aufzeichnen
Wenn die genehmigende Person auf „Genehmigen“ oder „Ablehnen“ klickt, zeichnen Sie auf, wer die Entscheidung getroffen hat und wann. Dadurch entsteht der vollständige Prüfpfad der Genehmigung: angefordert → benachrichtigt → entschieden → ausgeführt (oder abgebrochen).
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}Timeout-Behandlung: automatisch abbrechen
Wenn innerhalb des Timeout-Zeitfensters keine Entscheidung eintrifft, bricht der Agent die Aktion automatisch ab und benachrichtigt die ursprüngliche anfragende Person. Dadurch wird verhindert, dass Aktionen durch nicht reagierende genehmigende Personen dauerhaft blockiert werden.
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_messageVollständige Orchestrierung des Genehmigungsgatters
Führen Sie alle Schritte zusammen: Prüfen Sie, ob eine Genehmigung erforderlich ist, erstellen Sie die Anfrage, benachrichtigen Sie die zuständigen Personen, warten Sie und fahren Sie anschließend abhängig von der Entscheidung fort oder brechen Sie ab.
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}Prüfpfad für Genehmigungen
Jedes Ereignis im Lebenszyklus einer Genehmigung muss im Audit-Protokoll aufgezeichnet werden: Anfrage erstellt, Benachrichtigung gesendet, Entscheidung getroffen, Aktion ausgeführt oder abgebrochen. Dieser Prüfpfad ist für die SOC-2-Compliance erforderlich.
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'})
Eskalation, wenn die primäre genehmigende Person nicht verfügbar ist
Wenn die primäre genehmigende Person innerhalb der Hälfte des Timeout-Zeitfensters nicht antwortet, eskalieren Sie die Anfrage an eine sekundäre genehmigende Person. So wird verhindert, dass alle Genehmigungen blockiert werden, weil eine einzelne Person abwesend ist.
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}')Was geschieht mit einer ausstehenden Genehmigungsanfrage, wenn der Timeout abläuft?
Das Verhalten bei einem Timeout ist ein wichtiger Bestandteil des HITL-Designs. Die Wahl des falschen Standardverhaltens hat unterschiedliche Auswirkungen auf Sicherheit und Benutzerfreundlichkeit.
Zusammenfassung: Human-in-the-Loop-Genehmigungsgatter
HITL-Genehmigungsgatter funktionieren, indem sie Aktionen mit hohem Risiko klassifizieren, eine angehaltene Genehmigungsanfrage erstellen, die zuständigen Personen über Slack oder E-Mail benachrichtigen, Entscheidungen abfragen, bei einem Timeout automatisch abbrechen und jedes Ereignis im Lebenszyklus im Prüfpfad aufzeichnen.
Die Eskalation an sekundäre genehmigende Personen verhindert Genehmigungsblockaden aufgrund von Nichtverfügbarkeit.
Häufig gestellte Fragen
Ist die Lektion „Freigabestufen mit Human-in-the-Loop“ kostenlos?
Ja — der vollständige Text von „Freigabestufen mit Human-in-the-Loop“ ist hier im Web kostenlos zu lesen. Um sie interaktiv zu üben (integrierter Code-Editor und 24/7 KI-Tutor) und den Rest des AI Agents-Kurses freizuschalten, upgrade auf CoddyKit PRO. Der AI Agents-Kurs umfasst insgesamt 4 Lektionen.
Was lerne ich in „Freigabestufen mit Human-in-the-Loop“?
Muster zum Pausieren, Anfordern und Freigeben für kritische Agentenaktionen. Du übst AI Agents mit praktischem Code, den du direkt im Browser ausführst, und ein 24/7 KI-Tutor beantwortet deine Fragen während du die Lektion bearbeitest.
Brauche ich Erfahrung, um AI Agents zu starten?
Keine Vorkenntnisse erforderlich. AI Agents auf CoddyKit ist für Anfänger bis fortgeschrittene Lernende strukturiert, sodass du hier starten oder von Anfang an beginnen und in deinem eigenen Tempo voranschreiten kannst. Dies ist Lektion 4 von 4.
Wie lange dauert die Lektion „Freigabestufen mit Human-in-the-Loop“?
Die meisten CoddyKit-Lektionen dauern etwa 5–10 Minuten. Jede ist kompakt und interaktiv, sodass du stetig Fortschritte machst und genau dort weitermachst, wo du aufgehört hast – im Web und in der App.
Kann ich in dieser AI Agents-Lektion Code schreiben und ausführen?
Ja. Jede AI Agents-Lektion enthält einen integrierten Code-Editor, sodass du echten Code direkt in deinem Browser schreibst und ausführst und sofort KI-Feedback erhältst — ohne lokale Einrichtung erforderlich.
Alle Lektionen in diesem Kurs
- Unveränderliche Aktionsprotokollierung für Agenten
- Richtliniendurchsetzung für Agentenaktionen
- Einhaltung gesetzlicher Vorgaben: GDPR und SOC2
- Freigabestufen mit Human-in-the-Loop