Coverage for web_search / runtime.py: 23%

1import os 

2import time 

3from typing import Optional, Dict, Any, List 

4from qrclaw.web_search.types import WebSearchResponse, SearchResult, WebSearchProvider 

5from qrclaw.web_search.provider_registry import list_providers, get_provider 

6from qrclaw.config import TAVILY_API_KEY 

7from qrclaw.logger import get_logger 

8 

9logger = get_logger("qrclaw.web_search.runtime") 

10 

11class WebSearchError(Exception): 

12 """Web Search Runtime Error""" 

13 pass 

14 

15def resolve_web_search_provider_id(prefer_provider: Optional[str] = None) -> Optional[str]: 

16 """ 

17 确定使用哪个 Provider。 

18 优先级逻辑： 

19 1. 用户显式指定 (prefer_provider) 

20 2. Tavily (最强，如果有 Key) 

21 3. Google (次强，如果有 Key) 

22 4. DuckDuckGo (免费保底) 

23 """ 

24 if prefer_provider: 

25 p_cls = get_provider(prefer_provider) 

26 if p_cls and p_cls().is_available(): 

27 return prefer_provider 

28 

29 # 按优先级列表自动探测 

30 # 这里我们硬编码一个优先级顺序，而不是简单遍历注册表 

31 priority_order = ["tavily", "google", "duckduckgo"] 

32 

33 for pid in priority_order: 

34 p_cls = get_provider(pid) 

35 if p_cls: 

36 provider = p_cls() 

37 if provider.is_available(): 

38 logger.info(f"Auto-detected web search provider: {provider.name}") 

39 return provider.id 

40 

41 # 如果上面的都没选中（理论上 DuckDuckGo 总是可用），再尝试剩下的 

42 for p_cls in list_providers(): 

43 provider = p_cls() 

44 if provider.is_available(): 

45 # 避免重复选 

46 if provider.id not in priority_order: 

47 logger.info(f"Auto-detected fallback provider: {provider.name}") 

48 return provider.id 

49 

50 return None 

51 

52def run_web_search(query: str, max_results: int = 5, provider_id: Optional[str] = None, **kwargs) -> WebSearchResponse: 

53 """ 

54 执行 Web 搜索 

55 """ 

56 selected_id = provider_id or resolve_web_search_provider_id() 

57 

58 if not selected_id: 

59 error_msg = "No web search provider is available." 

60 logger.error(error_msg) 

61 raise WebSearchError(error_msg) 

62 

63 provider_cls = get_provider(selected_id) 

64 if not provider_cls: 

65 raise WebSearchError(f"Provider '{selected_id}' not found.") 

66 

67 provider = provider_cls() 

68 logger.info(f"Executing web search with provider: {provider.name}") 

69 

70 try: 

71 return provider.search(query=query, max_results=max_results, **kwargs) 

72 except Exception as e: 

73 logger.error(f"Web search failed with {provider.name}: {e}") 

74 

75 # 简单的故障转移逻辑：如果首选失败，尝试降级到 DuckDuckGo 

76 if provider.id != "duckduckgo": 

77 logger.warning("Attempting fallback to DuckDuckGo...") 

78 fallback_cls = get_provider("duckduckgo") 

79 if fallback_cls: 

80 fallback = fallback_cls() 

81 return fallback.search(query=query, max_results=max_results, **kwargs) 

82 

83 raise