Tekoälyagentit · Oppitunti

Ihmisen hyväksyntää edellyttävät portit

Keskeytä–pyydä–hyväksy-mallit korkean riskin agenttitoimintoihin.

Oppitunti 4/413 vaihetta

Ihmisen hyväksyntää edellyttävät portit on ilmainen Tekoälyagentit-oppitunti CoddyKitissä. Tämä on oppitunti 4/4. Voit lukea koko oppitunnin alta ilmaiseksi ja harjoitella sen jälkeen käytännössä selaimessa sisäänrakennetulla koodieditorilla ja ympäri vuorokauden käytettävissä olevan tekoälytuutorin avulla. Oppitunti kuuluu Tekoälyagentit-oppimispolkuun, ja edistymisesi synkronoituu verkon ja CoddyKit-sovelluksen välillä. Tekoälyagentit-kurssilla on yhteensä 4 oppituntia.

Milloin agentit tarvitsevat ihmisen hyväksynnän

Korkean riskin toiminnoissa — sopimuksen lähettämisessä, tuotantodatan poistamisessa, budjetin käyttämisessä tai julkisen ilmoituksen lähettämisessä — agentin ei pidä toimia itsenäisesti. Sen on pysähdyttävä ja pyydettävä ihmiseltä hyväksyntä ennen jatkamista.

Tätä kutsutaan Human-in-the-Loop (HITL) -malliksi.

Korkean riskin toimien määrittäminen

Luokittelufunktiolla määritetään, mitkä toimet edellyttävät ihmisen hyväksyntää. Kynnysarvoa voidaan säätää käyttöönottokohtaisesti riskinsietokyvyn perusteella.

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

Hyväksyntäpyynnön luominen

Kun hyväksyntä vaaditaan, agentti luo hyväksyntäpyyntötietueen ja keskeyttää suorituksen. Pyyntö sisältää kuvauksen suoritettavasta toimesta, parametrit ja määräajan.

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

Ilmoittaminen Slackissa

Lähettäkää hyväksyntäpyyntö hyväksyjälle Slackin kautta. Ilmoittakaa siinä lyhyesti, mitä agentti haluaa tehdä, antakaa linkki hyväksymistä tai hylkäämistä varten ja ilmoittakaa määräaika, jonka jälkeen pyyntö aikakatkaistaan.

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)

Ilmoittaminen sähköpostitse

Varakanavana (tai ensisijaisena kanavana organisaatioissa, joissa Slackia ei käytetä) hyväksyntäpyynnöt voidaan lähettää sähköpostitse. Viestissä on oltava selkeät linkit hyväksymistä ja hylkäämistä varten.

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)

Hyväksyntäpäätöksen kysely

Ilmoituksen lähettämisen jälkeen agentti odottaa päätöstä. Käyttäkää kyselysilmukkaa, jossa on lyhyt odotusväli. Lopettakaa kysely, kun tila muuttuu arvosta 'pending' tai määräaika umpeutuu.

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)

Hyväksyntäpäätöksen tallentaminen

Kun hyväksyjä napsauttaa Hyväksy- tai Hylkää-painiketta, tallentakaa päätöksen tekijä ja ajankohta. Näin muodostuu täydellinen hyväksynnän audit trail: pyydetty → ilmoitettu → päätetty → suoritettu (tai peruttu).

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}

Aikakatkaisun käsittely: automaattinen peruutus

Jos päätöstä ei saada aikakatkaisuaikana, agentti peruuttaa toiminnon automaattisesti ja ilmoittaa asiasta alkuperäiselle pyytäjälle. Näin estetään toimintojen pysyvä estyminen siksi, ettei hyväksyjä vastaa.

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

Hyväksyntäportin koko työnkulun orkestrointi

Yhdistäkää kaikki vaiheet: tarkistakaa, tarvitaanko hyväksyntää, luokaa pyyntö, lähettäkää ilmoitus, odottakaa ja jatkakaa tai peruuttakaa toiminto päätöksen perusteella.

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}

Hyväksynnän audit trail

Jokainen hyväksynnän elinkaaren tapahtuma on tallennettava audit-lokiin: pyynnön luominen, ilmoituksen lähettäminen, päätöksen tekeminen sekä toiminnon suorittaminen tai peruuttaminen. Tämä trail on edellytys SOC 2 -vaatimustenmukaisuudelle.

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

Eskalointi, kun ensisijainen hyväksyjä ei ole tavoitettavissa

Jos ensisijainen hyväksyjä ei vastaa puolen aikakatkaisuajan kuluessa, pyyntö on eskaloitava toissijaiselle hyväksyjälle. Näin estetään kaikkien hyväksyntöjen juuttuminen yhden henkilön poissaolon vuoksi.

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

Mitä odottavalle hyväksyntäpyynnölle tapahtuu määräajan umpeutuessa?

Aikakatkaisun käsittely on keskeinen osa HITL-suunnittelua. Väärän oletustoiminnan valitsemisella voi olla erilaisia seurauksia turvallisuudelle ja käytettävyydelle.

Vastaa interaktiivisessa oppitunnissa

Human-in-the-Loop-hyväksyntäporttien kertaus

HITL-hyväksyntäportit toimivat seuraavasti: korkean riskin toimien luokittelu, keskeytetyn hyväksyntäpyynnön luominen, hyväksyjille ilmoittaminen Slackin tai sähköpostin kautta, päätösten kysely, automaattinen peruuttaminen aikakatkaisun yhteydessä sekä jokaisen elinkaaritapahtuman tallentaminen audit trailiin.

Eskalointi toissijaisille hyväksyjille estää hyväksyntöjen juuttumisen tavoittamattomissa olevien henkilöiden vuoksi.

Aloita maksutta

Opi Tekoälyagentit tekoälytuutorin avulla — ilmaiseksi

Kirjoita ja suorita oikeaa koodia selaimessa, saa välitöntä apua tekoälytuutorilta ympäri vuorokauden ja jatka siitä, mihin jäit, verkossa tai sovelluksessa.

Kurssit: 60
Oppitunnit: 239

Aloita opiskelu maksutta Näytä koko kurssi

Usein kysytyt kysymykset

Onko oppitunti ”Ihmisen hyväksyntää edellyttävät portit” ilmainen?

Kyllä – oppitunnin ”Ihmisen hyväksyntää edellyttävät portit” koko tekstin voi lukea täällä verkossa ilmaiseksi. Jos haluat harjoitella interaktiivisesti sisäänrakennetulla koodieditorilla ja ympäri vuorokauden käytettävissä olevan tekoälytuutorin avulla sekä avata koko Tekoälyagentit-kurssin, päivitä CoddyKit PROhon. Tekoälyagentit-kurssilla on yhteensä 4 oppituntia.

Mitä opin oppitunnilla ”Ihmisen hyväksyntää edellyttävät portit”?

Keskeytä–pyydä–hyväksy-mallit korkean riskin agenttitoimintoihin. Harjoittelet Tekoälyagentit-aihetta koodilla, jonka suoritat suoraan selaimessa. Ympäri vuorokauden käytettävissä oleva tekoälytuutori vastaa kysymyksiisi oppitunnin aikana.

Tarvitsenko kokemusta aloittaakseni Tekoälyagentit-opiskelun?

Aiempi kokemus ei ole tarpeen. CoddyKitin Tekoälyagentit-oppimispolku sopii vasta-alkajista edistyneisiin, joten voit aloittaa tästä tai alusta ja edetä omaan tahtiisi. Tämä on oppitunti 4/4.

Kuinka kauan ”Ihmisen hyväksyntää edellyttävät portit”-oppitunnin suorittaminen kestää?

Useimmat CoddyKitin oppitunnit kestävät noin 5–10 minuuttia. Jokainen oppitunti on lyhyt ja interaktiivinen, joten edistyt tasaisesti ja voit jatkaa siitä, mihin jäit – sekä verkossa että sovelluksessa.

Voinko kirjoittaa ja suorittaa koodia tällä Tekoälyagentit-oppitunnilla?

Kyllä. Jokainen Tekoälyagentit-oppitunti sisältää sisäänrakennetun koodieditorin, joten voit kirjoittaa ja suorittaa oikeaa koodia suoraan selaimessa ja saada välitöntä palautetta tekoälyltä – paikallista asennusta ei tarvita.

Kaikki tämän kurssin oppitunnit

Agenttien muuttumaton toimintojen kirjaus
Agentin toimintojen käytäntöjen valvonta
Sääntelyn noudattaminen: GDPR ja SOC2
Ihmisen hyväksyntää edellyttävät portit

← Takaisin: Tekoälyagentit