0Pricing
AI Agents · 课时

执行智能体操作策略

操作前策略检查、允许列表/拒绝列表和动态策略规则

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

什么是代理策略执行

策略执行是在每次代理操作之前运行的运行时关卡，用于决定该操作是否获准执行。如果没有它，代理唯一的约束就是 LLM 遵循指令的能力——而这种能力可能被绕过或误解。

执行机制必须位于 LLM<​​strong>之外的基础设施中。

操作前检查模式

执行任何工具前，请调用 can_agent_do(action, context)。此函数是唯一的执行关卡——所有执行操作的路径都必须经过它。

def can_agent_do(action: str, context: dict) -> tuple[bool, str]:
    '''
    Returns (allowed: bool, reason: str).
    Context includes: user_id, agent_id, session_id, parameters, timestamp.
    '''
    # 1. Check denylist first (fast path for obvious violations)
    if action in DENIED_ACTIONS:
        return False, f'Action "{action}" is on the global denylist'

    # 2. Check allowlist
    if action not in ALLOWED_ACTIONS:
        return False, f'Action "{action}" is not on the allowlist'

    # 3. Context-specific checks
    return check_context_policy(action, context)

定义允许列表和拒绝列表

允许列表列举代理获准执行的所有操作。未列入列表的任何操作默认都会被阻止。拒绝列表则为那些无论上下文如何都绝不应获准执行的操作增加额外的安全保障。

# Allowlist: tools the agent can use
ALLOWED_ACTIONS = {
    'web_search',
    'read_file',
    'write_file',
    'send_email',
    'create_calendar_event',
    'query_database',
    'execute_python_sandbox',
    'fetch_url',
    'create_ticket'
}

# Denylist: actions that are always blocked, regardless of context
DENIED_ACTIONS = {
    'delete_all_records',
    'export_entire_database',
    'send_mass_email',
    'modify_system_config',
    'create_admin_user',
    'disable_audit_logging'
}

if __name__ == '__main__':
    for action in ('web_search', 'send_mass_email'):
        print(f"{action}: allowed={action in ALLOWED_ACTIONS} denied={action in DENIED_ACTIONS}")

特定上下文的策略检查

除了简单的允许/拒绝列表之外，策略通常还取决于上下文：用户是谁、用户的角色是什么、当前时间是什么、目标资源是什么？这些都是特定上下文的检查。

from datetime import datetime, timezone

def check_context_policy(action: str, context: dict) -> tuple[bool, str]:
    user_id   = context.get('user_id', '')
    params    = context.get('parameters', {})
    user_role = context.get('user_role', 'user')

    # send_email: only agents with email_sender role
    if action == 'send_email':
        if user_role not in ('email_agent', 'admin'):
            return False, f'Role "{user_role}" cannot send emails'
        recipient = params.get('to', '')
        if not recipient.endswith('@trusted-domain.com'):
            return False, 'Email recipient must be in @trusted-domain.com'

    # write_file: path restrictions
    if action == 'write_file':
        path = params.get('path', '')
        if not path.startswith('/tmp/') and not path.startswith('/workspace/'):
            return False, f'File writes outside /tmp/ and /workspace/ are not allowed'

    return True, 'Permitted'

if __name__ == '__main__':
    ctx = {'user_id': 'u1', 'user_role': 'user',
           'parameters': {'to': 'someone@gmail.com'}}
    print('send_email as user:', check_context_policy('send_email', ctx))

    ctx2 = {'user_id': 'u1', 'user_role': 'user',
            'parameters': {'path': '/etc/passwd'}}
    print('write_file outside sandbox:', check_context_policy('write_file', ctx2))

通过策略引擎实现动态策略

硬编码的策略在生产环境中难以更新。请使用策略引擎（例如 OPA——开放策略代理）来评估以数据而非代码形式定义的策略。无需重新部署代理即可更新策略。

import requests

OPA_URL = 'http://localhost:8181/v1/data/agent/allow'

def opa_policy_check(action: str, context: dict) -> tuple[bool, str]:
    payload = {
        'input': {
            'action':   action,
            'user_id':  context.get('user_id'),
            'role':     context.get('user_role', 'user'),
            'params':   context.get('parameters', {}),
            'time_utc': datetime.now(timezone.utc).isoformat()
        }
    }
    try:
        resp = requests.post(OPA_URL, json=payload, timeout=0.5)
        result = resp.json().get('result', {})
        allowed = result.get('allow', False)
        reason  = result.get('reason', 'Policy decision')
        return allowed, reason
    except Exception as e:
        # Fail closed: deny if policy engine is unreachable
        return False, f'Policy engine unavailable: {e}'

失败即关闭与失败即开放

当策略引擎不可用时，您有两种选择：

  • 失败即关闭：拒绝所有操作。安全，但代理会停止工作。
  • 失败即开放：允许所有操作。代理会继续工作，但策略不会得到执行。

对于安全敏感型代理，请始终采用失败即关闭。对于执行低风险操作的生产力代理，失败即开放可能是可以接受的。

FAIL_CLOSED = True  # Configure per agent

def safe_policy_check(action: str, context: dict) -> tuple[bool, str]:
    try:
        return can_agent_do(action, context)
    except Exception as e:
        if FAIL_CLOSED:
            return False, f'Policy check failed (fail-closed): {e}'
        else:
            # Log the failure but allow the action
            import logging
            logging.warning('Policy check error (fail-open): %s', e)
            return True, 'Policy check bypassed due to error (fail-open)'

限制操作频率

策略执行可以包括频率限制：代理可能获准发送电子邮件，但每个会话最多只能发送 5 封。超过限制后，该操作将被拒绝。

from collections import defaultdict
import time

action_counts: dict[str, dict[str, int]] = defaultdict(lambda: defaultdict(int))
action_window_start: dict[str, float] = defaultdict(float)

ACTION_RATE_LIMITS = {
    'send_email':   {'limit': 5,  'window_secs': 3600},  # 5/hour
    'write_file':   {'limit': 50, 'window_secs': 300},
    'web_search':   {'limit': 20, 'window_secs': 60}
}

def check_rate_limit(action: str, session_id: str) -> tuple[bool, str]:
    limit_config = ACTION_RATE_LIMITS.get(action)
    if not limit_config:
        return True, 'No rate limit defined'

    window = limit_config['window_secs']
    now = time.time()
    key = f'{session_id}:{action}'

    if now - action_window_start[key] > window:
        action_counts[key] = defaultdict(int)
        action_window_start[key] = now

    action_counts[key]['count'] += 1
    if action_counts[key]['count'] > limit_config['limit']:
        return False, f'Rate limit exceeded: {action} ({limit_config["limit"]}/{window}s)'
    return True, 'Within rate limit'

if __name__ == '__main__':
    for i in range(6):
        allowed, reason = check_rate_limit('send_email', 'session-1')
    print(f'After 6 send_email calls: allowed={allowed}, reason={reason}')

记录策略违规

每次策略拒绝都必须记录完整上下文。当调试意外的代理行为或调查安全事件时，应首先查看这些日志。

import logging, json, time

policy_logger = logging.getLogger('agent.policy')

def enforced_action(agent_id: str, user_id: str, action: str,
                    context: dict, audit_log) -> tuple[bool, str]:
    allowed, reason = safe_policy_check(action, context)

    log_entry = {
        'ts':       time.time(),
        'agent_id': agent_id,
        'user_id':  user_id,
        'action':   action,
        'allowed':  allowed,
        'reason':   reason,
        'params':   context.get('parameters', {})
    }

    if allowed:
        policy_logger.info('ALLOWED %s', json.dumps(log_entry))
    else:
        policy_logger.warning('DENIED %s', json.dumps(log_entry))

    audit_log.append(
        agent_id, user_id,
        f'POLICY_{"ALLOW" if allowed else "DENY"}',
        {'action': action},
        {'allowed': allowed, 'reason': reason},
        context.get('session_id', '')
    )
    return allowed, reason

限定资源范围

即使某项操作已获准，也要限定代理可以访问的资源范围。处理用户 A 文档的代理不应能够读取用户 B 的文件，即使 read_file 已列入允许列表。

def check_resource_scope(action: str, context: dict) -> tuple[bool, str]:
    user_id  = context.get('user_id', '')
    params   = context.get('parameters', {})

    if action == 'read_file':
        path = params.get('path', '')
        # Each user's files must be under their own namespace
        if not path.startswith(f'/workspace/{user_id}/'):
            return False, (
                f'User {user_id} cannot read files outside '
                f'/workspace/{user_id}/'
            )

    if action == 'query_database':
        table = params.get('table', '')
        allowed_tables = {'products', 'public_docs', f'user_{user_id}_data'}
        if table not in allowed_tables:
            return False, f'Table "{table}" not in scope for user {user_id}'

    return True, 'Resource scope check passed'

if __name__ == '__main__':
    ctx = {'user_id': 'u1', 'parameters': {'path': '/workspace/u2/secret.txt'}}
    print('Cross-user file read:', check_resource_scope('read_file', ctx))

    ctx2 = {'user_id': 'u1', 'parameters': {'path': '/workspace/u1/notes.txt'}}
    print('Own file read:      ', check_resource_scope('read_file', ctx2))

测试策略规则

策略规则属于代码，因此必须经过测试。请为每条规则编写单元测试，确保拒绝和允许行为正常工作，并且边界情况不会造成策略被意外绕过。

def test_policy_rules():
    # Denylist blocks unconditionally
    ok, msg = can_agent_do('delete_all_records', {'user_role': 'admin'})
    assert not ok, 'Denylist should block even for admin'

    # Email requires trusted domain
    ok, msg = can_agent_do('send_email', {
        'user_role': 'email_agent',
        'parameters': {'to': 'attacker@evil.com'}
    })
    assert not ok, 'Should block untrusted email recipient'

    # File write outside allowed paths
    ok, msg = can_agent_do('write_file', {
        'user_role': 'user',
        'parameters': {'path': '/etc/crontab'}
    })
    assert not ok, 'Should block write to /etc/'

    print('All policy tests passed')

test_policy_rules()

缓存策略决策

每次操作都调用策略引擎会增加延迟，尤其是在使用外部 OPA 服务时。请为（操作、上下文哈希值）对缓存近期决策，并设置较短的 TTL，以减少往返请求。

import hashlib, time

policy_cache: dict[str, dict] = {}
POLICY_CACHE_TTL = 10  # seconds — short TTL so policy updates take effect quickly

def cached_policy_check(action: str, context: dict) -> tuple[bool, str]:
    ctx_hash = hashlib.md5(
        f'{action}:{context.get("user_id")}:{context.get("user_role")}'
        .encode()
    ).hexdigest()
    key = f'{action}:{ctx_hash}'
    entry = policy_cache.get(key)
    if entry and time.time() - entry['ts'] < POLICY_CACHE_TTL:
        return entry['result']
    result = can_agent_do(action, context)
    policy_cache[key] = {'result': result, 'ts': time.time()}
    return result

策略执行中的“失败即关闭”方式是什么

失败即关闭与失败即开放之间的选择，是任何策略执行系统都必须面对的基本安全权衡。了解各自适用的场景，是治理的核心概念。

策略执行回顾

代理策略执行使用：以操作前检查作为唯一关卡；使用允许列表 + 拒绝列表做出基础决策；使用特定上下文规则限定资源范围并检查角色；使用动态策略引擎（OPA）实现可更新的规则；按每个操作和每个会话设置频率限制；并为安全关键型代理默认采用失败即关闭。

常见问题解答

「执行智能体操作策略」课时是免费的吗？

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

「执行智能体操作策略」这节课中我会学到什么？

操作前策略检查、允许列表/拒绝列表和动态策略规则 你通过在浏览器中直接运行的动手代码来练习 AI Agents，全天候 AI 导师会在你学习这节课的过程中回答你的问题。

学习 AI Agents 需要有经验吗？

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

「执行智能体操作策略」课时需要多长时间？

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

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

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

此课程中的所有课时

  1. 智能体的不可变操作日志
  2. 执行智能体操作策略
  3. 法规合规：GDPR 与 SOC2
  4. 人在回路中的审批关卡
← 返回 AI Agents