Spaces:
Sleeping
Sleeping
File size: 4,752 Bytes
226b286 |
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 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 |
import os
import requests
from agents import function_tool
from typing import Optional
# ---------------------------------------------------------
# Load environment variables
# ---------------------------------------------------------
# ============================================================
# 🔹 GOOGLE SEARCH TOOLSET (Serper.dev API)
# ============================================================
@function_tool
def google_search(query: str, num_results: int = 3) -> str:
"""
Perform a general Google search using Serper.dev API.
Parameters:
-----------
query : str
The search query string, e.g., "latest Tesla stock news".
num_results : int, optional (default=3)
Maximum number of search results to return.
Returns:
--------
str
Formatted string of top search results, each including:
- Title of the page
- URL link
- Snippet / description
If no results are found or API key is missing, returns an error message.
Example:
--------
google_search("AI in finance", num_results=2)
Output:
Title: How AI is Transforming Finance
Link: https://example.com/ai-finance
Snippet: AI is increasingly used for trading, risk management...
Title: AI Applications in Banking
Link: https://example.com/ai-banking
Snippet: Banks are leveraging AI for customer service, fraud detection...
"""
print(f"[DEBUG] google_search called with query='{query}', num_results={num_results}")
try:
api_key = os.getenv("SERPER_API_KEY")
if not api_key:
return "Error: SERPER_API_KEY missing in environment variables."
url = "https://google.serper.dev/search"
headers = {"X-API-KEY": api_key, "Content-Type": "application/json"}
payload = {"q": query, "num": num_results, "tbs": "qdr:d"} # results from last 24h
response = requests.post(url, headers=headers, json=payload, timeout=10)
response.raise_for_status()
data = response.json()
if "organic" not in data or not data["organic"]:
return f"No results found for query: '{query}'"
formatted_results = [
f"Title: {item.get('title')}\n"
f"Link: {item.get('link')}\n"
f"Snippet: {item.get('snippet', '')}\n"
for item in data["organic"][:num_results]
]
return "\n".join(formatted_results)
except requests.exceptions.RequestException as e:
print(f"[DEBUG] Network error during Google search: {e}")
return f"Network error during Google search: {e}"
except Exception as e:
print(f"[DEBUG] Error performing Google search: {e}")
return f"Error performing Google search: {e}"
@function_tool
def google_search_recent(query: str, num_results: int = 3, timeframe: str = "d") -> str:
"""
Perform a Google search with time-based filtering using Serper.dev API.
Parameters:
-----------
query : str
The search query string.
num_results : int, optional (default=3)
Maximum number of search results to return.
timeframe : str, optional (default="d")
Time range for results:
- "d" = past day
- "w" = past week
- "m" = past month
- "y" = past year
Returns:
--------
str
Formatted string of recent search results.
"""
print(f"[DEBUG] google_search_recent called with query='{query}', timeframe={timeframe}")
try:
api_key = os.getenv("SERPER_API_KEY")
if not api_key:
return "Error: SERPER_API_KEY missing in environment variables."
url = "https://google.serper.dev/search"
headers = {"X-API-KEY": api_key, "Content-Type": "application/json"}
payload = {"q": query, "num": num_results, "tbs": f"qdr:{timeframe}"}
response = requests.post(url, headers=headers, json=payload, timeout=10)
response.raise_for_status()
data = response.json()
if "organic" not in data or not data["organic"]:
return f"No recent results found for query: '{query}'"
formatted_results = [
f"Title: {item.get('title')}\n"
f"Link: {item.get('link')}\n"
f"Snippet: {item.get('snippet', '')}\n"
for item in data["organic"][:num_results]
]
return f"Recent results ({timeframe}):\n\n" + "\n".join(formatted_results)
except requests.exceptions.RequestException as e:
print(f"[DEBUG] Network error: {e}")
return f"Network error during Google search: {e}"
except Exception as e:
print(f"[DEBUG] Error: {e}")
return f"Error performing Google search: {e}"
|