"""用于使用SearxNG元搜索API的实用程序。
SearxNG是一个隐私友好的免费元搜索引擎，它从多个搜索引擎和数据库中聚合结果，并支持OpenSearch规范。
有关安装说明的更多详细信息，请参阅此处。
有关搜索API，请参阅https://docs.searxng.org/dev/search_api.html
快速开始
-----------
为了使用此实用程序，您需要提供searx主机。这可以通过传递命名参数:attr:`searx_host <SearxSearchWrapper.searx_host>`或导出环境变量SEARX_HOST来完成。
注意：这是唯一必需的参数。
然后像这样创建一个searx搜索实例：
.. code-block:: python
from langchain_community.utilities import SearxSearchWrapper
# 当主机以`http`开头时，SSL被禁用，连接被假定在私有网络上
searx_host='http://self.hosted'
search = SearxSearchWrapper(searx_host=searx_host)
现在您可以使用``search``实例来查询searx API。
搜索
---------
使用:meth:`run() <SearxSearchWrapper.run>`和:meth:`results() <SearxSearchWrapper.results>`方法来查询searx API。
其他方法可用于方便使用。
:class:`SearxResults`是对原始json结果的便捷包装。
使用``run``方法进行搜索的示例用法：
.. code-block:: python
s.run(query="what is the best search engine?")
引擎参数
-----------------
您可以将任何`接受的searx搜索API参数 <https://docs.searxng.org/dev/search_api.html>`_参数传递给:py:class:`SearxSearchWrapper`实例。
在以下示例中，我们使用了:attr:`engines <SearxSearchWrapper.engines>`和``language``参数：
.. code-block:: python
# 假设searx主机已设置如上或作为环境变量导出
s = SearxSearchWrapper(engines=['google', 'bing'],
language='es')
搜索提示
-----------
Searx提供了一种特殊的`搜索语法 <https://docs.searxng.org/user/index.html#search-syntax>`_，也可以用来代替传递引擎参数。
例如，以下查询：
.. code-block:: python
s = SearxSearchWrapper("langchain library", engines=['github'])
# 也可以写成：
s = SearxSearchWrapper("langchain library !github")
# 或者甚至：
s = SearxSearchWrapper("langchain library !gh")
在某些情况下，您可能希望将额外的字符串传递给搜索查询。例如，当``run()``方法由代理调用时。搜索后缀也可以用作传递额外参数给searx或基础搜索引擎的一种方式。
.. code-block:: python
# 选择github引擎并传递搜索后缀
s = SearchWrapper("langchain library", query_suffix="!gh")
s = SearchWrapper("langchain library")
# 选择github的传统google搜索语法
s.run("large language models", query_suffix="site:github.com")
*注意*：搜索后缀可以在实例级别和方法级别上定义。生成的查询将是两者的连接，前者优先。
有关更多详细信息，请参阅`SearxNG配置引擎 <https://docs.searxng.org/admin/engines/configured_engines.html>`_和`SearxNG搜索语法 <https://docs.searxng.org/user/index.html#id1>`_。
注
-----
此包装器基于SearxNG分支https://github.com/searxng/searxng，该分支比原始的Searx项目维护得更好，并提供更多功能。
公共searxNG实例通常对API使用设置了速率限制器，因此您可能希望使用自托管实例并禁用速率限制器。
如果您正在自托管一个实例，您可以根据自己的网络自定义速率限制器，如下所述。
有关公共SearxNG实例列表，请参见https://searx.space/
"""
import json
from typing import Any, Dict, List, Optional
import aiohttp
import requests
from langchain_core.pydantic_v1 import (
BaseModel,
Extra,
Field,
PrivateAttr,
root_validator,
validator,
)
from langchain_core.utils import get_from_dict_or_env
def _get_default_params() -> dict:
return {"language": "en", "format": "json"}
[docs]class SearxResults(dict):
"""在搜索API结果周围的类似字典的包装器。"""
_data: str = ""
[docs] def __init__(self, data: str):
"""从Searx获取原始结果，并将其转换为类似字典的对象。"""
json_data = json.loads(data)
super().__init__(json_data)
self.__dict__ = self
def __str__(self) -> str:
"""searx结果的文本表示。"""
return self._data
@property
def results(self) -> Any:
"""静音mypy以访问此字段。
:meta private:
"""
return self.get("results")
@property
def answers(self) -> Any:
"""JSON结果的辅助访问器。"""
return self.get("answers")
[docs]class SearxSearchWrapper(BaseModel):
"""封装了Searx API。
要使用，需要通过传递命名参数“searx_host”或导出环境变量“SEARX_HOST”来提供searx主机。
在某些情况下，您可能希望禁用SSL验证，例如在本地运行searx时。您可以通过传递命名参数“unsecure”来实现此目的。您还可以将主机URL方案传递为“http”以禁用SSL。
示例：
.. code-block:: python
from langchain_community.utilities import SearxSearchWrapper
searx = SearxSearchWrapper(searx_host="http://localhost:8888")
禁用SSL的示例：
.. code-block:: python
from langchain_community.utilities import SearxSearchWrapper
# 请注意，如果将url方案传递为http，则不需要unsecure参数
searx = SearxSearchWrapper(searx_host="http://localhost:8888",
unsecure=True)"""
_result: SearxResults = PrivateAttr()
searx_host: str = ""
unsecure: bool = False
params: dict = Field(default_factory=_get_default_params)
headers: Optional[dict] = None
engines: Optional[List[str]] = []
categories: Optional[List[str]] = []
query_suffix: Optional[str] = ""
k: int = 10
aiosession: Optional[Any] = None
@validator("unsecure")
def disable_ssl_warnings(cls, v: bool) -> bool:
"""禁用SSL警告。"""
if v:
# requests.urllib3.disable_warnings()
try:
import urllib3
urllib3.disable_warnings()
except ImportError as e:
print(e) # noqa: T201
return v
@root_validator()
def validate_params(cls, values: Dict) -> Dict:
"""验证自定义searx参数是否与默认参数合并。"""
user_params = values["params"]
default = _get_default_params()
values["params"] = {**default, **user_params}
engines = values.get("engines")
if engines:
values["params"]["engines"] = ",".join(engines)
categories = values.get("categories")
if categories:
values["params"]["categories"] = ",".join(categories)
searx_host = get_from_dict_or_env(values, "searx_host", "SEARX_HOST")
if not searx_host.startswith("http"):
print( # noqa: T201
f"Warning: missing the url scheme on host \
! assuming secure https://{searx_host} "
)
searx_host = "https://" + searx_host
elif searx_host.startswith("http://"):
values["unsecure"] = True
cls.disable_ssl_warnings(True)
values["searx_host"] = searx_host
return values
class Config:
"""此pydantic对象的配置。"""
extra = Extra.forbid
def _searx_api_query(self, params: dict) -> SearxResults:
"""实际请求到searx API。"""
raw_result = requests.get(
self.searx_host,
headers=self.headers,
params=params,
verify=not self.unsecure,
)
# test if http result is ok
if not raw_result.ok:
raise ValueError("Searx API returned an error: ", raw_result.text)
res = SearxResults(raw_result.text)
self._result = res
return res
async def _asearx_api_query(self, params: dict) -> SearxResults:
if not self.aiosession:
async with aiohttp.ClientSession() as session:
async with session.get(
self.searx_host,
headers=self.headers,
params=params,
ssl=(lambda: False if self.unsecure else None)(),
) as response:
if not response.ok:
raise ValueError("Searx API returned an error: ", response.text)
result = SearxResults(await response.text())
self._result = result
else:
async with self.aiosession.get(
self.searx_host,
headers=self.headers,
params=params,
verify=not self.unsecure,
) as response:
if not response.ok:
raise ValueError("Searx API returned an error: ", response.text)
result = SearxResults(await response.text())
self._result = result
return result
[docs] def run(
self,
query: str,
engines: Optional[List[str]] = None,
categories: Optional[List[str]] = None,
query_suffix: Optional[str] = "",
**kwargs: Any,
) -> str:
"""通过Searx API运行查询并解析结果。
您可以将任何其他参数传递给searx查询API。
参数:
query: 要搜索的查询。
query_suffix: 附加到查询的额外后缀。
engines: 用于查询的引擎列表。
categories: 用于查询的类别列表。
**kwargs: 传递给searx API的额外参数。
返回:
str: 查询的结果。
引发:
ValueError: 如果查询出现错误。
示例:
这将向qwant引擎发出查询:
.. code-block:: python
from langchain_community.utilities import SearxSearchWrapper
searx = SearxSearchWrapper(searx_host="http://my.searx.host")
searx.run("what is the weather in France ?", engine="qwant")
# 可以使用`query_suffix`使用searx的`!`语法选择引擎来实现相同的结果
searx.run("what is the weather in France ?", query_suffix="!qwant")
"""
_params = {
"q": query,
}
params = {**self.params, **_params, **kwargs}
if self.query_suffix and len(self.query_suffix) > 0:
params["q"] += " " + self.query_suffix
if isinstance(query_suffix, str) and len(query_suffix) > 0:
params["q"] += " " + query_suffix
if isinstance(engines, list) and len(engines) > 0:
params["engines"] = ",".join(engines)
if isinstance(categories, list) and len(categories) > 0:
params["categories"] = ",".join(categories)
res = self._searx_api_query(params)
if len(res.answers) > 0:
toret = res.answers[0]
# only return the content of the results list
elif len(res.results) > 0:
toret = "\n\n".join([r.get("content", "") for r in res.results[: self.k]])
else:
toret = "No good search result found"
return toret
[docs] async def arun(
self,
query: str,
engines: Optional[List[str]] = None,
query_suffix: Optional[str] = "",
**kwargs: Any,
) -> str:
"""`run`的异步版本。"""
_params = {
"q": query,
}
params = {**self.params, **_params, **kwargs}
if self.query_suffix and len(self.query_suffix) > 0:
params["q"] += " " + self.query_suffix
if isinstance(query_suffix, str) and len(query_suffix) > 0:
params["q"] += " " + query_suffix
if isinstance(engines, list) and len(engines) > 0:
params["engines"] = ",".join(engines)
res = await self._asearx_api_query(params)
if len(res.answers) > 0:
toret = res.answers[0]
# only return the content of the results list
elif len(res.results) > 0:
toret = "\n\n".join([r.get("content", "") for r in res.results[: self.k]])
else:
toret = "No good search result found"
return toret
[docs] def results(
self,
query: str,
num_results: int,
engines: Optional[List[str]] = None,
categories: Optional[List[str]] = None,
query_suffix: Optional[str] = "",
**kwargs: Any,
) -> List[Dict]:
"""通过Searx API运行查询并返回带有元数据的结果。
参数：
query: 要搜索的查询。
query_suffix: 附加到查询的额外后缀。
num_results: 限制要返回的结果数量。
engines: 用于查询的引擎列表。
categories: 用于查询的类别列表。
**kwargs: 传递给searx API的额外参数。
返回：
具有以下键的字典：
{
snippet: 结果的描述。
title: 结果的标题。
link: 结果的链接。
engines: 用于结果的引擎。
category: 结果的Searx类别。
}
"""
_params = {
"q": query,
}
params = {**self.params, **_params, **kwargs}
if self.query_suffix and len(self.query_suffix) > 0:
params["q"] += " " + self.query_suffix
if isinstance(query_suffix, str) and len(query_suffix) > 0:
params["q"] += " " + query_suffix
if isinstance(engines, list) and len(engines) > 0:
params["engines"] = ",".join(engines)
if isinstance(categories, list) and len(categories) > 0:
params["categories"] = ",".join(categories)
results = self._searx_api_query(params).results[:num_results]
if len(results) == 0:
return [{"Result": "No good Search Result was found"}]
return [
{
"snippet": result.get("content", ""),
"title": result["title"],
"link": result["url"],
"engines": result["engines"],
"category": result["category"],
}
for result in results
]
[docs] async def aresults(
self,
query: str,
num_results: int,
engines: Optional[List[str]] = None,
query_suffix: Optional[str] = "",
**kwargs: Any,
) -> List[Dict]:
"""使用JSON结果进行异步查询。
使用aiohttp。有关更多信息，请参阅`results`。
"""
_params = {
"q": query,
}
params = {**self.params, **_params, **kwargs}
if self.query_suffix and len(self.query_suffix) > 0:
params["q"] += " " + self.query_suffix
if isinstance(query_suffix, str) and len(query_suffix) > 0:
params["q"] += " " + query_suffix
if isinstance(engines, list) and len(engines) > 0:
params["engines"] = ",".join(engines)
results = (await self._asearx_api_query(params)).results[:num_results]
if len(results) == 0:
return [{"Result": "No good Search Result was found"}]
return [
{
"snippet": result.get("content", ""),
"title": result["title"],
"link": result["url"],
"engines": result["engines"],
"category": result["category"],
}
for result in results
]