传统服务：请求 → 固定逻辑 → 响应 Agent：请求 → 推理 → 可能调用工具A/B/C → 可能再推理 → 再调工具 → ... → 响应

┌─────────────────────────────────────────────┐ │ L5: 人工审核层 — 高风险操作需Human-in-Loop │ ├─────────────────────────────────────────────┤ │ L4: 审计追踪层 — 全链路行为日志 + 告警 │ ├─────────────────────────────────────────────┤ │ L3: 沙箱隔离层 — 工具执行环境隔离 │ ├─────────────────────────────────────────────┤ │ L2: 权限控制层 — 最小权限 + 临时凭证 │ ├─────────────────────────────────────────────┤ │ L1: 输入校验层 — Prompt注入防护 + 输出过滤 │ └─────────────────────────────────────────────┘

import re from typing import Optional

class InputGuard:
"""Agent输入安全校验"""

# 危险模式检测 DANGEROUS_PATTERNS = [ r"ignore\s+(previous|all|above)\s+(instructions|prompts)", r"you\s+are\s+now\s+(a|an)\s+", r"forget\s+(everything|your)\s+", r"system\s*prompt", r"<\|.*?\|>", # 特殊token标记 r"```\s*system", ] # 最大输入长度 MAX_INPUT_LENGTH = 4096 @classmethod def validate(cls, user_input: str) -> tuple[bool, Optional[str]]: """校验用户输入，返回 (是否安全, 风险描述)""" # 1. 长度限制 if len(user_input) > cls.MAX_INPUT_LENGTH: return False, f"输入超长：{len(user_input)} > {cls.MAX_INPUT_LENGTH}" # 2. 注入检测 for pattern in cls.DANGEROUS_PATTERNS: if re.search(pattern, user_input, re.IGNORECASE): return False, f"检测到注入模式：{pattern}" # 3. 特殊字符清理（保留业务需要的合法字符） cleaned = cls._sanitize(user_input) return True, None @classmethod def _sanitize(cls, text: str) -> str: """清理危险字符，保留合法内容""" # 移除控制字符 text = re.sub(r'[\x00-\x08\x0b\x0c\x0e-\x1f]', '', text) # 转义Markdown/HTML注入 text = text.replace('<script', '<script') return text

输出过滤同样重要——防止Agent泄露敏感信息：

class OutputGuard: """Agent输出安全过滤"""
# 敏感信息模式 SENSITIVE_PATTERNS = [ (r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b', '[EMAIL_REDACTED]'), (r'\b\d{3}-\d{2}-\d{4}\b', '[SSN_REDACTED]'), # 身份证号等 (r'\b(?:sk-|pk-)[a-zA-Z0-9]{20,}\b', '[KEY_REDACTED]'), # API Key (r'\b0x[a-fA-F0-9]{40}\b', '[ADDRESS_REDACTED]'), # 钱包地址 ] @classmethod def filter_output(cls, text: str) -> str: """过滤输出中的敏感信息""" for pattern, replacement in cls.SENSITIVE_PATTERNS: text = re.sub(pattern, replacement, text) return text

3.2 L2：权限控制层

核心原则：最小权限 + 临时凭证 + 范围隔离

from dataclasses import dataclass from enum import Enum from datetime import datetime, timedelta import secrets

class Permission(Enum):
"""细粒度权限定义"""
READ_DATA = "data:read"
WRITE_DATA = "data:write"
CALL_EXTERNAL_API = "api:call"
EXECUTE_CODE = "code:execute"
SEND_NOTIFICATION = "notify:send"
ACCESS_FILESYSTEM = "fs:access"

@dataclass
class AgentCredential:
"""临时凭证，自动过期"""
token: str
permissions: set[Permission]
expires_at: datetime
scope: str # 限定资源范围
max_calls: int # 最大调用次数
current_calls: int = 0

class CredentialManager:
"""凭证管理：临时、限权、可追溯"""

def __init__(self): self._active_credentials = {} def issue_credential( self, agent_id: str, permissions: set[Permission], scope: str, ttl_minutes: int = 30, max_calls: int = 100 ) -> AgentCredential: """为Agent签发临时凭证""" credential = AgentCredential( token=secrets.token_urlsafe(32), permissions=permissions, expires_at=datetime.utcnow() + timedelta(minutes=ttl_minutes), scope=scope, max_calls=max_calls ) self._active_credentials[credential.token] = credential return credential def validate(self, token: str, required_permission: Permission) -> bool: """验证凭证是否有效""" cred = self._active_credentials.get(token) if not cred: return False # 过期检查 if datetime.utcnow() > cred.expires_at: del self._active_credentials[token] return False # 权限检查 if required_permission not in cred.permissions: return False # 调用次数检查 if cred.current_calls >= cred.max_calls: return False cred.current_calls += 1 return True def revoke(self, token: str): """主动撤销凭证""" self._active_credentials.pop(token, None)

使用示例

manager = CredentialManager()

为"数据分析Agent"签发凭证——只给读权限，30分钟过期

cred = manager.issue_credential(
agent_id="data-analyst",
permissions={Permission.READ_DATA},
scope="database:sales_data",
ttl_minutes=30,
max_calls=50
)

关键点：不要让Agent持有长期有效的Token。每次任务签发临时凭证，任务结束自动过期。

3.3 L3：沙箱隔离层

Agent调用的每一个外部工具，都应该在隔离环境中执行。

import subprocess import tempfile import os from pathlib import Path

class SandboxExecutor:
"""工具执行沙箱"""

def __init__(self, timeout_seconds: int = 30): self.timeout = timeout_seconds self.sandbox_dir = tempfile.mkdtemp(prefix="agent_sandbox_") def execute_code(self, code: str, allowed_imports: set[str] = None) -> dict: """在沙箱中执行代码""" # 1. 白名单检查 if allowed_imports: for line in code.split('\n'): if line.strip().startswith('import ') or line.strip().startswith('from '): module = line.strip().split()[1].split('.')[0] if module not in allowed_imports: return {"success": False, "error": f"禁止导入: {module}"} # 2. 写入隔离目录 script_path = Path(self.sandbox_dir) / "task.py" script_path.write_text(code) # 3. 在受限环境中执行 try: result = subprocess.run( ["python3", "--isolated", str(script_path)], capture_output=True, text=True, timeout=self.timeout, cwd=self.sandbox_dir, env={ "PATH": "/usr/bin", # 最小化PATH "HOME": self.sandbox_dir, # 不传递任何敏感环境变量 } ) return { "success": result.returncode == 0, "stdout": result.stdout[:1000], # 限制输出大小 "stderr": result.stderr[:500] if not result.returncode else None } except subprocess.TimeoutExpired: return {"success": False, "error": "执行超时"} finally: # 4. 清理 script_path.unlink(missing_ok=True) def call_external_api(self, url: str, method: str = "GET", allowed_domains: set[str] = None) -> dict: """限制外部API调用范围""" from urllib.parse import urlparse domain = urlparse(url).hostname # 域名白名单 if allowed_domains and domain not in allowed_domains: return {"success": False, "error": f"域名不在白名单: {domain}"} # 禁止内网访问 if domain in ("localhost", "127.0.0.1", "0.0.0.0") or domain.startswith("10.") or domain.startswith("192.168."): return {"success": False, "error": "禁止访问内网地址"} # 实际调用... return {"success": True, "data": "..."}

3.4 L4：审计追踪层

每一步行为都要留痕，这是事后溯源的唯一依据。

import json import logging from datetime import datetime from dataclasses import dataclass, asdict from enum import Enum

class ActionType(Enum):
TOOL_CALL = "tool_call"
LLM_INFERENCE = "llm_inference"
DATA_ACCESS = "data_access"
EXTERNAL_API = "external_api"
USER_INTERACTION = "user_interaction"
ERROR = "error"

@dataclass
class AuditEntry:
"""审计日志条目"""
timestamp: str
agent_id: str
task_id: str
action_type: str
tool_name: str = ""
input_summary: str = ""
output_summary: str = ""
duration_ms: int = 0
risk_level: str = "low" # low / medium / high / critical
metadata: dict = None

class AuditLogger:
"""Agent行为审计日志"""

def __init__(self, agent_id: str): self.agent_id = agent_id self.logger = logging.getLogger(f"agent_audit.{agent_id}") # 配置独立日志文件 handler = logging.FileHandler(f"audit/{agent_id}.jsonl") handler.setFormatter(logging.Formatter('%(message)s')) self.logger.addHandler(handler) self.logger.setLevel(logging.INFO) def log_tool_call(self, task_id: str, tool_name: str, input_data: str, output_data: str, duration_ms: int, risk_level: str = "low"): """记录工具调用""" entry = AuditEntry( timestamp=datetime.utcnow().isoformat(), agent_id=self.agent_id, task_id=task_id, action_type=ActionType.TOOL_CALL.value, tool_name=tool_name, input_summary=input_data[:200], # 截断避免日志爆炸 output_summary=output_data[:200], duration_ms=duration_ms, risk_level=risk_level ) self.logger.info(json.dumps(asdict(entry), ensure_ascii=False)) def log_error(self, task_id: str, error_msg: str, risk_level: str = "medium"): """记录异常""" entry = AuditEntry( timestamp=datetime.utcnow().isoformat(), agent_id=self.agent_id, task_id=task_id, action_type=ActionType.ERROR.value, output_summary=error_msg[:500], risk_level=risk_level ) self.logger.info(json.dumps(asdict(entry), ensure_ascii=False))

使用示例

audit = AuditLogger(“data-analyst-agent”)
audit.log_tool_call(
task_id=“task-001”,
tool_name=“database_query”,
input_data=“SELECT * FROM sales WHERE date > ‘2026-01-01’”,
output_data=“返回 1,234 条记录”,
duration_ms=156,
risk_level=“medium”
)

3.5 L5：人工审核层

高风险操作必须经过Human-in-the-Loop确认。

from enum import Enum

class RiskLevel(Enum):
LOW = 1 # 只读查询
MEDIUM = 2 # 内部数据修改
HIGH = 3 # 外部API调用 / 批量操作
CRITICAL = 4 # 删除操作 / 资金相关 / 对外发送

class HumanApprovalGate:
"""人工审核门禁"""

# 需要人工确认的操作 REQUIRE_APPROVAL = { RiskLevel.HIGH, RiskLevel.CRITICAL } # 自动拒绝的操作（即使人工也不能做） AUTO_DENY = [ "delete_production_data", "transfer_funds_above_10000", "send_email_to_all_users", ] @classmethod def evaluate(cls, action: str, risk: RiskLevel, context: dict) -> dict: """评估是否需要人工审核""" # 自动拒绝 if action in cls.AUTO_DENY: return {"approved": False, "reason": "操作被安全策略自动拒绝"} # 低风险直接放行 if risk not in cls.REQUIRE_APPROVAL: return {"approved": True, "reason": "低风险操作，自动放行"} # 高风险/关键操作：等待人工确认 return { "approved": False, "reason": f"操作 {action} (风险等级: {risk.name}) 需要人工审核", "requires_human": True, "context": context }

使用示例

result = HumanApprovalGate.evaluate(
action="batch_delete_records",
risk=RiskLevel.CRITICAL,
context={"table": "orders", "count": 5000}
)

→ requires_human: True, 等待管理员确认

四、完整安全Agent工作流

把上面五层串起来，一个生产级Agent的完整工作流：

class SafeAgent: """生产级安全Agent实现"""
def __init__(self, agent_id: str, config: dict): self.agent_id = agent_id self.input_guard = InputGuard() self.output_guard = OutputGuard() self.credential_manager = CredentialManager() self.sandbox = SandboxExecutor() self.audit = AuditLogger(agent_id) self.approval_gate = HumanApprovalGate() async def execute_task(self, user_input: str, task_id: str) -> dict: """安全执行Agent任务""" # L1: 输入校验 is_safe, risk_desc = self.input_guard.validate(user_input) if not is_safe: self.audit.log_error(task_id, f"输入拦截: {risk_desc}", "high") return {"error": "输入未通过安全校验", "detail": risk_desc} # L2: 签发临时凭证 cred = self.credential_manager.issue_credential( agent_id=self.agent_id, permissions={p for p in Permission}, # 根据任务调整 scope=task_id, ttl_minutes=15 ) try: # L3: 在沙箱中执行工具调用 result = await self._run_tools(user_input, task_id, cred) # L4: 审计记录 self.audit.log_tool_call( task_id=task_id, tool_name="main_task", input_data=user_input, output_summary=str(result)[:200], duration_ms=result.get("duration_ms", 0) ) # L1: 输出过滤 safe_output = self.output_guard.filter_output(str(result)) return {"success": True, "result": safe_output} except Exception as e: self.audit.log_error(task_id, str(e), "high") return {"error": "执行异常", "detail": str(e)} finally: # 凭证自动过期 self.credential_manager.revoke(cred.token) async def _run_tools(self, input_data: str, task_id: str, cred) -> dict: """工具执行（简化示例）""" # 这里实际会调用LLM推理 → 决定工具 → 沙箱执行 # 每次工具调用前检查权限和风险等级 # 示例：数据库查询 query = "SELECT * FROM orders WHERE status = 'pending'" # 风险评估 risk = RiskLevel.MEDIUM # 只读查询 approval = self.approval_gate.evaluate("db_query", risk, {}) if not approval["approved"]: return {"requires_approval": True, "reason": approval["reason"]} # 执行查询... return {"data": "查询结果", "duration_ms": 89}

五、安全自检清单（Checklist）

这是我团队每个Agent上线前必过的清单：

输入安全

• Prompt注入检测：覆盖至少10种注入模式
• 输入长度限制：单条输入≤4096字符
• 特殊字符过滤：控制字符、HTML/JS注入

权限控制

• 凭证临时化：所有Token有效期≤30分钟
• 最小权限：每个Agent只拥有任务必需的权限
• 资源范围限定：限定Agent可访问的数据/服务/URL范围
• 调用次数限制：设置单次任务的API调用上限

执行隔离

• 工具执行沙箱化：代码执行、API调用在隔离环境
• 网络隔离：禁止Agent访问内网地址
• 文件系统隔离：Agent无法访问宿主文件系统

审计追踪

• 全链路日志：每次工具调用、LLM推理、数据访问都留痕
• 异常告警：高风险操作触发实时告警
• 日志保留：审计日志保留≥90天

人工审核

• 高风险操作需人工确认：批量操作、删除操作、对外发送
• 自动拒绝清单：明确定义绝对禁止的操作
• 审核超时机制：30分钟未响应自动降级处理

供应链安全

• 依赖白名单：Agent只能使用经过审查的依赖包
• 包完整性校验：锁定版本 + checksum验证
• 内部镜像仓库：不直接从公共Registry拉包

六、生产部署关键指标

（数据来源：团队内部Agent安全基线 + OWASP Top 10 for LLM Applications 2025）

七、总结

2026年的AI Agent开发，安全已经不是可选项，而是必选项。

核心就三句话：

1. 永远不要相信Agent的判断—— 用规则兜底，不要只靠LLM自律
2. 永远不要给Agent过多权限—— 临时凭证 + 最小权限 + 范围隔离
3. 永远不要对Agent的行为一无所知—— 全链路审计 + 实时告警

Agent的能力会越来越强，但安全防护必须同步跟上。否则，你的Agent不是生产力工具，而是一个高价值的攻击面。

💬 互动引导
你们团队在用AI Agent时遇到过什么安全问题？或者有什么安全实践想分享的？欢迎评论区交流～

完整的《AI Agent安全开发Toolkit》（含上述全部代码示例 + 安全自检清单Excel + 生产部署检查脚本）已整理，关注我私信可获取。

（本文基于公开安全事件和技术实践总结，不构成商业推荐。）