Source code for langchain_community.utilities.duckduckgo_search

"""调用DuckDuckGo搜索的工具。

无需设置。免费。
https://pypi.org/project/duckduckgo-search/
"""
from typing import Dict, List, Optional

from langchain_core.pydantic_v1 import BaseModel, Extra, root_validator


class DuckDuckGoSearchAPIWrapper(BaseModel):
    """DuckDuckGo搜索API的封装器。

免费且无需任何设置。"""

    region: Optional[str] = "wt-wt"
    safesearch: str = "moderate"
    time: Optional[str] = "y"
    max_results: int = 5
    backend: str = "api"  # which backend to use in DDGS.text() (api, html, lite)
    source: str = "text"  # which function to use in DDGS (DDGS.text() or DDGS.news())

    class Config:
        """此pydantic对象的配置。"""

        extra = Extra.forbid

    @root_validator()
    def validate_environment(cls, values: Dict) -> Dict:
        """验证Python包是否存在于环境中。"""
        try:
            from duckduckgo_search import DDGS  # noqa: F401
        except ImportError:
            raise ImportError(
                "Could not import duckduckgo-search python package. "
                "Please install it with `pip install -U duckduckgo-search`."
            )
        return values

    def _ddgs_text(
        self, query: str, max_results: Optional[int] = None
    ) -> List[Dict[str, str]]:
        """通过DuckDuckGo文本搜索运行查询并返回结果。"""
        from duckduckgo_search import DDGS

        with DDGS() as ddgs:
            ddgs_gen = ddgs.text(
                query,
                region=self.region,
                safesearch=self.safesearch,
                timelimit=self.time,
                max_results=max_results or self.max_results,
                backend=self.backend,
            )
            if ddgs_gen:
                return [r for r in ddgs_gen]
        return []

    def _ddgs_news(
        self, query: str, max_results: Optional[int] = None
    ) -> List[Dict[str, str]]:
        """运行查询通过DuckDuckGo新闻搜索并返回结果。"""
        from duckduckgo_search import DDGS

        with DDGS() as ddgs:
            ddgs_gen = ddgs.news(
                query,
                region=self.region,
                safesearch=self.safesearch,
                timelimit=self.time,
                max_results=max_results or self.max_results,
            )
            if ddgs_gen:
                return [r for r in ddgs_gen]
        return []

    def run(self, query: str) -> str:
        """通过DuckDuckGo运行查询并返回连接的结果。"""
        if self.source == "text":
            results = self._ddgs_text(query)
        elif self.source == "news":
            results = self._ddgs_news(query)
        else:
            results = []

        if not results:
            return "No good DuckDuckGo Search Result was found"
        return " ".join(r["body"] for r in results)

    def results(
        self, query: str, max_results: int, source: Optional[str] = None
    ) -> List[Dict[str, str]]:
        """运行查询通过DuckDuckGo并返回元数据。

参数：
    query: 要搜索的查询。
    max_results: 要返回的结果数量。
    source: 要查找的来源。

返回：
    一个包含以下键的字典列表：
        snippet - 结果的描述。
        title - 结果的标题。
        link - 结果的链接。
"""
        source = source or self.source
        if source == "text":
            results = [
                {"snippet": r["body"], "title": r["title"], "link": r["href"]}
                for r in self._ddgs_text(query, max_results=max_results)
            ]
        elif source == "news":
            results = [
                {
                    "snippet": r["body"],
                    "title": r["title"],
                    "link": r["url"],
                    "date": r["date"],
                    "source": r["source"],
                }
                for r in self._ddgs_news(query, max_results=max_results)
            ]
        else:
            results = []

        if results is None:
            results = [{"Result": "No good DuckDuckGo Search Result was found"}]

        return results