Coverage for src/local_deep_research/web_search_engines/engines/search_engine_semantic_scholar.py: 97%
292 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-22 21:20 +0000
1import re
2from typing import Any, Dict, List, Optional, Tuple
4import requests
5from langchain_core.language_models import BaseLLM
6from ...security.secure_logging import logger
7from requests.adapters import HTTPAdapter
8from urllib3.util import Retry
10from ...constants import SNIPPET_LENGTH_SHORT
11from ...settings.exceptions import SettingsValueError
12from ..rate_limiting import RateLimitError
13from ..search_engine_base import BaseSearchEngine, Exposure, Sensitivity
14from ...security import SafeSession
15from ...utilities.llm_utils import invoke_llm_sync
18# Longest Retry-After wait honored per HTTP retry.
19SEMANTIC_SCHOLAR_RETRY_AFTER_MAX_SECONDS = 30
22class SemanticScholarSearchEngine(BaseSearchEngine):
23 """
24 Semantic Scholar search engine implementation with two-phase approach.
25 Provides efficient access to scientific literature across all fields.
26 """
28 # Mark as public search engine
29 is_public = True
30 egress_sensitivity = Sensitivity.NON_SENSITIVE
31 egress_exposure = Exposure.EXPOSING
32 # Scientific/academic search engine
33 is_scientific = True
34 is_lexical = True
35 needs_llm_relevance_filter = True
36 # An invalid x-api-key gets 403 {"message":"Forbidden"} from its API
37 # gateway, without WWW-Authenticate. The key is sent on every request.
38 bare_403_means_bad_api_key = True
40 def __init__(
41 self,
42 max_results: int = 10,
43 api_key: Optional[str] = None,
44 year_range: Optional[Tuple[int, int]] = None,
45 get_abstracts: bool = True,
46 get_references: bool = False,
47 get_citations: bool = False,
48 get_embeddings: bool = False,
49 get_tldr: bool = True,
50 citation_limit: int = 10,
51 reference_limit: int = 10,
52 llm: Optional[BaseLLM] = None,
53 max_filtered_results: Optional[int] = None,
54 optimize_queries: bool = True,
55 max_retries: int = 5,
56 retry_backoff_factor: float = 1.0,
57 fields_of_study: Optional[List[str]] = None,
58 publication_types: Optional[List[str]] = None,
59 settings_snapshot: Optional[Dict[str, Any]] = None,
60 **kwargs,
61 ):
62 """
63 Initialize the Semantic Scholar search engine.
65 Args:
66 max_results: Maximum number of search results
67 api_key: Semantic Scholar API key for higher rate limits (optional)
68 year_range: Optional tuple of (start_year, end_year) to filter results
69 get_abstracts: Whether to fetch abstracts for all results
70 get_references: Whether to fetch references for papers
71 get_citations: Whether to fetch citations for papers
72 get_embeddings: Whether to fetch SPECTER embeddings for papers
73 get_tldr: Whether to fetch TLDR summaries for papers
74 citation_limit: Maximum number of citations to fetch per paper
75 reference_limit: Maximum number of references to fetch per paper
76 llm: Language model for relevance filtering
77 max_filtered_results: Maximum number of results to keep after filtering
78 optimize_queries: Whether to optimize natural language queries
79 max_retries: Maximum number of retries for API requests
80 retry_backoff_factor: Backoff factor for retries
81 fields_of_study: List of fields of study to filter results
82 publication_types: List of publication types to filter results
83 settings_snapshot: Settings snapshot for configuration
84 **kwargs: Additional parameters to pass to parent class
85 """
86 # Journal filter runs before LLM relevance (Tiers 1-3 are instant)
87 preview_filters = []
88 journal_filter = self._create_journal_filter(
89 "semantic_scholar", llm, settings_snapshot
90 )
91 if journal_filter is not None: 91 ↛ 92line 91 didn't jump to line 92 because the condition on line 91 was never true
92 preview_filters.append(journal_filter)
94 super().__init__(
95 llm=llm,
96 max_filtered_results=max_filtered_results,
97 max_results=max_results,
98 preview_filters=preview_filters, # type: ignore[arg-type]
99 settings_snapshot=settings_snapshot,
100 **kwargs,
101 )
103 # Get API key from settings if not provided
104 if not api_key and settings_snapshot:
105 from ...config.search_config import get_setting_from_snapshot
107 try:
108 api_key = get_setting_from_snapshot(
109 "search.engine.web.semantic_scholar.api_key",
110 settings_snapshot=settings_snapshot,
111 )
112 except SettingsValueError:
113 raise
114 except Exception:
115 logger.debug(
116 "Failed to read semantic_scholar.api_key from settings snapshot",
117 exc_info=True,
118 )
120 self.api_key = api_key
121 self.year_range = year_range
122 self.get_abstracts = get_abstracts
123 self.get_references = get_references
124 self.get_citations = get_citations
125 self.get_embeddings = get_embeddings
126 self.get_tldr = get_tldr
127 self.citation_limit = citation_limit
128 self.reference_limit = reference_limit
129 self.optimize_queries = optimize_queries
130 self.max_retries = max_retries
131 self.retry_backoff_factor = retry_backoff_factor
132 self.fields_of_study = (
133 self._ensure_list(fields_of_study)
134 if fields_of_study is not None
135 else None
136 )
137 self.publication_types = (
138 self._ensure_list(publication_types)
139 if publication_types is not None
140 else None
141 )
143 # Base API URLs
144 self.base_url = "https://api.semanticscholar.org/graph/v1"
145 self.paper_search_url = f"{self.base_url}/paper/search"
146 self.paper_details_url = f"{self.base_url}/paper"
148 # Create a session with retry capabilities
149 self.session: SafeSession | None = self._create_session()
151 # Log API key status
152 if self.api_key:
153 logger.info(
154 "Using Semantic Scholar with API key (higher rate limits)"
155 )
156 else:
157 logger.info(
158 "Using Semantic Scholar without API key (lower rate limits)"
159 )
161 def _create_session(self) -> SafeSession:
162 """Create and configure a requests session with retry capabilities"""
163 session = SafeSession()
165 # Configure automatic retries with exponential backoff
166 retry_strategy = Retry(
167 total=self.max_retries,
168 backoff_factor=self.retry_backoff_factor,
169 status_forcelist=[429, 500, 502, 503, 504],
170 allowed_methods={"HEAD", "GET", "POST", "OPTIONS"},
171 # urllib3 otherwise honors Retry-After up to six hours per retry
172 # while the agent's tool call blocks. Cap each wait; after the
173 # retries, the research-level rate-limit cooldown takes over.
174 retry_after_max=SEMANTIC_SCHOLAR_RETRY_AFTER_MAX_SECONDS,
175 )
177 adapter = HTTPAdapter(max_retries=retry_strategy)
178 session.mount("https://", adapter)
180 # Set up headers
181 headers = {"Accept": "application/json"}
182 if self.api_key:
183 headers["x-api-key"] = self.api_key
185 session.headers.update(headers)
187 return session
189 def close(self):
190 """
191 Close the HTTP session and clean up resources.
193 Call this method when done using the search engine to prevent
194 connection/file descriptor leaks.
195 """
196 if hasattr(self, "session") and self.session:
197 try:
198 self.session.close()
199 except Exception as e:
200 safe_msg = self._scrub_error(e)
201 logger.warning(
202 f"Error closing SemanticScholar session: {safe_msg}"
203 )
204 finally:
205 self.session = None
206 # Close content filters (JournalReputationFilter) via parent
207 super().close()
209 def __del__(self):
210 """Destructor to ensure session is closed."""
211 self.close()
213 def __enter__(self):
214 """Context manager entry."""
215 return self
217 def __exit__(self, exc_type, exc_val, exc_tb):
218 """Context manager exit - ensures session cleanup."""
219 self.close()
220 return False
222 def _respect_rate_limit(self):
223 """Apply rate limiting between requests"""
224 # Apply rate limiting before request
225 self._last_wait_time = self.rate_tracker.apply_rate_limit(
226 self.engine_type
227 )
228 logger.debug(f"Applied rate limit wait: {self._last_wait_time:.2f}s")
230 def _make_request(
231 self,
232 url: str,
233 params: Optional[Dict] = None,
234 data: Optional[Dict] = None,
235 method: str = "GET",
236 ) -> Dict:
237 """
238 Make a request to the Semantic Scholar API.
240 Args:
241 url: API endpoint URL
242 params: Query parameters
243 data: JSON data for POST requests
244 method: HTTP method (GET or POST)
246 Returns:
247 API response as dictionary
248 """
249 self._respect_rate_limit()
251 try:
252 if self.session is None: 252 ↛ 253line 252 didn't jump to line 253 because the condition on line 252 was never true
253 raise RuntimeError("Session is not initialized")
254 if method.upper() == "GET":
255 response = self.session.get(url, params=params, timeout=30)
256 elif method.upper() == "POST":
257 response = self.session.post(
258 url, params=params, json=data, timeout=30
259 )
260 else:
261 raise ValueError(f"Unsupported HTTP method: {method}")
263 # Handle rate limiting
264 if response.status_code == 429:
265 logger.warning("Semantic Scholar rate limit exceeded")
266 raise RateLimitError(
267 "Semantic Scholar rate limit exceeded",
268 retry_after=response.headers.get("Retry-After"),
269 )
271 response.raise_for_status()
272 return response.json() # type: ignore[no-any-return]
273 except requests.RequestException as e:
274 self._record_search_failure(e)
275 safe_msg = self._scrub_error(e)
276 logger.warning(f"API request failed: {safe_msg}")
277 return {}
279 def _optimize_query(self, query: str) -> str:
280 """
281 Optimize a natural language query for Semantic Scholar search.
282 If LLM is available, uses it to extract key terms and concepts.
284 Args:
285 query: Natural language query
287 Returns:
288 Optimized query string
289 """
290 if not self.llm or not self.optimize_queries:
291 return query
293 try:
294 prompt = f"""Transform this natural language question into an optimized academic search query.
296Original query: "{query}"
298INSTRUCTIONS:
2991. Extract key academic concepts, technical terms, and proper nouns
3002. Remove generic words, filler words, and non-technical terms
3013. Add quotation marks around specific phrases that should be kept together
3024. Return ONLY the optimized search query with no explanation
3035. Keep it under 100 characters if possible
305EXAMPLE TRANSFORMATIONS:
306"What are the latest findings about mRNA vaccines and COVID-19?" → "mRNA vaccines COVID-19 recent findings"
307"How does machine learning impact climate change prediction?" → "machine learning "climate change" prediction"
308"Tell me about quantum computing approaches for encryption" → "quantum computing encryption"
310Return ONLY the optimized search query with no explanation.
311"""
313 response = invoke_llm_sync(self.llm, prompt)
314 optimized_query = (
315 str(response.content)
316 if hasattr(response, "content")
317 else str(response)
318 ).strip()
320 # Clean up the query - remove any explanations
321 lines = optimized_query.split("\n")
322 optimized_query = lines[0].strip()
324 # Safety check - if query looks too much like an explanation, use original
325 if len(optimized_query.split()) > 15 or ":" in optimized_query:
326 logger.warning(
327 "Query optimization result looks too verbose, using original"
328 )
329 return query
331 logger.info(
332 f"Optimized query for Semantic Scholar "
333 f"({len(query)} -> {len(optimized_query)} chars)"
334 )
336 return optimized_query
337 except Exception as e:
338 safe_msg = self._scrub_error(e)
339 logger.warning(f"Error optimizing query: {safe_msg}")
340 return query # Fall back to original query on error
342 def _direct_search(self, query: str) -> List[Dict[str, Any]]:
343 """
344 Make a direct search request to the Semantic Scholar API.
346 Args:
347 query: The search query
349 Returns:
350 List of paper dictionaries
351 """
352 self.last_search_failure = None
353 try:
354 # Configure fields to retrieve
355 fields = [
356 "paperId",
357 "externalIds",
358 "url",
359 "title",
360 "abstract",
361 "venue",
362 "publicationVenue", # Structured venue with name/type/ISSN
363 "year",
364 "authors",
365 "citationCount", # Add citation count for ranking
366 "openAccessPdf", # PDF URL for open access papers
367 ]
369 if self.get_tldr:
370 fields.append("tldr")
372 params = {
373 "query": query,
374 "limit": min(
375 self.max_results, 100
376 ), # API limit is 100 per request
377 "fields": ",".join(fields),
378 }
380 # Add year filter if specified
381 if self.year_range:
382 start_year, end_year = self.year_range
383 params["year"] = f"{start_year}-{end_year}"
385 # Add fields of study filter if specified
386 if self.fields_of_study:
387 params["fieldsOfStudy"] = ",".join(self.fields_of_study)
389 # Add publication types filter if specified
390 if self.publication_types:
391 params["publicationTypes"] = ",".join(self.publication_types)
393 response = self._make_request(self.paper_search_url, params)
395 if "data" in response:
396 papers = response["data"]
397 logger.info(f"Found {len(papers)} papers with direct search")
398 return papers # type: ignore[no-any-return]
399 logger.warning("No data in response for direct search")
400 return []
402 except Exception as e:
403 self._record_search_failure(e)
404 safe_msg = self._scrub_error(e)
405 logger.warning(f"Error in direct search: {safe_msg}")
406 return []
408 def _adaptive_search(self, query: str) -> Tuple[List[Dict[str, Any]], str]:
409 """
410 Perform an adaptive search that adjusts based on result volume.
411 Uses LLM to generate better fallback queries when available.
413 Args:
414 query: The search query
416 Returns:
417 Tuple of (list of paper results, search strategy used)
418 """
419 # Start with a standard search
420 papers = self._direct_search(query)
421 strategy = "standard"
423 # If no results, try different variations
424 if not papers:
425 # Try removing quotes to broaden search
426 if '"' in query:
427 unquoted_query = query.replace('"', "")
428 logger.info(
429 "No results with quoted terms, retrying without quotes"
430 )
431 papers = self._direct_search(unquoted_query)
433 if papers:
434 strategy = "unquoted"
435 return papers, strategy
437 # If LLM is available, use it to generate better fallback queries
438 if self.llm:
439 try:
440 # Generate alternate search queries focusing on core concepts
441 prompt = f"""You are helping refine a search query that returned no results.
443Original query: "{query}"
445The query might be too specific or use natural language phrasing that doesn't match academic paper keywords.
447Please provide THREE alternative search queries that:
4481. Focus on the core academic concepts
4492. Use precise terminology commonly found in academic papers
4503. Break down complex queries into more searchable components
4514. Format each as a concise keyword-focused search term (not a natural language question)
453Format each query on a new line with no numbering or explanation. Keep each query under 8 words and very focused.
454"""
455 # Get the LLM's response
456 response = invoke_llm_sync(self.llm, prompt)
458 # Extract the alternative queries
459 alt_queries = []
460 if hasattr(
461 response, "content"
462 ): # Handle various LLM response formats
463 content = response.content
464 alt_queries = [
465 q.strip()
466 for q in content.strip().split("\n")
467 if q.strip()
468 ]
469 elif isinstance(response, str): 469 ↛ 477line 469 didn't jump to line 477 because the condition on line 469 was always true
470 alt_queries = [
471 q.strip()
472 for q in response.strip().split("\n")
473 if q.strip()
474 ]
476 # Try each alternative query
477 for alt_query in alt_queries[
478 :3
479 ]: # Limit to first 3 alternatives
480 logger.info("Trying LLM-suggested alternative query")
481 alt_papers = self._direct_search(alt_query)
483 if alt_papers:
484 logger.info(
485 "Found {} papers using the LLM-suggested query",
486 len(alt_papers),
487 )
488 strategy = "llm_alternative"
489 return alt_papers, strategy
490 except Exception as e:
491 safe_msg = self._scrub_error(e)
492 logger.warning(
493 f"Error using LLM for query refinement: {safe_msg}"
494 )
495 # Fall through to simpler strategies
497 # Fallback: Try with the longest words (likely specific terms)
498 words = re.findall(r"\w+", query)
499 longer_words = [word for word in words if len(word) > 6]
500 if longer_words:
501 # Use up to 3 of the longest words
502 longer_words = sorted(longer_words, key=len, reverse=True)[:3]
503 key_terms_query = " ".join(longer_words)
504 logger.info("Trying with key terms extracted from the query")
505 papers = self._direct_search(key_terms_query)
507 if papers:
508 strategy = "key_terms"
509 return papers, strategy
511 # Final fallback: Try with just the longest word
512 if words: 512 ↛ 522line 512 didn't jump to line 522 because the condition on line 512 was always true
513 longest_word = max(words, key=len)
514 if len(longest_word) > 5: # Only use if it's reasonably long
515 logger.info("Trying with the single longest key term")
516 papers = self._direct_search(longest_word)
518 if papers:
519 strategy = "single_term"
520 return papers, strategy
522 return papers, strategy
524 def _get_paper_details(self, paper_id: str) -> Dict[str, Any]:
525 """
526 Get detailed information about a specific paper.
528 Args:
529 paper_id: Semantic Scholar Paper ID
531 Returns:
532 Dictionary with paper details
533 """
534 try:
535 # Construct fields parameter
536 fields = [
537 "paperId",
538 "externalIds",
539 "corpusId",
540 "url",
541 "title",
542 "abstract",
543 "venue",
544 "publicationVenue", # Structured venue with name/type/ISSN
545 "year",
546 "authors",
547 "fieldsOfStudy",
548 "citationCount", # Add citation count
549 ]
551 if self.get_tldr:
552 fields.append("tldr")
554 if self.get_embeddings:
555 fields.append("embedding")
557 # Add citation and reference fields if requested
558 if self.get_citations:
559 fields.append(f"citations.limit({self.citation_limit})")
561 if self.get_references:
562 fields.append(f"references.limit({self.reference_limit})")
564 # Make the request
565 url = f"{self.paper_details_url}/{paper_id}"
566 params = {"fields": ",".join(fields)}
568 return self._make_request(url, params)
570 except Exception as e:
571 safe_msg = self._scrub_error(e)
572 logger.warning(f"Error getting paper details: {safe_msg}")
573 return {}
575 def _get_previews(self, query: str) -> List[Dict[str, Any]]:
576 """
577 Get preview information for Semantic Scholar papers.
579 Args:
580 query: The search query
582 Returns:
583 List of preview dictionaries
584 """
585 logger.info("Getting Semantic Scholar previews")
587 # Optimize the query if LLM is available
588 optimized_query = self._optimize_query(query)
590 # Use the adaptive search approach
591 papers, strategy = self._adaptive_search(optimized_query)
593 if not papers:
594 logger.warning("No Semantic Scholar results found")
595 return []
597 # Format as previews
598 previews = []
599 for paper in papers:
600 try:
601 # Format authors - ensure we have a valid list with string values
602 authors = []
603 if paper.get("authors"):
604 authors = [
605 author.get("name", "")
606 for author in paper["authors"]
607 if author and author.get("name")
608 ]
610 # Ensure we have valid strings for all fields
611 paper_id = paper.get("paperId", "")
612 title = paper.get("title", "")
613 url = paper.get("url", "")
615 # Handle abstract safely, ensuring we always have a string
616 abstract = paper.get("abstract")
617 snippet = ""
618 if abstract:
619 snippet = (
620 abstract[:SNIPPET_LENGTH_SHORT] + "..."
621 if len(abstract) > SNIPPET_LENGTH_SHORT
622 else abstract
623 )
625 # Prefer publicationVenue (structured, with ISSN) over
626 # venue (plain string, often empty for many papers).
627 pub_venue = paper.get("publicationVenue") or {}
628 venue_name = pub_venue.get("name") or paper.get("venue", "")
629 venue_issn = pub_venue.get("issn")
631 year = paper.get("year")
632 external_ids = paper.get("externalIds", {})
634 # Handle TLDR safely
635 tldr_text = ""
636 if paper.get("tldr") and isinstance(paper.get("tldr"), dict):
637 tldr_text = paper.get("tldr", {}).get("text", "")
639 # Create preview with basic information, ensuring no None values
640 preview = {
641 "id": paper_id if paper_id else "",
642 "title": title if title else "",
643 "link": url if url else "",
644 "snippet": snippet,
645 "authors": authors,
646 "venue": venue_name if venue_name else "",
647 "journal_ref": venue_name if venue_name else None,
648 "issn": venue_issn,
649 "year": year,
650 "external_ids": external_ids if external_ids else {},
651 "source": "Semantic Scholar",
652 "_paper_id": paper_id if paper_id else "",
653 "_search_strategy": strategy,
654 "tldr": tldr_text,
655 }
657 # Store the full paper object for later reference
658 preview["_full_paper"] = paper
660 previews.append(preview)
661 except Exception as e:
662 safe_msg = self._scrub_error(e)
663 logger.warning(f"Error processing paper preview: {safe_msg}")
664 # Continue with the next paper
666 # Sort by year (newer first) if available
667 def _year_key(p: dict[str, Any]) -> int:
668 year = p.get("year")
669 try:
670 return int(year) if year is not None else 0
671 except (TypeError, ValueError):
672 return 0
674 previews = sorted(previews, key=_year_key, reverse=True)
676 logger.info(
677 f"Found {len(previews)} Semantic Scholar previews using strategy: {strategy}"
678 )
679 return previews
681 def _get_full_content(
682 self, relevant_items: List[Dict[str, Any]]
683 ) -> List[Dict[str, Any]]:
684 """
685 Get full content for the relevant Semantic Scholar papers.
686 Gets additional details like citations, references, and full metadata.
688 Args:
689 relevant_items: List of relevant preview dictionaries
691 Returns:
692 List of result dictionaries with full content
693 """
694 # For Semantic Scholar, we already have most content from the preview
695 # Additional API calls are only needed for citations/references
697 logger.info(
698 f"Getting content for {len(relevant_items)} Semantic Scholar papers"
699 )
701 results = []
702 for item in relevant_items:
703 result = item.copy()
704 paper_id = item.get("_paper_id", "")
706 # Skip if no paper ID
707 if not paper_id:
708 results.append(result)
709 continue
711 # Get paper details if citations or references are requested
712 if self.get_citations or self.get_references or self.get_embeddings:
713 paper_details = self._get_paper_details(paper_id)
715 if paper_details:
716 # Add citation information
717 if self.get_citations and "citations" in paper_details:
718 result["citations"] = paper_details["citations"]
720 # Add reference information
721 if self.get_references and "references" in paper_details:
722 result["references"] = paper_details["references"]
724 # Add embedding if available
725 if self.get_embeddings and "embedding" in paper_details:
726 result["embedding"] = paper_details["embedding"]
728 # Add fields of study
729 if "fieldsOfStudy" in paper_details:
730 result["fields_of_study"] = paper_details[
731 "fieldsOfStudy"
732 ]
734 # Promote useful fields from _full_paper to top level before
735 # dropping the raw paper (consistent with NASA ADS/OpenAlex which
736 # expose citations/journal_ref at the top level).
737 full_paper = result.get("_full_paper") or {}
738 if ( 738 ↛ 742line 738 didn't jump to line 742 because the condition on line 738 was never true
739 full_paper.get("citationCount") is not None
740 and "citations" not in result
741 ):
742 result["citations"] = full_paper.get("citationCount")
744 # Remove temporary fields
745 if "_paper_id" in result: 745 ↛ 747line 745 didn't jump to line 747 because the condition on line 745 was always true
746 del result["_paper_id"]
747 if "_search_strategy" in result: 747 ↛ 749line 747 didn't jump to line 749 because the condition on line 747 was always true
748 del result["_search_strategy"]
749 if "_full_paper" in result: 749 ↛ 752line 749 didn't jump to line 752 because the condition on line 749 was always true
750 del result["_full_paper"]
752 results.append(result)
754 return results