from Agents.AIAgent import AIAgent
from tools.program_called_command_list import load_ai_agent_name_list
from flow.flow_controller import get_program_code_from_response
import requests
from duckduckgo_search import DDGS
from selenium import webdriver
from selenium.webdriver.chrome.options import Options
from selenium.webdriver.common.by import By
import time
class WebSerchAgent():
def __init__(self, serch_number=3):
name = "KeyWordCreater"
self.agetn_dict = load_ai_agent_name_list("WebSerchAgent")
print(self.agetn_dict)
self.agent = AIAgent(name,
self.__get_agent_prompt(name),
[],
False)
self.serch_result = []
self.serch_num = serch_number
def get_name(self):
return WebSerchAgent.__name__
# AIAgentと関数的互換性確保のために必要
def clear_memory(self):
self.agent.clear_memory()
def update_temperature(self, temperature):
self.agent.update_temperature(temperature)
def update_tools(self, tools):
self.agent.update_tools(tools)
def get_respons(self, command):
return self.serch(command)
################################################
def __get_agent_prompt(self, name):
return self.agetn_dict[name][1]
def serch(self, command):
self.agent.update_system_prompt(
self.__get_agent_prompt("KeyWordCreater"))
# コールバック関数の設定 (エージェントの動作の可視化用)
respons = self.agent.get_respons(command)
keyword_list = self.__get_keywords(respons)
if None is keyword_list:
respons = self.agent.get_respons(
command+"\r\n についてあなたの役割に沿ってpython\
の # から始まるコメントでpythonのプログラムコードとして出力して")
keyword_list = self.__get_keywords(respons)
if None is keyword_list:
return "キーワードが正常なデータ形式で出力されませんでした。"
web_texts = []
have_url = {}
# キーワードを使ってweb検索しテキストを読み込む
for keyword in keyword_list:
time.sleep(4)
serch_data = self.__web_search(keyword)
if None is not serch_data:
for data in serch_data:
if "href" in data:
if self.__check_url(data["href"]):
if data["href"] not in have_url: # 同じurlは除く
web_texts.append(
{"text": self.__get_web_text(data["href"]),
"href": data["href"]})
have_url[data["href"]] = True
# 読み込んだwebテキストから必要部分を抜粋
self.agent.update_system_prompt(self.__get_agent_prompt("excerpt"))
text = ""
for i in range(len(web_texts)):
prompt = web_texts[i]["text"] +\
"\r\nから以下の要望や質問に沿った部分を抜き出してください。\r\n" + command
text += self.agent.get_respons(prompt)
text += "\r\nurl:"+web_texts[i]["href"] + "\r\n" + "-"*20 + "\r\n"
self.agent.update_system_prompt(self.__get_agent_prompt("summary"))
prompt = text + "\r\nをあなたの役割に応じて、以下の要望や質問に沿うようにまとめてください。\r\n" + command
respons = self.agent.get_respons(prompt)
return respons
def __get_keywords(self, respons):
code = get_program_code_from_response(respons)
if "python" not in code:
return None
code_list = code["python"].split("\n")
keyword_list = []
for code in code_list:
st = code.strip()
if 2 < len(st):
if "#" == st[0]:
keyword_list.append(st[1:])
return keyword_list
def __web_search(self, keyword):
"""
DuckDuckGoでウェブ検索を実行します。
Args:
query: 検索キーワード：キーワードはクエリと書かれることがあります。
Returns:
str(
検索結果のテキスト
タイトル:....
URL:https://......
説明:....
--------------------
タイトル:....
URL:https://......
説明:....
--------------------
タイトル:....
URL:https://......
説明:....
--------------------
........
)
"""
try:
self.serch_result = DDGS().text(keyword,
region='wt-wt',
safesearch='off',
timelimit=None,
backend="html",
max_results=self.serch_num)
return self.serch_result
except Exception as e:
print(f"エラーが発生しました: {e}")
return None
def __check_url(self, url: str) -> bool:
"""指定されたURLが正常に読み込めるかどうかを判定します。
Args:
url: チェックするURL
Returns:
正常に読み込める場合はTrue、そうでない場合はFalse
"""
try:
response = requests.get(url)
print("get_web_text_links", url)
return response.status_code == 200
except requests.exceptions.RequestException:
print("get_web_text_links", url)
return False
def __get_web_text(self, url):
"""
この機能は以下のような文章でしようされるます。
・指定されたurlをクリックしテキストデータを得ます。
・指定されたURLのテキストデータとテキストに設定されたリンクを得ます。
・ホームページにアクセスして
・リンク先にアクセスして
・リンクをクリックして
urlは,ホームページアドレス、リンク、ページへのリンク、ページのurlなどと呼ばれることがあります。
ホームページは、サイト、webページ、webサイト,ページなどと呼ばれることがあります。
Args:
url: WebページのURL：urlはホームページアドレス、リンク、ページへのリンク、ページのurlなどと呼ばれることがあります。
Returns:
str(
テキストで得ます。
)
"""
# Chromeのオプションを設定
options = Options()
options.add_argument('--incognito')
options.add_argument(
'--user-agent=Mozilla/5.0 (Windows NT 10.0; Win64; x64)\
AppleWebKit/537.36 (KHTML, like Gecko) Chrome/128.0.0.0 Safari/537.36')
# Chromeドライバを起動
driver = webdriver.Chrome(options=options)
# 指定されたURLにアクセス
try:
driver.get(url)
except Exception as e:
print(f"Error accessing URL: {url}, {e}")
return "nodatanodata\rnodata:<nodata>"
# テキストデータを取得
try:
text_data = driver.find_element(By.TAG_NAME, "body").text
except Exception as e:
print(f"Error getting text data: {url}, {e}")
return "リンク切れです"
text_data = text_data.replace("<br>", "\n")
text_data = text_data.replace("<br/>", "\n")
# ブラウザを閉じる
try:
driver.quit()
except Exception as e:
print(f"Error closing browser: {url}, {e}")
return text_data
return text_data
