跳转至

附录 2B:工具调用的权限控制

Agent 能调工具,这是能力。Agent 能调任何工具,这是风险。

一个旅游助手如果只能查天气、搜景点,最多返回错误数据。但如果它还能发邮件、下订单、删文件——模型的一次误判可能造成真实损失:订了一张不能退的机票,删了用户的文件,给陌生人发了邮件。

这一页讨论怎么给工具调用加权限控制。核心思路:不是所有工具都应该被随便调,调用前应该根据工具的危险等级决定是直接执行、还是先确认再执行、还是拒绝执行


工具分级

按副作用和可逆性,工具可以分三级:

级别 描述 例子 默认策略
read 只读,无副作用 查天气、搜景点、读文件、查数据库 直接执行
write 有副作用但可逆 写文件、发消息(可撤回)、创建草稿 执行并记录
dangerous 有副作用且不可逆/高成本 下订单、付款、删文件、发正式邮件、执行 SQL DELETE 人工确认后才执行

在代码里怎么实现:

from enum import Enum
from dataclasses import dataclass
from typing import Callable


class ToolLevel(Enum):
    READ = "read"
    WRITE = "write"
    DANGEROUS = "dangerous"


@dataclass
class ToolDef:
    name: str
    description: str
    handler: Callable
    level: ToolLevel
    parameters: dict  # JSON Schema

每个工具注册时声明自己的级别。查天气是 READ,写文件是 WRITE,下订单是 DANGEROUS

tools = [
    ToolDef(
        name="get_weather",
        description="查询天气",
        handler=get_weather,
        level=ToolLevel.READ,
        parameters={...}
    ),
    ToolDef(
        name="book_hotel",
        description="预订酒店",
        handler=book_hotel,
        level=ToolLevel.DANGEROUS,
        parameters={...}
    ),
]

执行前拦截

在 Agent 循环里,每次模型返回 tool_calls,不是直接执行,而是先过一道权限检查:

def execute_with_permission(tool_def: ToolDef, args: dict) -> dict:
    """根据工具级别决定是否执行。"""

    # READ 级别:直接跑
    if tool_def.level == ToolLevel.READ:
        return tool_def.handler(**args)

    # WRITE 级别:跑,但记日志
    if tool_def.level == ToolLevel.WRITE:
        print(f"[LOG] 执行写操作: {tool_def.name}({args})")
        result = tool_def.handler(**args)
        log_tool_execution(tool_def.name, args, result)  # 写入审计日志
        return result

    # DANGEROUS 级别:暂停,等人确认
    if tool_def.level == ToolLevel.DANGEROUS:
        print(f"\n[WARNING] Agent 想执行高危操作:")
        print(f"   工具: {tool_def.name}")
        print(f"   参数: {args}")
        confirm = input("   是否允许?(y/n): ").strip().lower()
        if confirm == "y":
            result = tool_def.handler(**args)
            log_tool_execution(tool_def.name, args, result, approved_by="human")
            return result
        else:
            return {"error": "操作被用户拒绝", "tool": tool_def.name}

这段代码的逻辑很直白:READ 直接执行,WRITE 执行加日志,DANGEROUS 先问人。被拒绝的操作返回一个 error,Agent 循环会把这个 error 当作工具返回值喂回模型,模型会看到"操作被拒绝"并调整策略。


Human-in-the-Loop(HITL)触发条件

不是所有 DANGEROUS 操作都需要人工确认。更细粒度的 HITL 触发条件可以按参数判断:

def needs_human_approval(tool_name: str, args: dict) -> bool:
    """判断是否需要人工确认。"""

    # 金额超过阈值
    if tool_name == "book_hotel":
        price = args.get("price", 0)
        if price > 500:
            return True

    # 影响范围大
    if tool_name == "send_email":
        recipients = args.get("recipients", [])
        if len(recipients) > 5:  # 群发
            return True

    # 不可逆操作一律确认
    if tool_name in ("delete_file", "cancel_booking", "execute_sql"):
        return True

    return False

这样,订 200 元的快捷酒店可以直接过,订 2000 元的度假酒店会弹确认。发一封邮件直接过,群发 50 人会弹确认。

在 Agent 循环里这样用:

for tc in assistant_msg.tool_calls:
    func_name = tc.function.name
    func_args = json.loads(tc.function.arguments)
    tool_def = tool_registry[func_name]

    if tool_def.level == ToolLevel.DANGEROUS or needs_human_approval(func_name, func_args):
        # 进入 HITL 流程
        result = request_human_approval_and_execute(tool_def, func_args)
    else:
        result = tool_def.handler(**func_args)

Policy:工具允许列表

更前置的控制是 Policy——在模型请求发出之前,就决定哪些工具可用。

@dataclass
class ToolPolicy:
    """工具策略:按任务上下文决定哪些工具可用。"""
    allowed_tools: set[str]          # 允许调用的工具名
    blocked_tools: set[str]          # 禁止调用的工具名
    max_calls_per_tool: dict[str, int]  # 每个工具的最大调用次数
    require_approval: set[str]       # 需要人工确认的工具名


# 针对旅游咨询场景的策略
travel_consult_policy = ToolPolicy(
    allowed_tools={"get_weather", "search_places", "estimate_route"},
    blocked_tools={"delete_file", "execute_sql"},
    max_calls_per_tool={"get_weather": 3, "search_places": 5, "estimate_route": 3},
    require_approval={"book_hotel", "book_flight"},
)

这个策略说的是:旅游咨询阶段,Agent 只能查天气、搜景点、估路线,每个工具有调用次数上限。订酒店和机票需要人工确认。删文件和执行 SQL 直接禁止——不管模型怎么请求。

执行时:

def apply_policy(policy: ToolPolicy, tool_name: str, call_counts: dict) -> str | None:
    """检查工具调用是否符合策略。返回 None 表示通过,返回字符串表示拒绝原因。"""

    if tool_name in policy.blocked_tools:
        return f"工具 {tool_name} 被策略禁止"

    if policy.allowed_tools and tool_name not in policy.allowed_tools:
        return f"工具 {tool_name} 不在允许列表中"

    max_calls = policy.max_calls_per_tool.get(tool_name, float("inf"))
    current_calls = call_counts.get(tool_name, 0)
    if current_calls >= max_calls:
        return f"工具 {tool_name} 已达调用上限 ({max_calls} 次)"

    return None  # 通过

Policy 检查发生在工具执行之前,甚至可以发生在 tools 参数传给 API 之前——你可以根据 Policy 动态过滤传给模型的工具列表,让模型根本看不到不该用的工具。


Guardrails:运行时拦截

Policy 管的是"哪些工具能调"。Guardrails 管的是"调用的内容是否安全"。

def check_guardrails(tool_name: str, args: dict) -> str | None:
    """检查工具参数是否触犯 guardrail。返回 None 表示通过。"""

    # 防止路径穿越
    if tool_name in ("read_file", "write_file"):
        path = args.get("path", "")
        if ".." in path or path.startswith("/etc") or path.startswith("/root"):
            return f"路径 {path} 被 guardrail 拦截:禁止访问系统目录"

    # 防止 SQL 注入
    if tool_name == "query_database":
        sql = args.get("sql", "")
        dangerous_keywords = ["DROP", "DELETE", "TRUNCATE", "ALTER"]
        for kw in dangerous_keywords:
            if kw in sql.upper():
                return f"SQL 语句包含危险关键词: {kw}"

    # 防止发送敏感信息
    if tool_name == "send_email":
        body = args.get("body", "")
        sensitive_patterns = ["api_key", "secret"]
        if any(pattern in body.lower() for pattern in sensitive_patterns):
            return "邮件内容可能包含敏感信息,已拦截"

    return None

Guardrails 和 Policy 的区别:Policy 是静态规则("这个工具能不能调"),Guardrails 是动态检查("这次调用的参数有没有问题")。两者配合使用。


完整的权限检查流程

把工具分级、Policy、Guardrails、HITL 串在一起:

模型返回 tool_calls
  |
  v
Policy 检查 ---- 不在允许列表? ---- 拒绝,返回错误给模型
  |
  v 通过
Guardrails 检查 -- 参数危险? ---- 拒绝,返回错误给模型
  |
  v 通过
工具分级判断
  |
  |-- READ ---- 直接执行
  |
  |-- WRITE ---- 执行 + 审计日志
  |
  +-- DANGEROUS ---- HITL 确认
                      |
                      |-- 用户批准 ---- 执行 + 审计日志
                      |
                      +-- 用户拒绝 ---- 返回"被拒绝"给模型

在代码里是几行:

def safe_execute_tool(tool_def, args, policy, call_counts):
    # 1. Policy
    rejection = apply_policy(policy, tool_def.name, call_counts)
    if rejection:
        return {"error": rejection}

    # 2. Guardrails
    rejection = check_guardrails(tool_def.name, args)
    if rejection:
        return {"error": rejection}

    # 3. 分级执行
    return execute_with_permission(tool_def, args)

三层检查,最后才到真正的函数调用。每一层返回的错误都会被当作工具结果喂回模型——模型看到"操作被拒绝"后,通常会换一种方式完成任务或告知用户。


和第七章的关系

这页讲的是权限控制的基本机制。第七章(安全与评测)会展开讨论:

  • Policy 的动态管理:不同用户、不同会话可以有不同的工具策略。
  • Guardrails 的分类:输入 guardrail(拦截恶意 prompt)、输出 guardrail(拦截敏感输出)、工具 guardrail(拦截危险操作)。
  • HITL 的工程实现:异步审批、超时处理、审批上下文。
  • 审计和追溯:谁在什么时候批准了什么操作。

这里只需要记住一条原则:Agent 的工具调用默认应该是受限的。能力需要显式授予,不是默认拥有。 先从最小权限开始,遇到需求再逐步放开——别反过来。