Web Serch Agent:ソースコード:WebSerchAgent.py


from tools.AIAgent import AIAgent
from tools.command_list import get_ai_agent
from tools.fllow_controller import get_program_code_from_response
from langchain_community.callbacks import StreamlitCallbackHandler
import tools.command_list 
import streamlit as st
import requests


from duckduckgo_search import DDGS
import requests
from selenium import webdriver
from selenium.webdriver.chrome.options import Options
from selenium.webdriver.common.by import By


class WebSerchAgent():
    def __init__(self, serch_number=3):
        keyword_creater_name = "KeyWordCreater"
        self.agent = AIAgent(keyword_creater_name,
                             get_ai_agent(keyword_creater_name),
                             [],
                             False)
        self.serch_result = []
        self.serch_num = serch_number

    # AIAgentと関数的互換性確保のために必要
    def get_respons(self, command):
        return self.serch(command)
    
    def clear_memory(self):
        self.agent.clear_memory()
################################################

    def serch(self, command):
        self.agent.update_system_prompt(get_ai_agent("KeyWordCreater"))
        # コールバック関数の設定 (エージェントの動作の可視化用)
        st_cb = StreamlitCallbackHandler(
            st.container(), expand_new_thoughts=True)
        tools.command_list.g_time_keeper.wait()
        respons = self.agent.get_respons(command, st_cb)
        keyword_list = self.__get_keywords(respons)
        if None is keyword_list:
            respons = self.agent.get_respons(command+"\r\n についてあなたの役割に沿ってpython の # から始まるコメントでpythonのプログラムコードとして出力して", st_cb)
            keyword_list = self.__get_keywords(respons)
            if None is keyword_list:
                return "キーワードが正常なデータ形式で出力されませんでした。"
        web_texts = []
        have_url = {}
        #キーワードを使ってweb検索しテキストを読み込む
        for keyword in keyword_list:
            tools.command_list.g_time_keeper.wait()
            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(get_ai_agent("excerpt"))
        text = ""
        for i in range(len(web_texts)):
            tools.command_list.g_time_keeper.wait()
            prompt = web_texts[i]["text"] + "\r\nから以下の要望や質問に沿った部分を抜き出してください。\r\n" + command
            text += self.agent.get_respons(prompt, st_cb)
            text += "\r\nurl:"+web_texts[i]["href"] + "\r\n" + "-"*20 + "\r\n"
        self.agent.update_system_prompt(get_ai_agent("summary"))
        prompt = text + "\r\nをあなたの役割に応じて、以下の要望や質問に沿うようにまとめてください。\r\n" + command
        tools.command_list.g_time_keeper.wait()
        respons = self.agent.get_respons(prompt, st_cb)
        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,
                                       max_results=self.serch_num)
            return self.serch_result
            if self.serch_result:
                pass
                # result = ""
                # for data in serched_data:
                #     result += f"タイトル: {data['title']}\r\n"
                #     result += f"URL: {data['href']}\r\n"
                #     result += f"説明: {data['body']}\r\n"
                #     result += "-" * 20 + "\r\n"
    # 
                # return result +"\r\n" +"作業を続けてください。"
            else:
                return "検索結果が見つかりませんでした。"

        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

以下AIによるまとめ

WebSerchAgent クラスのリファレンスマニュアル

概要

WebSerchAgent クラスは、ユーザーの質問に対して、Web検索を行い、関連する情報を取得して回答を生成するクラスです。

メソッド

__init__(self, serch_number=3)

  • コンストラクタ
  • 引数:
    • serch_number: 検索する結果の数を指定します。デフォルトは3です。
  • 動作:
    • AIAgent オブジェクトを初期化します。
    • 検索結果を格納するリスト serch_result を初期化します。
    • 検索する結果の数を serch_num に格納します。

get_respons(self, command)

  • ユーザーからの質問を受け取り、Web検索を行い、回答を生成します。
  • 引数:
    • command: ユーザーからの質問
  • 動作:
    • AIAgent の get_respons メソッドを使用して、質問に対するキーワードを生成します。
    • 生成されたキーワードを使って Web 検索を行い、関連する情報を取得します。
    • 取得した情報から、質問に関連する部分を抜粋します。
    • 抜粋した情報をまとめ、回答を生成します。
  • 返り値:
    • 生成された回答

clear_memory(self)

  • AIAgent のメモリをクリアします。
  • 引数: なし
  • 動作:
    • AIAgent の clear_memory メソッドを呼び出します。

serch(self, command)

  • ユーザーからの質問を受け取り、Web検索を行い、回答を生成します。
  • 引数:
    • command: ユーザーからの質問
  • 動作:
    • AIAgent の get_respons メソッドを使用して、質問に対するキーワードを生成します。
    • 生成されたキーワードを使って Web 検索を行い、関連する情報を取得します。
    • 取得した情報から、質問に関連する部分を抜粋します。
    • 抜粋した情報をまとめ、回答を生成します。
  • 返り値:
    • 生成された回答

__get_keywords(self, respons)

  • AIAgent から返された応答から、キーワードを抽出します。
  • 引数:
    • respons: AIAgent から返された応答
  • 動作:
    • 応答から Python コードを抽出します。
    • Python コードから、# で始まるコメントをキーワードとして抽出します。
  • 返り値:
    • 抽出されたキーワードのリスト

__web_search(self, keyword)

  • 指定されたキーワードで Web 検索を行い、検索結果を取得します。
  • 引数:
    • keyword: 検索するキーワード
  • 動作:
    • DuckDuckGo 検索 API を使用して、キーワードで Web 検索を行います。
    • 検索結果から、タイトル、URL、説明文を抽出します。
  • 返り値:
    • 検索結果のリスト

__check_url(self, url: str)

  • 指定された URL が正常に読み込めるかどうかを判定します。
  • 引数:
    • url: チェックする URL
  • 動作:
    • requests ライブラリを使用して、URL にアクセスします。
    • ステータスコードが 200 であれば、正常に読み込めたと判断します。
  • 返り値:
    • 正常に読み込める場合は True、そうでない場合は False

__get_web_text(self, url)

  • 指定された URL の Web ページのテキストデータを取得します。
  • 引数:
    • url: Web ページの URL
  • 動作:
    • Selenium を使用して、指定された URL の Web ページにアクセスします。
    • Web ページのテキストデータを取得します。
  • 返り値:
    • Web ページのテキストデータ