Coverage for src/local_deep_research/web_search_engines/engines/search_engine_pubmed.py: 94%
733 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
4from defusedxml import ElementTree as ET
6from langchain_core.language_models import BaseLLM
8from ...constants import SNIPPET_LENGTH_LONG
9from ...security.safe_requests import safe_get
10from ...security.secure_logging import logger
11from ..rate_limiting import RateLimitError
12from ..engine_availability import retry_after_from_error
13from ..search_engine_base import BaseSearchEngine, Exposure, Sensitivity
14from ...utilities.llm_utils import invoke_llm_sync
17class PubMedSearchEngine(BaseSearchEngine):
18 """
19 PubMed search engine implementation with two-phase approach and adaptive search.
20 Provides efficient access to biomedical literature while minimizing API usage.
21 """
23 # Mark as public search engine
24 is_public = True
25 egress_sensitivity = Sensitivity.NON_SENSITIVE
26 egress_exposure = Exposure.EXPOSING
27 # Scientific/medical search engine
28 is_scientific = True
29 is_lexical = True
30 needs_llm_relevance_filter = True
32 def __init__(
33 self,
34 max_results: int = 10,
35 api_key: Optional[str] = None,
36 days_limit: Optional[int] = None,
37 get_abstracts: bool = True,
38 get_full_text: bool = False,
39 full_text_limit: int = 3,
40 llm: Optional[BaseLLM] = None,
41 max_filtered_results: Optional[int] = None,
42 optimize_queries: bool = True,
43 include_publication_type_in_context: bool = True,
44 include_journal_in_context: bool = True,
45 include_year_in_context: bool = True,
46 include_authors_in_context: bool = False,
47 include_full_date_in_context: bool = False,
48 include_mesh_terms_in_context: bool = True,
49 include_keywords_in_context: bool = True,
50 include_doi_in_context: bool = False,
51 include_pmid_in_context: bool = False,
52 include_pmc_availability_in_context: bool = False,
53 max_mesh_terms: int = 3,
54 max_keywords: int = 3,
55 include_citation_in_context: bool = False,
56 include_language_in_context: bool = False,
57 settings_snapshot: Optional[Dict[str, Any]] = None,
58 ):
59 """
60 Initialize the PubMed search engine.
62 Args:
63 max_results: Maximum number of search results
64 api_key: NCBI API key for higher rate limits (optional)
65 days_limit: Limit results to N days (optional)
66 get_abstracts: Whether to fetch abstracts for all results
67 get_full_text: Whether to fetch full text content (when available in PMC)
68 full_text_limit: Max number of full-text articles to retrieve
69 llm: Language model for relevance filtering
70 max_filtered_results: Maximum number of results to keep after filtering
71 optimize_queries: Whether to optimize natural language queries for PubMed
72 """
73 # Wire up the journal reputation filter as a preview filter so
74 # results are scored against bundled OpenAlex/DOAJ/predatory data
75 # before the (more expensive) LLM relevance pass.
76 preview_filters = []
77 journal_filter = self._create_journal_filter(
78 "pubmed", llm, settings_snapshot
79 )
80 if journal_filter is not None: 80 ↛ 81line 80 didn't jump to line 81 because the condition on line 80 was never true
81 preview_filters.append(journal_filter)
83 # Initialize the BaseSearchEngine with LLM, max_filtered_results, and max_results
84 super().__init__(
85 llm=llm,
86 max_filtered_results=max_filtered_results,
87 max_results=max_results,
88 preview_filters=preview_filters, # type: ignore[arg-type]
89 settings_snapshot=settings_snapshot,
90 )
91 self.max_results = max(self.max_results, 25)
92 self.api_key = api_key
93 self.days_limit = days_limit
94 self.get_abstracts = get_abstracts
95 self.get_full_text = get_full_text
96 self.full_text_limit = full_text_limit
97 self.optimize_queries = optimize_queries
98 self.include_publication_type_in_context = (
99 include_publication_type_in_context
100 )
101 self.include_journal_in_context = include_journal_in_context
102 self.include_year_in_context = include_year_in_context
103 self.include_authors_in_context = include_authors_in_context
104 self.include_full_date_in_context = include_full_date_in_context
105 self.include_mesh_terms_in_context = include_mesh_terms_in_context
106 self.include_keywords_in_context = include_keywords_in_context
107 self.include_doi_in_context = include_doi_in_context
108 self.include_pmid_in_context = include_pmid_in_context
109 self.include_pmc_availability_in_context = (
110 include_pmc_availability_in_context
111 )
112 self.max_mesh_terms = max_mesh_terms
113 self.max_keywords = max_keywords
114 self.include_citation_in_context = include_citation_in_context
115 self.include_language_in_context = include_language_in_context
117 # Base API URLs
118 self.base_url = "https://eutils.ncbi.nlm.nih.gov/entrez/eutils"
119 self.search_url = f"{self.base_url}/esearch.fcgi"
120 self.summary_url = f"{self.base_url}/esummary.fcgi"
121 self.fetch_url = f"{self.base_url}/efetch.fcgi"
122 self.link_url = f"{self.base_url}/elink.fcgi"
124 # PMC base URL for full text
125 self.pmc_url = "https://www.ncbi.nlm.nih.gov/pmc/articles/"
127 def _get_result_count(self, query: str) -> int:
128 """
129 Get the total number of results for a query without retrieving the results themselves.
131 Args:
132 query: The search query
134 Returns:
135 Total number of matching results
136 """
137 try:
138 # Prepare search parameters
139 params = {
140 "db": "pubmed",
141 "term": query,
142 "retmode": "json",
143 "retmax": 0, # Don't need actual results, just the count
144 }
146 # Add API key if available
147 if self.api_key:
148 params["api_key"] = self.api_key
150 self._last_wait_time = self.rate_tracker.apply_rate_limit(
151 self.engine_type
152 )
154 # Execute search request
155 response = safe_get(self.search_url, params=params)
156 response.raise_for_status()
158 # Parse response
159 data = response.json()
160 count = int(data["esearchresult"]["count"])
162 logger.info("PubMed reports {} total results for the query", count)
163 return count
165 except Exception as e:
166 safe_msg = self._scrub_error(e)
167 logger.warning(
168 f"Error getting result count ({type(e).__name__}): {safe_msg}"
169 )
170 return 0
172 def _extract_core_terms(self, query: str) -> str:
173 """
174 Extract core terms from a complex query for volume estimation.
176 Args:
177 query: PubMed query string
179 Returns:
180 Simplified query with core terms
181 """
182 # Remove field specifications and operators
183 simplified = re.sub(r"\[\w+\]", "", query) # Remove [Field] tags
184 simplified = re.sub(
185 r"\b(AND|OR|NOT)\b", "", simplified
186 ) # Remove operators
188 # Remove quotes and parentheses
189 simplified = (
190 simplified.replace('"', "").replace("(", "").replace(")", "")
191 )
193 # Split by whitespace and join terms with 4+ chars (likely meaningful)
194 terms = [term for term in simplified.split() if len(term) >= 4]
196 # Join with AND to create a basic search
197 return " ".join(terms[:5]) # Limit to top 5 terms
199 def _expand_time_window(self, time_filter: str) -> str:
200 """
201 Expand a time window to get more results.
203 Args:
204 time_filter: Current time filter
206 Returns:
207 Expanded time filter
208 """
209 # Parse current time window
210 import re
212 match = re.match(r'"last (\d+) (\w+)"[pdat]', time_filter)
213 if not match:
214 return '"last 10 years"[pdat]'
216 amount, unit = int(match.group(1)), match.group(2)
218 # Expand based on current unit
219 if unit == "months" or unit == "month":
220 if amount < 6:
221 return '"last 6 months"[pdat]'
222 if amount < 12:
223 return '"last 1 year"[pdat]'
224 return '"last 2 years"[pdat]'
225 if unit == "years" or unit == "year": 225 ↛ 232line 225 didn't jump to line 232 because the condition on line 225 was always true
226 if amount < 2:
227 return '"last 2 years"[pdat]'
228 if amount < 5:
229 return '"last 5 years"[pdat]'
230 return '"last 10 years"[pdat]'
232 return '"last 10 years"[pdat]'
234 def _optimize_query_for_pubmed(self, query: str) -> str:
235 """
236 Optimize a natural language query for PubMed search.
237 Uses LLM to transform questions into effective keyword-based queries.
239 Args:
240 query: Natural language query
242 Returns:
243 Optimized query string for PubMed
244 """
245 if not self.llm or not self.optimize_queries:
246 # Return original query if no LLM available or optimization disabled
247 return query
249 try:
250 # Prompt for query optimization
251 prompt = f"""Transform this natural language question into an optimized PubMed search query.
253Original query: "{query}"
255CRITICAL RULES:
2561. ONLY RETURN THE EXACT SEARCH QUERY - NO EXPLANATIONS, NO COMMENTS
2572. DO NOT wrap the entire query in quotes
2583. DO NOT include ANY date restrictions or year filters
2594. Use parentheses around OR statements: (term1[Field] OR term2[Field])
2605. Use only BASIC MeSH terms - stick to broad categories like "Vaccines"[Mesh]
2616. KEEP IT SIMPLE - use 2-3 main concepts maximum
2627. Focus on Title/Abstract searches for reliability: term[Title/Abstract]
2638. Use wildcards for variations: vaccin*[Title/Abstract]
265EXAMPLE QUERIES:
266✓ GOOD: (mRNA[Title/Abstract] OR "messenger RNA"[Title/Abstract]) AND vaccin*[Title/Abstract]
267✓ GOOD: (influenza[Title/Abstract] OR flu[Title/Abstract]) AND treatment[Title/Abstract]
268✗ BAD: (mRNA[Title/Abstract]) AND "specific disease"[Mesh] AND treatment[Title/Abstract] AND 2023[dp]
269✗ BAD: "Here's a query to find articles about vaccines..."
271Return ONLY the search query without any explanations.
272"""
274 # Get response from LLM
275 response = invoke_llm_sync(self.llm, prompt)
276 raw_response = (
277 str(response.content)
278 if hasattr(response, "content")
279 else str(response)
280 ).strip()
282 # Clean up the query - extract only the actual query and remove any explanations
283 # First check if there are multiple lines and take the first non-empty line
284 lines = raw_response.split("\n")
285 cleaned_lines = [line.strip() for line in lines if line.strip()]
287 if cleaned_lines: 287 ↛ 337line 287 didn't jump to line 337 because the condition on line 287 was always true
288 optimized_query = cleaned_lines[0]
290 # Remove any quotes that wrap the entire query
291 if optimized_query.startswith('"') and optimized_query.endswith(
292 '"'
293 ):
294 optimized_query = optimized_query[1:-1]
296 # Remove any explanation phrases that might be at the beginning
297 explanation_starters = [
298 "here is",
299 "here's",
300 "this query",
301 "the following",
302 ]
303 for starter in explanation_starters:
304 if optimized_query.lower().startswith(starter):
305 # Find the actual query part - typically after a colon
306 colon_pos = optimized_query.find(":")
307 if colon_pos > 0:
308 optimized_query = optimized_query[
309 colon_pos + 1 :
310 ].strip()
312 # Check if the query still seems to contain explanations
313 if (
314 len(optimized_query) > 200
315 or "this query will" in optimized_query.lower()
316 ):
317 # It's probably still an explanation - try to extract just the query part
318 # Look for common patterns in the explanation like parentheses
319 pattern = r"\([^)]+\)\s+AND\s+"
320 import re
322 matches = re.findall(pattern, optimized_query)
323 if matches: 323 ↛ 343line 323 didn't jump to line 343 because the condition on line 323 was always true
324 # Extract just the query syntax parts
325 query_parts = []
326 for part in re.split(r"\.\s+", optimized_query):
327 if (
328 "(" in part
329 and ")" in part
330 and ("AND" in part or "OR" in part)
331 ):
332 query_parts.append(part)
333 if query_parts: 333 ↛ 343line 333 didn't jump to line 343 because the condition on line 333 was always true
334 optimized_query = " ".join(query_parts)
335 else:
336 # Fall back to original query if cleaning fails
337 logger.warning(
338 "Failed to extract a clean query from LLM response"
339 )
340 optimized_query = query
342 # Final safety check - if query looks too much like an explanation, use original
343 if len(optimized_query.split()) > 30:
344 logger.warning(
345 "Query too verbose, falling back to simpler form"
346 )
347 # Create a simple query from the original
348 words = [
349 w
350 for w in query.split()
351 if len(w) > 3
352 and w.lower()
353 not in (
354 "what",
355 "are",
356 "the",
357 "and",
358 "for",
359 "with",
360 "from",
361 "have",
362 "been",
363 "recent",
364 )
365 ]
366 optimized_query = " AND ".join(words[:3])
368 # Basic cleanup: standardize field tag case for consistency
369 import re
371 optimized_query = re.sub(
372 r"\[mesh\]", "[Mesh]", optimized_query, flags=re.IGNORECASE
373 )
374 optimized_query = re.sub(
375 r"\[title/abstract\]",
376 "[Title/Abstract]",
377 optimized_query,
378 flags=re.IGNORECASE,
379 )
380 optimized_query = re.sub(
381 r"\[publication type\]",
382 "[Publication Type]",
383 optimized_query,
384 flags=re.IGNORECASE,
385 )
387 # Fix unclosed quotes followed by field tags
388 # Pattern: "term[Field] -> "term"[Field]
389 # The fallback consumes complete quoted segments. This heuristic
390 # can still mispair quotes after unmatched quotes or inside balanced
391 # bracket-containing phrases. Unclosed phrase bodies may have brackets.
392 optimized_query = re.sub(
393 r'"([^"]+)\[|"[^"]*"',
394 lambda m: f'"{m.group(1)}"[' if m.group(1) else m.group(0),
395 optimized_query,
396 )
398 # Simplify the query if still no results are found
399 self._simplify_query_cache = optimized_query
401 # Log original and optimized queries
402 logger.info(
403 f"Optimized query for PubMed "
404 f"({len(query)} -> {len(optimized_query)} chars)"
405 )
406 logger.debug("Query optimization complete")
408 return optimized_query
410 except Exception as e:
411 safe_msg = self._scrub_error(e)
412 logger.exception(
413 f"Error optimizing query ({type(e).__name__}): {safe_msg}"
414 )
415 logger.debug("Falling back to the original query")
416 return query # Fall back to original query on error
418 def _simplify_query(self, query: str) -> str:
419 """
420 Simplify a PubMed query that returned no results.
421 Progressively removes elements to get a more basic query.
423 Args:
424 query: The original query that returned no results
426 Returns:
427 Simplified query
428 """
429 logger.info(f"Simplifying PubMed query ({len(query)} chars)")
430 logger.debug("Query simplification started")
432 # Simple approach: remove field restrictions to broaden the search
433 import re
435 # Remove field tags to make search broader
436 simplified = query
438 # Remove [Mesh] tags - search in all fields instead
439 simplified = re.sub(r"\[Mesh\]", "", simplified, flags=re.IGNORECASE)
441 # Remove [Publication Type] tags
442 simplified = re.sub(
443 r"\[Publication Type\]", "", simplified, flags=re.IGNORECASE
444 )
446 # Keep [Title/Abstract] as it's usually helpful
447 # Clean up any double spaces
448 simplified = re.sub(r"\s+", " ", simplified).strip()
450 # If no simplification was possible, return the original query
451 if simplified == query:
452 logger.debug("No simplification possible, returning original query")
454 logger.info(f"Simplified PubMed query ({len(simplified)} chars)")
455 logger.debug(
456 f"Query simplified from {len(query)} to {len(simplified)} chars"
457 )
458 return simplified
460 def _is_historical_focused(self, query: str) -> bool:
461 """
462 Determine if a query is specifically focused on historical/older information using LLM.
463 Default assumption is that queries should prioritize recent information unless
464 explicitly asking for historical content.
466 Args:
467 query: The search query
469 Returns:
470 Boolean indicating if the query is focused on historical information
471 """
472 if not self.llm:
473 # Fall back to basic keyword check if no LLM available
474 historical_terms = [
475 "history",
476 "historical",
477 "early",
478 "initial",
479 "first",
480 "original",
481 "before",
482 "prior to",
483 "origins",
484 "evolution",
485 "development",
486 ]
487 historical_years = [str(year) for year in range(1900, 2020)]
489 query_lower = query.lower()
490 has_historical_term = any(
491 term in query_lower for term in historical_terms
492 )
493 has_past_year = any(year in query for year in historical_years)
495 return has_historical_term or has_past_year
497 try:
498 # Use LLM to determine if the query is focused on historical information
499 prompt = f"""Determine if this query is specifically asking for HISTORICAL or OLDER information.
501Query: "{query}"
503Answer ONLY "yes" if the query is clearly asking for historical, early, original, or past information from more than 5 years ago.
504Answer ONLY "no" if the query is asking about recent, current, or new information, or if it's a general query without a specific time focus.
506The default assumption should be that medical and scientific queries want RECENT information unless clearly specified otherwise.
507"""
509 response = invoke_llm_sync(self.llm, prompt)
510 answer = (
511 (
512 str(response.content)
513 if hasattr(response, "content")
514 else str(response)
515 )
516 .strip()
517 .lower()
518 )
520 # Log the determination. Only the derived boolean goes to the
521 # log: the prompt above embeds the query, and an LLM that ignores
522 # the "answer ONLY yes/no" instruction can echo it back inside
523 # ``answer``.
524 is_historical = "yes" in answer
525 logger.info(
526 f"Historical focus determination for query ({len(query)} chars)"
527 )
528 logger.info(f"LLM determined historical focus: {is_historical}")
530 return is_historical
532 except Exception as e:
533 safe_msg = self._scrub_error(e)
534 logger.exception(
535 f"Error determining historical focus ({type(e).__name__}): {safe_msg}"
536 )
537 # Fall back to basic keyword check
538 historical_terms = [
539 "history",
540 "historical",
541 "early",
542 "initial",
543 "first",
544 "original",
545 "before",
546 "prior to",
547 "origins",
548 "evolution",
549 "development",
550 ]
551 return any(term in query.lower() for term in historical_terms)
553 def _adaptive_search(self, query: str) -> Tuple[List[str], str]:
554 """
555 Perform an adaptive search that adjusts based on topic volume and whether
556 the query focuses on historical information.
558 Args:
559 query: The search query (already optimized)
561 Returns:
562 Tuple of (list of PMIDs, search strategy used)
563 """
564 # Estimate topic volume
565 estimated_volume = self._get_result_count(query)
567 # Determine if the query is focused on historical information
568 is_historical_focused = self._is_historical_focused(query)
570 if is_historical_focused:
571 # User wants historical information - no date filtering
572 time_filter = None
573 strategy = "historical_focus"
574 elif estimated_volume > 5000:
575 # Very common topic - use tighter recency filter
576 time_filter = '"last 1 year"[pdat]'
577 strategy = "high_volume"
578 elif estimated_volume > 1000:
579 # Common topic
580 time_filter = '"last 3 years"[pdat]'
581 strategy = "common_topic"
582 elif estimated_volume > 100:
583 # Moderate volume
584 time_filter = '"last 5 years"[pdat]'
585 strategy = "moderate_volume"
586 else:
587 # Rare topic - still use recency but with wider range
588 time_filter = '"last 10 years"[pdat]'
589 strategy = "rare_topic"
591 # Run search based on strategy
592 if time_filter:
593 # Try with adaptive time filter
594 query_with_time = f"({query}) AND {time_filter}"
595 logger.info(
596 f"Using adaptive search strategy: {strategy} with filter: {time_filter}"
597 )
598 results = self._search_pubmed(query_with_time)
600 # If too few results, gradually expand time window
601 if len(results) < 5 and '"last 10 years"[pdat]' not in time_filter:
602 logger.info(
603 f"Insufficient results ({len(results)}), expanding time window"
604 )
605 expanded_time = self._expand_time_window(time_filter)
606 query_with_expanded_time = f"({query}) AND {expanded_time}"
607 expanded_results = self._search_pubmed(query_with_expanded_time)
609 if len(expanded_results) > len(results):
610 logger.info(
611 f"Expanded time window yielded {len(expanded_results)} results"
612 )
613 return expanded_results, f"{strategy}_expanded"
615 # If still no results, try without time filter
616 if not results:
617 logger.info(
618 "No results with time filter, trying without time restrictions"
619 )
620 results = self._search_pubmed(query)
621 strategy = "no_time_filter"
622 else:
623 # Historical query - run without time filter
624 logger.info(
625 "Using historical search strategy without date filtering"
626 )
627 results = self._search_pubmed(query)
629 return results, strategy
631 def _search_pubmed(self, query: str) -> List[str]:
632 """
633 Search PubMed and return a list of article IDs.
635 Args:
636 query: The search query
638 Returns:
639 List of PubMed IDs matching the query
640 """
641 self.last_search_failure = None
642 try:
643 # Prepare search parameters
644 params = {
645 "db": "pubmed",
646 "term": query,
647 "retmode": "json",
648 # Cap PMIDs fetched per query so a large user-supplied
649 # max_results can't trigger an unbounded esearch/efetch load.
650 "retmax": min(self.max_results, 100),
651 "usehistory": "y",
652 }
654 # Add API key if available
655 if self.api_key:
656 params["api_key"] = self.api_key
657 logger.debug("Using PubMed API key for higher rate limits")
658 else:
659 logger.debug("No PubMed API key - using default rate limits")
661 # Add date restriction if specified
662 if self.days_limit:
663 params["reldate"] = self.days_limit
664 params["datetype"] = "pdat" # Publication date
665 logger.debug(f"Limiting results to last {self.days_limit} days")
667 logger.debug(
668 f"PubMed search query ({len(query)} chars) with "
669 f"max_results={self.max_results}"
670 )
672 self._last_wait_time = self.rate_tracker.apply_rate_limit(
673 self.engine_type
674 )
675 logger.debug(
676 f"Applied rate limit wait: {self._last_wait_time:.2f}s"
677 )
679 # Execute search request
680 logger.debug(f"Sending request to PubMed API: {self.search_url}")
681 response = safe_get(self.search_url, params=params)
682 response.raise_for_status()
683 logger.debug(f"PubMed API response status: {response.status_code}")
685 # Parse response
686 data = response.json()
687 id_list: list[str] = data["esearchresult"]["idlist"]
688 total_count = data["esearchresult"].get("count", "unknown")
690 logger.info(
691 f"PubMed search found {len(id_list)} results "
692 f"(total available: {total_count})"
693 )
694 if len(id_list) > 0:
695 logger.debug(f"First 5 PMIDs: {id_list[:5]}")
696 return id_list
698 except Exception as e:
699 self._record_search_failure(e)
700 safe_msg = self._scrub_error(e)
701 logger.warning(
702 f"Error searching PubMed ({type(e).__name__}): {safe_msg}"
703 )
704 return []
706 def _get_article_summaries(
707 self, id_list: List[str]
708 ) -> List[Dict[str, Any]]:
709 """
710 Get summaries for a list of PubMed article IDs.
712 Args:
713 id_list: List of PubMed IDs
715 Returns:
716 List of article summary dictionaries
717 """
718 if not id_list:
719 logger.debug("Empty ID list provided to _get_article_summaries")
720 return []
722 logger.debug(f"Fetching summaries for {len(id_list)} PubMed articles")
724 try:
725 # Prepare parameters
726 params = {
727 "db": "pubmed",
728 "id": ",".join(id_list),
729 "retmode": "json",
730 "rettype": "summary",
731 }
733 # Add API key if available
734 if self.api_key:
735 params["api_key"] = self.api_key
737 self._last_wait_time = self.rate_tracker.apply_rate_limit(
738 self.engine_type
739 )
740 logger.debug(
741 f"Applied rate limit wait: {self._last_wait_time:.2f}s"
742 )
744 # Execute request
745 logger.debug(f"Requesting summaries from: {self.summary_url}")
746 response = safe_get(self.summary_url, params=params)
747 response.raise_for_status()
748 logger.debug(f"Summary API response status: {response.status_code}")
750 # Parse response
751 data = response.json()
752 logger.debug(
753 f"PubMed API returned data for {len(id_list)} requested IDs"
754 )
755 summaries = []
757 for pmid in id_list:
758 if pmid in data["result"]: 758 ↛ 806line 758 didn't jump to line 806 because the condition on line 758 was always true
759 article = data["result"][pmid]
760 logger.debug(
761 f"Processing article {pmid}: {article.get('title', 'NO TITLE')[:50]}"
762 )
764 # Extract authors (if available)
765 authors = []
766 if "authors" in article: 766 ↛ 772line 766 didn't jump to line 772 because the condition on line 766 was always true
767 authors = [
768 author["name"] for author in article["authors"]
769 ]
771 # Extract DOI from articleids if not in main field
772 doi = article.get("doi", "")
773 if not doi and "articleids" in article:
774 for aid in article["articleids"]: 774 ↛ 780line 774 didn't jump to line 780 because the loop on line 774 didn't complete
775 if aid.get("idtype") == "doi":
776 doi = aid.get("value", "")
777 break
779 # Create summary dictionary with all available fields
780 summary = {
781 "id": pmid,
782 "title": article.get("title", ""),
783 "pubdate": article.get("pubdate", ""),
784 "epubdate": article.get("epubdate", ""),
785 "source": article.get("source", ""),
786 "authors": authors,
787 "lastauthor": article.get("lastauthor", ""),
788 "journal": article.get("fulljournalname", ""),
789 "volume": article.get("volume", ""),
790 "issue": article.get("issue", ""),
791 "pages": article.get("pages", ""),
792 "doi": doi,
793 "issn": article.get("issn", ""),
794 "essn": article.get("essn", ""),
795 "pubtype": article.get(
796 "pubtype", []
797 ), # Publication types from esummary
798 "recordstatus": article.get("recordstatus", ""),
799 "lang": article.get("lang", []),
800 "pmcrefcount": article.get("pmcrefcount", None),
801 "link": f"https://pubmed.ncbi.nlm.nih.gov/{pmid}/",
802 }
804 summaries.append(summary)
805 else:
806 logger.warning(
807 f"PMID {pmid} not found in PubMed API response"
808 )
810 return summaries
812 except Exception as e:
813 self._record_search_failure(e)
814 error_msg = str(e)
815 safe_msg = self._scrub_error(error_msg)
816 logger.warning(
817 f"Error getting article summaries for {len(id_list)} articles ({type(e).__name__}): {safe_msg}"
818 )
820 # Check for rate limiting patterns
821 if (
822 "429" in error_msg
823 or "too many requests" in error_msg.lower()
824 or "rate limit" in error_msg.lower()
825 or "service unavailable" in error_msg.lower()
826 or "503" in error_msg
827 or "403" in error_msg
828 ):
829 # `from None` suppresses the implicit __context__ chain:
830 # the original exception still carries the raw message, so
831 # a full traceback render (chain=True) would re-leak the
832 # secret that safe_msg just scrubbed.
833 raise RateLimitError(
834 f"PubMed rate limit hit: {safe_msg}",
835 retry_after=retry_after_from_error(e),
836 ) from None
838 return []
840 def _get_article_abstracts(self, id_list: List[str]) -> Dict[str, str]:
841 """
842 Get abstracts for a list of PubMed article IDs.
844 Args:
845 id_list: List of PubMed IDs
847 Returns:
848 Dictionary mapping PubMed IDs to their abstracts
849 """
850 if not id_list:
851 logger.debug("Empty ID list provided to _get_article_abstracts")
852 return {}
854 logger.debug(f"Fetching abstracts for {len(id_list)} PubMed articles")
856 try:
857 # Prepare parameters
858 params = {
859 "db": "pubmed",
860 "id": ",".join(id_list),
861 "retmode": "xml",
862 "rettype": "abstract",
863 }
865 # Add API key if available
866 if self.api_key:
867 params["api_key"] = self.api_key
869 self._last_wait_time = self.rate_tracker.apply_rate_limit(
870 self.engine_type
871 )
872 logger.debug(
873 f"Applied rate limit wait: {self._last_wait_time:.2f}s"
874 )
876 # Execute request
877 logger.debug(f"Requesting abstracts from: {self.fetch_url}")
878 response = safe_get(self.fetch_url, params=params)
879 response.raise_for_status()
880 logger.debug(
881 f"Abstract fetch response status: {response.status_code}, size: {len(response.text)} bytes"
882 )
884 # Parse XML response
885 root = ET.fromstring(response.text)
886 logger.debug(
887 f"Parsing abstracts from XML for {len(id_list)} articles"
888 )
890 # Extract abstracts
891 abstracts = {}
893 for article in root.findall(".//PubmedArticle"):
894 pmid_elem = article.find(".//PMID")
895 pmid = pmid_elem.text if pmid_elem is not None else None
897 if pmid is None:
898 continue
900 # Find abstract text
901 abstract_text = ""
902 abstract_elem = article.find(".//AbstractText")
904 if abstract_elem is not None: 904 ↛ 908line 904 didn't jump to line 908 because the condition on line 904 was always true
905 abstract_text = abstract_elem.text or ""
907 # Some abstracts are split into multiple sections
908 abstract_sections = article.findall(".//AbstractText")
909 if len(abstract_sections) > 1:
910 logger.debug(
911 f"Article {pmid} has {len(abstract_sections)} abstract sections"
912 )
914 for section in abstract_sections:
915 # Get section label if it exists
916 label = section.get("Label")
917 section_text = section.text or ""
919 if label and section_text:
920 if abstract_text: 920 ↛ 923line 920 didn't jump to line 923 because the condition on line 920 was always true
921 abstract_text += f"\n\n{label}: {section_text}"
922 else:
923 abstract_text = f"{label}: {section_text}"
924 elif section_text:
925 if abstract_text: 925 ↛ 928line 925 didn't jump to line 928 because the condition on line 925 was always true
926 abstract_text += f"\n\n{section_text}"
927 else:
928 abstract_text = section_text
930 # Store in dictionary
931 if pmid and abstract_text:
932 abstracts[pmid] = abstract_text
933 logger.debug(
934 f"Abstract for {pmid}: {len(abstract_text)} chars"
935 )
936 elif pmid: 936 ↛ 893line 936 didn't jump to line 893 because the condition on line 936 was always true
937 logger.warning(f"No abstract found for PMID {pmid}")
939 logger.info(
940 f"Successfully retrieved {len(abstracts)} abstracts out of {len(id_list)} requested"
941 )
942 return abstracts
944 except Exception as e:
945 safe_msg = self._scrub_error(e)
946 logger.warning(
947 f"Error getting article abstracts for {len(id_list)} articles ({type(e).__name__}): {safe_msg}"
948 )
949 return {}
951 def _get_article_detailed_metadata(
952 self, id_list: List[str]
953 ) -> Dict[str, Dict[str, Any]]:
954 """
955 Get detailed metadata for PubMed articles including publication types,
956 MeSH terms, keywords, and affiliations.
958 Args:
959 id_list: List of PubMed IDs
961 Returns:
962 Dictionary mapping PubMed IDs to their detailed metadata
963 """
964 if not id_list:
965 return {}
967 try:
968 # Prepare parameters
969 params = {
970 "db": "pubmed",
971 "id": ",".join(id_list),
972 "retmode": "xml",
973 "rettype": "medline",
974 }
976 # Add API key if available
977 if self.api_key:
978 params["api_key"] = self.api_key
980 self._last_wait_time = self.rate_tracker.apply_rate_limit(
981 self.engine_type
982 )
984 # Execute request
985 response = safe_get(self.fetch_url, params=params)
986 response.raise_for_status()
988 # Parse XML response
989 root = ET.fromstring(response.text)
991 metadata = {}
993 for article in root.findall(".//PubmedArticle"):
994 pmid_elem = article.find(".//PMID")
995 pmid = pmid_elem.text if pmid_elem is not None else None
997 if pmid is None: 997 ↛ 998line 997 didn't jump to line 998 because the condition on line 997 was never true
998 continue
1000 article_metadata: Dict[str, Any] = {}
1002 # Extract publication types
1003 pub_types = []
1004 for pub_type in article.findall(".//PublicationType"):
1005 if pub_type.text: 1005 ↛ 1004line 1005 didn't jump to line 1004 because the condition on line 1005 was always true
1006 pub_types.append(pub_type.text)
1007 if pub_types:
1008 article_metadata["publication_types"] = pub_types
1010 # Extract MeSH terms
1011 mesh_terms = []
1012 for mesh in article.findall(".//MeshHeading"):
1013 descriptor = mesh.find(".//DescriptorName")
1014 if descriptor is not None and descriptor.text: 1014 ↛ 1012line 1014 didn't jump to line 1012 because the condition on line 1014 was always true
1015 mesh_terms.append(descriptor.text)
1016 if mesh_terms:
1017 article_metadata["mesh_terms"] = mesh_terms
1019 # Extract keywords
1020 keywords = []
1021 for keyword in article.findall(".//Keyword"):
1022 if keyword.text: 1022 ↛ 1021line 1022 didn't jump to line 1021 because the condition on line 1022 was always true
1023 keywords.append(keyword.text)
1024 if keywords:
1025 article_metadata["keywords"] = keywords
1027 # Extract affiliations
1028 affiliations = []
1029 for affiliation in article.findall(".//Affiliation"):
1030 if affiliation.text: 1030 ↛ 1029line 1030 didn't jump to line 1029 because the condition on line 1030 was always true
1031 affiliations.append(affiliation.text)
1032 if affiliations:
1033 article_metadata["affiliations"] = affiliations
1035 # Extract grant information
1036 grants = []
1037 for grant in article.findall(".//Grant"):
1038 grant_info = {}
1039 grant_id = grant.find(".//GrantID")
1040 if grant_id is not None and grant_id.text: 1040 ↛ 1042line 1040 didn't jump to line 1042 because the condition on line 1040 was always true
1041 grant_info["id"] = grant_id.text
1042 agency = grant.find(".//Agency")
1043 if agency is not None and agency.text: 1043 ↛ 1045line 1043 didn't jump to line 1045 because the condition on line 1043 was always true
1044 grant_info["agency"] = agency.text
1045 if grant_info: 1045 ↛ 1037line 1045 didn't jump to line 1037 because the condition on line 1045 was always true
1046 grants.append(grant_info)
1047 if grants:
1048 article_metadata["grants"] = grants
1050 # Check for free full text in PMC
1051 pmc_elem = article.find(".//ArticleId[@IdType='pmc']")
1052 if pmc_elem is not None:
1053 article_metadata["has_free_full_text"] = True
1054 article_metadata["pmc_id"] = pmc_elem.text
1056 # Extract conflict of interest statement
1057 coi_elem = article.find(".//CoiStatement")
1058 if coi_elem is not None and coi_elem.text:
1059 article_metadata["conflict_of_interest"] = coi_elem.text
1061 metadata[pmid] = article_metadata
1063 return metadata
1065 except Exception as e:
1066 safe_msg = self._scrub_error(e)
1067 logger.warning(
1068 f"Error getting detailed article metadata ({type(e).__name__}): {safe_msg}"
1069 )
1070 return {}
1072 def _create_enriched_content(
1073 self, result: Dict[str, Any], base_content: str
1074 ) -> str:
1075 """
1076 Create enriched content by adding relevant metadata context to help the LLM.
1078 Args:
1079 result: The result dictionary with metadata
1080 base_content: The base content (abstract or full text)
1082 Returns:
1083 Enriched content string with metadata context
1084 """
1085 enriched_parts = []
1087 # Add study type information
1088 if "publication_types" in result:
1089 pub_types = result["publication_types"]
1090 # Filter for significant types
1091 significant_types = [
1092 pt
1093 for pt in pub_types
1094 if any(
1095 key in pt.lower()
1096 for key in [
1097 "clinical trial",
1098 "randomized",
1099 "meta-analysis",
1100 "systematic review",
1101 "case report",
1102 "guideline",
1103 "comparative study",
1104 "multicenter",
1105 ]
1106 )
1107 ]
1108 if significant_types:
1109 enriched_parts.append(
1110 f"[Study Type: {', '.join(significant_types)}]"
1111 )
1113 # Add the main content
1114 enriched_parts.append(base_content)
1116 # Add metadata footer
1117 metadata_footer = []
1119 # Add ALL MeSH terms
1120 if "mesh_terms" in result and len(result["mesh_terms"]) > 0:
1121 metadata_footer.append(
1122 f"Medical Topics (MeSH): {', '.join(result['mesh_terms'])}"
1123 )
1125 # Add ALL keywords
1126 if "keywords" in result and len(result["keywords"]) > 0:
1127 metadata_footer.append(f"Keywords: {', '.join(result['keywords'])}")
1129 # Add ALL affiliations
1130 if "affiliations" in result and len(result["affiliations"]) > 0:
1131 if len(result["affiliations"]) == 1:
1132 metadata_footer.append(
1133 f"Institution: {result['affiliations'][0]}"
1134 )
1135 else:
1136 affiliations_text = "\n - " + "\n - ".join(
1137 result["affiliations"]
1138 )
1139 metadata_footer.append(f"Institutions:{affiliations_text}")
1141 # Add ALL funding information with full details
1142 if "grants" in result and len(result["grants"]) > 0:
1143 grant_details = []
1144 for grant in result["grants"]:
1145 grant_text = []
1146 if "agency" in grant:
1147 grant_text.append(grant["agency"])
1148 if "id" in grant:
1149 grant_text.append(f"(Grant ID: {grant['id']})")
1150 if grant_text:
1151 grant_details.append(" ".join(grant_text))
1152 if grant_details:
1153 if len(grant_details) == 1:
1154 metadata_footer.append(f"Funded by: {grant_details[0]}")
1155 else:
1156 funding_text = "\n - " + "\n - ".join(grant_details)
1157 metadata_footer.append(f"Funding Sources:{funding_text}")
1159 # Add FULL conflict of interest statement
1160 if "conflict_of_interest" in result:
1161 coi_text = result["conflict_of_interest"]
1162 if coi_text:
1163 # Still skip trivial "no conflict" statements to reduce noise
1164 if not any(
1165 phrase in coi_text.lower()
1166 for phrase in [
1167 "no conflict",
1168 "no competing",
1169 "nothing to disclose",
1170 "none declared",
1171 "authors declare no",
1172 ]
1173 ):
1174 metadata_footer.append(f"Conflict of Interest: {coi_text}")
1175 elif (
1176 "but" in coi_text.lower()
1177 or "except" in coi_text.lower()
1178 or "however" in coi_text.lower()
1179 ):
1180 # Include if there's a "no conflict BUT..." type statement
1181 metadata_footer.append(f"Conflict of Interest: {coi_text}")
1183 # Combine everything
1184 if metadata_footer:
1185 enriched_parts.append("\n---\nStudy Metadata:")
1186 enriched_parts.extend(metadata_footer)
1188 return "\n".join(enriched_parts)
1190 def _find_pmc_ids(self, pmid_list: List[str]) -> Dict[str, str]:
1191 """
1192 Find PMC IDs for the given PubMed IDs (for full-text access).
1194 Args:
1195 pmid_list: List of PubMed IDs
1197 Returns:
1198 Dictionary mapping PubMed IDs to their PMC IDs (if available)
1199 """
1200 if not pmid_list or not self.get_full_text:
1201 return {}
1203 try:
1204 # Prepare parameters
1205 params = {
1206 "dbfrom": "pubmed",
1207 "db": "pmc",
1208 "linkname": "pubmed_pmc",
1209 "id": ",".join(pmid_list),
1210 "retmode": "json",
1211 }
1213 # Add API key if available
1214 if self.api_key:
1215 params["api_key"] = self.api_key
1217 self._last_wait_time = self.rate_tracker.apply_rate_limit(
1218 self.engine_type
1219 )
1221 # Execute request
1222 response = safe_get(self.link_url, params=params)
1223 response.raise_for_status()
1225 # Parse response
1226 data = response.json()
1228 # Map PubMed IDs to PMC IDs
1229 pmid_to_pmcid = {}
1231 for linkset in data.get("linksets", []):
1232 pmid = linkset.get("ids", [None])[0]
1234 if not pmid: 1234 ↛ 1235line 1234 didn't jump to line 1235 because the condition on line 1234 was never true
1235 continue
1237 for link in linkset.get("linksetdbs", []):
1238 if link.get("linkname") == "pubmed_pmc": 1238 ↛ 1237line 1238 didn't jump to line 1237 because the condition on line 1238 was always true
1239 pmcids = link.get("links", [])
1240 if pmcids: 1240 ↛ 1237line 1240 didn't jump to line 1237 because the condition on line 1240 was always true
1241 pmid_to_pmcid[str(pmid)] = f"PMC{pmcids[0]}"
1243 logger.info(
1244 f"Found {len(pmid_to_pmcid)} PMC IDs for full-text access"
1245 )
1246 return pmid_to_pmcid
1248 except Exception as e:
1249 safe_msg = self._scrub_error(e)
1250 logger.warning(
1251 f"Error finding PMC IDs ({type(e).__name__}): {safe_msg}"
1252 )
1253 return {}
1255 def _get_pmc_full_text(self, pmcid: str) -> str:
1256 """
1257 Get full text for a PMC article.
1259 Args:
1260 pmcid: PMC ID of the article
1262 Returns:
1263 Full text content or empty string if not available
1264 """
1265 try:
1266 # Prepare parameters
1267 params = {
1268 "db": "pmc",
1269 "id": pmcid,
1270 "retmode": "xml",
1271 "rettype": "full",
1272 }
1274 # Add API key if available
1275 if self.api_key:
1276 params["api_key"] = self.api_key
1278 self._last_wait_time = self.rate_tracker.apply_rate_limit(
1279 self.engine_type
1280 )
1282 # Execute request
1283 response = safe_get(self.fetch_url, params=params)
1284 response.raise_for_status()
1286 # Parse XML response
1287 root = ET.fromstring(response.text)
1289 # Extract full text
1290 full_text = []
1292 # Extract article title
1293 title_elem = root.find(".//article-title")
1294 title_text = (
1295 "".join(title_elem.itertext()).strip()
1296 if title_elem is not None
1297 else ""
1298 )
1299 if title_text: 1299 ↛ 1303line 1299 didn't jump to line 1303 because the condition on line 1299 was always true
1300 full_text.append(f"# {title_text}")
1302 # Extract abstract
1303 abstract_paras = root.findall(".//abstract//p")
1304 if abstract_paras:
1305 full_text.append("\n## Abstract\n")
1306 for p in abstract_paras:
1307 text = "".join(p.itertext())
1308 if text: 1308 ↛ 1306line 1308 didn't jump to line 1306 because the condition on line 1308 was always true
1309 full_text.append(text)
1311 # Extract body content in document order: the paragraphs of a section,
1312 # then its subsections one heading level down. Searching every section
1313 # for `.//p` read a subsection's paragraphs again under each section
1314 # around it, and skipped paragraphs outside any section.
1315 def add_section_content(element, depth):
1316 for child in element:
1317 if child.tag == "sec":
1318 title = child.find("title")
1319 title_text = (
1320 "".join(title.itertext()).strip()
1321 if title is not None
1322 else ""
1323 )
1324 if title_text: 1324 ↛ 1328line 1324 didn't jump to line 1328 because the condition on line 1324 was always true
1325 full_text.append(
1326 f"\n{'#' * min(depth + 2, 6)} {title_text}\n"
1327 )
1328 add_section_content(child, depth + 1)
1329 elif child.tag == "p":
1330 text = "".join(child.itertext())
1331 if text: 1331 ↛ 1316line 1331 didn't jump to line 1316 because the condition on line 1331 was always true
1332 full_text.append(text)
1333 elif child.tag != "title":
1334 # A figure, table, list or box: its paragraphs and any
1335 # section inside it, at the depth of the section around it
1336 add_section_content(child, depth)
1338 body = root.find(".//body")
1339 if body is not None: 1339 ↛ 1342line 1339 didn't jump to line 1342 because the condition on line 1339 was always true
1340 add_section_content(body, 0)
1342 result_text = "\n\n".join(full_text)
1343 logger.debug(
1344 f"Successfully extracted {len(result_text)} chars of PMC full text with {len(full_text)} sections"
1345 )
1346 return result_text
1348 except Exception as e:
1349 safe_msg = self._scrub_error(e)
1350 logger.warning(
1351 f"Error getting PMC full text ({type(e).__name__}): {safe_msg}"
1352 )
1353 return ""
1355 def _get_previews(self, query: str) -> List[Dict[str, Any]]:
1356 """
1357 Get preview information for PubMed articles.
1359 Args:
1360 query: The search query
1362 Returns:
1363 List of preview dictionaries
1364 """
1365 logger.info("Getting PubMed previews")
1367 # Optimize the query for PubMed if LLM is available
1368 optimized_query = self._optimize_query_for_pubmed(query)
1370 # Perform adaptive search
1371 pmid_list, strategy = self._adaptive_search(optimized_query)
1373 # If no results, try a simplified query
1374 if not pmid_list:
1375 logger.warning(
1376 f"No PubMed results found using strategy: {strategy}"
1377 )
1378 simplified_query = self._simplify_query(optimized_query)
1379 if simplified_query != optimized_query:
1380 logger.info("Trying with simplified query")
1381 pmid_list, strategy = self._adaptive_search(simplified_query)
1382 if pmid_list:
1383 logger.info(
1384 f"Simplified query found {len(pmid_list)} results"
1385 )
1387 if not pmid_list:
1388 logger.warning("No PubMed results found after query simplification")
1389 return []
1391 # Get article summaries
1392 logger.debug(f"Fetching article summaries for {len(pmid_list)} PMIDs")
1393 summaries = self._get_article_summaries(pmid_list)
1394 logger.debug(f"Retrieved {len(summaries)} summaries")
1396 # ALWAYS fetch abstracts for snippet-only mode to provide context for LLM
1397 logger.debug(
1398 f"Fetching abstracts for {len(pmid_list)} articles for snippet enrichment"
1399 )
1400 abstracts = self._get_article_abstracts(pmid_list)
1401 logger.debug(f"Retrieved {len(abstracts)} abstracts")
1403 # Format as previews
1404 previews = []
1405 for summary in summaries:
1406 # Build snippet from individual metadata preferences
1407 snippet_parts = []
1409 # Check for publication type from esummary (earlier than detailed metadata)
1410 pub_type_prefix = ""
1411 if self.include_publication_type_in_context and summary.get(
1412 "pubtype"
1413 ):
1414 # Use first publication type from esummary
1415 pub_type_prefix = f"[{summary['pubtype'][0]}] "
1417 # Add authors if enabled
1418 if self.include_authors_in_context and summary.get("authors"):
1419 authors_text = ", ".join(summary.get("authors", []))
1420 if len(authors_text) > 100:
1421 # Truncate long author lists
1422 authors_text = authors_text[:97] + "..."
1423 snippet_parts.append(authors_text)
1425 # Add journal if enabled
1426 if self.include_journal_in_context and summary.get("journal"): 1426 ↛ 1430line 1426 didn't jump to line 1430 because the condition on line 1426 was always true
1427 snippet_parts.append(summary["journal"])
1429 # Add date (full or year only)
1430 if summary.get("pubdate"): 1430 ↛ 1440line 1430 didn't jump to line 1440 because the condition on line 1430 was always true
1431 if self.include_full_date_in_context: 1431 ↛ 1432line 1431 didn't jump to line 1432 because the condition on line 1431 was never true
1432 snippet_parts.append(summary["pubdate"])
1433 elif ( 1433 ↛ 1440line 1433 didn't jump to line 1440 because the condition on line 1433 was always true
1434 self.include_year_in_context
1435 and len(summary["pubdate"]) >= 4
1436 ):
1437 snippet_parts.append(summary["pubdate"][:4])
1439 # Add citation details if enabled
1440 if self.include_citation_in_context:
1441 citation_parts = []
1442 if summary.get("volume"): 1442 ↛ 1444line 1442 didn't jump to line 1444 because the condition on line 1442 was always true
1443 citation_parts.append(f"Vol {summary['volume']}")
1444 if summary.get("issue"): 1444 ↛ 1446line 1444 didn't jump to line 1446 because the condition on line 1444 was always true
1445 citation_parts.append(f"Issue {summary['issue']}")
1446 if summary.get("pages"): 1446 ↛ 1448line 1446 didn't jump to line 1448 because the condition on line 1446 was always true
1447 citation_parts.append(f"pp {summary['pages']}")
1448 if citation_parts: 1448 ↛ 1452line 1448 didn't jump to line 1452 because the condition on line 1448 was always true
1449 snippet_parts.append(f"({', '.join(citation_parts)})")
1451 # Join snippet parts or provide default
1452 if snippet_parts: 1452 ↛ 1463line 1452 didn't jump to line 1463 because the condition on line 1452 was always true
1453 # Use different separators based on what's included
1454 if self.include_authors_in_context:
1455 snippet = ". ".join(
1456 snippet_parts
1457 ) # Authors need period separator
1458 else:
1459 snippet = " - ".join(
1460 snippet_parts
1461 ) # Journal and year use dash
1462 else:
1463 snippet = "Research article"
1465 # Add publication type prefix
1466 snippet = pub_type_prefix + snippet
1468 # Add language indicator if not English
1469 if self.include_language_in_context and summary.get("lang"):
1470 langs = summary["lang"]
1471 if langs and langs[0] != "eng" and langs[0]: 1471 ↛ 1475line 1471 didn't jump to line 1475 because the condition on line 1471 was always true
1472 snippet = f"{snippet} [{langs[0].upper()}]"
1474 # Add identifiers if enabled
1475 identifier_parts = []
1476 if self.include_pmid_in_context and summary.get("id"):
1477 identifier_parts.append(f"PMID: {summary['id']}")
1478 if self.include_doi_in_context and summary.get("doi"):
1479 identifier_parts.append(f"DOI: {summary['doi']}")
1481 if identifier_parts:
1482 snippet = f"{snippet} | {' | '.join(identifier_parts)}"
1484 # ALWAYS include title and abstract in snippet for LLM analysis
1485 pmid = summary["id"]
1486 title = summary["title"]
1487 abstract_text = abstracts.get(pmid, "")
1489 # Truncate abstract if too long
1490 if len(abstract_text) > 500: 1490 ↛ 1491line 1490 didn't jump to line 1491 because the condition on line 1490 was never true
1491 abstract_text = abstract_text[:497] + "..."
1493 # Build the enriched snippet with title and abstract
1494 if abstract_text:
1495 enriched_snippet = f"Title: {title}\n\nAbstract: {abstract_text}\n\nMetadata: {snippet}"
1496 else:
1497 enriched_snippet = f"Title: {title}\n\nMetadata: {snippet}"
1499 # Log the complete snippet for debugging
1500 logger.debug(f"Complete snippet for PMID {pmid}:")
1501 logger.debug(f" Title: {title[:100]}...")
1502 logger.debug(f" Abstract length: {len(abstract_text)} chars")
1503 logger.debug(f" Metadata: {snippet}")
1504 logger.debug(
1505 f" Full enriched snippet ({len(enriched_snippet)} chars): {enriched_snippet[:500]}..."
1506 )
1508 # Create preview with basic information
1509 preview = {
1510 "id": summary["id"],
1511 "title": summary["title"],
1512 "link": summary["link"],
1513 "snippet": enriched_snippet, # Use enriched snippet with title and abstract
1514 "authors": summary.get("authors", []),
1515 "journal": summary.get("journal", ""),
1516 # Alias for the journal reputation filter, which reads
1517 # `journal_ref` (the field name used by arXiv).
1518 # Use None (not empty string) to match other engines so the
1519 # filter treats missing journals consistently.
1520 "journal_ref": summary.get("journal") or None,
1521 # Forward the print / linking ISSN so the reputation
1522 # filter's Tier 2/3 lookups can key on it (faster and
1523 # more reliable than fuzzy name matching). essn is the
1524 # electronic ISSN; prefer it when issn is blank.
1525 "issn": summary.get("issn") or summary.get("essn") or None,
1526 "pubdate": summary.get("pubdate", ""),
1527 "doi": summary.get("doi", ""),
1528 "source": "PubMed",
1529 "_pmid": summary["id"], # Store PMID for later use
1530 "_search_strategy": strategy, # Store search strategy for analytics
1531 }
1533 previews.append(preview)
1535 logger.info(
1536 f"Found {len(previews)} PubMed previews using strategy: {strategy}"
1537 )
1538 if previews:
1539 logger.debug(
1540 f"Sample preview title: '{previews[0].get('title', 'NO TITLE')[:80]}...'"
1541 )
1542 return previews
1544 def _get_full_content(
1545 self, relevant_items: List[Dict[str, Any]]
1546 ) -> List[Dict[str, Any]]:
1547 """
1548 Get full content for the relevant PubMed articles.
1549 Efficiently manages which content to retrieve (abstracts and/or full text).
1551 Args:
1552 relevant_items: List of relevant preview dictionaries
1554 Returns:
1555 List of result dictionaries with full content
1556 """
1557 logger.info(
1558 f"Getting content for {len(relevant_items)} PubMed articles"
1559 )
1561 # Collect all PMIDs for relevant items
1562 pmids = []
1563 for item in relevant_items:
1564 if "_pmid" in item:
1565 pmids.append(item["_pmid"])
1567 # Get abstracts if requested and PMIDs exist
1568 # In snippet-only mode, always get abstracts as they serve as snippets
1569 abstracts = {}
1570 if self.get_abstracts and pmids:
1571 abstracts = self._get_article_abstracts(pmids)
1573 # Get detailed metadata for all articles (publication types, MeSH terms, etc.)
1574 detailed_metadata = {}
1575 if pmids:
1576 detailed_metadata = self._get_article_detailed_metadata(pmids)
1578 # Find PMC IDs for full-text retrieval (if enabled and not in snippet-only mode)
1579 pmid_to_pmcid = {}
1580 if self.get_full_text and pmids:
1581 pmid_to_pmcid = self._find_pmc_ids(pmids)
1583 # Add content to results
1584 results: List[Dict[str, Any]] = []
1585 for item in relevant_items:
1586 result = item.copy()
1587 pmid = item.get("_pmid", "")
1589 # Add detailed metadata if available
1590 if pmid in detailed_metadata:
1591 metadata = detailed_metadata[pmid]
1593 # Add publication types (e.g., "Clinical Trial", "Meta-Analysis")
1594 if "publication_types" in metadata:
1595 result["publication_types"] = metadata["publication_types"]
1597 # Add first publication type to snippet if enabled
1598 if ( 1598 ↛ 1610line 1598 didn't jump to line 1610 because the condition on line 1598 was always true
1599 self.include_publication_type_in_context
1600 and metadata["publication_types"]
1601 ):
1602 # Just take the first publication type as is
1603 pub_type = metadata["publication_types"][0]
1604 if "snippet" in result: 1604 ↛ 1610line 1604 didn't jump to line 1610 because the condition on line 1604 was always true
1605 result["snippet"] = (
1606 f"[{pub_type}] {result['snippet']}"
1607 )
1609 # Add MeSH terms for medical categorization
1610 if "mesh_terms" in metadata:
1611 result["mesh_terms"] = metadata["mesh_terms"]
1613 # Add MeSH terms to snippet if enabled
1614 if ( 1614 ↛ 1630line 1614 didn't jump to line 1630 because the condition on line 1614 was always true
1615 self.include_mesh_terms_in_context
1616 and metadata["mesh_terms"]
1617 ):
1618 mesh_to_show = (
1619 metadata["mesh_terms"][: self.max_mesh_terms]
1620 if self.max_mesh_terms > 0
1621 else metadata["mesh_terms"]
1622 )
1623 if mesh_to_show and "snippet" in result: 1623 ↛ 1630line 1623 didn't jump to line 1630 because the condition on line 1623 was always true
1624 mesh_text = "MeSH: " + ", ".join(mesh_to_show)
1625 result["snippet"] = (
1626 f"{result['snippet']} | {mesh_text}"
1627 )
1629 # Add keywords
1630 if "keywords" in metadata:
1631 result["keywords"] = metadata["keywords"]
1633 # Add keywords to snippet if enabled
1634 if ( 1634 ↛ 1652line 1634 didn't jump to line 1652 because the condition on line 1634 was always true
1635 self.include_keywords_in_context
1636 and metadata["keywords"]
1637 ):
1638 keywords_to_show = (
1639 metadata["keywords"][: self.max_keywords]
1640 if self.max_keywords > 0
1641 else metadata["keywords"]
1642 )
1643 if keywords_to_show and "snippet" in result: 1643 ↛ 1652line 1643 didn't jump to line 1652 because the condition on line 1643 was always true
1644 keywords_text = "Keywords: " + ", ".join(
1645 keywords_to_show
1646 )
1647 result["snippet"] = (
1648 f"{result['snippet']} | {keywords_text}"
1649 )
1651 # Add affiliations
1652 if "affiliations" in metadata: 1652 ↛ 1653line 1652 didn't jump to line 1653 because the condition on line 1652 was never true
1653 result["affiliations"] = metadata["affiliations"]
1655 # Add funding/grant information
1656 if "grants" in metadata: 1656 ↛ 1657line 1656 didn't jump to line 1657 because the condition on line 1656 was never true
1657 result["grants"] = metadata["grants"]
1659 # Add conflict of interest statement
1660 if "conflict_of_interest" in metadata: 1660 ↛ 1661line 1660 didn't jump to line 1661 because the condition on line 1660 was never true
1661 result["conflict_of_interest"] = metadata[
1662 "conflict_of_interest"
1663 ]
1665 # Add free full text availability
1666 if "has_free_full_text" in metadata:
1667 result["has_free_full_text"] = metadata[
1668 "has_free_full_text"
1669 ]
1670 if "pmc_id" in metadata: 1670 ↛ 1674line 1670 didn't jump to line 1674 because the condition on line 1670 was always true
1671 result["pmc_id"] = metadata["pmc_id"]
1673 # Add PMC availability to snippet if enabled
1674 if ( 1674 ↛ 1684line 1674 didn't jump to line 1684 because the condition on line 1674 was always true
1675 self.include_pmc_availability_in_context
1676 and metadata["has_free_full_text"]
1677 and "snippet" in result
1678 ):
1679 result["snippet"] = (
1680 f"{result['snippet']} | [Free Full Text]"
1681 )
1683 # Add abstract if available
1684 if pmid in abstracts:
1685 result["abstract"] = abstracts[pmid]
1687 # Create enriched content with metadata context
1688 enriched_content = self._create_enriched_content(
1689 result, abstracts[pmid]
1690 )
1692 # ALWAYS include title and abstract in snippet for LLM analysis
1693 # Build comprehensive snippet with title and abstract
1694 title = result.get("title", "")
1695 abstract_text = (
1696 abstracts[pmid][:SNIPPET_LENGTH_LONG]
1697 if len(abstracts[pmid]) > SNIPPET_LENGTH_LONG
1698 else abstracts[pmid]
1699 )
1701 # Prepend title and abstract to the existing metadata snippet
1702 if "snippet" in result:
1703 # Keep metadata snippet and add content
1704 result["snippet"] = (
1705 f"Title: {title}\n\nAbstract: {abstract_text}\n\nMetadata: {result['snippet']}"
1706 )
1707 else:
1708 # No metadata snippet, just title and abstract
1709 result["snippet"] = (
1710 f"Title: {title}\n\nAbstract: {abstract_text}"
1711 )
1713 # Use abstract as content if no full text
1714 if pmid not in pmid_to_pmcid:
1715 result["full_content"] = enriched_content
1716 result["full_content_is_complete"] = False
1717 result["content"] = enriched_content
1718 result["content_type"] = "abstract"
1720 # Add full text for a limited number of top articles
1721 if (
1722 pmid in pmid_to_pmcid
1723 and self.get_full_text
1724 and len(
1725 [r for r in results if r.get("content_type") == "full_text"]
1726 )
1727 < self.full_text_limit
1728 ):
1729 # Get full text content
1730 pmcid = pmid_to_pmcid[pmid]
1731 full_text = self._get_pmc_full_text(pmcid)
1733 if full_text:
1734 enriched_full_text = self._create_enriched_content(
1735 result, full_text
1736 )
1737 result["full_content"] = enriched_full_text
1738 result["full_content_is_complete"] = True
1739 result["content"] = enriched_full_text
1740 result["content_type"] = "full_text"
1741 result["pmcid"] = pmcid
1742 elif pmid in abstracts: 1742 ↛ 1753line 1742 didn't jump to line 1753 because the condition on line 1742 was always true
1743 # Fall back to abstract if full text retrieval fails
1744 enriched_content = self._create_enriched_content(
1745 result, abstracts[pmid]
1746 )
1747 result["full_content"] = enriched_content
1748 result["full_content_is_complete"] = False
1749 result["content"] = enriched_content
1750 result["content_type"] = "abstract"
1752 # Remove temporary fields
1753 if "_pmid" in result:
1754 del result["_pmid"]
1755 if "_search_strategy" in result:
1756 del result["_search_strategy"]
1758 results.append(result)
1760 return results
1762 def search_by_author(
1763 self, author_name: str, max_results: Optional[int] = None
1764 ) -> List[Dict[str, Any]]:
1765 """
1766 Search for articles by a specific author.
1768 Args:
1769 author_name: Name of the author
1770 max_results: Maximum number of results (defaults to self.max_results)
1772 Returns:
1773 List of articles by the author
1774 """
1775 original_max_results = self.max_results
1777 try:
1778 if max_results:
1779 self.max_results = max_results
1781 query = f"{author_name}[Author]"
1782 return self.run(query)
1784 finally:
1785 # Restore original value
1786 self.max_results = original_max_results
1788 def search_by_journal(
1789 self, journal_name: str, max_results: Optional[int] = None
1790 ) -> List[Dict[str, Any]]:
1791 """
1792 Search for articles in a specific journal.
1794 Args:
1795 journal_name: Name of the journal
1796 max_results: Maximum number of results (defaults to self.max_results)
1798 Returns:
1799 List of articles from the journal
1800 """
1801 original_max_results = self.max_results
1803 try:
1804 if max_results:
1805 self.max_results = max_results
1807 query = f"{journal_name}[Journal]"
1808 return self.run(query)
1810 finally:
1811 # Restore original value
1812 self.max_results = original_max_results
1814 def search_recent(
1815 self, query: str, days: int = 30, max_results: Optional[int] = None
1816 ) -> List[Dict[str, Any]]:
1817 """
1818 Search for recent articles matching the query.
1820 Args:
1821 query: The search query
1822 days: Number of days to look back
1823 max_results: Maximum number of results (defaults to self.max_results)
1825 Returns:
1826 List of recent articles matching the query
1827 """
1828 original_max_results = self.max_results
1829 original_days_limit = self.days_limit
1831 try:
1832 if max_results:
1833 self.max_results = max_results
1835 # Set days limit for this search
1836 self.days_limit = days
1838 return self.run(query)
1840 finally:
1841 # Restore original values
1842 self.max_results = original_max_results
1843 self.days_limit = original_days_limit
1845 def advanced_search(
1846 self, terms: Dict[str, str], max_results: Optional[int] = None
1847 ) -> List[Dict[str, Any]]:
1848 """
1849 Perform an advanced search with field-specific terms.
1851 Args:
1852 terms: Dictionary mapping fields to search terms
1853 Valid fields: Author, Journal, Title, MeSH, Affiliation, etc.
1854 max_results: Maximum number of results (defaults to self.max_results)
1856 Returns:
1857 List of articles matching the advanced query
1858 """
1859 original_max_results = self.max_results
1861 try:
1862 if max_results:
1863 self.max_results = max_results
1865 # Build advanced query string
1866 query_parts = []
1867 for field, term in terms.items():
1868 query_parts.append(f"{term}[{field}]")
1870 query = " AND ".join(query_parts)
1871 return self.run(query)
1873 finally:
1874 # Restore original value
1875 self.max_results = original_max_results