想象这样一个场景：你的Agent有一个execute_sql工具，可以查询数据库。某一天，用户问了一个看似无害的问题："帮我看看最近一周的订单总数。"
Agent决定调用execute_sql工具，生成了以下SQL：
SELECT COUNT(*) FROM orders WHERE created_at > '2026-08-28'
这个查询没问题。但同一天，另一个用户问："帮我把所有超过30天未支付的订单取消掉。"
Agent理解了用户的意图，生成了：
DELETE FROM orders WHERE status = 'pending' AND created_at < '2026-07-28'
如果这条SQL真的被执行了，会发生什么？ 所有符合条件的未支付订单被删除，而这些订单本应进入催款流程而非直接删除。
问题不在于Agent"理解了"用户的意图——用户确实想取消这些订单。问题在于：Agent的判断可能出错，而错误的后果不可逆。
这就是安全沙箱存在的意义：在工具执行和外部世界之间建立一道可控的屏障。 Agent的决策可能有误，但沙箱确保错误的影响范围是有限的、可恢复的、经过授权的。
工具的安全沙箱不是一个单一的机制，而是多层防御的组合。每一层解决不同类型的安全问题：
┌─────────────────────────────────────────────────┐
│ 第1层：权限声明（Permission Declaration） │
│ · 工具声明自己的危险级别 │
│ · 决定执行前是否需要确认 │
├─────────────────────────────────────────────────┤
│ 第2层：参数校验（Parameter Validation） │
│ · 在执行前验证参数是否合法 │
│ · 拒绝明显危险或异常的输入 │
├─────────────────────────────────────────────────┤
│ 第3层：资源限制（Resource Limitation） │
│ · 限制执行时间、内存、网络访问 │
│ · 防止工具失控或消耗过多资源 │
├─────────────────────────────────────────────────┤
│ 第4层：审计记录（Audit Logging） │
│ · 记录每次工具调用的完整信息 │
│ · 支持事后追溯和问题定位 │
└─────────────────────────────────────────────────┘
在4.1节中，你已经为工具定义了category和permission。现在把它们转化为实际的安全策略。
from enum import Enum
from dataclasses import dataclass
from typing import Any, Callable, Dict, Optional, List
class ToolCategory(Enum):
READ = "read"
WRITE = "write"
EXECUTE = "execute"
COMMUNICATE = "communicate"
DANGEROUS = "dangerous"
class ToolPermission(Enum):
AUTO = "auto"
USER_CONFIRM = "user_confirm"
ADMIN_ONLY = "admin_only"
class PermissionDeniedError(Exception):
"""权限被拒绝"""
def __init__(self, tool_name: str, reason: str,
required_permission: ToolPermission,
current_permission: ToolPermission):
self.tool_name = tool_name
self.reason = reason
self.required_permission = required_permission
self.current_permission = current_permission
super().__init__(
f"工具 '{tool_name}' 执行被拒绝。"
f"原因：{reason}。"
f"需要权限：{required_permission.value}，"
f"当前权限：{current_permission.value}"
)
权限确认的触发条件
并非所有工具都需要在执行前询问用户。确认的触发条件应该是基于风险而非固定规则的：
class PermissionManager:
"""管理工具执行的权限"""
def __init__(self):
# 需要确认的权限级别
self.confirmation_required = {
ToolPermission.USER_CONFIRM,
ToolPermission.ADMIN_ONLY,
}
# 需要确认的分类
self.confirmation_categories = {
ToolCategory.WRITE,
ToolCategory.DANGEROUS,
ToolCategory.COMMUNICATE,
}
# 需要确认的特定操作模式
self.dangerous_patterns = [
"delete", "drop", "truncate", "remove", "purge",
"execute", "shell", "command", "script",
"send", "publish", "broadcast",
"transfer", "withdraw", "payment",
]
def should_require_confirmation(self, tool_def, arguments: Dict[str, Any]) -> bool:
"""判断当前工具调用是否需要用户确认"""
# 规则1：权限级别要求确认
if tool_def.permission in self.confirmation_required:
return True
# 规则2：分类要求确认
if tool_def.category in self.confirmation_categories:
return True
# 规则3：参数中出现危险模式
args_str = str(arguments).lower()
for pattern in self.dangerous_patterns:
if pattern in args_str:
return True
# 规则4：工具自身标记需要确认
if tool_def.metadata.get("always_confirm", False):
return True
return False
def check_permission(self, tool_def, arguments: Dict[str, Any],
user_confirmed: bool = False,
user_role: str = "user") -> Optional[PermissionDeniedError]:
"""检查权限，返回错误（如果拒绝）"""
# ADMIN_ONLY权限检查
if tool_def.permission == ToolPermission.ADMIN_ONLY:
if user_role != "admin":
return PermissionDeniedError(
tool_name=tool_def.name,
reason="此工具仅管理员可用",
required_permission=ToolPermission.ADMIN_ONLY,
current_permission=ToolPermission.USER_CONFIRM,
)
# 需要确认但用户未确认
if self.should_require_confirmation(tool_def, arguments):
if not user_confirmed:
return PermissionDeniedError(
tool_name=tool_def.name,
reason=f"工具 '{tool_def.name}' 属于{tool_def.category.value}类操作，"
f"执行前需要用户确认",
required_permission=ToolPermission.USER_CONFIRM,
current_permission=ToolPermission.AUTO if not user_confirmed else ToolPermission.USER_CONFIRM,
)
return None
这个PermissionManager的设计有几个关键点：
第一：确认的触发是基于多条件的。 不仅仅看权限级别，还看工具分类和参数内容。一个READ类的工具，如果参数中出现了"delete"字样，也会触发确认——因为这可能是Prompt注入的尝试。
第二：拒绝和确认是分开的。 ADMIN_ONLY权限的拒绝是硬性的（用户无法通过确认来绕过），而USER_CONFIRM的确认是软性的（用户确认后可以执行）。这两种机制解决不同的问题：前者解决"谁有权执行"，后者解决"是否确认要执行"。
第三：参数内容也参与安全判断。 这很重要——LLM可能被注入攻击，在看似无害的工具调用中填充危险参数。检查参数内容是第二道防线。
权限确认在工具执行之前发生，但确认只能防止"已经意识到的风险"。还有一些风险是不可预见的——用户的输入可能包含异常字符、异常格式、或者意图不明确但后果严重的内容。
参数校验的目标是：在参数被传入工具函数之前，验证它们是否合理。
class ParameterValidator:
"""参数校验器"""
def __init__(self):
self.validators = {}
def register_validator(self, tool_name: str, param_name: str, validator_func: Callable):
"""为一个工具的参数注册校验器"""
key = f"{tool_name}.{param_name}"
self.validators[key] = validator_func
def validate(self, tool_name: str, arguments: Dict[str, Any]) -> List[str]:
"""验证参数，返回错误列表（空列表表示通过）"""
errors = []
# 通用校验
for key, value in arguments.items():
# 检查是否为空
if value is None or value == "":
errors.append(f"参数 '{key}' 不能为空")
continue
# 检查字符串长度
if isinstance(value, str) and len(value) > 10000:
errors.append(f"参数 '{key}' 长度（{len(value)}）超过最大允许长度（10000）")
# 检查嵌套结构（防止深度嵌套攻击）
if isinstance(value, (dict, list)):
depth = self._get_depth(value)
if depth > 5:
errors.append(f"参数 '{key}' 嵌套深度（{depth}）超过限制（5）")
# 工具特定的校验
for key, validator in self.validators.items():
validator_tool, validator_param = key.split(".", 1)
if validator_tool == tool_name and validator_param in arguments:
try:
result = validator(arguments[validator_param])
if isinstance(result, str):
errors.append(result)
elif result is False:
errors.append(f"参数 '{validator_param}' 校验失败")
except Exception as e:
errors.append(f"参数 '{validator_param}' 校验异常：{e}")
return errors
def _get_depth(self, obj) -> int:
"""递归计算嵌套深度"""
if not isinstance(obj, (dict, list)):
return 0
if isinstance(obj, dict):
if not obj:
return 1
return 1 + max(self._get_depth(v) for v in obj.values())
if isinstance(obj, list):
if not obj:
return 1
return 1 + max(self._get_depth(item) for item in obj)
return 0
使用示例——为SQL执行工具注册参数校验：
import re
validator = ParameterValidator()
def validate_sql_param(sql: str):
"""校验SQL参数"""
# 拒绝多语句（防止SQL注入）
if ";" in sql and not sql.strip().endswith(";"):
return "SQL包含多条语句，仅允许单条语句执行"
# 拒绝危险的SQL模式
dangerous_patterns = [
(r"DROP\s+TABLE", "不允许DROP TABLE操作"),
(r"DROP\s+DATABASE", "不允许DROP DATABASE操作"),
(r"TRUNCATE", "不允许TRUNCATE操作"),
(r"ALTER\s+TABLE", "不允许ALTER TABLE操作"),
(r"GRANT\s+", "不允许GRANT操作"),
(r"REVOKE\s+", "不允许REVOKE操作"),
(r"INTO\s+OUTFILE", "不允许导出文件操作"),
(r"LOAD_FILE", "不允许加载文件操作"),
]
sql_upper = sql.upper()
for pattern, error_msg in dangerous_patterns:
if re.search(pattern, sql_upper):
return error_msg
# 必须包含LIMIT（防止全表扫描）
if "SELECT" in sql_upper and "LIMIT" not in sql_upper:
return "SELECT查询必须包含LIMIT子句"
return None # 校验通过
validator.register_validator("execute_sql", "query", validate_sql_param)
这个SQL校验器展示了一个重要的设计原则：白名单比黑名单更安全。 与其列举所有不允许的操作（黑名单），不如定义允许的操作模式（白名单）。上面的校验器实际上在向白名单方向靠拢——它要求SELECT必须带LIMIT，拒绝所有结构性变更操作。
即使参数合法，工具执行仍然可能失控：网络请求挂起、数据库查询消耗大量内存、无限循环。资源限制确保工具的执行在物理上也是受控的。
import signal
import resource
import threading
import time
from contextlib import contextmanager
from typing import Any, Dict, Optional
class ResourceLimiter:
"""资源限制器"""
def __init__(self,
max_execution_time: int = 10,
max_memory_mb: int = 512,
max_output_size: int = 10000):
self.max_execution_time = max_execution_time
self.max_memory_mb = max_memory_mb
self.max_output_size = max_output_size
@contextmanager
def time_limit(self, seconds: int):
"""时间限制上下文管理器"""
def handler(signum, frame):
raise TimeoutError(f"执行超过{seconds}秒")
old_handler = signal.signal(signal.SIGALRM, handler)
signal.alarm(seconds)
try:
yield
finally:
signal.alarm(0)
signal.signal(signal.SIGALRM, old_handler)
@contextmanager
def memory_limit(self, mb: int):
"""内存限制上下文管理器（仅Unix）"""
try:
soft, hard = resource.getrlimit(resource.RLIMIT_AS)
limit = mb * 1024 * 1024
resource.setrlimit(resource.RLIMIT_AS, (limit, hard))
yield
except (ValueError, resource.error) as e:
# 系统不支持内存限制时，仅记录而不失败
print(f"警告：无法设置内存限制：{e}")
yield
finally:
try:
resource.setrlimit(resource.RLIMIT_AS, (soft, hard))
except:
pass
def limit_output(self, result: str) -> str:
"""限制输出大小"""
if len(result) > self.max_output_size:
truncated = result[:self.max_output_size]
return truncated + f"\n...[输出已截断：原{len(result)}字符，显示前{self.max_output_size}字符]"
return result
将资源限制集成到工具执行中：
class ToolRegistry:
def __init__(self,
default_timeout: int = 10,
default_max_memory: int = 512,
default_max_output: int = 10000):
self._tools = {}
self.default_timeout = default_timeout
self.default_max_memory = default_max_memory
self.default_max_output = default_max_output
self.limiter = ResourceLimiter(
max_execution_time=default_timeout,
max_memory_mb=default_max_memory,
max_output_size=default_max_output,
)
self.permission_manager = PermissionManager()
self.validator = ParameterValidator()
def execute(self, name: str, arguments: Dict[str, Any],
user_confirmed: bool = False,
user_role: str = "user") -> ToolResult:
"""执行工具，带完整的安全检查"""
import time
start = time.time()
if name not in self._tools:
return ToolResult(
success=False, content="",
error=f"未知工具：{name}",
tool_name=name
)
tool_def = self._tools[name]
# 第1层：权限检查
permission_error = self.permission_manager.check_permission(
tool_def, arguments, user_confirmed, user_role
)
if permission_error:
return ToolResult(
success=False, content="",
error=str(permission_error),
tool_name=name
)
# 第2层：参数校验
validation_errors = self.validator.validate(name, arguments)
if validation_errors:
return ToolResult(
success=False, content="",
error=f"参数校验失败：{'; '.join(validation_errors)}",
tool_name=name
)
# 第3层：资源限制下的执行
timeout = tool_def.timeout_seconds or self.default_timeout
try:
with self.limiter.time_limit(timeout):
result = tool_def.func(**arguments)
# 限制输出大小
result_str = str(result)
result_str = self.limiter.limit_output(result_str)
duration = int((time.time() - start) * 1000)
return ToolResult(
success=True,
content=result_str,
tool_name=name,
duration_ms=duration
)
except TimeoutError as e:
duration = int((time.time() - start) * 1000)
return ToolResult(
success=False, content="",
error=f"工具执行超时：{e}",
tool_name=name,
duration_ms=duration
)
except Exception as e:
duration = int((time.time() - start) * 1000)
return ToolResult(
success=False, content="",
error=f"工具执行异常：{type(e).__name__}: {e}",
tool_name=name,
duration_ms=duration
)
即使有前面三层防线，工具执行仍然可能出错。当错误发生时，你需要知道发生了什么、谁做的、什么时候、参数是什么。这就是审计记录的价值。
import json
import uuid
from dataclasses import dataclass, field, asdict
from datetime import datetime
@dataclass
class AuditRecord:
"""一次工具调用的审计记录"""
audit_id: str
timestamp: str
tool_name: str
category: str
permission_required: str
user_confirmed: bool
user_role: str
arguments: Dict[str, Any]
success: bool
error: Optional[str] = None
duration_ms: int = 0
result_preview: str = ""
trace_id: Optional[str] = None
def to_dict(self) -> Dict[str, Any]:
return asdict(self)
def to_json(self) -> str:
return json.dumps(self.to_dict(), ensure_ascii=False, indent=2)
class AuditLogger:
"""审计日志记录器"""
def __init__(self, log_file: Optional[str] = None):
self.log_file = log_file
self.records: List[AuditRecord] = []
def record(self, tool_name: str, category: str,
permission_required: str, user_confirmed: bool,
user_role: str, arguments: Dict[str, Any],
success: bool, error: Optional[str],
duration_ms: int, result_preview: str,
trace_id: Optional[str] = None) -> AuditRecord:
"""记录一次工具调用"""
record = AuditRecord(
audit_id=str(uuid.uuid4())[:12],
timestamp=datetime.now().isoformat(),
tool_name=tool_name,
category=category,
permission_required=permission_required,
user_confirmed=user_confirmed,
user_role=user_role,
arguments=arguments,
success=success,
error=error,
duration_ms=duration_ms,
result_preview=result_preview[:200],
trace_id=trace_id,
)
self.records.append(record)
# 写入文件（如果有配置）
if self.log_file:
with open(self.log_file, 'a', encoding='utf-8') as f:
f.write(record.to_json() + "\n")
return record
def get_records_by_tool(self, tool_name: str) -> List[AuditRecord]:
"""获取特定工具的所有审计记录"""
return [r for r in self.records if r.tool_name == tool_name]
def get_denied_records(self) -> List[AuditRecord]:
"""获取所有被拒绝的记录"""
return [r for r in self.records if not r.success]
def get_records_by_time_range(self, start: datetime, end: datetime) -> List[AuditRecord]:
"""获取时间范围内的记录"""
return [
r for r in self.records
if start <= datetime.fromisoformat(r.timestamp) <= end
]
审计记录与普通日志的区别：
普通日志：记录"系统发生了什么"，用于调试
审计记录：记录"谁做了什么操作"，用于合规和追溯
审计记录的关键特征：
不可变：一旦记录，不应被修改
完整：包含判断"这个操作是否合理"所需的全部信息
可查询：支持按工具、时间、结果等维度检索
持久化：写入文件或数据库，不能只存在内存中
现在将这些安全机制集成到Runtime的执行流程中：
class AgentRuntime:
def __init__(self, config, tool_registry, memory,
output_schema=None, tracer=None, logger=None,
user_role: str = "user",
confirm_callback: Optional[Callable] = None):
# ... 之前的初始化 ...
self.user_role = user_role
self.confirm_callback = confirm_callback # 用户确认回调
def _execute_with_safety(self, call: ToolCallRequest):
"""带安全沙箱的工具执行"""
tool_def = self.tools._tools.get(call.name)
if not tool_def:
return ToolResult(
success=False, content="",
error=f"未知工具：{call.name}",
tool_name=call.name
)
# 检查是否需要确认
user_confirmed = False
if self.tools.permission_manager.should_require_confirmation(tool_def, call.arguments):
if self.confirm_callback:
# 调用确认回调，让用户确认
user_confirmed = self.confirm_callback(
tool_name=call.name,
arguments=call.arguments,
category=tool_def.category.value,
)
if not user_confirmed:
return ToolResult(
success=False, content="",
error=f"用户取消了工具 '{call.name}' 的执行",
tool_name=call.name
)
else:
# 没有确认回调，直接拒绝
return ToolResult(
success=False, content="",
error=f"工具 '{call.name}' 需要用户确认，但当前没有确认机制",
tool_name=call.name
)
# 执行（带完整安全检查）
return self.tools.execute(
name=call.name,
arguments=call.arguments,
user_confirmed=user_confirmed,
user_role=self.user_role,
)
确认回调的设计：在实际应用中，用户确认是一个异步交互过程。Agent暂停执行，等待用户响应。在简单的CLI场景中，可以用input()实现：
def cli_confirm_callback(tool_name: str, arguments: Dict[str, Any],
category: str) -> bool:
"""CLI场景下的确认回调"""
print(f"\n⚠️ Agent请求执行{tool_name}工具")
print(f" 分类：{category}")
print(f" 参数：{json.dumps(arguments, ensure_ascii=False, indent=2)}")
response = input(" 是否允许执行？(y/n): ").strip().lower()
return response in ('y', 'yes', '是')
将以上各层整合，看一个完整的流程：
用户："帮我取消订单ORD12345678"
Agent循环：
1. LLM判断：需要调用cancel_order工具
2. Runtime准备执行cancel_order(order_id="ORD12345678")
3. 安全沙箱检查开始：
第1层-权限检查：
- cancel_order属于WRITE类
- 权限级别为USER_CONFIRM
- should_require_confirmation返回True
- 触发用户确认流程
用户看到：
"⚠️ Agent请求执行cancel_order工具
分类：write
参数：{"order_id": "ORD12345678"}
是否允许执行？(y/n)"
用户输入：n（发现订单号不对）
结果：工具执行被拒绝
4. 错误信息返回给LLM："用户取消了工具 'cancel_order' 的执行"
5. LLM看到错误后重新决策：
- 可能询问用户正确的订单号
- 可能告知用户取消操作已被阻止
这个流程展示了安全沙箱的核心价值：它在Agent的自主性和系统的安全性之间建立了一个可控的平衡点。Agent可以自由地提出执行请求，但实际执行需要经过安全检查。
安全沙箱是Agent与外部世界之间的受控边界。本节实现了四层防御：
核心设计原则：
在下一节中，我们将讨论工具错误处理与重试策略——当工具执行失败时，如何优雅地恢复。
评论专区
评论加载中...登录后即可发表评论