Coverage for src/local_deep_research/web_search_engines/engines/search_engine_pubmed.py: 93%
717 statements
« prev ^ index » next coverage.py v7.15.1, created at 2026-07-20 01:24 +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 ..search_engine_base import BaseSearchEngine, Exposure, Sensitivity
15class PubMedSearchEngine(BaseSearchEngine):
16 """
17 PubMed search engine implementation with two-phase approach and adaptive search.
18 Provides efficient access to biomedical literature while minimizing API usage.
19 """
21 # Mark as public search engine
22 is_public = True
23 egress_sensitivity = Sensitivity.NON_SENSITIVE
24 egress_exposure = Exposure.EXPOSING
25 # Scientific/medical search engine
26 is_scientific = True
27 is_lexical = True
28 needs_llm_relevance_filter = True
30 def __init__(
31 self,
32 max_results: int = 10,
33 api_key: Optional[str] = None,
34 days_limit: Optional[int] = None,
35 get_abstracts: bool = True,
36 get_full_text: bool = False,
37 full_text_limit: int = 3,
38 llm: Optional[BaseLLM] = None,
39 max_filtered_results: Optional[int] = None,
40 optimize_queries: bool = True,
41 include_publication_type_in_context: bool = True,
42 include_journal_in_context: bool = True,
43 include_year_in_context: bool = True,
44 include_authors_in_context: bool = False,
45 include_full_date_in_context: bool = False,
46 include_mesh_terms_in_context: bool = True,
47 include_keywords_in_context: bool = True,
48 include_doi_in_context: bool = False,
49 include_pmid_in_context: bool = False,
50 include_pmc_availability_in_context: bool = False,
51 max_mesh_terms: int = 3,
52 max_keywords: int = 3,
53 include_citation_in_context: bool = False,
54 include_language_in_context: bool = False,
55 settings_snapshot: Optional[Dict[str, Any]] = None,
56 ):
57 """
58 Initialize the PubMed search engine.
60 Args:
61 max_results: Maximum number of search results
62 api_key: NCBI API key for higher rate limits (optional)
63 days_limit: Limit results to N days (optional)
64 get_abstracts: Whether to fetch abstracts for all results
65 get_full_text: Whether to fetch full text content (when available in PMC)
66 full_text_limit: Max number of full-text articles to retrieve
67 llm: Language model for relevance filtering
68 max_filtered_results: Maximum number of results to keep after filtering
69 optimize_queries: Whether to optimize natural language queries for PubMed
70 """
71 # Wire up the journal reputation filter as a preview filter so
72 # results are scored against bundled OpenAlex/DOAJ/predatory data
73 # before the (more expensive) LLM relevance pass.
74 preview_filters = []
75 journal_filter = self._create_journal_filter(
76 "pubmed", llm, settings_snapshot
77 )
78 if journal_filter is not None: 78 ↛ 79line 78 didn't jump to line 79 because the condition on line 78 was never true
79 preview_filters.append(journal_filter)
81 # Initialize the BaseSearchEngine with LLM, max_filtered_results, and max_results
82 super().__init__(
83 llm=llm,
84 max_filtered_results=max_filtered_results,
85 max_results=max_results,
86 preview_filters=preview_filters, # type: ignore[arg-type]
87 settings_snapshot=settings_snapshot,
88 )
89 self.max_results = max(self.max_results, 25)
90 self.api_key = api_key
91 self.days_limit = days_limit
92 self.get_abstracts = get_abstracts
93 self.get_full_text = get_full_text
94 self.full_text_limit = full_text_limit
95 self.optimize_queries = optimize_queries
96 self.include_publication_type_in_context = (
97 include_publication_type_in_context
98 )
99 self.include_journal_in_context = include_journal_in_context
100 self.include_year_in_context = include_year_in_context
101 self.include_authors_in_context = include_authors_in_context
102 self.include_full_date_in_context = include_full_date_in_context
103 self.include_mesh_terms_in_context = include_mesh_terms_in_context
104 self.include_keywords_in_context = include_keywords_in_context
105 self.include_doi_in_context = include_doi_in_context
106 self.include_pmid_in_context = include_pmid_in_context
107 self.include_pmc_availability_in_context = (
108 include_pmc_availability_in_context
109 )
110 self.max_mesh_terms = max_mesh_terms
111 self.max_keywords = max_keywords
112 self.include_citation_in_context = include_citation_in_context
113 self.include_language_in_context = include_language_in_context
115 # Base API URLs
116 self.base_url = "https://eutils.ncbi.nlm.nih.gov/entrez/eutils"
117 self.search_url = f"{self.base_url}/esearch.fcgi"
118 self.summary_url = f"{self.base_url}/esummary.fcgi"
119 self.fetch_url = f"{self.base_url}/efetch.fcgi"
120 self.link_url = f"{self.base_url}/elink.fcgi"
122 # PMC base URL for full text
123 self.pmc_url = "https://www.ncbi.nlm.nih.gov/pmc/articles/"
125 def _get_result_count(self, query: str) -> int:
126 """
127 Get the total number of results for a query without retrieving the results themselves.
129 Args:
130 query: The search query
132 Returns:
133 Total number of matching results
134 """
135 try:
136 # Prepare search parameters
137 params = {
138 "db": "pubmed",
139 "term": query,
140 "retmode": "json",
141 "retmax": 0, # Don't need actual results, just the count
142 }
144 # Add API key if available
145 if self.api_key:
146 params["api_key"] = self.api_key
148 self._last_wait_time = self.rate_tracker.apply_rate_limit(
149 self.engine_type
150 )
152 # Execute search request
153 response = safe_get(self.search_url, params=params)
154 response.raise_for_status()
156 # Parse response
157 data = response.json()
158 count = int(data["esearchresult"]["count"])
160 logger.info(
161 "Query '{}' has {} total results in PubMed", query, count
162 )
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 = self.llm.invoke(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 optimized_query = re.sub(r'"([^"]+)\[', r'"\1"[', optimized_query)
391 # Simplify the query if still no results are found
392 self._simplify_query_cache = optimized_query
394 # Log original and optimized queries
395 logger.info("Original query: '{}'", query)
396 logger.info(f"Optimized for PubMed: '{optimized_query}'")
397 logger.debug(
398 f"Query optimization complete: '{query[:50]}...' -> '{optimized_query[:100]}...'"
399 )
401 return optimized_query
403 except Exception as e:
404 safe_msg = self._scrub_error(e)
405 logger.exception(
406 f"Error optimizing query ({type(e).__name__}): {safe_msg}"
407 )
408 logger.debug(f"Falling back to original query: '{query}'")
409 return query # Fall back to original query on error
411 def _simplify_query(self, query: str) -> str:
412 """
413 Simplify a PubMed query that returned no results.
414 Progressively removes elements to get a more basic query.
416 Args:
417 query: The original query that returned no results
419 Returns:
420 Simplified query
421 """
422 logger.info(f"Simplifying query: {query}")
423 logger.debug(f"Query simplification started for: '{query[:100]}...'")
425 # Simple approach: remove field restrictions to broaden the search
426 import re
428 # Remove field tags to make search broader
429 simplified = query
431 # Remove [Mesh] tags - search in all fields instead
432 simplified = re.sub(r"\[Mesh\]", "", simplified, flags=re.IGNORECASE)
434 # Remove [Publication Type] tags
435 simplified = re.sub(
436 r"\[Publication Type\]", "", simplified, flags=re.IGNORECASE
437 )
439 # Keep [Title/Abstract] as it's usually helpful
440 # Clean up any double spaces
441 simplified = re.sub(r"\s+", " ", simplified).strip()
443 # If no simplification was possible, return the original query
444 if simplified == query:
445 logger.debug("No simplification possible, returning original query")
447 logger.info(f"Simplified query: {simplified}")
448 logger.debug(
449 f"Query simplified from {len(query)} to {len(simplified)} chars"
450 )
451 return simplified
453 def _is_historical_focused(self, query: str) -> bool:
454 """
455 Determine if a query is specifically focused on historical/older information using LLM.
456 Default assumption is that queries should prioritize recent information unless
457 explicitly asking for historical content.
459 Args:
460 query: The search query
462 Returns:
463 Boolean indicating if the query is focused on historical information
464 """
465 if not self.llm:
466 # Fall back to basic keyword check if no LLM available
467 historical_terms = [
468 "history",
469 "historical",
470 "early",
471 "initial",
472 "first",
473 "original",
474 "before",
475 "prior to",
476 "origins",
477 "evolution",
478 "development",
479 ]
480 historical_years = [str(year) for year in range(1900, 2020)]
482 query_lower = query.lower()
483 has_historical_term = any(
484 term in query_lower for term in historical_terms
485 )
486 has_past_year = any(year in query for year in historical_years)
488 return has_historical_term or has_past_year
490 try:
491 # Use LLM to determine if the query is focused on historical information
492 prompt = f"""Determine if this query is specifically asking for HISTORICAL or OLDER information.
494Query: "{query}"
496Answer ONLY "yes" if the query is clearly asking for historical, early, original, or past information from more than 5 years ago.
497Answer 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.
499The default assumption should be that medical and scientific queries want RECENT information unless clearly specified otherwise.
500"""
502 response = self.llm.invoke(prompt)
503 answer = (
504 (
505 str(response.content)
506 if hasattr(response, "content")
507 else str(response)
508 )
509 .strip()
510 .lower()
511 )
513 # Log the determination
514 logger.info(f"Historical focus determination for query: '{query}'")
515 logger.info(f"LLM determined historical focus: {answer}")
517 return "yes" in answer
519 except Exception as e:
520 safe_msg = self._scrub_error(e)
521 logger.exception(
522 f"Error determining historical focus ({type(e).__name__}): {safe_msg}"
523 )
524 # Fall back to basic keyword check
525 historical_terms = [
526 "history",
527 "historical",
528 "early",
529 "initial",
530 "first",
531 "original",
532 "before",
533 "prior to",
534 "origins",
535 "evolution",
536 "development",
537 ]
538 return any(term in query.lower() for term in historical_terms)
540 def _adaptive_search(self, query: str) -> Tuple[List[str], str]:
541 """
542 Perform an adaptive search that adjusts based on topic volume and whether
543 the query focuses on historical information.
545 Args:
546 query: The search query (already optimized)
548 Returns:
549 Tuple of (list of PMIDs, search strategy used)
550 """
551 # Estimate topic volume
552 estimated_volume = self._get_result_count(query)
554 # Determine if the query is focused on historical information
555 is_historical_focused = self._is_historical_focused(query)
557 if is_historical_focused:
558 # User wants historical information - no date filtering
559 time_filter = None
560 strategy = "historical_focus"
561 elif estimated_volume > 5000:
562 # Very common topic - use tighter recency filter
563 time_filter = '"last 1 year"[pdat]'
564 strategy = "high_volume"
565 elif estimated_volume > 1000:
566 # Common topic
567 time_filter = '"last 3 years"[pdat]'
568 strategy = "common_topic"
569 elif estimated_volume > 100:
570 # Moderate volume
571 time_filter = '"last 5 years"[pdat]'
572 strategy = "moderate_volume"
573 else:
574 # Rare topic - still use recency but with wider range
575 time_filter = '"last 10 years"[pdat]'
576 strategy = "rare_topic"
578 # Run search based on strategy
579 if time_filter:
580 # Try with adaptive time filter
581 query_with_time = f"({query}) AND {time_filter}"
582 logger.info(
583 f"Using adaptive search strategy: {strategy} with filter: {time_filter}"
584 )
585 results = self._search_pubmed(query_with_time)
587 # If too few results, gradually expand time window
588 if len(results) < 5 and '"last 10 years"[pdat]' not in time_filter:
589 logger.info(
590 f"Insufficient results ({len(results)}), expanding time window"
591 )
592 expanded_time = self._expand_time_window(time_filter)
593 query_with_expanded_time = f"({query}) AND {expanded_time}"
594 expanded_results = self._search_pubmed(query_with_expanded_time)
596 if len(expanded_results) > len(results):
597 logger.info(
598 f"Expanded time window yielded {len(expanded_results)} results"
599 )
600 return expanded_results, f"{strategy}_expanded"
602 # If still no results, try without time filter
603 if not results:
604 logger.info(
605 "No results with time filter, trying without time restrictions"
606 )
607 results = self._search_pubmed(query)
608 strategy = "no_time_filter"
609 else:
610 # Historical query - run without time filter
611 logger.info(
612 "Using historical search strategy without date filtering"
613 )
614 results = self._search_pubmed(query)
616 return results, strategy
618 def _search_pubmed(self, query: str) -> List[str]:
619 """
620 Search PubMed and return a list of article IDs.
622 Args:
623 query: The search query
625 Returns:
626 List of PubMed IDs matching the query
627 """
628 try:
629 # Prepare search parameters
630 params = {
631 "db": "pubmed",
632 "term": query,
633 "retmode": "json",
634 # Cap PMIDs fetched per query so a large user-supplied
635 # max_results can't trigger an unbounded esearch/efetch load.
636 "retmax": min(self.max_results, 100),
637 "usehistory": "y",
638 }
640 # Add API key if available
641 if self.api_key:
642 params["api_key"] = self.api_key
643 logger.debug("Using PubMed API key for higher rate limits")
644 else:
645 logger.debug("No PubMed API key - using default rate limits")
647 # Add date restriction if specified
648 if self.days_limit:
649 params["reldate"] = self.days_limit
650 params["datetype"] = "pdat" # Publication date
651 logger.debug(f"Limiting results to last {self.days_limit} days")
653 logger.debug(
654 f"PubMed search query: '{query}' with max_results={self.max_results}"
655 )
657 self._last_wait_time = self.rate_tracker.apply_rate_limit(
658 self.engine_type
659 )
660 logger.debug(
661 f"Applied rate limit wait: {self._last_wait_time:.2f}s"
662 )
664 # Execute search request
665 logger.debug(f"Sending request to PubMed API: {self.search_url}")
666 response = safe_get(self.search_url, params=params)
667 response.raise_for_status()
668 logger.debug(f"PubMed API response status: {response.status_code}")
670 # Parse response
671 data = response.json()
672 id_list: list[str] = data["esearchresult"]["idlist"]
673 total_count = data["esearchresult"].get("count", "unknown")
675 logger.info(
676 f"PubMed search for '{query}' found {len(id_list)} results (total available: {total_count})"
677 )
678 if len(id_list) > 0:
679 logger.debug(f"First 5 PMIDs: {id_list[:5]}")
680 return id_list
682 except Exception as e:
683 safe_msg = self._scrub_error(e)
684 logger.warning(
685 f"Error searching PubMed for query '{query}' ({type(e).__name__}): {safe_msg}"
686 )
687 return []
689 def _get_article_summaries(
690 self, id_list: List[str]
691 ) -> List[Dict[str, Any]]:
692 """
693 Get summaries for a list of PubMed article IDs.
695 Args:
696 id_list: List of PubMed IDs
698 Returns:
699 List of article summary dictionaries
700 """
701 if not id_list:
702 logger.debug("Empty ID list provided to _get_article_summaries")
703 return []
705 logger.debug(f"Fetching summaries for {len(id_list)} PubMed articles")
707 try:
708 # Prepare parameters
709 params = {
710 "db": "pubmed",
711 "id": ",".join(id_list),
712 "retmode": "json",
713 "rettype": "summary",
714 }
716 # Add API key if available
717 if self.api_key:
718 params["api_key"] = self.api_key
720 self._last_wait_time = self.rate_tracker.apply_rate_limit(
721 self.engine_type
722 )
723 logger.debug(
724 f"Applied rate limit wait: {self._last_wait_time:.2f}s"
725 )
727 # Execute request
728 logger.debug(f"Requesting summaries from: {self.summary_url}")
729 response = safe_get(self.summary_url, params=params)
730 response.raise_for_status()
731 logger.debug(f"Summary API response status: {response.status_code}")
733 # Parse response
734 data = response.json()
735 logger.debug(
736 f"PubMed API returned data for {len(id_list)} requested IDs"
737 )
738 summaries = []
740 for pmid in id_list:
741 if pmid in data["result"]: 741 ↛ 789line 741 didn't jump to line 789 because the condition on line 741 was always true
742 article = data["result"][pmid]
743 logger.debug(
744 f"Processing article {pmid}: {article.get('title', 'NO TITLE')[:50]}"
745 )
747 # Extract authors (if available)
748 authors = []
749 if "authors" in article: 749 ↛ 755line 749 didn't jump to line 755 because the condition on line 749 was always true
750 authors = [
751 author["name"] for author in article["authors"]
752 ]
754 # Extract DOI from articleids if not in main field
755 doi = article.get("doi", "")
756 if not doi and "articleids" in article: 756 ↛ 763line 756 didn't jump to line 763 because the condition on line 756 was always true
757 for aid in article["articleids"]: 757 ↛ 763line 757 didn't jump to line 763 because the loop on line 757 didn't complete
758 if aid.get("idtype") == "doi": 758 ↛ 757line 758 didn't jump to line 757 because the condition on line 758 was always true
759 doi = aid.get("value", "")
760 break
762 # Create summary dictionary with all available fields
763 summary = {
764 "id": pmid,
765 "title": article.get("title", ""),
766 "pubdate": article.get("pubdate", ""),
767 "epubdate": article.get("epubdate", ""),
768 "source": article.get("source", ""),
769 "authors": authors,
770 "lastauthor": article.get("lastauthor", ""),
771 "journal": article.get("fulljournalname", ""),
772 "volume": article.get("volume", ""),
773 "issue": article.get("issue", ""),
774 "pages": article.get("pages", ""),
775 "doi": doi,
776 "issn": article.get("issn", ""),
777 "essn": article.get("essn", ""),
778 "pubtype": article.get(
779 "pubtype", []
780 ), # Publication types from esummary
781 "recordstatus": article.get("recordstatus", ""),
782 "lang": article.get("lang", []),
783 "pmcrefcount": article.get("pmcrefcount", None),
784 "link": f"https://pubmed.ncbi.nlm.nih.gov/{pmid}/",
785 }
787 summaries.append(summary)
788 else:
789 logger.warning(
790 f"PMID {pmid} not found in PubMed API response"
791 )
793 return summaries
795 except Exception as e:
796 error_msg = str(e)
797 safe_msg = self._scrub_error(error_msg)
798 logger.warning(
799 f"Error getting article summaries for {len(id_list)} articles ({type(e).__name__}): {safe_msg}"
800 )
802 # Check for rate limiting patterns
803 if (
804 "429" in error_msg
805 or "too many requests" in error_msg.lower()
806 or "rate limit" in error_msg.lower()
807 or "service unavailable" in error_msg.lower()
808 or "503" in error_msg
809 or "403" in error_msg
810 ):
811 # `from None` suppresses the implicit __context__ chain:
812 # the original exception still carries the raw message, so
813 # a full traceback render (chain=True) would re-leak the
814 # secret that safe_msg just scrubbed.
815 raise RateLimitError(
816 f"PubMed rate limit hit: {safe_msg}"
817 ) from None
819 return []
821 def _get_article_abstracts(self, id_list: List[str]) -> Dict[str, str]:
822 """
823 Get abstracts for a list of PubMed article IDs.
825 Args:
826 id_list: List of PubMed IDs
828 Returns:
829 Dictionary mapping PubMed IDs to their abstracts
830 """
831 if not id_list:
832 logger.debug("Empty ID list provided to _get_article_abstracts")
833 return {}
835 logger.debug(f"Fetching abstracts for {len(id_list)} PubMed articles")
837 try:
838 # Prepare parameters
839 params = {
840 "db": "pubmed",
841 "id": ",".join(id_list),
842 "retmode": "xml",
843 "rettype": "abstract",
844 }
846 # Add API key if available
847 if self.api_key:
848 params["api_key"] = self.api_key
850 self._last_wait_time = self.rate_tracker.apply_rate_limit(
851 self.engine_type
852 )
853 logger.debug(
854 f"Applied rate limit wait: {self._last_wait_time:.2f}s"
855 )
857 # Execute request
858 logger.debug(f"Requesting abstracts from: {self.fetch_url}")
859 response = safe_get(self.fetch_url, params=params)
860 response.raise_for_status()
861 logger.debug(
862 f"Abstract fetch response status: {response.status_code}, size: {len(response.text)} bytes"
863 )
865 # Parse XML response
866 root = ET.fromstring(response.text)
867 logger.debug(
868 f"Parsing abstracts from XML for {len(id_list)} articles"
869 )
871 # Extract abstracts
872 abstracts = {}
874 for article in root.findall(".//PubmedArticle"):
875 pmid_elem = article.find(".//PMID")
876 pmid = pmid_elem.text if pmid_elem is not None else None
878 if pmid is None:
879 continue
881 # Find abstract text
882 abstract_text = ""
883 abstract_elem = article.find(".//AbstractText")
885 if abstract_elem is not None: 885 ↛ 889line 885 didn't jump to line 889 because the condition on line 885 was always true
886 abstract_text = abstract_elem.text or ""
888 # Some abstracts are split into multiple sections
889 abstract_sections = article.findall(".//AbstractText")
890 if len(abstract_sections) > 1:
891 logger.debug(
892 f"Article {pmid} has {len(abstract_sections)} abstract sections"
893 )
895 for section in abstract_sections:
896 # Get section label if it exists
897 label = section.get("Label")
898 section_text = section.text or ""
900 if label and section_text:
901 if abstract_text: 901 ↛ 904line 901 didn't jump to line 904 because the condition on line 901 was always true
902 abstract_text += f"\n\n{label}: {section_text}"
903 else:
904 abstract_text = f"{label}: {section_text}"
905 elif section_text:
906 if abstract_text: 906 ↛ 909line 906 didn't jump to line 909 because the condition on line 906 was always true
907 abstract_text += f"\n\n{section_text}"
908 else:
909 abstract_text = section_text
911 # Store in dictionary
912 if pmid and abstract_text:
913 abstracts[pmid] = abstract_text
914 logger.debug(
915 f"Abstract for {pmid}: {len(abstract_text)} chars"
916 )
917 elif pmid: 917 ↛ 874line 917 didn't jump to line 874 because the condition on line 917 was always true
918 logger.warning(f"No abstract found for PMID {pmid}")
920 logger.info(
921 f"Successfully retrieved {len(abstracts)} abstracts out of {len(id_list)} requested"
922 )
923 return abstracts
925 except Exception as e:
926 safe_msg = self._scrub_error(e)
927 logger.warning(
928 f"Error getting article abstracts for {len(id_list)} articles ({type(e).__name__}): {safe_msg}"
929 )
930 return {}
932 def _get_article_detailed_metadata(
933 self, id_list: List[str]
934 ) -> Dict[str, Dict[str, Any]]:
935 """
936 Get detailed metadata for PubMed articles including publication types,
937 MeSH terms, keywords, and affiliations.
939 Args:
940 id_list: List of PubMed IDs
942 Returns:
943 Dictionary mapping PubMed IDs to their detailed metadata
944 """
945 if not id_list:
946 return {}
948 try:
949 # Prepare parameters
950 params = {
951 "db": "pubmed",
952 "id": ",".join(id_list),
953 "retmode": "xml",
954 "rettype": "medline",
955 }
957 # Add API key if available
958 if self.api_key:
959 params["api_key"] = self.api_key
961 self._last_wait_time = self.rate_tracker.apply_rate_limit(
962 self.engine_type
963 )
965 # Execute request
966 response = safe_get(self.fetch_url, params=params)
967 response.raise_for_status()
969 # Parse XML response
970 root = ET.fromstring(response.text)
972 metadata = {}
974 for article in root.findall(".//PubmedArticle"):
975 pmid_elem = article.find(".//PMID")
976 pmid = pmid_elem.text if pmid_elem is not None else None
978 if pmid is None: 978 ↛ 979line 978 didn't jump to line 979 because the condition on line 978 was never true
979 continue
981 article_metadata: Dict[str, Any] = {}
983 # Extract publication types
984 pub_types = []
985 for pub_type in article.findall(".//PublicationType"):
986 if pub_type.text: 986 ↛ 985line 986 didn't jump to line 985 because the condition on line 986 was always true
987 pub_types.append(pub_type.text)
988 if pub_types:
989 article_metadata["publication_types"] = pub_types
991 # Extract MeSH terms
992 mesh_terms = []
993 for mesh in article.findall(".//MeshHeading"):
994 descriptor = mesh.find(".//DescriptorName")
995 if descriptor is not None and descriptor.text: 995 ↛ 993line 995 didn't jump to line 993 because the condition on line 995 was always true
996 mesh_terms.append(descriptor.text)
997 if mesh_terms:
998 article_metadata["mesh_terms"] = mesh_terms
1000 # Extract keywords
1001 keywords = []
1002 for keyword in article.findall(".//Keyword"):
1003 if keyword.text: 1003 ↛ 1002line 1003 didn't jump to line 1002 because the condition on line 1003 was always true
1004 keywords.append(keyword.text)
1005 if keywords:
1006 article_metadata["keywords"] = keywords
1008 # Extract affiliations
1009 affiliations = []
1010 for affiliation in article.findall(".//Affiliation"):
1011 if affiliation.text: 1011 ↛ 1010line 1011 didn't jump to line 1010 because the condition on line 1011 was always true
1012 affiliations.append(affiliation.text)
1013 if affiliations:
1014 article_metadata["affiliations"] = affiliations
1016 # Extract grant information
1017 grants = []
1018 for grant in article.findall(".//Grant"):
1019 grant_info = {}
1020 grant_id = grant.find(".//GrantID")
1021 if grant_id is not None and grant_id.text: 1021 ↛ 1023line 1021 didn't jump to line 1023 because the condition on line 1021 was always true
1022 grant_info["id"] = grant_id.text
1023 agency = grant.find(".//Agency")
1024 if agency is not None and agency.text: 1024 ↛ 1026line 1024 didn't jump to line 1026 because the condition on line 1024 was always true
1025 grant_info["agency"] = agency.text
1026 if grant_info: 1026 ↛ 1018line 1026 didn't jump to line 1018 because the condition on line 1026 was always true
1027 grants.append(grant_info)
1028 if grants:
1029 article_metadata["grants"] = grants
1031 # Check for free full text in PMC
1032 pmc_elem = article.find(".//ArticleId[@IdType='pmc']")
1033 if pmc_elem is not None:
1034 article_metadata["has_free_full_text"] = True
1035 article_metadata["pmc_id"] = pmc_elem.text
1037 # Extract conflict of interest statement
1038 coi_elem = article.find(".//CoiStatement")
1039 if coi_elem is not None and coi_elem.text:
1040 article_metadata["conflict_of_interest"] = coi_elem.text
1042 metadata[pmid] = article_metadata
1044 return metadata
1046 except Exception as e:
1047 safe_msg = self._scrub_error(e)
1048 logger.warning(
1049 f"Error getting detailed article metadata ({type(e).__name__}): {safe_msg}"
1050 )
1051 return {}
1053 def _create_enriched_content(
1054 self, result: Dict[str, Any], base_content: str
1055 ) -> str:
1056 """
1057 Create enriched content by adding relevant metadata context to help the LLM.
1059 Args:
1060 result: The result dictionary with metadata
1061 base_content: The base content (abstract or full text)
1063 Returns:
1064 Enriched content string with metadata context
1065 """
1066 enriched_parts = []
1068 # Add study type information
1069 if "publication_types" in result:
1070 pub_types = result["publication_types"]
1071 # Filter for significant types
1072 significant_types = [
1073 pt
1074 for pt in pub_types
1075 if any(
1076 key in pt.lower()
1077 for key in [
1078 "clinical trial",
1079 "randomized",
1080 "meta-analysis",
1081 "systematic review",
1082 "case report",
1083 "guideline",
1084 "comparative study",
1085 "multicenter",
1086 ]
1087 )
1088 ]
1089 if significant_types:
1090 enriched_parts.append(
1091 f"[Study Type: {', '.join(significant_types)}]"
1092 )
1094 # Add the main content
1095 enriched_parts.append(base_content)
1097 # Add metadata footer
1098 metadata_footer = []
1100 # Add ALL MeSH terms
1101 if "mesh_terms" in result and len(result["mesh_terms"]) > 0:
1102 metadata_footer.append(
1103 f"Medical Topics (MeSH): {', '.join(result['mesh_terms'])}"
1104 )
1106 # Add ALL keywords
1107 if "keywords" in result and len(result["keywords"]) > 0:
1108 metadata_footer.append(f"Keywords: {', '.join(result['keywords'])}")
1110 # Add ALL affiliations
1111 if "affiliations" in result and len(result["affiliations"]) > 0:
1112 if len(result["affiliations"]) == 1:
1113 metadata_footer.append(
1114 f"Institution: {result['affiliations'][0]}"
1115 )
1116 else:
1117 affiliations_text = "\n - " + "\n - ".join(
1118 result["affiliations"]
1119 )
1120 metadata_footer.append(f"Institutions:{affiliations_text}")
1122 # Add ALL funding information with full details
1123 if "grants" in result and len(result["grants"]) > 0:
1124 grant_details = []
1125 for grant in result["grants"]:
1126 grant_text = []
1127 if "agency" in grant:
1128 grant_text.append(grant["agency"])
1129 if "id" in grant:
1130 grant_text.append(f"(Grant ID: {grant['id']})")
1131 if grant_text:
1132 grant_details.append(" ".join(grant_text))
1133 if grant_details:
1134 if len(grant_details) == 1:
1135 metadata_footer.append(f"Funded by: {grant_details[0]}")
1136 else:
1137 funding_text = "\n - " + "\n - ".join(grant_details)
1138 metadata_footer.append(f"Funding Sources:{funding_text}")
1140 # Add FULL conflict of interest statement
1141 if "conflict_of_interest" in result:
1142 coi_text = result["conflict_of_interest"]
1143 if coi_text:
1144 # Still skip trivial "no conflict" statements to reduce noise
1145 if not any(
1146 phrase in coi_text.lower()
1147 for phrase in [
1148 "no conflict",
1149 "no competing",
1150 "nothing to disclose",
1151 "none declared",
1152 "authors declare no",
1153 ]
1154 ):
1155 metadata_footer.append(f"Conflict of Interest: {coi_text}")
1156 elif (
1157 "but" in coi_text.lower()
1158 or "except" in coi_text.lower()
1159 or "however" in coi_text.lower()
1160 ):
1161 # Include if there's a "no conflict BUT..." type statement
1162 metadata_footer.append(f"Conflict of Interest: {coi_text}")
1164 # Combine everything
1165 if metadata_footer:
1166 enriched_parts.append("\n---\nStudy Metadata:")
1167 enriched_parts.extend(metadata_footer)
1169 return "\n".join(enriched_parts)
1171 def _find_pmc_ids(self, pmid_list: List[str]) -> Dict[str, str]:
1172 """
1173 Find PMC IDs for the given PubMed IDs (for full-text access).
1175 Args:
1176 pmid_list: List of PubMed IDs
1178 Returns:
1179 Dictionary mapping PubMed IDs to their PMC IDs (if available)
1180 """
1181 if not pmid_list or not self.get_full_text:
1182 return {}
1184 try:
1185 # Prepare parameters
1186 params = {
1187 "dbfrom": "pubmed",
1188 "db": "pmc",
1189 "linkname": "pubmed_pmc",
1190 "id": ",".join(pmid_list),
1191 "retmode": "json",
1192 }
1194 # Add API key if available
1195 if self.api_key:
1196 params["api_key"] = self.api_key
1198 self._last_wait_time = self.rate_tracker.apply_rate_limit(
1199 self.engine_type
1200 )
1202 # Execute request
1203 response = safe_get(self.link_url, params=params)
1204 response.raise_for_status()
1206 # Parse response
1207 data = response.json()
1209 # Map PubMed IDs to PMC IDs
1210 pmid_to_pmcid = {}
1212 for linkset in data.get("linksets", []):
1213 pmid = linkset.get("ids", [None])[0]
1215 if not pmid: 1215 ↛ 1216line 1215 didn't jump to line 1216 because the condition on line 1215 was never true
1216 continue
1218 for link in linkset.get("linksetdbs", []):
1219 if link.get("linkname") == "pubmed_pmc": 1219 ↛ 1218line 1219 didn't jump to line 1218 because the condition on line 1219 was always true
1220 pmcids = link.get("links", [])
1221 if pmcids: 1221 ↛ 1218line 1221 didn't jump to line 1218 because the condition on line 1221 was always true
1222 pmid_to_pmcid[str(pmid)] = f"PMC{pmcids[0]}"
1224 logger.info(
1225 f"Found {len(pmid_to_pmcid)} PMC IDs for full-text access"
1226 )
1227 return pmid_to_pmcid
1229 except Exception as e:
1230 safe_msg = self._scrub_error(e)
1231 logger.warning(
1232 f"Error finding PMC IDs ({type(e).__name__}): {safe_msg}"
1233 )
1234 return {}
1236 def _get_pmc_full_text(self, pmcid: str) -> str:
1237 """
1238 Get full text for a PMC article.
1240 Args:
1241 pmcid: PMC ID of the article
1243 Returns:
1244 Full text content or empty string if not available
1245 """
1246 try:
1247 # Prepare parameters
1248 params = {
1249 "db": "pmc",
1250 "id": pmcid,
1251 "retmode": "xml",
1252 "rettype": "full",
1253 }
1255 # Add API key if available
1256 if self.api_key:
1257 params["api_key"] = self.api_key
1259 self._last_wait_time = self.rate_tracker.apply_rate_limit(
1260 self.engine_type
1261 )
1263 # Execute request
1264 response = safe_get(self.fetch_url, params=params)
1265 response.raise_for_status()
1267 # Parse XML response
1268 root = ET.fromstring(response.text)
1270 # Extract full text
1271 full_text = []
1273 # Extract article title
1274 title_elem = root.find(".//article-title")
1275 if title_elem is not None and title_elem.text: 1275 ↛ 1279line 1275 didn't jump to line 1279 because the condition on line 1275 was always true
1276 full_text.append(f"# {title_elem.text}")
1278 # Extract abstract
1279 abstract_paras = root.findall(".//abstract//p")
1280 if abstract_paras:
1281 full_text.append("\n## Abstract\n")
1282 for p in abstract_paras:
1283 text = "".join(p.itertext())
1284 if text: 1284 ↛ 1282line 1284 didn't jump to line 1282 because the condition on line 1284 was always true
1285 full_text.append(text)
1287 # Extract body content
1288 body = root.find(".//body")
1289 if body is not None: 1289 ↛ 1302line 1289 didn't jump to line 1302 because the condition on line 1289 was always true
1290 for section in body.findall(".//sec"):
1291 # Get section title
1292 title = section.find(".//title")
1293 if title is not None and title.text: 1293 ↛ 1297line 1293 didn't jump to line 1297 because the condition on line 1293 was always true
1294 full_text.append(f"\n## {title.text}\n")
1296 # Get paragraphs
1297 for p in section.findall(".//p"):
1298 text = "".join(p.itertext())
1299 if text: 1299 ↛ 1297line 1299 didn't jump to line 1297 because the condition on line 1299 was always true
1300 full_text.append(text)
1302 result_text = "\n\n".join(full_text)
1303 logger.debug(
1304 f"Successfully extracted {len(result_text)} chars of PMC full text with {len(full_text)} sections"
1305 )
1306 return result_text
1308 except Exception as e:
1309 safe_msg = self._scrub_error(e)
1310 logger.warning(
1311 f"Error getting PMC full text ({type(e).__name__}): {safe_msg}"
1312 )
1313 return ""
1315 def _get_previews(self, query: str) -> List[Dict[str, Any]]:
1316 """
1317 Get preview information for PubMed articles.
1319 Args:
1320 query: The search query
1322 Returns:
1323 List of preview dictionaries
1324 """
1325 logger.info(f"Getting PubMed previews for query: {query}")
1327 # Optimize the query for PubMed if LLM is available
1328 optimized_query = self._optimize_query_for_pubmed(query)
1330 # Perform adaptive search
1331 pmid_list, strategy = self._adaptive_search(optimized_query)
1333 # If no results, try a simplified query
1334 if not pmid_list:
1335 logger.warning(
1336 f"No PubMed results found using strategy: {strategy}"
1337 )
1338 simplified_query = self._simplify_query(optimized_query)
1339 if simplified_query != optimized_query:
1340 logger.info(f"Trying with simplified query: {simplified_query}")
1341 pmid_list, strategy = self._adaptive_search(simplified_query)
1342 if pmid_list:
1343 logger.info(
1344 f"Simplified query found {len(pmid_list)} results"
1345 )
1347 if not pmid_list:
1348 logger.warning("No PubMed results found after query simplification")
1349 return []
1351 # Get article summaries
1352 logger.debug(f"Fetching article summaries for {len(pmid_list)} PMIDs")
1353 summaries = self._get_article_summaries(pmid_list)
1354 logger.debug(f"Retrieved {len(summaries)} summaries")
1356 # ALWAYS fetch abstracts for snippet-only mode to provide context for LLM
1357 logger.debug(
1358 f"Fetching abstracts for {len(pmid_list)} articles for snippet enrichment"
1359 )
1360 abstracts = self._get_article_abstracts(pmid_list)
1361 logger.debug(f"Retrieved {len(abstracts)} abstracts")
1363 # Format as previews
1364 previews = []
1365 for summary in summaries:
1366 # Build snippet from individual metadata preferences
1367 snippet_parts = []
1369 # Check for publication type from esummary (earlier than detailed metadata)
1370 pub_type_prefix = ""
1371 if self.include_publication_type_in_context and summary.get( 1371 ↛ 1375line 1371 didn't jump to line 1375 because the condition on line 1371 was never true
1372 "pubtype"
1373 ):
1374 # Use first publication type from esummary
1375 pub_type_prefix = f"[{summary['pubtype'][0]}] "
1377 # Add authors if enabled
1378 if self.include_authors_in_context and summary.get("authors"):
1379 authors_text = ", ".join(summary.get("authors", []))
1380 if len(authors_text) > 100:
1381 # Truncate long author lists
1382 authors_text = authors_text[:97] + "..."
1383 snippet_parts.append(authors_text)
1385 # Add journal if enabled
1386 if self.include_journal_in_context and summary.get("journal"): 1386 ↛ 1390line 1386 didn't jump to line 1390 because the condition on line 1386 was always true
1387 snippet_parts.append(summary["journal"])
1389 # Add date (full or year only)
1390 if summary.get("pubdate"): 1390 ↛ 1400line 1390 didn't jump to line 1400 because the condition on line 1390 was always true
1391 if self.include_full_date_in_context: 1391 ↛ 1392line 1391 didn't jump to line 1392 because the condition on line 1391 was never true
1392 snippet_parts.append(summary["pubdate"])
1393 elif ( 1393 ↛ 1400line 1393 didn't jump to line 1400 because the condition on line 1393 was always true
1394 self.include_year_in_context
1395 and len(summary["pubdate"]) >= 4
1396 ):
1397 snippet_parts.append(summary["pubdate"][:4])
1399 # Add citation details if enabled
1400 if self.include_citation_in_context:
1401 citation_parts = []
1402 if summary.get("volume"): 1402 ↛ 1404line 1402 didn't jump to line 1404 because the condition on line 1402 was always true
1403 citation_parts.append(f"Vol {summary['volume']}")
1404 if summary.get("issue"): 1404 ↛ 1406line 1404 didn't jump to line 1406 because the condition on line 1404 was always true
1405 citation_parts.append(f"Issue {summary['issue']}")
1406 if summary.get("pages"): 1406 ↛ 1408line 1406 didn't jump to line 1408 because the condition on line 1406 was always true
1407 citation_parts.append(f"pp {summary['pages']}")
1408 if citation_parts: 1408 ↛ 1412line 1408 didn't jump to line 1412 because the condition on line 1408 was always true
1409 snippet_parts.append(f"({', '.join(citation_parts)})")
1411 # Join snippet parts or provide default
1412 if snippet_parts: 1412 ↛ 1423line 1412 didn't jump to line 1423 because the condition on line 1412 was always true
1413 # Use different separators based on what's included
1414 if self.include_authors_in_context:
1415 snippet = ". ".join(
1416 snippet_parts
1417 ) # Authors need period separator
1418 else:
1419 snippet = " - ".join(
1420 snippet_parts
1421 ) # Journal and year use dash
1422 else:
1423 snippet = "Research article"
1425 # Add publication type prefix
1426 snippet = pub_type_prefix + snippet
1428 # Add language indicator if not English
1429 if self.include_language_in_context and summary.get("lang"):
1430 langs = summary["lang"]
1431 if langs and langs[0] != "eng" and langs[0]: 1431 ↛ 1435line 1431 didn't jump to line 1435 because the condition on line 1431 was always true
1432 snippet = f"{snippet} [{langs[0].upper()}]"
1434 # Add identifiers if enabled
1435 identifier_parts = []
1436 if self.include_pmid_in_context and summary.get("id"):
1437 identifier_parts.append(f"PMID: {summary['id']}")
1438 if self.include_doi_in_context and summary.get("doi"):
1439 identifier_parts.append(f"DOI: {summary['doi']}")
1441 if identifier_parts:
1442 snippet = f"{snippet} | {' | '.join(identifier_parts)}"
1444 # ALWAYS include title and abstract in snippet for LLM analysis
1445 pmid = summary["id"]
1446 title = summary["title"]
1447 abstract_text = abstracts.get(pmid, "")
1449 # Truncate abstract if too long
1450 if len(abstract_text) > 500: 1450 ↛ 1451line 1450 didn't jump to line 1451 because the condition on line 1450 was never true
1451 abstract_text = abstract_text[:497] + "..."
1453 # Build the enriched snippet with title and abstract
1454 if abstract_text:
1455 enriched_snippet = f"Title: {title}\n\nAbstract: {abstract_text}\n\nMetadata: {snippet}"
1456 else:
1457 enriched_snippet = f"Title: {title}\n\nMetadata: {snippet}"
1459 # Log the complete snippet for debugging
1460 logger.debug(f"Complete snippet for PMID {pmid}:")
1461 logger.debug(f" Title: {title[:100]}...")
1462 logger.debug(f" Abstract length: {len(abstract_text)} chars")
1463 logger.debug(f" Metadata: {snippet}")
1464 logger.debug(
1465 f" Full enriched snippet ({len(enriched_snippet)} chars): {enriched_snippet[:500]}..."
1466 )
1468 # Create preview with basic information
1469 preview = {
1470 "id": summary["id"],
1471 "title": summary["title"],
1472 "link": summary["link"],
1473 "snippet": enriched_snippet, # Use enriched snippet with title and abstract
1474 "authors": summary.get("authors", []),
1475 "journal": summary.get("journal", ""),
1476 # Alias for the journal reputation filter, which reads
1477 # `journal_ref` (the field name used by arXiv).
1478 # Use None (not empty string) to match other engines so the
1479 # filter treats missing journals consistently.
1480 "journal_ref": summary.get("journal") or None,
1481 # Forward the print / linking ISSN so the reputation
1482 # filter's Tier 2/3 lookups can key on it (faster and
1483 # more reliable than fuzzy name matching). essn is the
1484 # electronic ISSN; prefer it when issn is blank.
1485 "issn": summary.get("issn") or summary.get("essn") or None,
1486 "pubdate": summary.get("pubdate", ""),
1487 "doi": summary.get("doi", ""),
1488 "source": "PubMed",
1489 "_pmid": summary["id"], # Store PMID for later use
1490 "_search_strategy": strategy, # Store search strategy for analytics
1491 }
1493 previews.append(preview)
1495 logger.info(
1496 f"Found {len(previews)} PubMed previews using strategy: {strategy}"
1497 )
1498 if previews: 1498 ↛ 1502line 1498 didn't jump to line 1502 because the condition on line 1498 was always true
1499 logger.debug(
1500 f"Sample preview title: '{previews[0].get('title', 'NO TITLE')[:80]}...'"
1501 )
1502 return previews
1504 def _get_full_content(
1505 self, relevant_items: List[Dict[str, Any]]
1506 ) -> List[Dict[str, Any]]:
1507 """
1508 Get full content for the relevant PubMed articles.
1509 Efficiently manages which content to retrieve (abstracts and/or full text).
1511 Args:
1512 relevant_items: List of relevant preview dictionaries
1514 Returns:
1515 List of result dictionaries with full content
1516 """
1517 logger.info(
1518 f"Getting content for {len(relevant_items)} PubMed articles"
1519 )
1521 # Collect all PMIDs for relevant items
1522 pmids = []
1523 for item in relevant_items:
1524 if "_pmid" in item:
1525 pmids.append(item["_pmid"])
1527 # Get abstracts if requested and PMIDs exist
1528 # In snippet-only mode, always get abstracts as they serve as snippets
1529 abstracts = {}
1530 if self.get_abstracts and pmids:
1531 abstracts = self._get_article_abstracts(pmids)
1533 # Get detailed metadata for all articles (publication types, MeSH terms, etc.)
1534 detailed_metadata = {}
1535 if pmids:
1536 detailed_metadata = self._get_article_detailed_metadata(pmids)
1538 # Find PMC IDs for full-text retrieval (if enabled and not in snippet-only mode)
1539 pmid_to_pmcid = {}
1540 if self.get_full_text and pmids:
1541 pmid_to_pmcid = self._find_pmc_ids(pmids)
1543 # Add content to results
1544 results: List[Dict[str, Any]] = []
1545 for item in relevant_items:
1546 result = item.copy()
1547 pmid = item.get("_pmid", "")
1549 # Add detailed metadata if available
1550 if pmid in detailed_metadata:
1551 metadata = detailed_metadata[pmid]
1553 # Add publication types (e.g., "Clinical Trial", "Meta-Analysis")
1554 if "publication_types" in metadata:
1555 result["publication_types"] = metadata["publication_types"]
1557 # Add first publication type to snippet if enabled
1558 if ( 1558 ↛ 1570line 1558 didn't jump to line 1570 because the condition on line 1558 was always true
1559 self.include_publication_type_in_context
1560 and metadata["publication_types"]
1561 ):
1562 # Just take the first publication type as is
1563 pub_type = metadata["publication_types"][0]
1564 if "snippet" in result: 1564 ↛ 1570line 1564 didn't jump to line 1570 because the condition on line 1564 was always true
1565 result["snippet"] = (
1566 f"[{pub_type}] {result['snippet']}"
1567 )
1569 # Add MeSH terms for medical categorization
1570 if "mesh_terms" in metadata:
1571 result["mesh_terms"] = metadata["mesh_terms"]
1573 # Add MeSH terms to snippet if enabled
1574 if ( 1574 ↛ 1590line 1574 didn't jump to line 1590 because the condition on line 1574 was always true
1575 self.include_mesh_terms_in_context
1576 and metadata["mesh_terms"]
1577 ):
1578 mesh_to_show = (
1579 metadata["mesh_terms"][: self.max_mesh_terms]
1580 if self.max_mesh_terms > 0
1581 else metadata["mesh_terms"]
1582 )
1583 if mesh_to_show and "snippet" in result: 1583 ↛ 1590line 1583 didn't jump to line 1590 because the condition on line 1583 was always true
1584 mesh_text = "MeSH: " + ", ".join(mesh_to_show)
1585 result["snippet"] = (
1586 f"{result['snippet']} | {mesh_text}"
1587 )
1589 # Add keywords
1590 if "keywords" in metadata:
1591 result["keywords"] = metadata["keywords"]
1593 # Add keywords to snippet if enabled
1594 if ( 1594 ↛ 1612line 1594 didn't jump to line 1612 because the condition on line 1594 was always true
1595 self.include_keywords_in_context
1596 and metadata["keywords"]
1597 ):
1598 keywords_to_show = (
1599 metadata["keywords"][: self.max_keywords]
1600 if self.max_keywords > 0
1601 else metadata["keywords"]
1602 )
1603 if keywords_to_show and "snippet" in result: 1603 ↛ 1612line 1603 didn't jump to line 1612 because the condition on line 1603 was always true
1604 keywords_text = "Keywords: " + ", ".join(
1605 keywords_to_show
1606 )
1607 result["snippet"] = (
1608 f"{result['snippet']} | {keywords_text}"
1609 )
1611 # Add affiliations
1612 if "affiliations" in metadata: 1612 ↛ 1613line 1612 didn't jump to line 1613 because the condition on line 1612 was never true
1613 result["affiliations"] = metadata["affiliations"]
1615 # Add funding/grant information
1616 if "grants" in metadata: 1616 ↛ 1617line 1616 didn't jump to line 1617 because the condition on line 1616 was never true
1617 result["grants"] = metadata["grants"]
1619 # Add conflict of interest statement
1620 if "conflict_of_interest" in metadata: 1620 ↛ 1621line 1620 didn't jump to line 1621 because the condition on line 1620 was never true
1621 result["conflict_of_interest"] = metadata[
1622 "conflict_of_interest"
1623 ]
1625 # Add free full text availability
1626 if "has_free_full_text" in metadata:
1627 result["has_free_full_text"] = metadata[
1628 "has_free_full_text"
1629 ]
1630 if "pmc_id" in metadata: 1630 ↛ 1634line 1630 didn't jump to line 1634 because the condition on line 1630 was always true
1631 result["pmc_id"] = metadata["pmc_id"]
1633 # Add PMC availability to snippet if enabled
1634 if ( 1634 ↛ 1644line 1634 didn't jump to line 1644 because the condition on line 1634 was always true
1635 self.include_pmc_availability_in_context
1636 and metadata["has_free_full_text"]
1637 and "snippet" in result
1638 ):
1639 result["snippet"] = (
1640 f"{result['snippet']} | [Free Full Text]"
1641 )
1643 # Add abstract if available
1644 if pmid in abstracts:
1645 result["abstract"] = abstracts[pmid]
1647 # Create enriched content with metadata context
1648 enriched_content = self._create_enriched_content(
1649 result, abstracts[pmid]
1650 )
1652 # ALWAYS include title and abstract in snippet for LLM analysis
1653 # Build comprehensive snippet with title and abstract
1654 title = result.get("title", "")
1655 abstract_text = (
1656 abstracts[pmid][:SNIPPET_LENGTH_LONG]
1657 if len(abstracts[pmid]) > SNIPPET_LENGTH_LONG
1658 else abstracts[pmid]
1659 )
1661 # Prepend title and abstract to the existing metadata snippet
1662 if "snippet" in result: 1662 ↛ 1669line 1662 didn't jump to line 1669 because the condition on line 1662 was always true
1663 # Keep metadata snippet and add content
1664 result["snippet"] = (
1665 f"Title: {title}\n\nAbstract: {abstract_text}\n\nMetadata: {result['snippet']}"
1666 )
1667 else:
1668 # No metadata snippet, just title and abstract
1669 result["snippet"] = (
1670 f"Title: {title}\n\nAbstract: {abstract_text}"
1671 )
1673 # Use abstract as content if no full text
1674 if pmid not in pmid_to_pmcid:
1675 result["full_content"] = enriched_content
1676 result["content"] = enriched_content
1677 result["content_type"] = "abstract"
1679 # Add full text for a limited number of top articles
1680 if (
1681 pmid in pmid_to_pmcid
1682 and self.get_full_text
1683 and len(
1684 [r for r in results if r.get("content_type") == "full_text"]
1685 )
1686 < self.full_text_limit
1687 ):
1688 # Get full text content
1689 pmcid = pmid_to_pmcid[pmid]
1690 full_text = self._get_pmc_full_text(pmcid)
1692 if full_text:
1693 enriched_full_text = self._create_enriched_content(
1694 result, full_text
1695 )
1696 result["full_content"] = enriched_full_text
1697 result["content"] = enriched_full_text
1698 result["content_type"] = "full_text"
1699 result["pmcid"] = pmcid
1700 elif pmid in abstracts: 1700 ↛ 1710line 1700 didn't jump to line 1710 because the condition on line 1700 was always true
1701 # Fall back to abstract if full text retrieval fails
1702 enriched_content = self._create_enriched_content(
1703 result, abstracts[pmid]
1704 )
1705 result["full_content"] = enriched_content
1706 result["content"] = enriched_content
1707 result["content_type"] = "abstract"
1709 # Remove temporary fields
1710 if "_pmid" in result:
1711 del result["_pmid"]
1712 if "_search_strategy" in result:
1713 del result["_search_strategy"]
1715 results.append(result)
1717 return results
1719 def search_by_author(
1720 self, author_name: str, max_results: Optional[int] = None
1721 ) -> List[Dict[str, Any]]:
1722 """
1723 Search for articles by a specific author.
1725 Args:
1726 author_name: Name of the author
1727 max_results: Maximum number of results (defaults to self.max_results)
1729 Returns:
1730 List of articles by the author
1731 """
1732 original_max_results = self.max_results
1734 try:
1735 if max_results:
1736 self.max_results = max_results
1738 query = f"{author_name}[Author]"
1739 return self.run(query)
1741 finally:
1742 # Restore original value
1743 self.max_results = original_max_results
1745 def search_by_journal(
1746 self, journal_name: str, max_results: Optional[int] = None
1747 ) -> List[Dict[str, Any]]:
1748 """
1749 Search for articles in a specific journal.
1751 Args:
1752 journal_name: Name of the journal
1753 max_results: Maximum number of results (defaults to self.max_results)
1755 Returns:
1756 List of articles from the journal
1757 """
1758 original_max_results = self.max_results
1760 try:
1761 if max_results:
1762 self.max_results = max_results
1764 query = f"{journal_name}[Journal]"
1765 return self.run(query)
1767 finally:
1768 # Restore original value
1769 self.max_results = original_max_results
1771 def search_recent(
1772 self, query: str, days: int = 30, max_results: Optional[int] = None
1773 ) -> List[Dict[str, Any]]:
1774 """
1775 Search for recent articles matching the query.
1777 Args:
1778 query: The search query
1779 days: Number of days to look back
1780 max_results: Maximum number of results (defaults to self.max_results)
1782 Returns:
1783 List of recent articles matching the query
1784 """
1785 original_max_results = self.max_results
1786 original_days_limit = self.days_limit
1788 try:
1789 if max_results:
1790 self.max_results = max_results
1792 # Set days limit for this search
1793 self.days_limit = days
1795 return self.run(query)
1797 finally:
1798 # Restore original values
1799 self.max_results = original_max_results
1800 self.days_limit = original_days_limit
1802 def advanced_search(
1803 self, terms: Dict[str, str], max_results: Optional[int] = None
1804 ) -> List[Dict[str, Any]]:
1805 """
1806 Perform an advanced search with field-specific terms.
1808 Args:
1809 terms: Dictionary mapping fields to search terms
1810 Valid fields: Author, Journal, Title, MeSH, Affiliation, etc.
1811 max_results: Maximum number of results (defaults to self.max_results)
1813 Returns:
1814 List of articles matching the advanced query
1815 """
1816 original_max_results = self.max_results
1818 try:
1819 if max_results:
1820 self.max_results = max_results
1822 # Build advanced query string
1823 query_parts = []
1824 for field, term in terms.items():
1825 query_parts.append(f"{term}[{field}]")
1827 query = " AND ".join(query_parts)
1828 return self.run(query)
1830 finally:
1831 # Restore original value
1832 self.max_results = original_max_results