Spaces:
Sleeping
Sleeping
IngoTB303
Refactor app structure: add agent and tools, enhance question fetching with attachments, and update requirements
1aa16a4
from smolagents import tool, Tool | |
from tavily import TavilyClient | |
import app_tokens | |
def web_search(query: str, proxy: bool = False) -> str: | |
"""Searches the web for your query. | |
Args: | |
query: Your query. | |
proxy: An optional boolean parameter, if a local proxy should be used or not. | |
""" | |
# tavily_client = TavilyClient(api_key=os.getenv("TAVILY_API_KEY")) | |
if proxy: | |
tavily_client = TavilyClient(api_key=app_tokens.get_tavily_api_key(), proxies={'http': 'http://localhost:3128', 'https': 'http://localhost:3128'}, verify=False) | |
else: | |
tavily_client = TavilyClient(api_key=app_tokens.get_tavily_api_key()) | |
response = tavily_client.search(query) | |
return str(response["results"]) | |
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.", | |
}, | |
"proxy": { | |
"type": "boolean", | |
"description": "An optional boolean parameter, if a local proxy should be used or not. Should be True, if the request timed out.", | |
"nullable": "True", | |
}, | |
} | |
output_type = "string" | |
def forward(self, url: str, proxy: bool = True) -> str: | |
try: | |
import re | |
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: | |
if proxy: | |
response = requests.get(url, timeout=20, proxies={'http': 'http://localhost:3128', 'https': 'http://localhost:3128'}, verify=False) | |
else: | |
response = requests.get(url, timeout=20) | |
response.raise_for_status() # Raise an exception for bad status codes | |
markdown_content = markdownify(response.text).strip() | |
markdown_content = re.sub(r"\n{3,}", "\n\n", markdown_content) | |
return truncate_content(markdown_content, 40000) | |
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)}" | |
# test = VisitWebpageTool() | |
# print(test.forward("https://www.wikipedia.de", proxy=True)) |