Spaces:
Sleeping
Sleeping
File size: 2,564 Bytes
9598d95 35b3426 9598d95 35b3426 9598d95 bacd706 35b3426 9598d95 35b3426 9598d95 35b3426 9598d95 |
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 |
import os
from smolagents import Tool
from googleapiclient.discovery import build
class YouTubeSearchTool(Tool):
"""
A tool to search for YouTube videos using the Google Custom Search Engine (CSE) API,
by specifically restricting results to the youtube.com domain.
"""
name = "youtube_search"
description = "Searches for and returns links and snippets for YouTube videos related to a specific query."
inputs = {
# Corrected the typo/broken line to read correctly
"query": {"type": "string", "description": "The search term to find relevant YouTube videos for."}
}
output_type = "string"
def __init__(self, **kwargs):
super().__init__(**kwargs)
# Retrieve credentials from environment variables
self.api_key = os.getenv("GOOGLE_API_KEY")
self.cse_id = os.getenv("GOOGLE_CSE_ID")
if not self.api_key or not self.cse_id:
# Note: This check relies on environment variables being set up
raise ValueError("GOOGLE_API_KEY or GOOGLE_CSE_ID secret not found.")
# Initialize the Google Custom Search service
self.service = build(
"customsearch", "v1", developerKey=self.api_key
)
def forward(self, query: str) -> str:
"""
Executes a Google search query, restricting results to youtube.com.
Args:
query: The search term provided by the agent.
Returns:
A formatted string of search results (max 3), or an error message.
"""
print(f"Executing YouTube search for: '{query}'")
# Modify the query to ensure results are restricted to YouTube
youtube_query = f"{query} site:youtube.com"
try:
# Execute the search request for up to 3 results
res = self.service.cse().list(
q=youtube_query,
cx=self.cse_id,
num=3
).execute()
items = res.get('items', [])
if not items:
return "XX record info: No YouTube results found."
search_results = []
for i, item in enumerate(items):
search_results.append(
f"RESULT {i+1}: '{item.get('title')}'\n"
f"CONTENT: {item.get('snippet')}\n"
f"SOURCE: {item.get('link')}"
)
return "\n\n---SEPARATOR---\n\n".join(search_results)
except Exception as e:
return f"Error during YouTube Search API call: {e}" |