Coverage for src/local_deep_research/web_search_engines/engines/search_engine_sofya.py: 100%
76 statements
« prev ^ index » next coverage.py v7.15.1, created at 2026-07-19 23:35 +0000
1"""Sofya search engine implementation for Local Deep Research.
3Sofya (https://sofya.co) is a hosted web-search API for AI agents. Its
4``POST /v1/search`` endpoint returns ranked results with a SERP snippet
5(``description``) and, in ``basic`` depth, the extracted page content
6(``content``) in the same response. A single request therefore covers both
7phases of LDR's two-phase retrieval model: previews come from
8``description`` and full content from ``content`` — no separate fetch round
9trip is needed.
11Sign-up gives 1000 free credits/month to GitHub-authenticated accounts, so
12the engine is usable without a paid plan. Get an API key at
13https://sofya.co/dashboard.
14"""
16from typing import Any, Dict, List, Optional
18import requests
19from langchain_core.language_models import BaseLLM
21from ...security.safe_requests import safe_post
22from ...security.secure_logging import logger
23from ..rate_limiting import RateLimitError
24from ..search_engine_base import BaseSearchEngine, Exposure, Sensitivity
27class SofyaSearchEngine(BaseSearchEngine):
28 """Sofya search engine implementation with two-phase approach."""
30 # Mark as a public, generic (general web) search engine.
31 is_public = True
32 egress_sensitivity = Sensitivity.NON_SENSITIVE
33 egress_exposure = Exposure.EXPOSING
34 is_generic = True
36 #: Base URL for the Sofya REST API.
37 BASE_URL = "https://sofya.co/v1"
38 #: Sofya caps ``max_results`` at 20 per request.
39 MAX_RESULTS_CAP = 20
40 #: Sofya accepts at most 10 include/exclude domains per request.
41 MAX_DOMAINS = 10
42 #: Characters of extracted content to use as a preview snippet when the
43 #: SERP ``description`` is missing.
44 SNIPPET_PREVIEW_CHARS = 500
45 #: HTTP request timeout in seconds.
46 REQUEST_TIMEOUT = 30
48 def __init__(
49 self,
50 max_results: int = 10,
51 api_key: Optional[str] = None,
52 llm: Optional[BaseLLM] = None,
53 topic: str = "general",
54 freshness: Optional[str] = None,
55 include_domains: Optional[List[str]] = None,
56 exclude_domains: Optional[List[str]] = None,
57 include_full_content: bool = True,
58 search_snippets_only: Optional[bool] = None,
59 max_filtered_results: Optional[int] = None,
60 settings_snapshot: Optional[Dict[str, Any]] = None,
61 **kwargs,
62 ):
63 """Initialize the Sofya search engine.
65 Args:
66 max_results: Maximum number of search results to request.
67 api_key: Sofya API key. May also be supplied via the
68 ``search.engine.web.sofya.api_key`` setting or the
69 ``LDR_SEARCH_ENGINE_WEB_SOFYA_API_KEY`` environment variable.
70 llm: Language model for relevance filtering.
71 topic: ``"general"`` or ``"news"``.
72 freshness: Recency filter — ``"day"``, ``"week"``, ``"month"``,
73 ``"year"``, or a ``"YYYY-MM-DD:YYYY-MM-DD"`` range. ``None``
74 means no filter.
75 include_domains: Restrict results to these domains (max 10).
76 exclude_domains: Exclude results from these domains (max 10).
77 include_full_content: Whether to fetch extracted page content.
78 When ``True`` (and full content will be consumed, see
79 ``search_snippets_only``) the request uses Sofya's ``basic``
80 depth (content included); otherwise it uses the cheaper
81 ``snippets`` depth (SERP snippets only).
82 search_snippets_only: Whether to return only search snippets.
83 ``None`` (default) derives it from ``include_full_content``,
84 like TinyFish; an explicit value (e.g. forwarded from the
85 ``search.snippets_only`` setting by the factory) wins.
86 max_filtered_results: Maximum number of results to keep after
87 relevance filtering.
88 settings_snapshot: Settings snapshot for thread context.
89 **kwargs: Additional base-class parameters (e.g.
90 ``programmatic_mode``); forwarded to ``super().__init__``.
91 """
92 if search_snippets_only is None:
93 search_snippets_only = not include_full_content
94 super().__init__(
95 llm=llm,
96 max_filtered_results=max_filtered_results,
97 max_results=max_results,
98 include_full_content=include_full_content,
99 search_snippets_only=search_snippets_only,
100 settings_snapshot=settings_snapshot,
101 **kwargs,
102 )
104 self.topic = topic if topic in ("general", "news") else "general"
105 self.freshness = freshness or None
106 self.include_domains = self._ensure_list(include_domains)[
107 : self.MAX_DOMAINS
108 ]
109 self.exclude_domains = self._ensure_list(exclude_domains)[
110 : self.MAX_DOMAINS
111 ]
113 # Resolve API key from params, settings snapshot, or env var.
114 self.api_key = self._resolve_api_key(
115 api_key,
116 "search.engine.web.sofya.api_key",
117 engine_name="Sofya",
118 settings_snapshot=settings_snapshot,
119 )
121 def _build_payload(self, query: str) -> Dict[str, Any]:
122 """Build the request body for the Sofya ``/search`` endpoint."""
123 # "basic" depth (3 credits) returns extracted page content; "snippets"
124 # (1 credit) returns SERP snippets only. Only pay for content when it
125 # will actually be consumed: run() calls _get_full_content (which
126 # promotes the content) only when search_snippets_only is False.
127 wants_content = (
128 self.include_full_content and not self.search_snippets_only
129 )
130 payload: Dict[str, Any] = {
131 "query": query,
132 "search_depth": "basic" if wants_content else "snippets",
133 "max_results": min(self.MAX_RESULTS_CAP, self.max_results),
134 "topic": self.topic,
135 }
136 if self.freshness:
137 payload["freshness"] = self.freshness
138 if self.include_domains:
139 payload["include_domains"] = self.include_domains
140 if self.exclude_domains:
141 payload["exclude_domains"] = self.exclude_domains
142 return payload
144 def _get_previews(self, query: str) -> List[Dict[str, Any]]:
145 """Get preview information from the Sofya search API.
147 Args:
148 query: The search query.
150 Returns:
151 List of preview dictionaries.
152 """
153 logger.info("Getting search results from Sofya")
155 try:
156 # Apply rate limiting before the request.
157 self._last_wait_time = self.rate_tracker.apply_rate_limit(
158 self.engine_type
159 )
161 response = safe_post(
162 f"{self.BASE_URL}/search",
163 json=self._build_payload(query),
164 headers={
165 "Authorization": f"Bearer {self.api_key}",
166 "Content-Type": "application/json",
167 },
168 timeout=self.REQUEST_TIMEOUT,
169 )
171 # Detect rate limiting from the status code.
172 self._raise_if_rate_limit(response.status_code)
173 response.raise_for_status()
175 data = response.json()
176 results = data.get("results", []) or []
178 previews = []
179 for i, result in enumerate(results):
180 url = result.get("url", "")
181 # Keep the short SERP ``description`` as the preview snippet
182 # (used for relevance filtering and display); fall back to a
183 # slice of the extracted content when no description is
184 # provided. The extracted page content itself stays in the
185 # ``_full_result`` stash until _get_full_content promotes it,
186 # so snippets-only mode never carries full page text.
187 snippet = result.get("description") or ""
188 if not snippet:
189 snippet = (result.get("content") or "")[
190 : self.SNIPPET_PREVIEW_CHARS
191 ]
193 preview = {
194 "id": url or str(i),
195 "title": result.get("title", ""),
196 "link": url,
197 "snippet": snippet,
198 "displayed_link": url,
199 "position": i,
200 "published_date": result.get("published_date"),
201 # Store the full Sofya result for _get_full_content().
202 "_full_result": result,
203 }
204 previews.append(preview)
206 self._search_results = previews
207 return previews
209 except RateLimitError:
210 raise # Re-raise rate limit errors
211 except requests.exceptions.RequestException as e:
212 safe_msg = self._scrub_error(e)
213 logger.warning(f"Error getting Sofya results: {safe_msg}")
214 # Re-check for rate limiting using the response status code, not
215 # the exception object, so _raise_if_rate_limit gets an int.
216 status_code = getattr(
217 getattr(e, "response", None), "status_code", None
218 )
219 if status_code is not None:
220 self._raise_if_rate_limit(status_code)
221 return []
222 except Exception as e:
223 safe_msg = self._scrub_error(e)
224 logger.warning(
225 f"Unexpected error getting Sofya results: {safe_msg}"
226 )
227 return []
229 def _get_full_content(
230 self, relevant_items: List[Dict[str, Any]]
231 ) -> List[Dict[str, Any]]:
232 """Return full content for the relevant results.
234 Sofya returns the extracted page text inline (``basic`` depth), and
235 ``_get_previews`` stashes it on each preview under ``_full_result``.
236 Promote it here to ``full_content`` (the field the report's citation
237 handler reads before falling back to ``snippet``) while dropping the
238 internal ``_full_result`` marker and preserving the preview fields,
239 notably ``link``, which the citation handler uses as the source URL.
240 Falls back to the snippet when no extracted content is available
241 (e.g. ``snippets`` depth), so ``full_content`` is always usable.
243 Args:
244 relevant_items: List of relevant preview dictionaries.
246 Returns:
247 List of result dictionaries with full content where available.
248 """
249 results = []
250 for item in relevant_items:
251 result = {k: v for k, v in item.items() if k != "_full_result"}
252 content = (item.get("_full_result") or {}).get("content") or ""
253 result["full_content"] = (
254 content
255 or result.get("full_content")
256 or result.get("snippet", "")
257 )
258 results.append(result)
259 return results