Source code for langchain_community.utilities.you

"""Util that calls you.com Search API.

In order to set this up, follow instructions at:
"""
from typing import Any, Dict, List, Literal, Optional

import aiohttp
import requests
from langchain_core.documents import Document
from langchain_core.pydantic_v1 import BaseModel, Field, root_validator
from langchain_core.utils import get_from_dict_or_env

YOU_API_URL = "https://api.ydc-index.io"


[docs]class YouHitMetadata(BaseModel): """Metadata on a single hit from you.com""" title: str = Field(description="The title of the result") url: str = Field(description="The url of the result") thumbnail_url: str = Field(description="Thumbnail associated with the result") description: str = Field(description="Details about the result")
[docs]class YouHit(YouHitMetadata): """A single hit from you.com, which may contain multiple snippets""" snippets: List[str] = Field(description="One or snippets of text")
[docs]class YouAPIOutput(BaseModel): """The output from you.com api""" hits: List[YouHit] = Field( description="A list of dictionaries containing the results" )
[docs]class YouDocument(BaseModel): """The output of parsing one snippet""" page_content: str = Field(description="One snippet of text") metadata: YouHitMetadata
[docs]class YouSearchAPIWrapper(BaseModel): """Wrapper for you.com Search API. To connect to the You.com api requires an API key which you can get at https://api.you.com. You can check out the docs at https://documentation.you.com. You need to set the environment variable `YDC_API_KEY` for retriever to operate. Attributes ---------- ydc_api_key: str, optional you.com api key, if YDC_API_KEY is not set in the environment num_web_results: int, optional The max number of web results to return, must be under 20 safesearch: str, optional Safesearch settings, one of off, moderate, strict, defaults to moderate country: str, optional Country code, ex: 'US' for united states, see api docs for list k: int, optional max number of Documents to return using `results()` n_hits: int, optional, deprecated Alias for num_web_results n_snippets_per_hit: int, optional limit the number of snippets returned per hit endpoint_type: str, optional you.com endpoints: search, news, rag; `web` and `snippet` alias `search` `rag` returns `{'message': 'Forbidden'}` @todo `news` endpoint """ ydc_api_key: Optional[str] = None num_web_results: Optional[int] = None safesearch: Optional[str] = None country: Optional[str] = None k: Optional[int] = None n_snippets_per_hit: Optional[int] = None # @todo deprecate `snippet`, not part of API endpoint_type: Literal["search", "news", "rag", "snippet"] = "search" # should deprecate n_hits n_hits: Optional[int] = None @root_validator(pre=True) def validate_environment(cls, values: Dict) -> Dict: """Validate that api key exists in environment.""" ydc_api_key = get_from_dict_or_env(values, "ydc_api_key", "YDC_API_KEY") values["ydc_api_key"] = ydc_api_key return values def _parse_results(self, raw_search_results: Dict) -> List[Document]: """ Extracts snippets from each hit and puts them in a Document Parameters: raw_search_results: A dict containing list of hits Returns: List[YouDocument]: A dictionary of parsed results """ # return news results if self.endpoint_type == "news": return [ Document(page_content=result["description"], metadata=result) for result in raw_search_results["news"]["results"] ] docs = [] for hit in raw_search_results["hits"]: n_snippets_per_hit = self.n_snippets_per_hit or len(hit.get("snippets")) for snippet in hit.get("snippets")[:n_snippets_per_hit]: docs.append( Document( page_content=snippet, metadata={ "url": hit.get("url"), "thumbnail_url": hit.get("thumbnail_url"), "title": hit.get("title"), "description": hit.get("description"), }, ) ) if self.k is not None and len(docs) >= self.k: return docs return docs
[docs] def raw_results( self, query: str, **kwargs: Any, ) -> Dict: """Run query through you.com Search and return hits. Args: query: The query to search for. num_web_results: The maximum number of results to return. safesearch: Safesearch settings, one of off, moderate, strict, defaults to moderate country: Country code Returns: YouAPIOutput """ headers = {"X-API-Key": self.ydc_api_key or ""} params = { "query": query, "num_web_results": self.num_web_results, "safesearch": self.safesearch, "country": self.country, **kwargs, } params = {k: v for k, v in params.items() if v is not None} # news endpoint expects `q` instead of `query` if self.endpoint_type == "news": params["q"] = params["query"] del params["query"] # @todo deprecate `snippet`, not part of API if self.endpoint_type == "snippet": self.endpoint_type = "search" response = requests.get( # type: ignore f"{YOU_API_URL}/{self.endpoint_type}", params=params, headers=headers, ) response.raise_for_status() return response.json()
[docs] def results( self, query: str, **kwargs: Any, ) -> List[Document]: """Run query through you.com Search and parses results into Documents.""" raw_search_results = self.raw_results( query, **{key: value for key, value in kwargs.items() if value is not None}, ) return self._parse_results(raw_search_results)
[docs] async def raw_results_async( self, query: str, **kwargs: Any, ) -> Dict: """Get results from the you.com Search API asynchronously.""" headers = {"X-API-Key": self.ydc_api_key or ""} params = { "query": query, "num_web_results": self.num_web_results, "safesearch": self.safesearch, "country": self.country, **kwargs, } params = {k: v for k, v in params.items() if v is not None} # news endpoint expects `q` instead of `query` if self.endpoint_type == "news": params["q"] = params["query"] del params["query"] # @todo deprecate `snippet`, not part of API if self.endpoint_type == "snippet": self.endpoint_type = "search" async with aiohttp.ClientSession() as session: async with session.get( url=f"{YOU_API_URL}/{self.endpoint_type}", params=params, headers=headers, ) as res: if res.status == 200: results = await res.json() return results else: raise Exception(f"Error {res.status}: {res.reason}")
[docs] async def results_async( self, query: str, **kwargs: Any, ) -> List[Document]: raw_search_results_async = await self.raw_results_async( query, **{key: value for key, value in kwargs.items() if value is not None}, ) return self._parse_results(raw_search_results_async)