""" title: Web Search author: EntropyYue funding_url: https://github.com/EntropyYue/web_search version: 9.1 license: MIT """ import asyncio import json import re import unicodedata from collections.abc import Callable from typing import Any from urllib.parse import ParseResult, urlparse from aiohttp import ClientError, ClientSession, ClientTimeout from bs4 import BeautifulSoup from pydantic import BaseModel, Field from tiktoken import get_encoding class PageCleaner: def __init__(self, token_limit: int = 1000): self.token_limit = token_limit self.tokenizer = get_encoding("cl100k_base") self.invisible_chars = ["\ufeff", "\u200b", "\u2028", "\u2060"] def extract_title(self, soup: BeautifulSoup) -> str: title = ( soup.title.string if soup.title and soup.title.string else "No title found" ) return self._normalize_text(title) def extract_text(self, soup: BeautifulSoup) -> str: return soup.get_text(separator="\n", strip=True) def clean_text(self, text: str) -> str: text = self._normalize_text(text) text = re.sub("[ \\t]+", " ", text) text = self._remove_emojis(text) text = self._remove_invisible_chars(text) return text.strip() def truncate_tokens(self, text: str) -> str: tokens = self.tokenizer.encode(text) truncated = self.tokenizer.decode(tokens[: self.token_limit]) return self._remove_invisible_chars(truncated).strip() def _normalize_text(self, text: str) -> str: return unicodedata.normalize("NFKC", text).strip() def _remove_emojis(self, text: str) -> str: return "".join( c for c in text if not unicodedata.category(c).startswith("So") ) def _remove_invisible_chars(self, text: str) -> str: for ch in self.invisible_chars: text = text.replace(ch, "") return text class WebLoader: def __init__(self, valves, headers: dict, token_limit: int) -> None: self.valves = valves self.headers = headers self.cleaner = PageCleaner(token_limit=token_limit) def get_base_url(self, url: str) -> str: parsed_url: ParseResult = urlparse(url) return f"{parsed_url.scheme}://{parsed_url.netloc}" async def fetch_and_process_page( self, url: str, session: ClientSession ) -> dict[str, str]: try: async with session.get(url, headers=self.headers) as response: response.raise_for_status() html = await response.text() except ClientError as e: return {"url": url, "error": f"检索页面失败, 网络错误: {str(e)}"} except Exception as e: return {"url": url, "error": f"检索页面失败: {str(e)}"} soup = BeautifulSoup(html, "html.parser") title = self.cleaner.extract_title(soup) raw_text = self.cleaner.extract_text(soup) clean_text = self.cleaner.clean_text(raw_text) truncated = self.cleaner.truncate_tokens(clean_text) return {"title": title, "url": url, "content": truncated} async def process_search_result( self, result: dict[str, str], session: ClientSession ) -> dict[str, str] | None: url = result["url"] snippet = result.get("content", "") if self.valves.IGNORED_WEBSITES: base_url = self.get_base_url(url) ignored_sites = [s.strip() for s in self.valves.IGNORED_WEBSITES.split(",")] if any(site in base_url for site in ignored_sites): return None result_data = await self.fetch_and_process_page(url, session) if "content" in result_data and "error" not in result_data: result_data["snippet"] = self.cleaner._remove_emojis(snippet) return result_data return None class EventEmitter: def __init__(self, valves, event_emitter: Callable[[dict], Any] | None = None): self.valves = valves self.event_emitter = event_emitter async def _emit(self, type, data: dict[str, Any]) -> None: if not self.event_emitter: return await self.event_emitter({"type": type, "data": data}) async def status( self, description: str = "未知状态", status: str = "in_progress", done: bool = False, action: str | None = "web_search", queries: list[str] | None = None, count: int | None = None, urls: list[str] | None = None, ) -> None: if not self.valves.STATUS: return await self._emit( type="status", data={ "description": description, "status": status, "done": done, "action": action, "queries": queries, "count": count, "urls": urls, }, ) async def queries(self, queries: list[str]) -> None: await self.status(action="web_search_queries_generated", queries=queries) async def urls(self, urls: list[str]) -> None: await self.status( action="web_search", description="Searched {{count}} sites", urls=urls ) async def fetched(self, count: int) -> None: await self.status(action="sources_retrieved", count=count, done=True) async def citation( self, document: list[str], metadata: list[dict[str, str]], source: dict[str, str], ) -> None: if not self.valves.CITATION_LINKS: return await self._emit( type="citation", data={"document": document, "metadata": metadata, "source": source}, ) class Tools: class Valves(BaseModel): SEARXNG_ENGINE_API_BASE_URL: str = Field( default="https://example.com/search", description="搜索引擎的基础URL" ) IGNORED_WEBSITES: str = Field( default="", description="以逗号分隔的要忽略的网站列表" ) MAX_SEARCH_RESULTS: int = Field(default=3, description="要分析的搜索引擎结果数") SEARCH_PAGE_TOKENS_LIMIT: int = Field( default=2000, description="搜索结果每页的限制Token数" ) GET_WEBSITE_TOKENS_LIMIT: int = Field( default=5000, description="获取网站的限制Token数" ) USE_ENV_PROXY: bool = Field(default=False, description="使用环境变量中的代理") WEB_LOAD_TIMEOUT: int = Field(default=5, description="网页抓取超时时间 (秒)") CITATION_LINKS: bool = Field( default=False, description="发送带有链接的自定义引用" ) STATUS: bool = Field(default=True, description="发送状态") def __init__(self): self.valves = self.Valves() self.timeout = ClientTimeout(total=self.valves.WEB_LOAD_TIMEOUT) self.headers = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/58.0.3029.110 Safari/537.3" } async def search_web( self, query: str, __event_emitter__: Callable[[dict], Any] | None = None ) -> str: """ 搜索网络并获取相关页面的内容，搜索未知知识、新闻、信息、公共联系信息、天气等 :params query: 搜索中使用的关键词 :return: The content of the pages in json format. """ loader = WebLoader( valves=self.valves, headers=self.headers, token_limit=self.valves.SEARCH_PAGE_TOKENS_LIMIT, ) emitter = EventEmitter(self.valves, __event_emitter__) await emitter.status("Searching the web") await emitter.queries([query]) search_engine_url = self.valves.SEARXNG_ENGINE_API_BASE_URL params = {"q": query, "format": "json"} try: async with ( ClientSession(trust_env=self.valves.USE_ENV_PROXY) as session, session.get( search_engine_url, params=params, headers=self.headers ) as resp, ): resp.raise_for_status() data = await resp.json() except ClientError as e: await emitter.status( status="error", description=f"搜索时出错: {str(e)}", done=True ) return json.dumps({"error": str(e)}) results = data.get("results", []) await emitter.urls([result.get("url", "") for result in results]) results_json: list[dict[str, str]] = [] if not results: return json.dumps( {"error": "No search results found"}, indent=4, ensure_ascii=False ) async with ClientSession( trust_env=self.valves.USE_ENV_PROXY, timeout=self.timeout ) as session: tasks = [ asyncio.create_task(loader.process_search_result(result, session)) for result in results ] for done in asyncio.as_completed(tasks): try: result_json = await done except BaseException: continue if not result_json: continue results_json.append(result_json) if len(results_json) >= self.valves.MAX_SEARCH_RESULTS: for task in tasks: if not task.done(): task.cancel() await asyncio.gather(*tasks, return_exceptions=True) break results_json = results_json[: self.valves.MAX_SEARCH_RESULTS] if not len(results_json): await emitter.fetched(0) return json.dumps( {"error": "No fetched results found"}, indent=4, ensure_ascii=False ) for result in results_json: await emitter.citation( document=[result["content"]], metadata=[{"source": result["url"]}], source={"name": result["title"]}, ) urls: list[str] = [] for result in results_json: urls.append(result["url"]) await emitter.fetched(len(results_json)) return json.dumps(results_json, indent=4, ensure_ascii=False) async def get_website( self, urls: list[str], __event_emitter__: Callable[[dict], Any] | None = None ) -> str: """ 打开输入的网站并获取其内容 :params url: 需要打开的网站 :return: The content of the website in json format. """ loader = WebLoader( valves=self.valves, headers=self.headers, token_limit=self.valves.GET_WEBSITE_TOKENS_LIMIT, ) emitter = EventEmitter(self.valves, __event_emitter__) await emitter.queries(urls) results_json = [] if urls == []: return "" async with ClientSession( trust_env=self.valves.USE_ENV_PROXY, timeout=self.timeout ) as session: tasks = [ asyncio.create_task(loader.fetch_and_process_page(url, session)) for url in urls ] for task in tasks: try: result_site = await task except BaseException: continue if not result_site: continue results_json.append(result_site) if "content" in result_site: await emitter.citation( document=[result_site["content"]], metadata=[{"source": result_site["url"]}], source={"name": result_site["title"]}, ) await emitter.urls([result.get("url", "") for result in results_json]) await emitter.fetched(len(results_json)) return json.dumps(results_json, indent=4, ensure_ascii=False)