Coverage for src/local_deep_research/web_search_engines/engines/search_engine_brave.py: 98%
51 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-22 21:20 +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: Retained for backward compatibility; full-content
59 retrieval is handled via the factory wrapper.
60 max_filtered_results: Maximum number of results to keep after filtering
61 settings_snapshot: Settings snapshot for thread context
62 **kwargs: Additional parameters (ignored but accepted for compatibility)
63 """
64 # Initialize the BaseSearchEngine with LLM, max_filtered_results, and max_results
65 super().__init__(
66 llm=llm,
67 max_filtered_results=max_filtered_results,
68 max_results=max_results,
69 include_full_content=include_full_content,
70 settings_snapshot=settings_snapshot,
71 )
73 # Set up language code mapping
74 if language_code_mapping is None:
75 from ...utilities.search_utilities import LANGUAGE_CODE_MAP
77 language_code_mapping = LANGUAGE_CODE_MAP
79 # Get API key - check params, settings, or env vars
80 brave_api_key = self._resolve_api_key(
81 api_key,
82 "search.engine.web.brave.api_key",
83 engine_name="Brave Search",
84 settings_snapshot=settings_snapshot,
85 )
86 self._brave_api_key = brave_api_key
88 # Get language code
89 language_code = language_code_mapping.get(search_language.lower(), "en")
91 # Convert safe search to Brave's format
92 brave_safe_search = "moderate" if safe_search else "off"
94 # Initialize Brave Search
95 search_kwargs = {
96 "count": min(20, max_results),
97 "country": region.upper(),
98 "search_lang": language_code,
99 "safesearch": brave_safe_search,
100 }
101 # Brave's freshness param prefixes time_period with "p" (pd/pw/pm/py).
102 # Only forward recognized codes; "all", any unrecognized value
103 # (typo, None, "") and non-string types omit freshness entirely so
104 # the search is unfiltered rather than sending an invalid value
105 # like "pall" (the isinstance guard also keeps unhashable values
106 # from raising on the frozenset lookup).
107 if isinstance(time_period, str) and time_period in VALID_TIME_PERIODS:
108 search_kwargs["freshness"] = f"p{time_period}"
109 self.engine = BraveSearch.from_api_key(
110 api_key=brave_api_key,
111 search_kwargs=search_kwargs,
112 )
114 # User agent is not needed for Brave Search API
116 # Full-content retrieval is the factory wrapper's job — the
117 # wrapper fetches pages via ``batch_fetch_and_extract`` and
118 # populates ``full_content`` on each result; the inner engine
119 # only emits snippets from ``_get_previews``.
121 def _get_previews(self, query: str) -> List[Dict[str, Any]]:
122 """
123 Get preview information from Brave Search.
125 Args:
126 query: The search query
128 Returns:
129 List of preview dictionaries
130 """
131 logger.info("Getting search results from Brave Search")
133 try:
134 # Get search results from Brave Search
135 raw_results = self.engine.run(query[:400])
137 # Parse results if they're in string format
138 if isinstance(raw_results, str):
139 try:
140 import json
142 raw_results = json.loads(raw_results)
143 except json.JSONDecodeError as e:
144 safe_msg = self._scrub_error(e)
145 logger.warning(
146 f"Unable to parse BraveSearch response as JSON: {safe_msg}"
147 )
148 return []
150 # Format results as previews
151 previews = []
152 for i, result in enumerate(raw_results):
153 preview = {
154 "id": i, # Use index as ID
155 "title": result.get("title", ""),
156 "link": result.get("link", ""),
157 "snippet": result.get("snippet", ""),
158 "displayed_link": result.get("link", ""),
159 "position": i,
160 }
162 # Store full Brave result for later
163 preview["_full_result"] = result
165 previews.append(preview)
167 # Store the previews for potential full content retrieval
168 self._search_results = previews
170 return previews
172 except RateLimitError:
173 raise
174 except Exception as e:
175 safe_msg = self._scrub_error(e)
176 logger.warning(f"Error getting Brave Search results: {safe_msg}")
177 self._raise_if_rate_limit(e)
178 return []