附录 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 的工具调用默认应该是受限的。能力需要显式授予,不是默认拥有。 先从最小权限开始,遇到需求再逐步放开——别反过来。