Source code for langchain_community.utilities.brave_search

import json
from typing import List

import requests
from langchain_core.documents import Document
from langchain_core.pydantic_v1 import BaseModel, Field


[docs]class BraveSearchWrapper(BaseModel): """封装了Brave搜索引擎。""" api_key: str """用于Brave搜索引擎的API密钥。""" search_kwargs: dict = Field(default_factory=dict) """传递给搜索请求的其他关键字参数。""" base_url: str = "https://api.search.brave.com/res/v1/web/search" """Brave搜索引擎的基本URL。"""
[docs] def run(self, query: str) -> str: """查询Brave搜索引擎并将结果作为JSON字符串返回。 参数: query: 要搜索的查询内容。 返回:结果作为JSON字符串。 """ web_search_results = self._search_request(query=query) final_results = [ { "title": item.get("title"), "link": item.get("url"), "snippet": item.get("description"), } for item in web_search_results ] return json.dumps(final_results)
[docs] def download_documents(self, query: str) -> List[Document]: """查询勇敢搜索引擎并将结果作为文档列表返回。 参数: query: 要搜索的查询内容。 返回:作为文档列表的结果。 """ results = self._search_request(query) return [ Document( page_content=item.get("description"), # type: ignore[arg-type] metadata={"title": item.get("title"), "link": item.get("url")}, ) for item in results ]
def _search_request(self, query: str) -> List[dict]: headers = { "X-Subscription-Token": self.api_key, "Accept": "application/json", } req = requests.PreparedRequest() params = {**self.search_kwargs, **{"q": query}} req.prepare_url(self.base_url, params) if req.url is None: raise ValueError("prepared url is None, this should not happen") response = requests.get(req.url, headers=headers) if not response.ok: raise Exception(f"HTTP error {response.status_code}") return response.json().get("web", {}).get("results", [])