multi search

gemini_agent.py

@@ -54,6 +54,10 @@ class GeminiAgent:
             # Check if question is about music discography or albums
             if self._is_discography_question(question):
                 return await self._handle_discography_question(question)
+                
+            # Check if question is about competitions, awards, or recipients
+            if self._is_competition_question(question):
+                return await self._handle_competition_question(question)
             
             # Regular text-based question
             return await self._handle_text_question(question)
@@ -96,8 +100,30 @@ class GeminiAgent:
         # If it has a music term and either an artist term or a date range, it's likely a discography question
         return has_music_term and (has_artist_term or has_date_range)
         
-    async def _google_search(self, query: str, num_results: int = 5) -> str:
-        """Perform a Google search using the Custom Search API"""
+    def _is_competition_question(self, question: str) -> bool:
+        """Determine if a question is about competitions, awards, or recipients"""
+        q = question.lower()
+        competition_patterns = [
+            "competition", "award", "prize", "medal", "recipient", "winner", "laureate",
+            "finalist", "champion", "trophy", "recognition", "honor", "honour", "nominee"
+        ]
+        
+        # Check for competition-related terms
+        has_competition_term = any(pattern in q for pattern in competition_patterns)
+        
+        # Check for specific patterns that indicate complex competition questions
+        complex_patterns = [
+            "first name", "last name", "nationality", "country", "no longer exists",
+            "century", "decade", "after\s+\d{4}", "before\s+\d{4}", "between\s+\d{4}",
+            "youngest", "oldest", "only", "ever", "never"
+        ]
+        
+        has_complex_pattern = any(re.search(pattern, q) for pattern in complex_patterns)
+        
+        return has_competition_term and has_complex_pattern
+        
+    async def _google_search(self, query: str, num_results: int = 5, exact_terms: str = None, site_restrict: str = None) -> str:
+        """Perform a Google search using the Custom Search API with enhanced options"""
         if not self.google_search_api_key or not self.google_search_cx:
             print("Google Search API key or CX not configured, falling back to DuckDuckGo")
             return self.ddg_tool.run(query)
@@ -111,6 +137,14 @@ class GeminiAgent:
                 'num': num_results
             }
             
+            # Add exact terms if provided
+            if exact_terms:
+                params['exactTerms'] = exact_terms
+                
+            # Add site restriction if provided
+            if site_restrict:
+                params['siteSearch'] = site_restrict
+            
             response = requests.get(url, params=params)
             if response.status_code != 200:
                 print(f"Google Search API error: {response.status_code}")
@@ -127,6 +161,17 @@ class GeminiAgent:
                 title = item.get('title', 'No title')
                 snippet = item.get('snippet', 'No description')
                 link = item.get('link', 'No link')
+                
+                # Try to get more content if available
+                page_map = item.get('pagemap', {})
+                meta_desc = ""
+                if 'metatags' in page_map and page_map['metatags']:
+                    meta_desc = page_map['metatags'][0].get('og:description', '')
+                    
+                # Add the meta description if it provides additional information
+                if meta_desc and meta_desc not in snippet:
+                    snippet += " " + meta_desc
+                    
                 formatted_results += f"Title: {title}\nDescription: {snippet}\nURL: {link}\n\n"
                 
             return formatted_results
@@ -210,6 +255,204 @@ If the answer is a person's name, provide ONLY their first name as requested."""
                 
         return answer
     
+    async def _multi_search(self, queries: list, num_results: int = 5, include_sites: list = None) -> str:
+        """Perform multiple searches and combine the results with enhanced options"""
+        combined_results = ""
+        
+        # Define authoritative sites for different domains
+        authoritative_sites = {
+            "music": ["grammy.org", "billboard.com", "allmusic.com", "musicbrainz.org"],
+            "competition": ["wikipedia.org", "britannica.com"],
+            "awards": ["nobelprize.org", "pulitzer.org", "oscars.org"],
+            "classical": ["classicalmusic.org", "gramophone.co.uk", "medici.tv"]
+        }
+        
+        # Process each query
+        for i, query in enumerate(queries):
+            print(f"Searching for query {i+1}/{len(queries)}: {query[:50]}...")
+            try:
+                # Standard search
+                result = await self._google_search(query, num_results)
+                if result:
+                    combined_results += f"=== Results for query: {query} ===\n{result}\n\n"
+                    
+                # If specific sites are provided, search those too
+                if include_sites:
+                    for site in include_sites:
+                        site_result = await self._google_search(query, num_results=3, site_restrict=site)
+                        if site_result and "no results" not in site_result.lower():
+                            combined_results += f"=== Results from {site} for: {query} ===\n{site_result}\n\n"
+                            
+                # For competition questions, try some authoritative sites
+                if "competition" in query.lower() or "award" in query.lower() or "prize" in query.lower():
+                    for site in authoritative_sites["competition"] + authoritative_sites["awards"]:
+                        site_result = await self._google_search(query, num_results=2, site_restrict=site)
+                        if site_result and "no results" not in site_result.lower():
+                            combined_results += f"=== Results from {site} for: {query} ===\n{site_result}\n\n"
+                            
+                # For classical music questions, try classical music sites
+                if "classical" in query.lower() or "conductor" in query.lower() or "orchestra" in query.lower():
+                    for site in authoritative_sites["classical"]:
+                        site_result = await self._google_search(query, num_results=2, site_restrict=site)
+                        if site_result and "no results" not in site_result.lower():
+                            combined_results += f"=== Results from {site} for: {query} ===\n{site_result}\n\n"
+                            
+                # Try exact term matching for key entities
+                key_terms = self._extract_key_terms(query)
+                if key_terms:
+                    exact_result = await self._google_search(query, num_results=3, exact_terms=key_terms)
+                    if exact_result and "no results" not in exact_result.lower():
+                        combined_results += f"=== Results with exact match for '{key_terms}' ===\n{exact_result}\n\n"
+                        
+            except Exception as e:
+                print(f"Search failed for query {i+1}: {e}")
+                
+        return combined_results
+        
+    def _extract_key_terms(self, query: str) -> str:
+        """Extract key terms from a query for exact matching"""
+        # Extract competition names
+        competition_match = re.search(r'(\w+\s+Competition|\w+\s+Award|\w+\s+Prize)', query, re.IGNORECASE)
+        if competition_match:
+            return competition_match.group(1)
+            
+        # Extract dates
+        date_match = re.search(r'(\d{4})', query)
+        if date_match:
+            return date_match.group(1)
+            
+        # Extract countries
+        country_patterns = ["Soviet Union", "Yugoslavia", "Czechoslovakia", "East Germany"]
+        for country in country_patterns:
+            if country.lower() in query.lower():
+                return country
+                
+        return ""
+        
+    async def _handle_competition_question(self, question: str) -> str:
+        """Handle questions about competitions, awards, and recipients with advanced search"""
+        print(f"Processing competition question: {question[:50]}...")
+        
+        # Extract key entities from the question
+        competition_name = ""
+        time_period = ""
+        nationality_info = ""
+        
+        # Try to extract competition name
+        competition_patterns = [
+            r'(\w+\s+Competition)',  # "Malko Competition"
+            r'(\w+\s+Award)',       # "Nobel Award"
+            r'(\w+\s+Prize)'        # "Pulitzer Prize"
+        ]
+        
+        for pattern in competition_patterns:
+            match = re.search(pattern, question, re.IGNORECASE)
+            if match:
+                competition_name = match.group(1)
+                break
+        
+        # Extract time period information
+        time_patterns = [
+            r'(\d{2}(?:st|nd|rd|th)\s+[Cc]entury)',  # "20th Century"
+            r'(after\s+\d{4})',                      # "after 1977"
+            r'(before\s+\d{4})',                     # "before 1990"
+            r'(between\s+\d{4}\s+and\s+\d{4})'       # "between 1977 and 2000"
+        ]
+        
+        for pattern in time_patterns:
+            match = re.search(pattern, question, re.IGNORECASE)
+            if match:
+                time_period = match.group(1)
+                break
+        
+        # Extract nationality information
+        if "nationality" in question.lower() or "country" in question.lower():
+            if "no longer exists" in question.lower():
+                nationality_info = "country that no longer exists"
+        
+        # Construct specialized search queries
+        search_queries = []
+        
+        # Generic competition queries
+        if competition_name:
+            base_query = f"{competition_name} winners list"
+            search_queries.append(base_query)
+            
+            if time_period:
+                search_queries.append(f"{competition_name} winners {time_period}")
+                
+            if nationality_info:
+                search_queries.append(f"{competition_name} winners {nationality_info}")
+                
+                # For questions about countries that no longer exist, add general queries
+                if "no longer exists" in nationality_info:
+                    # Add queries for common dissolved countries without hardcoding specific competitions
+                    dissolved_countries = ["Soviet Union", "Yugoslavia", "Czechoslovakia", "East Germany"]
+                    for country in dissolved_countries:
+                        search_queries.append(f"{competition_name} winners from {country}")
+                
+            # Add more specific queries
+            if time_period and nationality_info:
+                search_queries.append(f"{competition_name} winners {time_period} {nationality_info}")
+        else:
+            # If we couldn't extract competition name, use the original question
+            search_queries.append(question)
+        
+        # Perform multiple searches with different queries
+        combined_context = await self._multi_search(search_queries)
+        
+        # Also try Wikipedia for general information
+        wiki_context = ""
+        try:
+            if competition_name:
+                wiki_context = self.wiki_tool.run(competition_name)
+                print("Wikipedia search completed")
+        except Exception as e:
+            print(f"Wikipedia tool failed: {e}")
+            
+        # Add Wikipedia context if available
+        if wiki_context and not any(x in wiki_context.lower() for x in ["not found", "no results", "does not contain"]):
+            combined_context += f"Wikipedia context: {wiki_context}\n\n"
+        
+        # Create a specialized prompt for competition questions
+        prompt = f"""Based on the following search results, answer this question about a competition or award:
+
+{combined_context}
+
+Question: {question}
+
+Analyze the search results carefully to find information about competition winners, their nationalities, and the time periods.
+If the question asks about a country that no longer exists, look for winners from countries like the Soviet Union, Yugoslavia, Czechoslovakia, East Germany, etc.
+If asked for a first name only, extract just the first name from the full name.
+
+Provide ONLY the specific information requested with no explanations."""
+        
+        await self._rate_limit()
+        response = self.model.generate_content(
+            prompt,
+            generation_config=genai.types.GenerationConfig(
+                max_output_tokens=100,
+                temperature=0.0
+            )
+        )
+        answer = response.text.strip()
+        
+        # Clean up the answer
+        prefixes = ['The answer is', 'Based on', 'According to', 'The first name is', 'The recipient is']
+        for prefix in prefixes:
+            if answer.lower().startswith(prefix.lower()):
+                answer = answer[len(prefix):].strip()
+                if answer.startswith(','):
+                    answer = answer[1:].strip()
+        
+        # If the question asks for just a first name, extract it
+        if "first name" in question.lower():
+            name_parts = answer.split()
+            if name_parts:
+                answer = name_parts[0].rstrip(',.')
+                
+        return answer
+        
     async def _handle_discography_question(self, question: str) -> str:
         """Handle questions about music discography with enhanced search capabilities"""
         print(f"Processing discography question: {question[:50]}...")