0Pricing
AI Agents · Lekcja

Bramki zatwierdzania z udziałem człowieka

Wzorce wstrzymaj–poproś–zatwierdź dla działań agentów o dużym znaczeniu.

Bramki zatwierdzania z udziałem człowieka to bezpłatna lekcja AI Agents na CoddyKit. To lekcja 4 z 4. Możesz przeczytać całą lekcję poniżej za darmo — a potem ćwiczyć ją interaktywnie w przeglądarce z wbudowanym edytorem kodu i tutorem AI dostępnym 24/7. To część ścieżki edukacyjnej AI Agents, a Twój postęp synchronizuje się między webem a aplikacją CoddyKit. Kurs AI Agents zawiera 4 lekcji w sumie.

Kiedy agenci potrzebują zatwierdzenia przez człowieka

W przypadku działań o wysokiej stawce — wysłania umowy, usunięcia danych produkcyjnych, wydania środków z budżetu czy wysłania publicznego komunikatu — agent nie powinien działać autonomicznie. Powinien wstrzymać działanie i poprosić człowieka o zatwierdzenie przed kontynuowaniem.

Jest to wzorzec Human-in-the-Loop (HITL).

Definiowanie działań o wysokiej stawce

Zdefiniuj funkcję klasyfikującą, która określa, które działania wymagają zatwierdzenia przez człowieka. Próg można dostosować do każdego wdrożenia na podstawie poziomu akceptowanego ryzyka.

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

Tworzenie żądania zatwierdzenia

Gdy wymagane jest zatwierdzenie, agent tworzy rekord żądania zatwierdzenia i wstrzymuje wykonywanie. Żądanie zawiera opis tego, co się wydarzy, parametry oraz termin.

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

Powiadamianie za pośrednictwem Slacka

Wyślij żądanie zatwierdzenia do osoby zatwierdzającej za pośrednictwem Slacka. Dołącz podsumowanie działania, które agent chce wykonać, łącze umożliwiające zatwierdzenie lub odrzucenie oraz termin wygaśnięcia.

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)

Powiadamianie za pośrednictwem poczty e-mail

Jako kanał zapasowy (lub podstawowy w organizacjach, które nie korzystają ze Slacka) wysyłaj żądania zatwierdzenia pocztą e-mail, używając jednoznacznych łączy zatwierdzenia i odrzucenia.

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)

Odpytywanie o decyzję dotyczącą zatwierdzenia

Po wysłaniu powiadomienia agent czeka na decyzję. Użyj pętli odpytywania z krótkim odstępem uśpienia. Zakończ odpytywanie, gdy status przestanie być równy 'pending' lub upłynie termin.

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)

Rejestrowanie decyzji dotyczącej zatwierdzenia

Gdy osoba zatwierdzająca kliknie przycisk Zatwierdź lub Odrzuć, zapisz, kto podjął decyzję i kiedy. Tworzy to pełny ślad audytowy zatwierdzenia: żądanie → powiadomienie → decyzja → wykonanie (lub anulowanie).

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}

Obsługa limitu czasu: automatyczne anulowanie

Jeśli w wyznaczonym czasie nie nadejdzie żadna decyzja, agent automatycznie anuluje działanie i powiadamia pierwotnego wnioskodawcę. Zapobiega to trwałemu blokowaniu działań przez osoby zatwierdzające, które nie odpowiadają.

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

Pełna orkiestracja procesu zatwierdzania

Połącz wszystkie kroki: sprawdź, czy zatwierdzenie jest wymagane, utwórz żądanie, wyślij powiadomienie, poczekaj, a następnie kontynuuj lub anuluj działanie na podstawie decyzji.

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}

Ślad audytowy zatwierdzenia

Każde zdarzenie w cyklu życia zatwierdzenia musi zostać zapisane w dzienniku audytowym: utworzenie żądania, wysłanie powiadomienia, podjęcie decyzji oraz wykonanie lub anulowanie działania. Taki ślad jest wymagany do zachowania zgodności z 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'})

Eskalacja w przypadku niedostępności głównej osoby zatwierdzającej

Jeśli główna osoba zatwierdzająca nie odpowie w ciągu połowy wyznaczonego czasu, przekaż sprawę zastępczej osobie zatwierdzającej. Zapobiega to blokowaniu wszystkich zatwierdzeń z powodu nieobecności jednej osoby.

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

Co dzieje się z oczekującym żądaniem zatwierdzenia po upływie limitu czasu?

Sposób obsługi limitu czasu jest kluczowym elementem projektowania systemu HITL. Wybór niewłaściwego ustawienia domyślnego ma różne konsekwencje dla bezpieczeństwa i użyteczności.

Podsumowanie bramek zatwierdzania Human-in-the-Loop

Bramki zatwierdzania HITL działają poprzez: klasyfikowanie działań o wysokiej stawce, tworzenie wstrzymanego żądania zatwierdzenia, powiadamianie osób zatwierdzających za pośrednictwem Slacka lub poczty e-mail, odpytywanie o decyzje, automatyczne anulowanie po przekroczeniu limitu czasu oraz rejestrowanie każdego zdarzenia cyklu życia w śladzie audytowym.

Eskalacja do zastępczych osób zatwierdzających zapobiega blokadom procesu zatwierdzania wynikającym z niedostępności.

Często zadawane pytania

Czy lekcja „Bramki zatwierdzania z udziałem człowieka” jest bezpłatna?

Tak — pełny tekst „Bramki zatwierdzania z udziałem człowieka” jest dostępny za darmo tutaj w sieci. Aby ćwiczyć ją interaktywnie (wbudowany edytor kodu i tutor AI dostępny 24/7) i odblokować resztę kursu AI Agents, przejdź na CoddyKit PRO. Kurs AI Agents zawiera 4 lekcji w sumie.

Co nauczysz się w „Bramki zatwierdzania z udziałem człowieka”?

Wzorce wstrzymaj–poproś–zatwierdź dla działań agentów o dużym znaczeniu. Ćwiczysz AI Agents z praktycznym kodem, który uruchamiasz bezpośrednio w przeglądarce, a tutor AI dostępny 24/7 odpowiada na Twoje pytania podczas pracy nad lekcją.

Czy potrzebuję doświadczenia, aby zacząć AI Agents?

Nie wymagamy żadnego doświadczenia. AI Agents w CoddyKit jest strukturyzowany dla początkujących i zaawansowanych użytkowników, więc możesz zacząć tutaj lub od początku i uczyć się w swoim tempie. To lekcja 4 z 4.

Ile czasu zajmuje lekcja „Bramki zatwierdzania z udziałem człowieka”?

Większość lekcji CoddyKit trwa około 5–10 minut. Każda lekcja to mały, interaktywny krok, dzięki czemu robisz systematyczne postępy i zawsze wracasz dokładnie do tego samego miejsca — na webie i w aplikacji.

Czy mogę pisać i uruchamiać kod w tej lekcji AI Agents?

Tak. Każda lekcja AI Agents zawiera wbudowany edytor kodu, więc piszesz i uruchamiasz prawdziwy kod bezpośrednio w przeglądarce i od razu otrzymujesz sprzężenie zwrotne od AI — bez konfiguracji na komputerze.

Wszystkie lekcje w tym kursie

  1. Niezmienny rejestr działań agentów
  2. Egzekwowanie zasad dotyczących działań agentów
  3. Zgodność z regulacjami: GDPR i SOC2
  4. Bramki zatwierdzania z udziałem człowieka
← Powrót do AI Agents