Refining internet search to use sites

src/main/askai/core/component/internet_service.py

@@ -12,17 +12,17 @@
 
    Copyright·(c)·2024,·HSPyLib
 """
-from askai.core.askai_events import AskAiEvents
-from askai.core.askai_messages import msg
+import logging as log
+import os
 from functools import lru_cache
+from typing import Optional
+
 from hspylib.core.metaclass.singleton import Singleton
-from hspylib.core.zoned_datetime import now
 from langchain_community.utilities import GoogleSearchAPIWrapper
 from langchain_core.tools import Tool
-from typing import Optional
 
-import logging as log
-import os
+from askai.core.askai_events import AskAiEvents
+from askai.core.askai_messages import msg
 
 
 class InternetService(metaclass=Singleton):
@@ -43,14 +43,32 @@ def _top_results(self, query: str, max_results: int = 5) -> str:
         return ln.join([f"{i}- {r['snippet']}" for i, r in enumerate(results)])
 
     @lru_cache
-    def search(self, query: str) -> Optional[str]:
+    def search(self, query: str, after: str) -> Optional[str]:
         """Search the web using google search API."""
-        after = f"after: {now('%Y-%m-%d')}"
+        after = f"after: {after}"
         log.info("Searching GOOGLE for '%s' '%s'", query, after)
         AskAiEvents.ASKAI_BUS.events.reply.emit(message=msg.searching())
         search_results = self._tool.run(f"{query} + {after}")
         log.debug(f"Internet search returned: %s", search_results)
         return os.linesep.join(search_results) if isinstance(search_results, list) else search_results
 
+    @lru_cache
+    def search_sites(self, query: str, after: str, *urls: str) -> Optional[str]:
+        """Search the web using google search API.
+        :param query: TODO
+        :param after: TODO
+        :param urls: TODO
+        """
+        search_results = ''
+        AskAiEvents.ASKAI_BUS.events.reply.emit(message=msg.searching())
+        after = f"after: {after}"
+        for url in urls:
+            site = f"site: {url}"
+            log.info("Searching GOOGLE for '%s'  '%s'  '%s'", query, after, site)
+            results = self._tool.run(f"{query} + {site} + {after}")
+            search_results += str(results)
+        log.debug(f"Internet search returned: %s", search_results)
+        return os.linesep.join(search_results) if isinstance(search_results, list) else search_results
+
 
 assert (internet := InternetService().INSTANCE) is not None

src/main/askai/core/processor/generic_processor.py

@@ -44,7 +44,7 @@ def process(self, query_response: QueryResponse) -> Tuple[bool, Optional[str]]:
         final_prompt: str = msg.translate(template.format(user=prompt.user))
         shared.context.set("SETUP", final_prompt, "system")
         shared.context.set("QUESTION", query_response.question)
-        context: ContextRaw = shared.context.join("GENERAL", "SETUP", "QUESTION")
+        context: ContextRaw = shared.context.join("GENERAL", "INTERNET", "SETUP", "QUESTION")
         log.info("Setup::[GENERIC] '%s'  context=%s", query_response.question, context)
         try:
             if (response := shared.engine.ask(context, temperature=1, top_p=1)) and response.is_success:

src/main/askai/core/processor/internet_processor.py

@@ -15,6 +15,7 @@
 import logging as log
 from typing import Optional, Tuple
 
+from hspylib.core.zoned_datetime import now
 from langchain_core.prompts import PromptTemplate
 
 from askai.core.askai_messages import msg
@@ -46,19 +47,19 @@ def process(self, query_response: QueryResponse) -> Tuple[bool, Optional[str]]:
         try:
             if not (response := cache.read_reply(query_response.question)):
                 if (response := shared.engine.ask(context, temperature=0.0, top_p=0.0)) and response.is_success:
-                    search_result: SearchResult = object_mapper.of_json(response.message, SearchResult)
-                    if results := internet.search(" + ".join(search_result.keywords)):
-                        search_result.results = results
-                        output = self._wrap_output(query_response, search_result)
-                        shared.context.set("CONTEXT", output, "assistant")
+                    search: SearchResult = object_mapper.of_json(response.message, SearchResult)
+                    if results := internet.search_sites(" + ".join(search.keywords), now('%Y-%d-%m'), *search.urls):
+                        search.results = results
+                        output = self._wrap_output(query_response, search)
+                        shared.context.set("INTERNET", output, "assistant")
                         cache.save_reply(query_response.question, output)
                         status = True
                 else:
                     output = msg.llm_error(response.message)
             else:
                 log.debug('Reply found for "%s" in cache.', query_response.question)
                 output = response
-                shared.context.set("CONTEXT", output, "assistant")
+                shared.context.set("INTERNET", output, "assistant")
                 status = True
         except Exception as err:
             output = msg.llm_error(str(err))