diff --git a/config.example.yaml b/config.example.yaml index d41fe79..ef181f2 100644 --- a/config.example.yaml +++ b/config.example.yaml @@ -81,6 +81,15 @@ watchers: Review the following RSS feed entries and select only the most relevant and interesting ones. Provide a concise digest with titles and links. If no entries are relevant, respond with: "No relevant entries found." + enable_llm_filter: false # Use LLM to filter entries for relevance + llm_filter: + model_name: "gpt-4o-mini" # LLM model to use + prompt: | + Evaluate the recommendation score of the following RSS feed entry to current events. + The score ranges from 0 (not relevant) to 5 (highly relevant). + Respond with only the score in the format: \\boxed{score}. + max_concurrent_queries: 50 # Max concurrent LLM filter queries (default: 50) + score_threshold: 3 # Minimum score (0-5) to consider entry relevant # OpenReview Watcher # Monitors OpenReview submissions for new reviews, comments, and status changes diff --git a/watchers/rss_watcher.py b/watchers/rss_watcher.py index 508f2fc..1c3d1a0 100644 --- a/watchers/rss_watcher.py +++ b/watchers/rss_watcher.py @@ -7,6 +7,7 @@ import asyncio import hashlib +import re from collections.abc import Coroutine import aiohttp @@ -26,6 +27,11 @@ "Return a concise digest with entry titles and links. " "If no entries are relevant, respond with 'No relevant entries found.'" ) +DEFAULT_LLM_FILTER_PROMPT = ( + "Evaluate the recommendation score of the following RSS feed entry to current events. " + "The score ranges from 0 (not relevant) to 5 (highly relevant). " + "Respond with only the score in the format: \\boxed{score}." +) type EntryData = dict[str, str] type ConfigDict = dict[str, str | int | bool | dict[str, str]] @@ -56,8 +62,18 @@ def __init__(self, feed_config: ConfigDict, cache_id: str = "") -> None: digest_config = feed_config.get("digest", {}) digest_dict: dict[str, str] = digest_config if isinstance(digest_config, dict) else {} self.digest_prompt: str = str(digest_dict.get("prompt", DEFAULT_DIGEST_PROMPT)) - model_name = digest_dict.get("model_name") - self.digest_model: str = str(model_name) if model_name else "" + digest_model_name = digest_dict.get("model_name") + self.digest_model: str = str(digest_model_name) if digest_model_name else "" + + # LLM Filter configuration + self.enable_llm_filter: bool = bool(feed_config.get("enable_llm_filter", False)) + llm_filter_config = feed_config.get("llm_filter", {}) + llm_filter_dict: dict[str, str | int] = llm_filter_config if isinstance(llm_filter_config, dict) else {} + self.llm_filter_prompt: str = str(llm_filter_dict.get("prompt", DEFAULT_LLM_FILTER_PROMPT)) + llm_filter_model_name = llm_filter_dict.get("model_name") + self.llm_filter_model: str = str(llm_filter_model_name) if llm_filter_model_name else "" + self.max_concurrent_llm_filter_queries: int = int(llm_filter_dict.get("max_concurrent_queries", 50)) + self.llm_filter_score_threshold: int = int(llm_filter_dict.get("score_threshold", 3)) # Initialize StateDiff for persistent state tracking # Use hash of URL combined with cache_id to ensure uniqueness per user per feed @@ -147,10 +163,6 @@ async def fetch_feed(self) -> feedparser.FeedParserDict | None: try: proxy_info = f" (via proxy {self.proxy})" if self.proxy else "" logger.warning(f"Fetching RSS feed: {self}{proxy_info}") - # Use asyncio.to_thread for blocking fetch call - # feed: feedparser.FeedParserDict = await asyncio.to_thread( - # self._fetch_feed_sync - # ) feed: feedparser.FeedParserDict = await self._fetch_feed_aiohttp() # Check for feed errors @@ -203,7 +215,7 @@ async def get_new_entries(self) -> list[EntryData]: total_entries = len(feed.entries) # type: ignore[attr-defined] if total_entries == 0: - logger.warning("There is no entries in the fetched feed!") + logger.debug("There are no entries in the fetched feed!") # Extract all current entries current_entries: list[JsonSerializable] = [ @@ -226,7 +238,7 @@ async def get_new_entries(self) -> list[EntryData]: f"Found {len(new_entries)} new entries (out of {total_entries} total) in {self}" ) else: - logger.warning( + logger.debug( f"No new entries found (checked {total_entries} entries) in {self}" ) @@ -302,8 +314,30 @@ async def process_entries( if not entries: return + if feed.enable_llm_filter and feed.llm_filter_model: + # Use LLM to filter relevant entries + logger.info(f"Filtering {len(entries)} entries with LLM for {feed}") + try: + filtered_entries = await self.filter_entries_with_llm( + feed=feed, + entries=entries, + ) + + # Check if no entries were deemed relevant + if len(filtered_entries) == 0: + logger.info(f"No relevant entries found by LLM for {feed}") + return + + except Exception as e: + logger.error(f"Failed to filter entries with LLM: {e}", exc_info=True) + # Fall back to unfiltered entries + filtered_entries = entries + else: + # All entries without filtering + filtered_entries = entries + # Format entries for LLM - entries_text = feed.format_entries_for_llm(entries) + entries_text = feed.format_entries_for_llm(filtered_entries) if feed.enable_digest and feed.digest_model: # Use LLM to create digest @@ -331,14 +365,66 @@ async def process_entries( except Exception as e: logger.error(f"Failed to generate digest: {e}", exc_info=True) # Fallback to simple list - markdown_content = self.format_simple_list_markdown(feed, entries) + markdown_content = self.format_simple_list_markdown(feed, filtered_entries) else: - # Simple notification without LLM - markdown_content = self.format_simple_list_markdown(feed, entries) + # Simple notification without LLM digest + markdown_content = self.format_simple_list_markdown(feed, filtered_entries) # Send notification await self.notifier.send(markdown_content) + async def filter_entries_with_llm( + self, feed: RSSFeed, entries: list[EntryData] + ) -> list[EntryData]: + """ + Filter entries using LLM based on relevance. + + Args: + feed: RSSFeed instance + entries: List of entry dictionaries + + Returns: + List of relevant entry dictionaries + """ + filtered_entries: list[EntryData] = [] + semaphore = asyncio.Semaphore(feed.max_concurrent_llm_filter_queries) + + async def is_entry_relevant(entry: EntryData) -> bool: + """Check if a single entry is relevant using LLM.""" + async with semaphore: + entry_text = f"Title: {entry['title']}\nSummary: {entry['summary']}" + response = await llm_query( + message=f"Entry:\n\n{entry_text}", + model=feed.llm_filter_model, + system_message=feed.llm_filter_prompt, + ) + # Extract score from LLM response in format: \boxed{score} + matches = re.findall(r"\\boxed{\s*(\d+)\s*}", response) + if matches: + score = int(matches[0]) + return score >= feed.llm_filter_score_threshold + else: + logger.warning( + "LLM filter response missing valid \\boxed{} score for entry '{title}'. " + "Falling back to including entry. Raw response: {response_snippet}", + title=entry.get("title", ""), + response_snippet=response[:200], + ) + return True + + # Create tasks for all entries + tasks = [is_entry_relevant(entry) for entry in entries] + results = await asyncio.gather(*tasks, return_exceptions=True) + + # Collect relevant entries + for entry, result in zip(entries, results): + if isinstance(result, Exception): + logger.debug(f"Error checking relevance for entry {entry['title']}: {result}", exc_info=True) + continue + if result: + filtered_entries.append(entry) + return filtered_entries + def format_simple_list_markdown( self, feed: RSSFeed, entries: list[EntryData] ) -> str: @@ -368,11 +454,8 @@ async def check_all_feeds(self) -> None: """Check all configured RSS feeds for new entries.""" for feed in self.feeds: try: - logger.info("I am going to get new entries") new_entries = await feed.get_new_entries() - logger.info("I have got new entries and am going to process") await self.process_entries(feed, new_entries) - logger.info("I have processed entries") except Exception as e: logger.error(f"Error checking feed {feed}: {e}", exc_info=True) @@ -391,9 +474,7 @@ async def monitor_loop(self) -> None: while True: try: - logger.info("I am going to check all feeds") await self.check_all_feeds() - logger.info("I have checked all feeds") await asyncio.sleep(self.interval) except asyncio.CancelledError: