Coverage for src / local_deep_research / web_search_engines / rate_limiting / llm / detection.py: 69%

1""" 

2LLM-specific rate limit error detection. 

3""" 

4 

5from loguru import logger 

6 

7 

8def is_llm_rate_limit_error(error: Exception) -> bool: 

9 """ 

10 Detect if an error is a rate limit error from an LLM provider. 

11 

12 Args: 

13 error: The exception to check 

14 

15 Returns: 

16 True if this is a rate limit error, False otherwise 

17 """ 

18 error_str = str(error).lower() 

19 error_type = type(error).__name__.lower() 

20 

21 # Check for explicit HTTP 429 status codes 

22 if hasattr(error, "response") and hasattr(error.response, "status_code"): 

23 if error.response.status_code == 429: 23 ↛ 28line 23 didn't jump to line 28 because the condition on line 23 was always true

24 logger.debug(f"Detected HTTP 429 rate limit error: {error}") 

25 return True 

26 

27 # Check for common rate limit error messages 

28 rate_limit_indicators = [ 

29 "rate limit", 

30 "rate_limit", 

31 "ratelimit", 

32 "too many requests", 

33 "quota exceeded", 

34 "quota has been exhausted", 

35 "resource has been exhausted", 

36 "429", 

37 "threshold", 

38 "try again later", 

39 "slow down", 

40 ] 

41 

42 if any(indicator in error_str for indicator in rate_limit_indicators): 

43 logger.debug(f"Detected rate limit error from message: {error}") 

44 return True 

45 

46 # Check for specific provider error types 

47 if "ratelimiterror" in error_type or "quotaexceeded" in error_type: 47 ↛ 48line 47 didn't jump to line 48 because the condition on line 47 was never true

48 logger.debug(f"Detected rate limit error from type: {type(error)}") 

49 return True 

50 

51 # Check for OpenAI specific rate limit errors 

52 if hasattr(error, "__class__") and error.__class__.__module__ == "openai": 52 ↛ 53line 52 didn't jump to line 53 because the condition on line 52 was never true

53 if error_type in ["ratelimiterror", "apierror"] and "429" in error_str: 

54 logger.debug(f"Detected OpenAI rate limit error: {error}") 

55 return True 

56 

57 # Check for Anthropic specific rate limit errors 

58 if ( 58 ↛ 62line 58 didn't jump to line 62 because the condition on line 58 was never true

59 hasattr(error, "__class__") 

60 and "anthropic" in error.__class__.__module__ 

61 ): 

62 if any(x in error_str for x in ["rate_limit", "429", "too many"]): 

63 logger.debug(f"Detected Anthropic rate limit error: {error}") 

64 return True 

65 

66 return False 

67 

68 

69def extract_retry_after(error: Exception) -> float: 

70 """ 

71 Extract retry-after time from rate limit error if available. 

72 

73 Args: 

74 error: The rate limit error 

75 

76 Returns: 

77 Retry after time in seconds, or 0 if not found 

78 """ 

79 # Check for Retry-After header in response 

80 if hasattr(error, "response") and hasattr(error.response, "headers"): 

81 retry_after = error.response.headers.get("Retry-After") 

82 if retry_after: 82 ↛ 91line 82 didn't jump to line 91 because the condition on line 82 was always true

83 try: 

84 return float(retry_after) 

85 except ValueError: 

86 logger.debug( 

87 f"Could not parse Retry-After header: {retry_after}" 

88 ) 

89 

90 # Try to extract from error message 

91 error_str = str(error) 

92 

93 # Look for patterns like "try again in X seconds" 

94 import re 

95 

96 patterns = [ 

97 r"try again in (\d+(?:\.\d+)?)\s*seconds?", 

98 r"retry after (\d+(?:\.\d+)?)\s*seconds?", 

99 r"wait (\d+(?:\.\d+)?)\s*seconds?", 

100 ] 

101 

102 for pattern in patterns: 

103 match = re.search(pattern, error_str, re.IGNORECASE) 

104 if match: 

105 try: 

106 return float(match.group(1)) 

107 except ValueError: 

108 pass 

109 

110 return 0