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

1""" 

2LLM-specific rate limit error detection. 

3""" 

4 

5from ....security.log_sanitizer import scrub_error 

6from ....security.secure_logging import logger 

7 

8 

9def is_llm_rate_limit_error(error: BaseException) -> bool: 

10 """ 

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

12 

13 Args: 

14 error: The exception to check 

15 

16 Returns: 

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

18 """ 

19 error_str = str(error).lower() 

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

21 

22 # Check for explicit HTTP 429 status codes 

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

24 if error.response.status_code == 429: 

25 logger.debug( 

26 f"Detected HTTP 429 rate limit error: {scrub_error(error)}" 

27 ) 

28 return True 

29 

30 # Check for common rate limit error messages 

31 rate_limit_indicators = [ 

32 "rate limit", 

33 "rate_limit", 

34 "ratelimit", 

35 "too many requests", 

36 "quota exceeded", 

37 "quota has been exhausted", 

38 "resource has been exhausted", 

39 "429", 

40 "threshold", 

41 "try again later", 

42 "slow down", 

43 ] 

44 

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

46 logger.debug( 

47 f"Detected rate limit error from message: {scrub_error(error)}" 

48 ) 

49 return True 

50 

51 # Check for specific provider error types 

52 if "ratelimiterror" in error_type or "quotaexceeded" in error_type: 

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

54 return True 

55 

56 # Check for OpenAI specific rate limit errors 

57 if hasattr(error, "__class__") and error.__class__.__module__ == "openai": 

58 if error_type in ["ratelimiterror", "apierror"] and "429" in error_str: 58 ↛ 59line 58 didn't jump to line 59 because the condition on line 58 was never true

59 logger.debug( 

60 f"Detected OpenAI rate limit error: {scrub_error(error)}" 

61 ) 

62 return True 

63 

64 # Check for Anthropic specific rate limit errors 

65 if ( 

66 hasattr(error, "__class__") 

67 and "anthropic" in error.__class__.__module__ 

68 ): 

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

70 logger.debug( 

71 f"Detected Anthropic rate limit error: {scrub_error(error)}" 

72 ) 

73 return True 

74 

75 return False 

76 

77 

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

79 """ 

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

81 

82 Args: 

83 error: The rate limit error 

84 

85 Returns: 

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

87 """ 

88 # Check for Retry-After header in response 

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

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

91 if retry_after: 

92 try: 

93 return float(retry_after) 

94 except ValueError: 

95 logger.debug( 

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

97 ) 

98 

99 # Try to extract from error message 

100 error_str = str(error) 

101 

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

103 import re 

104 

105 patterns = [ 

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

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

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

109 ] 

110 

111 for pattern in patterns: 

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

113 if match: 

114 try: 

115 return float(match.group(1)) 

116 except ValueError: 

117 pass 

118 

119 return 0