Update tools/WebSearchTool.py
Browse files- tools/WebSearchTool.py +50 -57
tools/WebSearchTool.py
CHANGED
|
@@ -1,35 +1,34 @@
|
|
| 1 |
import requests
|
| 2 |
from smolagents import Tool
|
|
|
|
| 3 |
import json
|
| 4 |
|
| 5 |
-
class
|
| 6 |
-
"""
|
| 7 |
-
|
| 8 |
-
|
| 9 |
-
|
| 10 |
-
when specific search APIs are overloaded or require payment/keys. It attempts
|
| 11 |
-
multiple public instances for high resilience against temporary server outages or blocks.
|
| 12 |
"""
|
| 13 |
-
name = "
|
| 14 |
-
description = "Use
|
| 15 |
|
| 16 |
inputs = {
|
| 17 |
"query": {"type": "string", "description": "The search term to look up."}
|
| 18 |
}
|
| 19 |
output_type = "string"
|
| 20 |
|
| 21 |
-
|
| 22 |
-
|
| 23 |
-
|
| 24 |
-
|
| 25 |
-
"https://
|
| 26 |
-
|
| 27 |
-
|
| 28 |
-
|
| 29 |
|
| 30 |
def forward(self, query: str) -> str:
|
| 31 |
"""
|
| 32 |
-
Executes a
|
| 33 |
|
| 34 |
Args:
|
| 35 |
query: The search term provided by the agent.
|
|
@@ -37,52 +36,46 @@ class SearxngSearchTool(Tool):
|
|
| 37 |
Returns:
|
| 38 |
A formatted string of search results, or an error message.
|
| 39 |
"""
|
|
|
|
| 40 |
|
| 41 |
params = {
|
|
|
|
| 42 |
"q": query,
|
| 43 |
-
"
|
| 44 |
-
"
|
| 45 |
-
"pageno": 1,
|
| 46 |
}
|
| 47 |
|
| 48 |
-
|
| 49 |
-
|
| 50 |
-
|
| 51 |
-
|
| 52 |
-
|
| 53 |
-
try:
|
| 54 |
-
# Make the API request
|
| 55 |
-
response = requests.get(instance_url, params=params, timeout=5) # Added timeout
|
| 56 |
-
response.raise_for_status()
|
| 57 |
-
data = response.json()
|
| 58 |
-
|
| 59 |
-
# Extract the search results
|
| 60 |
-
items = data.get('results', [])
|
| 61 |
|
| 62 |
-
|
| 63 |
-
|
| 64 |
-
print(f"Instance {instance_url} returned no results. Trying next instance.")
|
| 65 |
-
continue
|
| 66 |
|
| 67 |
-
|
| 68 |
-
#
|
| 69 |
-
|
| 70 |
-
|
| 71 |
-
|
| 72 |
-
|
| 73 |
-
|
| 74 |
-
|
| 75 |
|
| 76 |
-
|
| 77 |
-
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 78 |
|
| 79 |
-
|
| 80 |
-
|
| 81 |
-
last_error = f"Instance {instance_url} failed: {e}"
|
| 82 |
-
print(last_error)
|
| 83 |
-
except Exception as e:
|
| 84 |
-
last_error = f"Error processing Searxng results for {instance_url}: {e}"
|
| 85 |
-
print(last_error)
|
| 86 |
|
| 87 |
-
|
| 88 |
-
|
|
|
|
|
|
|
|
|
| 1 |
import requests
|
| 2 |
from smolagents import Tool
|
| 3 |
+
import os
|
| 4 |
import json
|
| 5 |
|
| 6 |
+
class ScaleSerpSearchTool(Tool):
|
| 7 |
+
"""
|
| 8 |
+
A reliable web search tool using the Scale SERP API. This tool is chosen
|
| 9 |
+
for its high reliability and accessible free tier when other search providers
|
| 10 |
+
fail due to quota or instability.
|
|
|
|
|
|
|
| 11 |
"""
|
| 12 |
+
name = "scaleserp_search"
|
| 13 |
+
description = "Use the reliable Scale SERP API to fetch current and general knowledge from Google Search results. Requires an API key."
|
| 14 |
|
| 15 |
inputs = {
|
| 16 |
"query": {"type": "string", "description": "The search term to look up."}
|
| 17 |
}
|
| 18 |
output_type = "string"
|
| 19 |
|
| 20 |
+
def __init__(self, **kwargs):
|
| 21 |
+
super().__init__(**kwargs)
|
| 22 |
+
# Retrieve API key from environment variables
|
| 23 |
+
self.api_key = os.getenv("SCALESERP_API_KEY")
|
| 24 |
+
self.endpoint = "https://api.scaleserp.com/search"
|
| 25 |
+
|
| 26 |
+
if not self.api_key:
|
| 27 |
+
raise ValueError("SCALESERP_API_KEY secret not found. You need a key from the Scale SERP free tier.")
|
| 28 |
|
| 29 |
def forward(self, query: str) -> str:
|
| 30 |
"""
|
| 31 |
+
Executes a Scale SERP search query and formats the top results (up to 3).
|
| 32 |
|
| 33 |
Args:
|
| 34 |
query: The search term provided by the agent.
|
|
|
|
| 36 |
Returns:
|
| 37 |
A formatted string of search results, or an error message.
|
| 38 |
"""
|
| 39 |
+
print(f"Executing Scale SERP Search for: '{query}'")
|
| 40 |
|
| 41 |
params = {
|
| 42 |
+
"api_key": self.api_key,
|
| 43 |
"q": query,
|
| 44 |
+
"num": 3, # Request up to 3 results
|
| 45 |
+
"gl": "us" # Limit results to US region for consistency
|
|
|
|
| 46 |
}
|
| 47 |
|
| 48 |
+
try:
|
| 49 |
+
# Make the API request
|
| 50 |
+
response = requests.get(self.endpoint, params=params, timeout=10)
|
| 51 |
+
response.raise_for_status()
|
| 52 |
+
data = response.json()
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 53 |
|
| 54 |
+
# Extract the organic results
|
| 55 |
+
organic_results = data.get('organic_results', [])
|
|
|
|
|
|
|
| 56 |
|
| 57 |
+
if not organic_results:
|
| 58 |
+
# Check for a knowledge panel or featured snippet as a fallback
|
| 59 |
+
featured_snippet = data.get('knowledge_graph', {}).get('description')
|
| 60 |
+
if featured_snippet:
|
| 61 |
+
return f"RESULT 1 (Featured Snippet): '{query}'\nCONTENT: {featured_snippet}\nSOURCE: Knowledge Panel"
|
| 62 |
+
|
| 63 |
+
# If nothing useful is found:
|
| 64 |
+
return "XX record info: No results found."
|
| 65 |
|
| 66 |
+
search_results = []
|
| 67 |
+
# Process the top 3 organic results
|
| 68 |
+
for i, item in enumerate(organic_results[:3]):
|
| 69 |
+
search_results.append(
|
| 70 |
+
f"RESULT {i+1}: '{item.get('title', 'N/A')}'\n"
|
| 71 |
+
f"CONTENT: {item.get('snippet', 'No snippet available.')}\n"
|
| 72 |
+
f"SOURCE: {item.get('url', 'N/A')}"
|
| 73 |
+
)
|
| 74 |
|
| 75 |
+
# Join the results with a clear separator
|
| 76 |
+
return "\n\n---SEPARATOR---\n\n".join(search_results)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 77 |
|
| 78 |
+
except requests.exceptions.RequestException as e:
|
| 79 |
+
return f"Error during Scale SERP API Request: {e}"
|
| 80 |
+
except Exception as e:
|
| 81 |
+
return f"Error processing Scale SERP results: {e}"
|