AI系统的错误特征 AI系统的错误与传统软件不同——不仅有网络超时、服务不可用等基础设施错误，还有模型幻觉、输出格式错误、安全违规等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]), } 实践建议 区分错误类型：基础设施错误用重试，模型错误用降级，安全错误用阻断 设置合理超时：LLM调用超时建议30-60秒 指数退避：重试间隔使用指数退避，避免雪崩 兜底方案：始终准备兜底响应，确保用户体验 错误分类上报：不同类型错误分别统计，便于定位问题 结语 AI系统的错误处理需要比传统软件更精细的设计——不仅要处理基础设施故障，还要处理模型输出质量问题和安全风险。重试、降级、断路器和输出验证的组合使用，可以构建出既健壮又优雅的AI系统。 加入讨论 这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。
...
