จุดอนุมัติโดยมนุษย์ในวงจรการทำงาน

เมื่อเอเจนต์ต้องได้รับการอนุมัติจากมนุษย์

การกำหนดการดำเนินการที่มีความเสี่ยงสูง

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

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)

การแจ้งเตือนผ่านอีเมล

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)

การตรวจสอบการตัดสินใจอนุมัติเป็นระยะ

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)

การบันทึกการตัดสินใจอนุมัติ

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}

ร่องรอยการตรวจสอบการอนุมัติ

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'})

การยกระดับเมื่อผู้อนุมัติหลักไม่พร้อมใช้งาน

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}')

จะเกิดอะไรขึ้นกับคำขออนุมัติที่รอดำเนินการเมื่อหมดเวลารอ

ทบทวนประตูการอนุมัติแบบมนุษย์ร่วมในกระบวนการ