class FallbackManager:
def __init__(self):
self.fallback_level = 0 # 当前降级层级
self.error_window = [] # 滑动窗口错误计数
self.canned_responses = load_canned_responses()
def handle_request(self, request: Request) -> Response:
if self.fallback_level >= 3:
return self.get_canned_response(request.category)
try:
response = self.primary_model.generate(request)
# 输出安全检查
if not self.safety_check(response):
return self.get_canned_response(request.category)
self.record_success()
return response
except ModelUnavailableError:
self.record_failure()
return self.escalate_fallback(request)
def escalate_fallback(self, request: Request) -> Response:
self.fallback_level = min(self.fallback_level + 1, 4)
if self.fallback_level == 1:
return self.retry_with_lower_temperature(request)
elif self.fallback_level == 2:
return self.use_backup_model(request)
elif self.fallback_level == 3:
return self.get_canned_response(request.category)
else:
return self.route_to_human(request)