from abc import abstractmethod import uuid from text2vec import semantic_search from utils import ( get_relevant_history, load_knowledge_base_qa, load_knowledge_base_UnstructuredFile, get_embedding, extract, ) import json from typing import Dict, List import os from googleapiclient.discovery import build import requests from selenium import webdriver from selenium.webdriver.common.by import By from selenium.webdriver.support.ui import WebDriverWait from selenium.webdriver.support import expected_conditions as EC from bs4 import BeautifulSoup import base64 import re from datetime import datetime, timedelta from typing import Tuple, List, Any, Dict from email.mime.text import MIMEText from email.mime.multipart import MIMEMultipart from google.auth.transport.requests import Request from google.oauth2.credentials import Credentials from google_auth_oauthlib.flow import InstalledAppFlow from googleapiclient.discovery import build from googleapiclient.errors import HttpError from tqdm import tqdm class ToolComponent: def __init__(self): pass @abstractmethod def func(self): pass class KnowledgeBaseComponent(ToolComponent): """ Inject knowledge base top_k : Top_k with the highest matching degree type : "QA" or others knowledge_base(json_path) : knowledge_base_path """ def __init__(self, top_k, type, knowledge_base): super().__init__() self.top_k = top_k self.type = type self.knowledge_base = knowledge_base if self.type == "QA": ( self.kb_embeddings, self.kb_questions, self.kb_answers, self.kb_chunks, ) = load_knowledge_base_qa(self.knowledge_base) else: self.kb_embeddings, self.kb_chunks = load_knowledge_base_UnstructuredFile( self.knowledge_base ) def func(self, agent): query = ( agent.long_term_memory[-1]["content"] if len(agent.long_term_memory) > 0 else "" ) knowledge = "" query = extract(query, "query") query_embedding = get_embedding(query) hits = semantic_search(query_embedding, self.kb_embeddings, top_k=50) hits = hits[0] temp = [] if self.type == "QA": for hit in hits: matching_idx = hit["corpus_id"] if self.kb_chunks[matching_idx] in temp: pass else: knowledge = ( knowledge + f"question:{self.kb_questions[matching_idx]},answer:{self.kb_answers[matching_idx]}\n\n" ) temp.append(self.kb_answers[matching_idx]) if len(temp) == 1: break print(hits[0]["score"]) score = hits[0]["score"] if score < 0.5: return {"prompt": "No matching knowledge base"} else: return {"prompt": "The relevant content is: " + knowledge + "\n"} else: for hit in hits: matching_idx = hit["corpus_id"] if self.kb_chunks[matching_idx] in temp: pass else: knowledge = knowledge + f"{self.kb_answers[matching_idx]}\n\n" temp.append(self.kb_answers[matching_idx]) if len(temp) == self.top_k: break print(hits[0]["score"]) score = hits[0]["score"] if score < 0.5: return {"prompt": "No matching knowledge base"} else: print(knowledge) return {"prompt": "The relevant content is: " + knowledge + "\n"} class StaticComponent(ToolComponent): "Return static response" def __init__(self, output): super().__init__() self.output = output def func(self, agent): outputdict = {"response": self.output} return outputdict class ExtractComponent(ToolComponent): """ Extract keywords based on the current scene and store them in the environment extract_words(list) : Keywords to be extracted system_prompt & last_prompt : Prompt to extract keywords """ def __init__( self, extract_words, system_prompt, last_prompt=None, ): super().__init__() self.extract_words = extract_words self.system_prompt = system_prompt self.default_prompt = ( "Please strictly adhere to the following format for outputting:\n" ) for extract_word in extract_words: self.default_prompt += ( f"<{extract_word}> the content you need to extract " ) self.last_prompt = last_prompt if last_prompt else self.default_prompt def func(self, agent): response = agent.LLM.get_response( agent.long_term_memory, self.system_prompt, self.last_prompt, stream=False, ) for extract_word in self.extract_words: key = extract(response, extract_word) key = key if key else response agent.environment.shared_memory[extract_word] = key return {} """Search sources: chatgpt/search engines/specific search sources/can even be multimodal (if it comes to clothing)""" class WebSearchComponent(ToolComponent): """search engines""" __ENGINE_NAME__: List = ["google", "bing"] def __init__(self, engine_name: str, api: Dict): """ :param engine_name: The name of the search engine used :param api: Pass in a dictionary, such as {"bing":"key1", "google":"key2", ...}, of course each value can also be a list, or more complicated """ super(WebSearchComponent, self).__init__() """Determine whether the key and engine_name of the api are legal""" assert engine_name in WebSearchComponent.__ENGINE_NAME__ for api_name in api: assert api_name in WebSearchComponent.__ENGINE_NAME__ self.api = api self.engine_name = engine_name self.search: Dict = {"bing": self._bing_search, "google": self._google_search} def _bing_search(self, query: str, **kwargs): """Initialize search hyperparameters""" subscription_key = self.api["bing"] search_url = "https://api.bing.microsoft.com/v7.0/search" headers = {"Ocp-Apim-Subscription-Key": subscription_key} params = { "q": query, "textDecorations": True, "textFormat": "HTML", "count": 10, } """start searching""" response = requests.get(search_url, headers=headers, params=params) response.raise_for_status() results = response.json()["webPages"]["value"] """execute""" metadata_results = [] for result in results: metadata_result = { "snippet": result["snippet"], "title": result["name"], "link": result["url"], } metadata_results.append(metadata_result) return {"meta data": metadata_results} def _google_search(self, query: str, **kwargs): """Initialize search hyperparameters""" api_key = self.api[self.engine_name]["api_key"] cse_id = self.api[self.engine_name]["cse_id"] service = build("customsearch", "v1", developerKey=api_key) """start searching""" results = ( service.cse().list(q=query, cx=cse_id, num=10, **kwargs).execute()["items"] ) """execute""" metadata_results = [] for result in results: metadata_result = { "snippet": result["snippet"], "title": result["title"], "link": result["link"], } metadata_results.append(metadata_result) return {"meta data": metadata_results} def func(self, agent, **kwargs) -> Dict: query = ( agent.long_term_memory[-1]["content"] if len(agent.long_term_memory) > 0 else " " ) response = agent.LLM.get_response( None, system_prompt=f"Please analyze the provided conversation and identify keywords that can be used for a search engine query. Format the output as extracted keywords:\nConversation:\n{query}", stream=False, ) response = extract(response, "keywords") query = response if response else query search_results = self.search[self.engine_name](query=query, **kwargs) information = "" for i in search_results["meta data"][:5]: information += i["snippet"] return { "prompt": "You can refer to the following information to reply:\n" + information } def convert_search_engine_to(self, engine_name): assert engine_name in WebSearchComponent.__ENGINE_NAME__ self.engine_name = engine_name class WebCrawlComponent(ToolComponent): """Open a single web page for crawling""" def __init__(self): super(WebCrawlComponent, self).__init__() def func(self, agent_dict) -> Dict: url = agent_dict["url"] print(f"crawling {url} ......") content = "" """Crawling content from url may need to be carried out according to different websites, such as wiki, baidu, zhihu, etc.""" driver = webdriver.Chrome() try: """open url""" driver.get(url) """wait 20 second""" wait = WebDriverWait(driver, 20) wait.until(EC.presence_of_element_located((By.TAG_NAME, "body"))) """crawl code""" page_source = driver.page_source """parse""" soup = BeautifulSoup(page_source, "html.parser") """concatenate""" for paragraph in soup.find_all("p"): content = f"{content}\n{paragraph.get_text()}" except Exception as e: print("Error:", e) finally: """quit""" driver.quit() return {"content": content.strip()} class MailComponent(ToolComponent): __VALID_ACTION__ = ["read", "send"] def __init__( self, cfg_file: str, default_action: str = "read", name: str = "e-mail" ): """'../config/google_mail.json'""" super(MailComponent, self).__init__(name) self.name = name assert ( default_action.lower() in self.__VALID_ACTION__ ), f"Action `{default_action}` is not allowed! The valid action is in `{self.__VALID_ACTION__}`" self.action = default_action.lower() self.credential = self._login(cfg_file) def _login(self, cfg_file: str): SCOPES = [ "https://www.googleapis.com/auth/gmail.readonly", "https://www.googleapis.com/auth/gmail.send", ] creds = None if os.path.exists("token.json"): print("Login Successfully!") creds = Credentials.from_authorized_user_file("token.json", SCOPES) if not creds or not creds.valid: print("Please authorize in an open browser.") if creds and creds.expired and creds.refresh_token: creds.refresh(Request()) else: flow = InstalledAppFlow.from_client_secrets_file(cfg_file, SCOPES) creds = flow.run_local_server(port=0) # Save the credentials for the next run with open("token.json", "w") as token: token.write(creds.to_json()) return creds def _read(self, mail_dict: dict): credential = self.credential state = mail_dict["state"] if "state" in mail_dict else None time_between = ( mail_dict["time_between"] if "time_between" in mail_dict else None ) sender_mail = mail_dict["sender_mail"] if "sender_mail" in mail_dict else None only_both = mail_dict["only_both"] if "only_both" in mail_dict else False order_by_time = ( mail_dict["order_by_time"] if "order_by_time" in mail_dict else "descend" ) include_word = ( mail_dict["include_word"] if "include_word" in mail_dict else None ) exclude_word = ( mail_dict["exclude_word"] if "exclude_word" in mail_dict else None ) MAX_SEARCH_CNT = ( mail_dict["MAX_SEARCH_CNT"] if "MAX_SEARCH_CNT" in mail_dict else 50 ) number = mail_dict["number"] if "number" in mail_dict else 10 if state is None: state = "all" if time_between is not None: assert isinstance(time_between, tuple) assert len(time_between) == 2 assert state in ["all", "unread", "read", "sent"] if only_both: assert sender_mail is not None if sender_mail is not None: assert isinstance(sender_mail, str) assert credential assert order_by_time in ["descend", "ascend"] def generate_query(): query = "" if state in ["unread", "read"]: query = f"is:{state}" if state in ["sent"]: query = f"in:{state}" if only_both: query = f"{query} from:{sender_mail} OR to:{sender_mail}" if sender_mail is not None and not only_both: query = f"{query} from:({sender_mail})" if include_word is not None: query = f"{query} {include_word}" if exclude_word is not None: query = f"{query} -{exclude_word}" if time_between is not None: TIME_FORMAT = "%Y/%m/%d" t1, t2 = time_between if t1 == "now": t1 = datetime.now().strftime(TIME_FORMAT) if t2 == "now": t2 = datetime.now().strftime(TIME_FORMAT) if isinstance(t1, str) and isinstance(t2, str): t1 = datetime.strptime(t1, TIME_FORMAT) t2 = datetime.strptime(t2, TIME_FORMAT) elif isinstance(t1, str) and isinstance(t2, int): t1 = datetime.strptime(t1, TIME_FORMAT) t2 = t1 + timedelta(days=t2) elif isinstance(t1, int) and isinstance(t2, str): t2 = datetime.strptime(t2, TIME_FORMAT) t1 = t2 + timedelta(days=t1) else: assert False, "invalid time" if t1 > t2: t1, t2 = t2, t1 query = f"{query} after:{t1.strftime(TIME_FORMAT)} before:{t2.strftime(TIME_FORMAT)}" return query.strip() def sort_by_time(data: List[Dict]): if order_by_time == "descend": reverse = True else: reverse = False sorted_data = sorted( data, key=lambda x: datetime.strptime(x["time"], "%Y-%m-%d %H:%M:%S"), reverse=reverse, ) return sorted_data try: service = build("gmail", "v1", credentials=credential) results = ( service.users() .messages() .list(userId="me", labelIds=["INBOX"], q=generate_query()) .execute() ) messages = results.get("messages", []) email_data = list() if not messages: print("No eligible emails.") return None else: pbar = tqdm(total=min(MAX_SEARCH_CNT, len(messages))) for cnt, message in enumerate(messages): pbar.update(1) if cnt >= MAX_SEARCH_CNT: break msg = ( service.users() .messages() .get( userId="me", id=message["id"], format="full", metadataHeaders=None, ) .execute() ) subject = "" for header in msg["payload"]["headers"]: if header["name"] == "Subject": subject = header["value"] break sender = "" for header in msg["payload"]["headers"]: if header["name"] == "From": sender = re.findall( r"\b[\w\.-]+@[\w\.-]+\.\w+\b", header["value"] )[0] break body = "" if "parts" in msg["payload"]: for part in msg["payload"]["parts"]: if part["mimeType"] == "text/plain": data = part["body"]["data"] body = base64.urlsafe_b64decode(data).decode("utf-8") break email_info = { "sender": sender, "time": datetime.fromtimestamp( int(msg["internalDate"]) / 1000 ).strftime("%Y-%m-%d %H:%M:%S"), "subject": subject, "body": body, } email_data.append(email_info) pbar.close() email_data = sort_by_time(email_data)[0:number] return {"results": email_data} except Exception as e: print(e) return None def _send(self, mail_dict: dict): recipient_mail = mail_dict["recipient_mail"] subject = mail_dict["subject"] body = mail_dict["body"] credential = self.credential service = build("gmail", "v1", credentials=credential) message = MIMEMultipart() message["to"] = recipient_mail message["subject"] = subject message.attach(MIMEText(body, "plain")) raw_message = base64.urlsafe_b64encode(message.as_bytes()).decode("utf-8") try: message = ( service.users() .messages() .send(userId="me", body={"raw": raw_message}) .execute() ) return {"state": True} except HttpError as error: print(error) return {"state": False} def func(self, mail_dict: dict): if "action" in mail_dict: assert mail_dict["action"].lower() in self.__VALID_ACTION__ self.action = mail_dict["action"] functions = {"read": self._read, "send": self._send} return functions[self.action](mail_dict) def convert_action_to(self, action_name: str): assert ( action_name.lower() in self.__VALID_ACTION__ ), f"Action `{action_name}` is not allowed! The valid action is in `{self.__VALID_ACTION__}`" self.action = action_name.lower() class WeatherComponet(ToolComponent): def __init__(self, api_key, name="weather", TIME_FORMAT="%Y-%m-%d"): super(WeatherComponet, self).__init__(name) self.name = name self.TIME_FORMAT = TIME_FORMAT self.api_key = api_key def _parse(self, data): dict_data: dict = {} for item in data["data"]: date = item["datetime"] dict_data[date] = {} if "weather" in item: dict_data[date]["description"] = item["weather"]["description"] mapping = { "temp": "temperature", "max_temp": "max_temperature", "min_temp": "min_temperature", "precip": "accumulated_precipitation", } for key in ["temp", "max_temp", "min_temp", "precip"]: if key in item: dict_data[date][mapping[key]] = item[key] return dict_data def _query(self, city_name, country_code, start_date, end_date): """https://www.weatherbit.io/api/historical-weather-daily""" # print(datetime.strftime(start_date, self.TIME_FORMAT), datetime.strftime(datetime.now(), self.TIME_FORMAT), end_date, datetime.strftime(datetime.now()+timedelta(days=1), self.TIME_FORMAT)) if start_date == datetime.strftime( datetime.now(), self.TIME_FORMAT ) and end_date == datetime.strftime( datetime.now() + timedelta(days=1), self.TIME_FORMAT ): """today""" url = f"https://api.weatherbit.io/v2.0/current?city={city_name}&country={country_code}&key={self.api_key}" else: url = f"https://api.weatherbit.io/v2.0/history/daily?&city={city_name}&country={country_code}&start_date={start_date}&end_date={end_date}&key={self.api_key}" response = requests.get(url) data = response.json() return self._parse(data) def func(self, weather_dict: Dict) -> Dict: TIME_FORMAT = self.TIME_FORMAT # Beijing, Shanghai city_name = weather_dict["city_name"] # CN, US country_code = weather_dict["country_code"] # 2020-02-02 start_date = datetime.strftime( datetime.strptime(weather_dict["start_date"], self.TIME_FORMAT), self.TIME_FORMAT, ) end_date = weather_dict["end_date"] if "end_date" in weather_dict else None if end_date is None: end_date = datetime.strftime( datetime.strptime(start_date, TIME_FORMAT) + timedelta(days=-1), TIME_FORMAT, ) else: end_date = datetime.strftime( datetime.strptime(weather_dict["end_date"], self.TIME_FORMAT), self.TIME_FORMAT, ) if datetime.strptime(start_date, TIME_FORMAT) > datetime.strptime( end_date, TIME_FORMAT ): start_date, end_date = end_date, start_date assert start_date != end_date return self._query(city_name, country_code, start_date, end_date) class TranslateComponent(ToolComponent): __SUPPORT_LANGUAGE__ = [ "af", "am", "ar", "as", "az", "ba", "bg", "bn", "bo", "bs", "ca", "cs", "cy", "da", "de", "dsb", "dv", "el", "en", "es", "et", "eu", "fa", "fi", "fil", "fj", "fo", "fr", "fr-CA", "ga", "gl", "gom", "gu", "ha", "he", "hi", "hr", "hsb", "ht", "hu", "hy", "id", "ig", "ikt", "is", "it", "iu", "iu-Latn", "ja", "ka", "kk", "km", "kmr", "kn", "ko", "ku", "ky", "ln", "lo", "lt", "lug", "lv", "lzh", "mai", "mg", "mi", "mk", "ml", "mn-Cyrl", "mn-Mong", "mr", "ms", "mt", "mww", "my", "nb", "ne", "nl", "nso", "nya", "or", "otq", "pa", "pl", "prs", "ps", "pt", "pt-PT", "ro", "ru", "run", "rw", "sd", "si", "sk", "sl", "sm", "sn", "so", "sq", "sr-Cyrl", "sr-Latn", "st", "sv", "sw", "ta", "te", "th", "ti", "tk", "tlh-Latn", "tlh-Piqd", "tn", "to", "tr", "tt", "ty", "ug", "uk", "ur", "uz", "vi", "xh", "yo", "yua", "yue", "zh-Hans", "zh-Hant", "zu", ] def __init__( self, api_key, location, default_target_language="zh-cn", name="translate" ): super(TranslateComponent, self).__init__(name) self.name = name self.api_key = api_key self.location = location self.default_target_language = default_target_language def func(self, translate_dict: Dict) -> Dict: content = translate_dict["content"] target_language = self.default_target_language if "target_language" in translate_dict: target_language = translate_dict["target_language"] assert ( target_language in self.__SUPPORT_LANGUAGE__ ), f"language `{target_language}` is not supported." endpoint = "https://api.cognitive.microsofttranslator.com" path = "/translate" constructed_url = endpoint + path params = {"api-version": "3.0", "to": target_language} headers = { "Ocp-Apim-Subscription-Key": self.api_key, "Ocp-Apim-Subscription-Region": self.location, "Content-type": "application/json", "X-ClientTraceId": str(uuid.uuid4()), } body = [{"text": content}] request = requests.post( constructed_url, params=params, headers=headers, json=body ) response = request.json() response = json.dumps( response, sort_keys=True, ensure_ascii=False, indent=4, separators=(",", ": "), ) response = eval(response) return {"result": response[0]["translations"][0]["text"]} class APIComponent(ToolComponent): def __init__(self): super(APIComponent, self).__init__() def func(self, agent) -> Dict: pass class FunctionComponent(ToolComponent): def __init__( self, functions, function_call="auto", response_type="response", your_function=None, ): super().__init__() self.functions = functions self.function_call = function_call self.parameters = {} self.available_functions = {} self.response_type = response_type if your_function: function_name = your_function["name"] function_content = your_function["content"] exec(function_content) self.available_functions[function_name] = eval(function_name) for function in self.functions: self.parameters[function["name"]] = list( function["parameters"]["properties"].keys() ) self.available_functions[function["name"]] = eval(function["name"]) def func(self, agent): messages = agent.long_term_memory outputdict = {} query = agent.long_term_memory[-1].content if len(agent.long_term_memory) > 0 else " " relevant_history = get_relevant_history( query, agent.long_term_memory[:-1], agent.chat_embeddings[:-1], ) response = agent.LLM.get_response( messages, None, functions=self.functions, stream=False, function_call=self.function_call, relevant_history=relevant_history, ) response_message = response if response_message.get("function_call"): function_name = response_message["function_call"]["name"] fuction_to_call = self.available_functions[function_name] function_args = json.loads(response_message["function_call"]["arguments"]) input_args = {} for args_name in self.parameters[function_name]: input_args[args_name] = function_args.get(args_name) function_response = fuction_to_call(**input_args) if self.response_type == "response": outputdict["response"] = function_response elif self.response_type == "prompt": outputdict["prompt"] = function_response return outputdict class CodeComponent(ToolComponent): def __init__(self, file_name, keyword) -> None: super().__init__() self.file_name = file_name self.keyword = keyword self.system_prompt = ( "you need to extract the modified code as completely as possible." ) self.last_prompt = ( f"Please strictly adhere to the following format for outputting: \n" ) self.last_prompt += ( f"<{self.keyword}> the content you need to extract " ) def func(self, agent): response = agent.LLM.get_response( agent.long_term_memory, self.system_prompt, self.last_prompt, stream=False, ) code = extract(response, self.keyword) code = code if code else response os.makedirs("output_code", exist_ok=True) file_name = "output_code/" + self.file_name codes = code.split("\n") if codes[0] == "```python": codes.remove(codes[0]) if codes[-1] == "```": codes.remove(codes[-1]) code = "\n".join(codes) with open(file_name, "w", encoding="utf-8") as f: f.write(code) return {}