"""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 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)