AI错误处理设计模式

from enum import Enum

class ErrorType(Enum):
    # 基础设施错误
    TIMEOUT = "timeout"
    RATE_LIMIT = "rate_limit"
    SERVICE_UNAVAILABLE = "service_unavailable"
    
    # 模型错误
    HALLUCINATION = "hallucination"
    REFUSAL = "refusal"              # 模型拒绝回答
    TRUNCATION = "truncation"        # 输出被截断
    INVALID_FORMAT = "invalid_format" # 格式不合规
    
    # 安全错误
    PROMPT_INJECTION = "prompt_injection"
    TOXIC_OUTPUT = "toxic_output"
    PII_LEAK = "pii_leak"

from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type

class AIRetryPolicy:
    @retry(
        stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=1, min=2, max=30),
        retry=retry_if_exception_type((TimeoutError, ConnectionError)),
        retry_error_callback=lambda _: {"error": "Max retries exceeded"}
    )
    async def call_with_retry(self, llm, messages, **kwargs):
        try:
            response = await asyncio.wait_for(
                llm.ainvoke(messages, **kwargs),
                timeout=30
            )
            return response
        except TimeoutError:
            logger.warning("LLM call timeout, retrying...")
            raise
    
    async def call_with_smart_retry(self, llm, messages, **kwargs):
        """智能重试：根据错误类型调整策略"""
        for attempt in range(3):
            try:
                response = await llm.ainvoke(messages, **kwargs)
                
                # 检查输出质量
                if self.is_truncated(response):
                    # 截断错误：增加max_tokens重试
                    kwargs['max_tokens'] = kwargs.get('max_tokens', 2048) * 2
                    continue
                
                if self.is_refusal(response):
                    # 拒绝回答：调整prompt重试
                    messages = self.adjust_prompt_for_refusal(messages)
                    continue
                
                return response
                
            except RateLimitError:
                await asyncio.sleep(2 ** attempt)
            except TimeoutError:
                if attempt < 2:
                    continue
                raise
        
        return {"error": "All retries failed"}

class GracefulDegradation:
    """逐级降级策略"""
    
    def __init__(self, models):
        self.models = models  # 按优先级排序的模型列表
    
    async def generate(self, messages, **kwargs):
        errors = []
        
        for model in self.models:
            try:
                response = await model.generate(messages, **kwargs)
                
                # 验证响应质量
                if self.validate_response(response):
                    return response
                else:
                    errors.append(f"{model.name}: invalid response")
                    
            except Exception as e:
                errors.append(f"{model.name}: {str(e)}")
                logger.warning(f"Falling back from {model.name}: {e}")
                continue
        
        # 所有模型都失败，返回兜底响应
        logger.error(f"All models failed: {errors}")
        return self.fallback_response(messages)
    
    def fallback_response(self, messages):
        """兜底响应"""
        return {
            "content": "抱歉，服务暂时遇到问题。请稍后重试。",
            "status": "degraded",
            "timestamp": datetime.now().isoformat()
        }

class CircuitBreaker:
    """断路器：防止持续请求故障服务"""
    
    def __init__(self, failure_threshold=5, recovery_timeout=60):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.failure_count = 0
        self.last_failure_time = None
        self.state = "closed"  # closed, open, half-open
    
    async def call(self, func, *args, **kwargs):
        if self.state == "open":
            if time.time() - self.last_failure_time > self.recovery_timeout:
                self.state = "half-open"
            else:
                raise CircuitOpenError("Circuit breaker is open")
        
        try:
            result = await func(*args, **kwargs)
            self.on_success()
            return result
        except Exception as e:
            self.on_failure()
            raise
    
    def on_success(self):
        self.failure_count = 0
        self.state = "closed"
    
    def on_failure(self):
        self.failure_count += 1
        self.last_failure_time = time.time()
        if self.failure_count >= self.failure_threshold:
            self.state = "open"
            logger.error("Circuit breaker opened")

class OutputValidator:
    """验证LLM输出的合法性和质量"""
    
    async def validate(self, response, expected_format=None):
        checks = [
            self.check_empty(response),
            self.check_length(response),
            self.check_safety(response),
            self.check_format(response, expected_format),
        ]
        
        for check in checks:
            if not check["passed"]:
                return {"valid": False, "error": check["error"]}
        
        return {"valid": True}
    
    def check_safety(self, response):
        # 检查是否包含敏感信息
        sensitive_patterns = [
            r'\d{18}',           # 身份证号
            r'\d{16,19}',        # 银行卡号
            r'password\s*[:=]',  # 密码
        ]
        
        for pattern in sensitive_patterns:
            if re.search(pattern, response, re.IGNORECASE):
                return {"passed": False, "error": "Sensitive info detected"}
        
        return {"passed": True}
    
    def check_format(self, response, expected_format):
        if expected_format == "json":
            try:
                json.loads(response)
            except:
                return {"passed": False, "error": "Invalid JSON"}
        
        return {"passed": True}

class ErrorMonitor:
    def __init__(self):
        self.error_counts = defaultdict(int)
        self.error_history = deque(maxlen=1000)
    
    def record(self, error_type, context=None):
        self.error_counts[error_type] += 1
        self.error_history.append({
            "type": error_type,
            "timestamp": datetime.now().isoformat(),
            "context": context or {},
        })
        
        # 错误率突增告警
        if self.error_counts[error_type] > 100:
            self.alert(f"Error {error_type} exceeded 100 occurrences")
    
    def health_check(self):
        """系统健康检查"""
        total_errors = sum(self.error_counts.values())
        recent_errors = [
            e for e in self.error_history
            if (datetime.now() - datetime.fromisoformat(e["timestamp"])).seconds < 300
        ]
        
        return {
            "total_errors": total_errors,
            "recent_errors_5min": len(recent_errors),
            "top_errors": dict(sorted(self.error_counts.items(), 
                                     key=lambda x: x[1], reverse=True)[:5]),
        }