Spaces:
Running
on
Zero
Running
on
Zero
| from skills.skill import Skill | |
| from serpapi import GoogleSearch | |
| import openai | |
| from bs4 import BeautifulSoup | |
| import requests | |
| import re | |
| headers = { | |
| "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/94.0.4606.81 Safari/537.36" | |
| } | |
| class WebSearch(Skill): | |
| name = 'web_search' | |
| description = 'A tool that performs web searches.' | |
| api_keys_required = [['openai'],['serpapi']] | |
| def __init__(self, api_keys): | |
| super().__init__(api_keys) | |
| def execute(self, params, dependent_task_outputs, objective): | |
| # Your function goes here | |
| # Modify the query based on the dependent task output | |
| if dependent_task_outputs != "": | |
| dependent_task = f"Use the dependent task output below as reference to help craft the correct search query for the provided task above. Dependent task output:{dependent_task_outputs}." | |
| else: | |
| dependent_task = "." | |
| query = self.text_completion_tool("You are an AI assistant tasked with generating a Google search query based on the following task: "+params+". If the task looks like a search query, return the identical search query as your response. " + dependent_task + "\nSearch Query:") | |
| print("\033[90m\033[3m"+"Search query: " +str(query)+"\033[0m") | |
| # Set the search parameters | |
| search_params = { | |
| "engine": "google", | |
| "q": query, | |
| "api_key": self.serpapi_api_key, | |
| "num": 3 | |
| } | |
| # Perform the web search | |
| search_results = GoogleSearch(search_params).get_dict() | |
| # Simplify the search results | |
| search_results = self.simplify_search_results(search_results.get('organic_results', [])) | |
| print("\033[90m\033[3mCompleted search. Now scraping results.\n\033[0m") | |
| # Store the results from web scraping | |
| results = "" | |
| for result in search_results: | |
| url = result.get('link') | |
| print("\033[90m\033[3m" + "Scraping: "+url+"" + "...\033[0m") | |
| content = self.web_scrape_tool({"url": url, "task": params,"objective":objective}) | |
| results += str(content) + ". " | |
| print("\033[90m\033[3m"+str(results[0:100])[0:100]+"...\033[0m") | |
| # Process the results and generate a report | |
| results = self.text_completion_tool(f"You are an expert analyst combining the results of multiple web scrapes. Rewrite the following information as one cohesive report without removing any facts. Ignore any reports of not having info, unless all reports say so - in which case explain that the search did not work and suggest other web search queries to try. \n###INFORMATION:{results}.\n###REPORT:") | |
| return results | |
| def simplify_search_results(self, search_results): | |
| simplified_results = [] | |
| for result in search_results: | |
| simplified_result = { | |
| "position": result.get("position"), | |
| "title": result.get("title"), | |
| "link": result.get("link"), | |
| "snippet": result.get("snippet") | |
| } | |
| simplified_results.append(simplified_result) | |
| return simplified_results | |
| def text_completion_tool(self, prompt: str): | |
| messages = [ | |
| {"role": "user", "content": prompt} | |
| ] | |
| response = openai.ChatCompletion.create( | |
| model="gpt-3.5-turbo-16k-0613", | |
| messages=messages, | |
| temperature=0.2, | |
| max_tokens=1500, | |
| top_p=1, | |
| frequency_penalty=0, | |
| presence_penalty=0 | |
| ) | |
| return response.choices[0].message['content'].strip() | |
| def web_scrape_tool(self, params): | |
| content = self.fetch_url_content(params['url']) | |
| if content is None: | |
| return None | |
| text = self.extract_text(content) | |
| print("\033[90m\033[3m"+"Scrape completed. Length:" +str(len(text))+".Now extracting relevant info..."+"...\033[0m") | |
| info = self.extract_relevant_info(params['objective'], text[0:11000], params['task']) | |
| links = self.extract_links(content) | |
| #result = f"{info} URLs: {', '.join(links)}" | |
| result = info | |
| return result | |
| def fetch_url_content(self,url: str): | |
| try: | |
| response = requests.get(url, headers=headers, timeout=10) | |
| response.raise_for_status() | |
| return response.content | |
| except requests.exceptions.RequestException as e: | |
| print(f"Error while fetching the URL: {e}") | |
| return "" | |
| def extract_links(self,content: str): | |
| soup = BeautifulSoup(content, "html.parser") | |
| links = [link.get('href') for link in soup.findAll('a', attrs={'href': re.compile("^https?://")})] | |
| return links | |
| def extract_text(self,content: str): | |
| soup = BeautifulSoup(content, "html.parser") | |
| text = soup.get_text(strip=True) | |
| return text | |
| def extract_relevant_info(self, objective, large_string, task): | |
| chunk_size = 12000 | |
| overlap = 500 | |
| notes = "" | |
| if len(large_string) == 0: | |
| print("error scraping") | |
| return "Error scraping." | |
| for i in range(0, len(large_string), chunk_size - overlap): | |
| print("\033[90m\033[3m"+"Reading chunk..."+"\033[0m") | |
| chunk = large_string[i:i + chunk_size] | |
| messages = [ | |
| {"role": "system", "content": f"You are an AI assistant."}, | |
| {"role": "user", "content": f"You are an expert AI research assistant tasked with creating or updating the current notes. If the current note is empty, start a current-notes section by exracting relevant data to the task and objective from the chunk of text to analyze. If there is a current note, add new relevant info frol the chunk of text to analyze. Make sure the new or combined notes is comprehensive and well written. Here's the current chunk of text to analyze: {chunk}. ### Here is the current task: {task}.### For context, here is the objective: {objective}.### Here is the data we've extraced so far that you need to update: {notes}.### new-or-updated-note:"} | |
| ] | |
| response = openai.ChatCompletion.create( | |
| model="gpt-3.5-turbo-16k-0613", | |
| messages=messages, | |
| max_tokens=800, | |
| n=1, | |
| stop="###", | |
| temperature=0.7, | |
| ) | |
| notes += response.choices[0].message['content'].strip()+". "; | |
| return notes |