aniket47/document-intelligence-chatbot
0
1import requests2import json3from typing import List, Dict, Optional4import os5from dotenv import load_dotenv6 7# Load environment variables8load_dotenv()9 10class WebSearcher:11 """12 Serper.dev API integration for web search functionality13 """14 15 def __init__(self, api_key: Optional[str] = None):16 self.api_key = api_key or os.getenv("SERPER_API_KEY")17 self.base_url = "https://google.serper.dev/search"18 19 if not self.api_key:20 raise ValueError("Serper API key is required. Please set SERPER_API_KEY in your .env file")21 22 def search(self, query: str, num_results: int = 5) -> Dict:23 """24 Perform web search using Serper API25 26 Args:27 query: Search query28 num_results: Number of results to return29 30 Returns:31 Dictionary containing search results32 """33 headers = {34 'X-API-KEY': self.api_key,35 'Content-Type': 'application/json'36 }37 38 payload = {39 'q': query,40 'num': num_results,41 'page': 142 }43 44 try:45 response = requests.post(46 self.base_url,47 headers=headers,48 data=json.dumps(payload),49 timeout=1050 )51 52 response.raise_for_status()53 return response.json()54 55 except requests.exceptions.RequestException as e:56 raise Exception(f"Web search failed: {str(e)}")57 58 def format_search_results(self, search_response: Dict) -> List[Dict]:59 """60 Format search results into a standardized structure61 62 Args:63 search_response: Raw response from Serper API64 65 Returns:66 List of formatted search results67 """68 formatted_results = []69 70 # Process organic results71 organic_results = search_response.get('organic', [])72 73 for i, result in enumerate(organic_results):74 formatted_result = {75 'rank': i + 1,76 'title': result.get('title', ''),77 'snippet': result.get('snippet', ''),78 'link': result.get('link', ''),79 'source': result.get('displayLink', ''),80 'type': 'organic'81 }82 formatted_results.append(formatted_result)83 84 # Process answer box if available85 answer_box = search_response.get('answerBox')86 if answer_box:87 formatted_result = {88 'rank': 0, # Answer box gets top priority89 'title': answer_box.get('title', 'Direct Answer'),90 'snippet': answer_box.get('answer', answer_box.get('snippet', '')),91 'link': answer_box.get('link', ''),92 'source': answer_box.get('displayLink', 'Google'),93 'type': 'answer_box'94 }95 formatted_results.insert(0, formatted_result)96 97 # Process knowledge graph if available98 knowledge_graph = search_response.get('knowledgeGraph')99 if knowledge_graph:100 formatted_result = {101 'rank': 0,102 'title': knowledge_graph.get('title', 'Knowledge Graph'),103 'snippet': knowledge_graph.get('description', ''),104 'link': knowledge_graph.get('descriptionLink', ''),105 'source': knowledge_graph.get('source', 'Google Knowledge Graph'),106 'type': 'knowledge_graph'107 }108 formatted_results.insert(0 if not answer_box else 1, formatted_result)109 110 return formatted_results111 112 def search_and_format(self, query: str, num_results: int = 5) -> List[Dict]:113 """114 Perform search and return formatted results115 116 Args:117 query: Search query118 num_results: Number of results to return119 120 Returns:121 List of formatted search results122 """123 try:124 # Perform search125 search_response = self.search(query, num_results)126 127 # Format results128 formatted_results = self.format_search_results(search_response)129 130 return formatted_results131 132 except Exception as e:133 print(f"Error in web search: {str(e)}")134 return []135 136 def create_search_summary(self, results: List[Dict], max_length: int = 1000) -> str:137 """138 Create a summary from search results139 140 Args:141 results: List of search results142 max_length: Maximum length of summary143 144 Returns:145 Summary text with sources146 """147 if not results:148 return "No web search results found."149 150 summary_parts = []151 sources = []152 current_length = 0153 154 for result in results[:3]: # Use top 3 results for summary155 snippet = result.get('snippet', '')156 title = result.get('title', '')157 source = result.get('source', '')158 link = result.get('link', '')159 160 if snippet and current_length + len(snippet) < max_length:161 summary_parts.append(f"**{title}**: {snippet}")162 if source and link:163 sources.append(f"- [{source}]({link})")164 current_length += len(snippet) + len(title) + 4165 166 # Combine summary parts167 summary = "\n\n".join(summary_parts)168 169 if sources:170 summary += "\n\n**Sources:**\n" + "\n".join(sources)171 172 return summary