|
|
from smolagents import Tool |
|
|
from google import genai |
|
|
import requests |
|
|
import os |
|
|
|
|
|
class ReadVideoTool(Tool): |
|
|
name = "video_inspector" |
|
|
description = "Analyze video content to answer questions." |
|
|
inputs = { |
|
|
"video_url": {"type": "string", "description": "URL of video."}, |
|
|
"user_query": {"type": "string", "description": "Question about video."} |
|
|
} |
|
|
output_type = "string" |
|
|
|
|
|
MODEL_PRIORITY = ["gemini-2.5-flash", "gemini-2.5-flash-other-project"] |
|
|
|
|
|
def forward(self, video_url: str, user_query: str) -> str: |
|
|
for model_name in self.MODEL_PRIORITY: |
|
|
|
|
|
|
|
|
req = { |
|
|
'model': f'models/{model_name}', |
|
|
|
|
|
} |
|
|
|
|
|
|
|
|
url = f'https://generativelanguage.googleapis.com/v1beta/models/{model_name}:generateContent?key={os.getenv("GOOGLE_API_KEY")}' |
|
|
res = requests.post(url, json=req, headers={'Content-Type': 'application/json'}) |
|
|
if res.status_code != 200: |
|
|
return f"Video error {res.status_code}: {res.text}" |
|
|
if res.status_code == 429: |
|
|
|
|
|
return f"Tool unavailable: Gemini API Quota Limit Reached (Status 429). Cannot process video analysis right now." |
|
|
|
|
|
if res.status_code == 503: |
|
|
|
|
|
return f"Tool unavailable: Gemini API Service Unavailable (Status 503). Try again later." |
|
|
|
|
|
parts = res.json()['candidates'][0]['content']['parts'] |
|
|
return "".join([p.get('text', '') for p in parts]) |