File size: 5,629 Bytes
8829b43 3657cd4 8829b43 b648690 0c57329 8829b43 0c57329 b648690 8829b43 b648690 8829b43 b648690 0c57329 3657cd4 0c57329 8829b43 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 |
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 # seconds
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 # Exponential backoff
continue
else:
return "## Search Results\n\nRate limit exceeded. Please try again later or use a different search strategy."
else:
# For non-rate-limit errors, don't retry
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:
# Send a GET request to the URL with a 20-second timeout
response = requests.get(url, timeout=20)
response.raise_for_status() # Raise an exception for bad status codes
# Convert the HTML content to Markdown
markdown_content = markdownify(response.text).strip()
# Remove multiple line breaks
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}"
|