Coverage for src/local_deep_research/web_search_engines/engines/search_engine_brave.py: 98%
52 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-06 15:42 +0000
1from typing import Any, Dict, List, Optional
3from langchain_community.tools import BraveSearch
4from langchain_core.language_models import BaseLLM
6from ...security.secure_logging import logger
8from ..rate_limiting import RateLimitError
9from ..search_engine_base import (
10 VALID_TIME_PERIODS,
11 BaseSearchEngine,
12 Exposure,
13 Sensitivity,
14)
17class BraveSearchEngine(BaseSearchEngine):
18 """Brave search engine implementation with two-phase approach"""
20 # Mark as public search engine
21 is_public = True
22 egress_sensitivity = Sensitivity.NON_SENSITIVE
23 egress_exposure = Exposure.EXPOSING
24 # Mark as generic search engine (general web search)
25 is_generic = True
26 # secrets to redact from error messages (see BaseSearchEngine._scrub_error)
27 _secret_attrs = ("_brave_api_key",)
29 def __init__(
30 self,
31 max_results: int = 10,
32 region: str = "US",
33 time_period: str = "y",
34 safe_search: bool = True,
35 search_language: str = "English",
36 api_key: Optional[str] = None,
37 language_code_mapping: Optional[Dict[str, str]] = None,
38 llm: Optional[BaseLLM] = None,
39 include_full_content: bool = True,
40 max_filtered_results: Optional[int] = None,
41 settings_snapshot: Optional[Dict[str, Any]] = None,
42 **kwargs,
43 ):
44 """
45 Initialize the Brave search engine.
47 Args:
48 max_results: Maximum number of search results
49 region: Region code for search results
50 time_period: Time period filter (d/w/m/y); forwarded to Brave's
51 ``freshness`` param as p{d,w,m,y}. ``all`` (or any
52 unrecognized value) omits the filter.
53 safe_search: Whether to enable safe search
54 search_language: Language for search results
55 api_key: Brave Search API key (can also be set via LDR_SEARCH_ENGINE_WEB_BRAVE_API_KEY env var or in UI settings)
56 language_code_mapping: Mapping from language names to codes
57 llm: Language model for relevance filtering
58 include_full_content: Whether to include full webpage content in results
59 max_filtered_results: Maximum number of results to keep after filtering
60 settings_snapshot: Settings snapshot for thread context
61 **kwargs: Additional parameters (ignored but accepted for compatibility)
62 """
63 # Initialize the BaseSearchEngine with LLM, max_filtered_results, and max_results
64 super().__init__(
65 llm=llm,
66 max_filtered_results=max_filtered_results,
67 max_results=max_results,
68 include_full_content=include_full_content,
69 settings_snapshot=settings_snapshot,
70 )
72 # Set up language code mapping
73 if language_code_mapping is None:
74 from ...utilities.search_utilities import LANGUAGE_CODE_MAP
76 language_code_mapping = LANGUAGE_CODE_MAP
78 # Get API key - check params, settings, or env vars
79 brave_api_key = self._resolve_api_key(
80 api_key,
81 "search.engine.web.brave.api_key",
82 engine_name="Brave Search",
83 settings_snapshot=settings_snapshot,
84 )
85 self._brave_api_key = brave_api_key
87 # Get language code
88 language_code = language_code_mapping.get(search_language.lower(), "en")
90 # Convert safe search to Brave's format
91 brave_safe_search = "moderate" if safe_search else "off"
93 # Initialize Brave Search
94 search_kwargs = {
95 "count": min(20, max_results),
96 "country": region.upper(),
97 "search_lang": language_code,
98 "safesearch": brave_safe_search,
99 }
100 # Brave's freshness param prefixes time_period with "p" (pd/pw/pm/py).
101 # Only forward recognized codes; "all", any unrecognized value
102 # (typo, None, "") and non-string types omit freshness entirely so
103 # the search is unfiltered rather than sending an invalid value
104 # like "pall" (the isinstance guard also keeps unhashable values
105 # from raising on the frozenset lookup).
106 if isinstance(time_period, str) and time_period in VALID_TIME_PERIODS:
107 search_kwargs["freshness"] = f"p{time_period}"
108 self.engine = BraveSearch.from_api_key(
109 api_key=brave_api_key,
110 search_kwargs=search_kwargs,
111 )
113 # User agent is not needed for Brave Search API
115 # If full content is requested, initialize FullSearchResults
116 self._init_full_search(
117 web_search=self.engine,
118 language=search_language,
119 max_results=max_results,
120 region=region,
121 time_period=time_period,
122 safe_search=brave_safe_search,
123 )
125 def _get_previews(self, query: str) -> List[Dict[str, Any]]:
126 """
127 Get preview information from Brave Search.
129 Args:
130 query: The search query
132 Returns:
133 List of preview dictionaries
134 """
135 logger.info("Getting search results from Brave Search")
137 try:
138 # Get search results from Brave Search
139 raw_results = self.engine.run(query[:400])
141 # Parse results if they're in string format
142 if isinstance(raw_results, str):
143 try:
144 import json
146 raw_results = json.loads(raw_results)
147 except json.JSONDecodeError as e:
148 safe_msg = self._scrub_error(e)
149 logger.warning(
150 f"Unable to parse BraveSearch response as JSON: {safe_msg}"
151 )
152 return []
154 # Format results as previews
155 previews = []
156 for i, result in enumerate(raw_results):
157 preview = {
158 "id": i, # Use index as ID
159 "title": result.get("title", ""),
160 "link": result.get("link", ""),
161 "snippet": result.get("snippet", ""),
162 "displayed_link": result.get("link", ""),
163 "position": i,
164 }
166 # Store full Brave result for later
167 preview["_full_result"] = result
169 previews.append(preview)
171 # Store the previews for potential full content retrieval
172 self._search_results = previews
174 return previews
176 except RateLimitError:
177 raise
178 except Exception as e:
179 safe_msg = self._scrub_error(e)
180 logger.warning(f"Error getting Brave Search results: {safe_msg}")
181 self._raise_if_rate_limit(e)
182 return []