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 None