Spaces:
Sleeping
Sleeping
File size: 3,430 Bytes
60d836b c7aea9b 60d836b a008b1a 60d836b |
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 |
import json
import requests
from typing import List, Optional
# from langchain_community.document_transformers import MarkdownifyTransformer
from langchain_core.documents import Document
from langchain_core.pydantic_v1 import BaseModel, Field
class BraveAIWrapper(BaseModel):
api_key: str = Field(..., description="API key for Brave Search")
base_search_url: str = Field("https://api.search.brave.com/res/v1/web/search", const=True)
base_summarize_url: str = Field("https://api.search.brave.com/res/v1/summarizer/search", const=True)
headers: dict = Field(default_factory=dict, description="HTTP headers for API requests")
class Config:
arbitrary_types_allowed = True
def __init__(self, **data):
super().__init__(**data)
self.headers = {
"X-Subscription-Token": self.api_key,
"Accept": "application/json",
"Accept-Encoding": "gzip",
}
def get_brave_results(self, query: str, count: int = 3, safe_search: str = 'off') -> Optional[dict]:
"""
Get search results from Brave Search.
Args:
query (str): The search query.
count (int): Number of results to return.
safe_search (str): Safe search filter (off, moderate, strict).
Returns:
Optional[dict]: JSON response from Brave Search API or None if an error occurs.
"""
params = {
"q": query,
"count": count,
"summary": True,
"safe_search": safe_search,
"extra_snippets": True,
}
try:
response = requests.get(self.base_search_url, headers=self.headers, params=params)
response.raise_for_status()
return response.json()
except requests.exceptions.RequestException as e:
print(f"Error: {e}")
return None
def summarize_results(self, summarizer_key: str) -> Optional[dict]:
"""
Summarize search results using Brave Summarizer.
Args:
summarizer_key (str): The key for the summarizer.
Returns:
Optional[dict]: JSON response from Brave Summarizer API or None if an error occurs.
"""
params = {"key": summarizer_key}
try:
response = requests.get(self.base_summarize_url, headers=self.headers, params=params)
response.raise_for_status()
return response.json()
except requests.exceptions.RequestException as e:
print(f"Error: {e}")
return None
def get_and_summarize(self, query: str, count: int = 3, safe_search: str = 'off') -> Optional[str]:
"""
Get and summarize search results from Brave Search.
Args:
query (str): The search query.
count (int): Number of results to return.
safe_search (str): Safe search filter (off, moderate, strict).
Returns:
Optional[str]: Summarized result or None if an error occurs.
"""
results = self.get_brave_results(query, count, safe_search)
if results and 'summarizer' in results:
summarizer_key = results['summarizer']['key']
summary = self.summarize_results(summarizer_key)
if summary and 'summary' in summary and len(summary['summary']) > 0:
return summary['summary'][0]['data']
return None
|