mirror of
https://github.com/hwchase17/langchain
synced 2024-11-06 03:20:49 +00:00
99 lines
3.2 KiB
Python
99 lines
3.2 KiB
Python
"""Util that calls Bing Search."""
|
|
from typing import Dict, List
|
|
|
|
import requests
|
|
from langchain_core.pydantic_v1 import BaseModel, Extra, Field, root_validator
|
|
from langchain_core.utils import get_from_dict_or_env
|
|
|
|
|
|
class BingSearchAPIWrapper(BaseModel):
|
|
"""Wrapper for Bing Search API."""
|
|
|
|
bing_subscription_key: str
|
|
bing_search_url: str
|
|
k: int = 10
|
|
search_kwargs: dict = Field(default_factory=dict)
|
|
"""Additional keyword arguments to pass to the search request."""
|
|
|
|
class Config:
|
|
"""Configuration for this pydantic object."""
|
|
|
|
extra = Extra.forbid
|
|
|
|
def _bing_search_results(self, search_term: str, count: int) -> List[dict]:
|
|
headers = {"Ocp-Apim-Subscription-Key": self.bing_subscription_key}
|
|
params = {
|
|
"q": search_term,
|
|
"count": count,
|
|
"textDecorations": True,
|
|
"textFormat": "HTML",
|
|
**self.search_kwargs,
|
|
}
|
|
response = requests.get(
|
|
self.bing_search_url,
|
|
headers=headers,
|
|
params=params, # type: ignore
|
|
)
|
|
response.raise_for_status()
|
|
search_results = response.json()
|
|
if "webPages" in search_results:
|
|
return search_results["webPages"]["value"]
|
|
return []
|
|
|
|
@root_validator(pre=True)
|
|
def validate_environment(cls, values: Dict) -> Dict:
|
|
"""Validate that api key and endpoint exists in environment."""
|
|
bing_subscription_key = get_from_dict_or_env(
|
|
values, "bing_subscription_key", "BING_SUBSCRIPTION_KEY"
|
|
)
|
|
values["bing_subscription_key"] = bing_subscription_key
|
|
|
|
bing_search_url = get_from_dict_or_env(
|
|
values,
|
|
"bing_search_url",
|
|
"BING_SEARCH_URL",
|
|
# default="https://api.bing.microsoft.com/v7.0/search",
|
|
)
|
|
|
|
values["bing_search_url"] = bing_search_url
|
|
|
|
return values
|
|
|
|
def run(self, query: str) -> str:
|
|
"""Run query through BingSearch and parse result."""
|
|
snippets = []
|
|
results = self._bing_search_results(query, count=self.k)
|
|
if len(results) == 0:
|
|
return "No good Bing Search Result was found"
|
|
for result in results:
|
|
snippets.append(result["snippet"])
|
|
|
|
return " ".join(snippets)
|
|
|
|
def results(self, query: str, num_results: int) -> List[Dict]:
|
|
"""Run query through BingSearch and return metadata.
|
|
|
|
Args:
|
|
query: The query to search for.
|
|
num_results: The number of results to return.
|
|
|
|
Returns:
|
|
A list of dictionaries with the following keys:
|
|
snippet - The description of the result.
|
|
title - The title of the result.
|
|
link - The link to the result.
|
|
"""
|
|
metadata_results = []
|
|
results = self._bing_search_results(query, count=num_results)
|
|
if len(results) == 0:
|
|
return [{"Result": "No good Bing Search Result was found"}]
|
|
for result in results:
|
|
metadata_result = {
|
|
"snippet": result["snippet"],
|
|
"title": result["name"],
|
|
"link": result["url"],
|
|
}
|
|
metadata_results.append(metadata_result)
|
|
|
|
return metadata_results
|