Spaces:
Runtime error
Runtime error
File size: 3,309 Bytes
58d33f0 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 |
"""Util that calls Google Search using the Serper.dev API."""
from typing import Dict, Optional
import requests
from pydantic.class_validators import root_validator
from pydantic.main import BaseModel
from langchain.utils import get_from_dict_or_env
class GoogleSerperAPIWrapper(BaseModel):
"""Wrapper around the Serper.dev Google Search API.
You can create a free API key at https://serper.dev.
To use, you should have the environment variable ``SERPER_API_KEY``
set with your API key, or pass `serper_api_key` as a named parameter
to the constructor.
Example:
.. code-block:: python
from langchain import GoogleSerperAPIWrapper
google_serper = GoogleSerperAPIWrapper()
"""
k: int = 10
gl: str = "us"
hl: str = "en"
serper_api_key: Optional[str] = None
@root_validator()
def validate_environment(cls, values: Dict) -> Dict:
"""Validate that api key exists in environment."""
serper_api_key = get_from_dict_or_env(
values, "serper_api_key", "SERPER_API_KEY"
)
values["serper_api_key"] = serper_api_key
return values
def run(self, query: str) -> str:
"""Run query through GoogleSearch and parse result."""
results = self._google_serper_search_results(query, gl=self.gl, hl=self.hl)
return self._parse_results(results)
def _parse_results(self, results: dict) -> str:
snippets = []
if results.get("answerBox"):
answer_box = results.get("answerBox", {})
if answer_box.get("answer"):
return answer_box.get("answer")
elif answer_box.get("snippet"):
return answer_box.get("snippet").replace("\n", " ")
elif answer_box.get("snippetHighlighted"):
return ", ".join(answer_box.get("snippetHighlighted"))
if results.get("knowledgeGraph"):
kg = results.get("knowledgeGraph", {})
title = kg.get("title")
entity_type = kg.get("type")
if entity_type:
snippets.append(f"{title}: {entity_type}.")
description = kg.get("description")
if description:
snippets.append(description)
for attribute, value in kg.get("attributes", {}).items():
snippets.append(f"{title} {attribute}: {value}.")
for result in results["organic"][: self.k]:
if "snippet" in result:
snippets.append(result["snippet"])
for attribute, value in result.get("attributes", {}).items():
snippets.append(f"{attribute}: {value}.")
if len(snippets) == 0:
return "No good Google Search Result was found"
return " ".join(snippets)
def _google_serper_search_results(self, search_term: str, gl: str, hl: str) -> dict:
headers = {
"X-API-KEY": self.serper_api_key or "",
"Content-Type": "application/json",
}
params = {"q": search_term, "gl": gl, "hl": hl}
response = requests.post(
"https://google.serper.dev/search", headers=headers, params=params
)
response.raise_for_status()
search_results = response.json()
return search_results
|