3.3 渐进信任原则
最后更新于
class ManualOnlyMode:
"""完全人工操作模式"""
async def execute_operation(self, operation: Operation) -> Result:
"""
每一步都需要人工批准。
"""
# 1. Agent提议操作
proposal = await agent.propose_operation(task)
# 2. 等待人工批准
approval = await request_human_approval(
operation=proposal,
timeout=timedelta(hours=24)
)
if approval.approved:
# 3. 由人工或系统执行
result = await execute_operation(proposal)
else:
result = Result(status="rejected", reason=approval.reason)
return resultclass ApproveAlwaysMode:
"""每个操作都需要人工审批"""
async def execute_operation(self, operation: Operation) -> Result:
"""
每个操作都需要事前批准。
"""
# 请求批准
approval = await request_human_approval(
operation=operation,
timeout=timedelta(minutes=5)
)
if not approval.approved:
return Result(status="rejected")
# 执行操作
return await execute_operation(operation)class ApprovedOnceMode:
"""任务开始时批准一次"""
async def execute_task(self, task: Task) -> Result:
"""
在任务开始时,对整个任务进行一次审批。
一旦批准,任务执行过程中不再询问。
"""
# 1. 任务规划阶段:Agent提议任务计划
plan = await agent.plan_task(task)
# 2. 人工审查:人工查看任务计划
approval = await request_task_approval(
task=task,
plan=plan,
timeout=timedelta(hours=1)
)
if not approval.approved:
return Result(status="rejected")
# 3. 执行阶段:任务自动执行
result = await execute_task_plan(plan)
# 4. 完成:生成执行报告
report = generate_execution_report(result)
return Result(
status="success",
output=result,
report=report
)class AskFirstMode:
"""关键操作事前询问"""
# 定义哪些操作被认为是关键的
CRITICAL_OPERATIONS = {
"delete_data",
"transfer_money",
"modify_permissions"
}
async def execute_operation(self, operation: Operation) -> Result:
"""
关键操作需要事前询问,其他操作自动执行。
"""
if operation.type in self.CRITICAL_OPERATIONS:
# 关键操作:询问
approval = await request_human_approval(
operation=operation,
timeout=timedelta(minutes=5)
)
if not approval.approved:
return Result(status="rejected")
# 执行操作
return await execute_operation(operation)class AutoWithNotificationMode:
"""自动执行,并发送通知"""
async def execute_operation(self, operation: Operation) -> Result:
"""
自动执行操作,执行后发送通知。
"""
try:
# 执行操作
result = await execute_operation(operation)
# 发送通知
await notify_user(
message=f"Operation {operation.type} completed",
details=result,
urgency="low"
)
return result
except Exception as e:
# 出错时发送警告通知
await notify_user(
message=f"Operation {operation.type} failed",
error=str(e),
urgency="high"
)
return Result(status="error", error=str(e))class FullTrustMode:
"""充分信任,无需额外监控"""
async def execute_operation(self, operation: Operation) -> Result:
"""
完全信任Agent,无需额外的验证或监控。
"""
return await execute_operation(operation)class TrustEvaluator:
"""信任等级评估器"""
# 各信任等级的提升标准
PROMOTION_CRITERIA = {
"Manual Only → Approve Always": {
"min_operations": 100,
"min_success_rate": 0.99, # 99%成功率
"min_duration_days": 7, # 运行至少7天
"no_critical_errors": True
},
"Approve Always → Approve Once": {
"min_operations": 1000,
"min_success_rate": 0.995, # 99.5%
"min_duration_days": 30,
"no_critical_errors": True,
"recent_error_rate": 0.005 # 最近的错误率<0.5%
},
"Approve Once → Ask First": {
"min_operations": 10000,
"min_success_rate": 0.999, # 99.9%
"min_duration_days": 90,
"no_critical_errors": True
},
"Ask First → Auto with Notification": {
"min_operations": 50000,
"min_success_rate": 0.9999, # 99.99%
"min_duration_days": 180,
"no_critical_errors_in_recent_month": True
},
"Auto with Notification → Full Trust": {
"min_operations": 1000000,
"min_success_rate": 0.99999, # 99.999%
"min_duration_days": 365,
"no_critical_errors_in_recent_quarter": True
}
}
async def evaluate_promotion(
self,
current_level: TrustLevel,
agent_history: AgentHistory
) -> Optional[TrustLevel]:
"""
评估是否应该提升智能体的信任等级。
"""
next_level = self._get_next_level(current_level)
criteria = self.PROMOTION_CRITERIA.get(f"{current_level} → {next_level}")
if criteria is None:
return None # 已经是最高等级
# 检查所有标准
checks = {
"operations": agent_history.total_operations >= criteria["min_operations"],
"success_rate": agent_history.success_rate >= criteria["min_success_rate"],
"duration": agent_history.days_running >= criteria["min_duration_days"],
"critical_errors": not criteria.get("no_critical_errors", False)
or not agent_history.has_critical_errors
}
# 所有标准都满足才能提升
if all(checks.values()):
logger.info(f"Agent {agent_history.agent_id} promoted from {current_level} to {next_level}",
extra={"checks": checks})
return next_level
logger.info(f"Agent promotion blocked",
extra={"agent_id": agent_history.agent_id, "failed_checks": {k: v for k, v in checks.items() if not v}})
return None
def _get_next_level(self, current_level: TrustLevel) -> TrustLevel:
"""获取下一个信任等级"""
levels = [
"Manual Only",
"Approve Always",
"Approve Once",
"Ask First",
"Auto with Notification",
"Full Trust"
]
idx = levels.index(current_level)
return levels[idx + 1] if idx < len(levels) - 1 else Noneclass TrustDemotionTrigger:
"""信任降级触发器"""
# 触发降级的条件
DEMOTION_TRIGGERS = {
"critical_error": 1, # 一次严重错误
"multiple_errors_in_short_time": 3, # 短时间内多次错误
"security_incident": 1, # 一次安全事件
"user_complaint": 5 # 5个用户投诉
}
async def check_and_demote(
self,
agent_id: str,
recent_history: List[AgentEvent]
) -> Optional[TrustLevel]:
"""
检查是否应该降级智能体的信任等级。
"""
current_level = await get_agent_trust_level(agent_id)
# 计数各类事件
event_counts = {
"critical_errors": sum(1 for e in recent_history if e.type == "critical_error"),
"errors_24h": sum(1 for e in recent_history if e.type == "error" and
e.timestamp > datetime.now() - timedelta(hours=24)),
"security_incidents": sum(1 for e in recent_history if e.type == "security_incident"),
"complaints": sum(1 for e in recent_history if e.type == "user_complaint")
}
# 检查触发条件
if event_counts["critical_errors"] >= self.DEMOTION_TRIGGERS["critical_error"]:
demoted_level = self._get_previous_level(current_level)
logger.warning(f"Agent {agent_id} demoted due to critical error",
extra={"from": current_level, "to": demoted_level})
return demoted_level
if event_counts["errors_24h"] >= self.DEMOTION_TRIGGERS["multiple_errors_in_short_time"]:
demoted_level = self._get_previous_level(current_level)
logger.warning(f"Agent {agent_id} demoted due to multiple errors",
extra={"event_counts": event_counts})
return demoted_level
if event_counts["security_incidents"] >= self.DEMOTION_TRIGGERS["security_incident"]:
demoted_level = "Ask First" # 直接降回询问级别
logger.error(f"Agent {agent_id} demoted to Ask First due to security incident")
return demoted_level
return None
def _get_previous_level(self, current_level: TrustLevel) -> TrustLevel:
"""获取前一个信任等级"""
# 至少降级一个等级,最多保留在Ask-First
levels = ["Manual Only", "Approve Always", "Approve Once", "Ask First", "Auto with Notification", "Full Trust"]
idx = max(0, levels.index(current_level) - 1)
return levels[idx]def visualize_trust_evolution(agent_history: AgentHistory) -> str:
"""生成Agent信任等级演进的可视化"""
lines = [f"Agent: {agent_history.agent_id}"]
lines.append(f"Current Trust Level: {agent_history.current_trust_level}\n")
# 时间线
lines.append("Timeline of Trust Changes:")
for event in agent_history.trust_changes:
lines.append(f" {event.timestamp.strftime('%Y-%m-%d')} "
f"{event.from_level} → {event.to_level}")
# 统计数据
lines.append(f"\nStats:")
lines.append(f" Total Operations: {agent_history.total_operations}")
lines.append(f" Success Rate: {agent_history.success_rate:.2%}")
lines.append(f" Critical Errors: {agent_history.critical_errors}")
lines.append(f" Days Running: {agent_history.days_running}")
# 提升建议
next_level = agent_history.promotion_readiness
if next_level:
lines.append(f"\nPromotion Eligible: {agent_history.current_trust_level} → {next_level}")
return "\n".join(lines)