AI Agents · 课时

人在回路中的审批关卡

Q: 「人在回路中的审批关卡」这节课中我会学到什么？

适用于高风险智能体操作的暂停—请求—批准模式 你通过在浏览器中直接运行的动手代码来练习 AI Agents，全天候 AI 导师会在你学习这节课的过程中回答你的问题。

适用于高风险智能体操作的暂停—请求—批准模式

人在回路中的审批关卡 是 CoddyKit 上的免费 AI Agents 课时。这是第 4 节课，共 4 节。你可以在下方免费阅读本课时的完整内容 — 然后在浏览器中使用内置代码编辑器和全天候 AI 导师进行实践。这是 AI Agents 学习路径的一部分，你的进度在网页和 CoddyKit 应用中同步。 AI Agents 课程共包含 4 节课。

代理何时需要人工审批

对于发送合同、删除生产数据、使用预算或发布公告等高风险操作，代理不应自主行动，而应在继续操作前暂停并请求人工审批。

这就是人机协作（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'])

通过即时通讯工具发送通知

通过即时通讯工具将审批请求发送给审批人。请包含代理希望执行的操作摘要、用于批准或拒绝的链接，以及超时截止时间。

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}

审批审计轨迹

每个审批生命周期事件都必须记录在审计日志中：请求已创建、通知已发送、决定已作出、操作已执行或已取消。这条轨迹是 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'})

主要审批人不可用时的升级处理

如果主要审批人在超时期限过半之前仍未响应，请将请求升级给次要审批人。这样可以避免所有审批因某个人不在岗而被阻塞。

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 设计的关键部分。选择错误的默认处理方式，会对安全性和易用性产生不同影响。

人机协作审批闸门回顾

HITL 审批闸门的工作方式包括：对高风险操作进行分类、创建暂停执行的审批请求、通过即时通讯工具或电子邮件通知审批人、轮询决定、超时后自动取消，以及在审计轨迹中记录每个生命周期事件。

将请求升级给次要审批人，可以防止因审批人不可用而造成审批死锁。

常见问题解答

「人在回路中的审批关卡」课时是免费的吗？

是的 — 「人在回路中的审批关卡」的完整文本可在网页上免费阅读。要进行交互式练习（内置代码编辑器和全天候 AI 导师）并解锁 AI Agents 课程的其余内容，请升级到 CoddyKit PRO。 AI Agents 课程共包含 4 节课。

「人在回路中的审批关卡」这节课中我会学到什么？

适用于高风险智能体操作的暂停—请求—批准模式你通过在浏览器中直接运行的动手代码来练习 AI Agents，全天候 AI 导师会在你学习这节课的过程中回答你的问题。

学习 AI Agents 需要有经验吗？

无需任何先前经验。CoddyKit 上的 AI Agents 课程适合初学者到高级学习者，你可以从这里开始或从头开始，按照自己的节奏学习。这是第 4 节课，共 4 节。

「人在回路中的审批关卡」课时需要多长时间？

大多数 CoddyKit 课程大约需要 5–10 分钟。每节课都很精短且互动，所以你能稳步进步，并在网页和应用中从离开的地方继续。

我能在这节 AI Agents 课中编写并运行代码吗？

能。每节 AI Agents 课都包含内置代码编辑器，你可以在浏览器中直接编写并运行真实代码，并获得即时 AI 反馈 — 无需本地设置。

此课程中的所有课时

← 返回 AI Agents