| import re |
| import time |
| from typing import Any |
|
|
| from smolagents import tool, Tool |
| import duckduckgo_search |
| import wikipediaapi |
|
|
|
|
| class FinalAnswerTool(Tool): |
| name = "final_answer" |
| description = "Provides a final answer to the given problem." |
| inputs = {'answer': {'type': 'string', 'description': 'The final answer to the problem'}} |
| output_type = "string" |
|
|
| def forward(self, answer: str) -> str: |
| """ |
| Submits the final answer to the user. |
| """ |
| print(f"Final Answer: {answer}") |
| return answer |
|
|
| class DuckDuckGoSearchTool(Tool): |
| name = "web_search" |
| description = "Performs a duckduckgo web search based on your query (think a Google search) then returns the top search results." |
| inputs = {'query': {'type': 'string', 'description': 'The search query to perform.'}} |
| output_type = "string" |
|
|
| def __init__(self, max_results=10, **kwargs): |
| super().__init__() |
| self.max_results = max_results |
| try: |
| from duckduckgo_search import DDGS |
| except ImportError as e: |
| raise ImportError( |
| "You must install package `duckduckgo_search` to run this tool: for instance run `pip install duckduckgo-search`." |
| ) from e |
| self.ddgs = DDGS(**kwargs) |
|
|
| def forward(self, query: str) -> str: |
| max_retries = 3 |
| retry_delay = 2 |
| |
| for attempt in range(max_retries): |
| try: |
| results = self.ddgs.text(query, max_results=self.max_results) |
| if len(results) == 0: |
| raise Exception("No results found! Try a less restrictive/shorter query.") |
| postprocessed_results = [f"[{result['title']}]({result['href']})\n{result['body']}" for result in results] |
| return "## Search Results\n\n" + "\n\n".join(postprocessed_results) |
| |
| except Exception as e: |
| error_msg = str(e) |
| if "202" in error_msg or "rate" in error_msg.lower() or "ratelimit" in error_msg.lower(): |
| if attempt < max_retries - 1: |
| print(f"Rate limit hit, retrying in {retry_delay} seconds... (attempt {attempt + 1}/{max_retries})") |
| time.sleep(retry_delay) |
| retry_delay *= 2 |
| continue |
| else: |
| return "## Search Results\n\nRate limit exceeded. Please try again later or use a different search strategy." |
| else: |
| |
| raise e |
| |
| return "## Search Results\n\nFailed to get search results after multiple attempts." |
| |
| class VisitWebpageTool(Tool): |
| name = "visit_webpage" |
| description = "Visits a webpage at the given url and reads its content as a markdown string. Use this to browse webpages." |
| inputs = {'url': {'type': 'string', 'description': 'The url of the webpage to visit.'}} |
| output_type = "string" |
|
|
| def forward(self, url: str) -> str: |
| try: |
| import requests |
| from markdownify import markdownify |
| from requests.exceptions import RequestException |
|
|
| from smolagents.utils import truncate_content |
| except ImportError as e: |
| raise ImportError( |
| "You must install packages `markdownify` and `requests` to run this tool: for instance run `pip install markdownify requests`." |
| ) from e |
| try: |
| |
| response = requests.get(url, timeout=20) |
| response.raise_for_status() |
|
|
| |
| markdown_content = markdownify(response.text).strip() |
|
|
| |
| markdown_content = re.sub(r"\n{3,}", "\n\n", markdown_content) |
|
|
| return truncate_content(markdown_content, 10000) |
|
|
| except requests.exceptions.Timeout: |
| return "The request timed out. Please try again later or check the URL." |
| except RequestException as e: |
| return f"Error fetching the webpage: {str(e)}" |
| except Exception as e: |
| return f"An unexpected error occurred: {str(e)}" |
|
|
| def __init__(self, *args, **kwargs): |
| self.is_initialized = False |
|
|
| class WikipediaSearchTool(Tool): |
| name = "wikipedia_search" |
| description = "Searches Wikipedia for a given query and returns the summary of the most relevant page. Use this for questions about specific facts, people, places, or events." |
| inputs = {'query': {'type': 'string', 'description': 'The search query to perform.'}} |
| output_type = "string" |
|
|
| def __init__(self, **kwargs): |
| super().__init__(**kwargs) |
| self.wiki_wiki = wikipediaapi.Wikipedia( |
| language='en', |
| user_agent='SmolAgents-Course-Assignment/1.0' |
| ) |
|
|
| def forward(self, query: str) -> str: |
| """ |
| Searches Wikipedia for a given query and returns the summary of the most relevant page. |
| Use this for questions about specific facts, people, places, or events. |
| """ |
| try: |
| page = self.wiki_wiki.page(query) |
| if page.exists(): |
| return f"Title: {page.title}\nSummary: {page.summary}" |
| else: |
| return f"Could not find a Wikipedia page for '{query}'." |
| except Exception as e: |
| return f"An error occurred while searching Wikipedia: {e}" |
|
|
