Add Exa Answer method (#2078)

This adds "Answer" to the Exa toolkit

---------

Co-authored-by: chirag gupta <[email protected]>

cookbook/tools/exa_tools.py

@@ -21,3 +21,8 @@
     "Find me similar papers to https://arxiv.org/pdf/2307.06435 and provide a summary of what they contain",
     markdown=True,
 )
+
+agent.print_response(
+    "What is the latest valuation of SpaceX?",
+    markdown=True,
+)

libs/agno/agno/tools/calcom.py

@@ -174,7 +174,6 @@ def get_upcoming_bookings(self, email: Optional[str] = None) -> str:
             querystring = {"status": "upcoming"}
             if email:
                 querystring["attendeeEmail"] = email
-
 
             response = requests.get(url, headers=self._get_headers(), params=querystring)
             if response.status_code == 200:

libs/agno/agno/tools/exa.py

@@ -21,6 +21,7 @@ class ExaTools(Toolkit):
         text (bool): Retrieve text content from results. Default is True.
         text_length_limit (int): Max length of text content per result. Default is 1000.
         highlights (bool): Include highlighted snippets. Default is True.
+        answer (bool): Enable answer generation. Default is True.
         api_key (Optional[str]): Exa API key. Retrieved from `EXA_API_KEY` env variable if not provided.
         num_results (Optional[int]): Default number of search results. Overrides individual searches if set.
         start_crawl_date (Optional[str]): Include results crawled on/after this date (`YYYY-MM-DD`).
@@ -33,13 +34,15 @@ class ExaTools(Toolkit):
         include_domains (Optional[List[str]]): Restrict results to these domains.
         exclude_domains (Optional[List[str]]): Exclude results from these domains.
         show_results (bool): Log search results for debugging. Default is False.
+        model (Optional[str]): The search model to use. Options are 'exa' or 'exa-pro'.
     """
 
     def __init__(
         self,
         search: bool = True,
         get_contents: bool = True,
         find_similar: bool = True,
+        answer: bool = True,
         text: bool = True,
         text_length_limit: int = 1000,
         highlights: bool = True,
@@ -57,6 +60,7 @@ def __init__(
         include_domains: Optional[List[str]] = None,
         exclude_domains: Optional[List[str]] = None,
         show_results: bool = False,
+        model: Optional[str] = None,
     ):
         super().__init__(name="exa")
 
@@ -82,13 +86,16 @@ def __init__(
         self.category: Optional[str] = category
         self.include_domains: Optional[List[str]] = include_domains
         self.exclude_domains: Optional[List[str]] = exclude_domains
+        self.model: Optional[str] = model
 
         if search:
             self.register(self.search_exa)
         if get_contents:
             self.register(self.get_contents)
         if find_similar:
             self.register(self.find_similar)
+        if answer:
+            self.register(self.exa_answer)
 
     def _parse_results(self, exa_results: SearchResponse) -> str:
         exa_results_parsed = []
@@ -227,3 +234,47 @@ def find_similar(self, url: str, num_results: int = 5) -> str:
         except Exception as e:
             logger.error(f"Failed to get similar links from Exa: {e}")
             return f"Error: {e}"
+
+    def exa_answer(self, query: str, text: bool = False) -> str:
+        """
+        Get an LLM answer to a question informed by Exa search results.
+
+        Args:
+            query (str): The question or query to answer.
+            text (bool): Include full text from citation. Default is False.
+        Returns:
+            str: The answer results in JSON format with both generated answer and sources.
+        """
+
+        if self.model and self.model not in ["exa", "exa-pro"]:
+            raise ValueError("Model must be either 'exa' or 'exa-pro'")
+        try:
+            logger.info(f"Generating answer for query: {query}")
+            answer_kwargs: Dict[str, Any] = {
+                "model": self.model,
+                "text": text,
+            }
+            answer_kwargs = {k: v for k, v in answer_kwargs.items() if v is not None}
+            answer = self.exa.answer(query=query, **answer_kwargs)
+            result = {
+                "answer": answer.answer,
+                "citations": [
+                    {
+                        "id": citation.id,
+                        "url": citation.url,
+                        "title": citation.title,
+                        "published_date": citation.published_date,
+                        "author": citation.author,
+                        "text": citation.text if text else None,
+                    }
+                    for citation in answer.citations
+                ],
+            }
+            if self.show_results:
+                logger.info(json.dumps(result))
+
+            return json.dumps(result, indent=4)
+
+        except Exception as e:
+            logger.error(f"Failed to get answer from Exa: {e}")
+            return f"Error: {e}"

libs/agno/agno/tools/googlesheets.py

@@ -298,23 +298,23 @@ def create_duplicate_sheet(
         """
         if not self.creds:
             return "Not authenticated. Call auth() first."
-        
+
         if not self.service:
             return "Service not initialized"
-        
+
         try:
             # Ensure the drive scope is included
             if "https://www.googleapis.com/auth/drive" not in self.scopes:
                 self.scopes.append("https://www.googleapis.com/auth/drive")
                 self._auth()  # Re-authenticate with updated scopes
-            
+
             drive_service = build("drive", "v3", credentials=self.creds)
-            
+
             # Use new_title if provided, otherwise fetch the title from the source spreadsheet
             if not new_title:
                 source_sheet = self.service.spreadsheets().get(spreadsheetId=source_id).execute()
                 new_title = source_sheet["properties"]["title"]
-            
+
             body = {"name": new_title}
             new_file = drive_service.files().copy(fileId=source_id, body=body).execute()
             new_spreadsheet_id = new_file.get("id")