Source code for langchain_community.utilities.google_serper

"""Util that calls Google Search using the Serper.dev API."""

from typing import Any, Dict, List, Optional

import aiohttp
import requests
from langchain_core.utils import get_from_dict_or_env
from pydantic import BaseModel, ConfigDict, model_validator
from typing_extensions import Literal


[docs] 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_community.utilities import GoogleSerperAPIWrapper google_serper = GoogleSerperAPIWrapper() """ k: int = 10 gl: str = "us" hl: str = "en" # "places" and "images" is available from Serper but not implemented in the # parser of run(). They can be used in results() type: Literal["news", "search", "places", "images"] = "search" result_key_for_type: dict = { "news": "news", "places": "places", "images": "images", "search": "organic", } tbs: Optional[str] = None serper_api_key: Optional[str] = None aiosession: Optional[aiohttp.ClientSession] = None model_config = ConfigDict( arbitrary_types_allowed=True, ) @model_validator(mode="before") @classmethod def validate_environment(cls, values: Dict) -> Any: """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
[docs] def results(self, query: str, **kwargs: Any) -> Dict: """Run query through GoogleSearch.""" return self._google_serper_api_results( query, gl=self.gl, hl=self.hl, num=self.k, tbs=self.tbs, search_type=self.type, **kwargs, )
[docs] def run(self, query: str, **kwargs: Any) -> str: """Run query through GoogleSearch and parse result.""" results = self._google_serper_api_results( query, gl=self.gl, hl=self.hl, num=self.k, tbs=self.tbs, search_type=self.type, **kwargs, ) return self._parse_results(results)
[docs] async def aresults(self, query: str, **kwargs: Any) -> Dict: """Run query through GoogleSearch.""" results = await self._async_google_serper_search_results( query, gl=self.gl, hl=self.hl, num=self.k, search_type=self.type, tbs=self.tbs, **kwargs, ) return results
[docs] async def arun(self, query: str, **kwargs: Any) -> str: """Run query through GoogleSearch and parse result async.""" results = await self._async_google_serper_search_results( query, gl=self.gl, hl=self.hl, num=self.k, search_type=self.type, tbs=self.tbs, **kwargs, ) return self._parse_results(results)
def _parse_snippets(self, results: dict) -> List[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 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[self.result_key_for_type[self.type]][: 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 snippets def _parse_results(self, results: dict) -> str: return " ".join(self._parse_snippets(results)) def _google_serper_api_results( self, search_term: str, search_type: str = "search", **kwargs: Any ) -> dict: headers = { "X-API-KEY": self.serper_api_key or "", "Content-Type": "application/json", } params = { "q": search_term, **{key: value for key, value in kwargs.items() if value is not None}, } response = requests.post( f"https://google.serper.dev/{search_type}", headers=headers, params=params ) response.raise_for_status() search_results = response.json() return search_results async def _async_google_serper_search_results( self, search_term: str, search_type: str = "search", **kwargs: Any ) -> dict: headers = { "X-API-KEY": self.serper_api_key or "", "Content-Type": "application/json", } url = f"https://google.serper.dev/{search_type}" params = { "q": search_term, **{key: value for key, value in kwargs.items() if value is not None}, } if not self.aiosession: async with aiohttp.ClientSession() as session: async with session.post( url, params=params, headers=headers, raise_for_status=False ) as response: search_results = await response.json() else: async with self.aiosession.post( url, params=params, headers=headers, raise_for_status=True ) as response: search_results = await response.json() return search_results