nynuzz/SamyAgent
0
1import os
2import subprocess
3import base64
4import mimetypes
5from typing import List, Union
6from pathlib import Path
7import math
8import requests
9import pandas as pd
10import cv2
11from pytubefix import YouTube
12from youtube_transcript_api import YouTubeTranscriptApi
13from openai import OpenAI, APIError
14
15from langchain_core.tools import tool
16from langchain_community.document_loaders import CSVLoader
17
18from web_search_agent import web_search_graph
19
20
21#-----------------------------------------------------------------------------
22# Helper Tools
23#-----------------------------------------------------------------------------
24@tool("sort_tool")
25def sort_tool(items: List[Union[float, str]], order: str = "ascending") -> Union[List[Union[float, str]], str]:
26 """
27 Sort a list of numbers (in numeric order) or strings (in alphabetical order).
28 Use this tool whenever you need to sort a list of items.
29
30 Args:
31 items (List[Union[float, str]]): The list of items to sort. The list must contain either only numbers or only strings.
32 order (str, optional): The sorting order. Valid values: 'ascending' or 'descending'. Default is 'ascending'.
33
34 Returns:
35 Union[List[Union[float, str]], str]: The sorted list if successful, or an error message string in case of failure.
36 """
37 print(f"--- ESECUZIONE DEL TOOL 'sort_list' CON INPUT: items={items}, order='{order}' ---")
38
39 # 1. Controlla se la lista è vuota
40 if not items:
41 return []
42
43 # 2. Valida l'argomento 'order' e imposta il flag per l'ordinamento
44 normalized_order = order.lower().strip()
45 if normalized_order == "ascending":
46 reverse_flag = False
47 elif normalized_order == "descending":
48 reverse_flag = True
49 else:
50 return "Error: The 'order' value is invalid. Allowed values are 'ascending' or 'descending'."
51
52 # 3. Esegui l'ordinamento, gestendo possibili errori di tipo
53 try:
54 # Usiamo la funzione built-in sorted(), che è molto efficiente
55 # e gestisce correttamente sia numeri che stringhe (ma non mischiati).
56 sorted_items = sorted(items, reverse=reverse_flag)
57 return sorted_items
58 except TypeError:
59 # Questa eccezione viene sollevata se si cerca di ordinare una lista
60 # con tipi non confrontabili, es. [1, "mela", 3]
61 return "Error: The list contains incompatible data types that cannot be sorted together (e.g., numbers and strings)."
62 except Exception as e:
63 return f"An unexpected error occurred during sorting: {e}"
64
65
66@tool("download_tool")
67def download_tool(task_id: str) -> Union[str, str]:
68 """
69 Download a file associated with a task_id from a predefined URL.
70 The file is saved in the 'uploads' directory with the name 'task_id.ext',
71 where the extension (.ext) is determined dynamically from the server response.
72
73 Args:
74 task_id (str): The unique identifier for the task and the file to download.
75
76 Returns:
77 Union[str, str]: The filename of the downloaded file if successful or an error message string in case of failure.
78 """
79 print(f"--- ESECUZIONE DEL TOOL 'download_file' CON INPUT: task_id={task_id} ---")
80
81 # 1. Impostazioni di base
82 BASE_URL = "https://agents-course-unit4-scoring.hf.space/files/"
83 UPLOADS_DIR = "./uploads/"
84
85 # 2. Assicurarsi che la directory di destinazione esista
86 try:
87 os.makedirs(UPLOADS_DIR, exist_ok=True)
88 except OSError as e:
89 error_message = f"Error: Unable to create the destination directory '{UPLOADS_DIR}'. Details: {e}"
90 print(error_message)
91 return error_message
92
93 # 3. Eseguire la richiesta HTTP per scaricare il file
94 url = f"{BASE_URL}{task_id}"
95 try:
96 # Usare 'stream=True' è una buona pratica per scaricare file
97 with requests.get(url, stream=True, timeout=30) as response:
98 # Controlla se la richiesta ha avuto successo (es. status code 200)
99 response.raise_for_status()
100
101 # 4. Estrarre il nome del file originale per ottenere l'estensione
102 content_disposition = response.headers.get('content-disposition')
103 if not content_disposition:
104 error_message = "Error: The server response does not contain the 'content-disposition' header to get the file name."
105 print(error_message)
106 return error_message
107
108 # Parsing dell'header per trovare il filename. Es: 'attachment; filename="nomefile.ext"'
109 parts = content_disposition.split(';')
110 filename_part = next((part for part in parts if 'filename=' in part), None)
111
112 if not filename_part:
113 error_message = "Error: Unable to find 'filename' in the 'content-disposition' header."
114 print(error_message)
115 return error_message
116
117 original_filename = filename_part.split('=')[1].strip().strip('"')
118 _, extension = os.path.splitext(original_filename)
119
120 if not extension:
121 error_message = f"Error: Unable to get the file extension from '{original_filename}'."
122 print(error_message)
123 return error_message
124
125 # 5. Costruire il percorso di salvataggio e salvare il file
126 local_filename = f"{task_id}{extension}"
127 local_filepath = os.path.join(UPLOADS_DIR, local_filename)
128
129 with open(local_filepath, 'wb') as f:
130 # Scrive il contenuto a pezzi per gestire file di grandi dimensioni
131 for chunk in response.iter_content(chunk_size=8192):
132 f.write(chunk)
133
134 success_message = f"File scaricato con successo e salvato in: {local_filepath}"
135 print(success_message)
136 return local_filename
137
138 except requests.exceptions.RequestException as e:
139 # Gestisce errori di rete, timeout, DNS, etc.
140 error_message = f"Network error occurred while downloading the file: {e}"
141 print(error_message)
142 return error_message
143 except Exception as e:
144 # Cattura qualsiasi altra eccezione imprevista
145 error_message = f"An unexpected error occurred: {e}"
146 print(error_message)
147 return error_message
148
149
150#-----------------------------------------------------------------------------
151# Math Tools
152#-----------------------------------------------------------------------------
153@tool("add_tool")
154def add_tool(numbers: List[float]) -> float:
155 """
156 Calculate the sum of a list of numbers.
157 Use this tool when you need to perform a sum operation on multiple numbers.
158
159 Args:
160 numbers (List[float]): The list of numbers to be summed.
161 """
162 print(f"--- ESECUZIONE DEL TOOL 'sum_numbers' CON INPUT: {numbers} ---")
163 return sum(numbers)
164
165
166@tool("multiply_tool")
167def multiply_tool(numbers: List[float]) -> float:
168 """
169 Calculate the product of a list of numbers.
170 Use this tool when you need to multiply two or more numbers together.
171
172 Args:
173 numbers (List[float]): The list of numbers to be multiplied.
174 """
175 print(f"--- ESECUZIONE DEL TOOL 'multiply_numbers' CON INPUT: {numbers} ---")
176 if not numbers:
177 return 0
178 return math.prod(numbers)
179
180
181@tool("subtract_tool")
182def subtract_tool(minuend: float, subtrahend: float) -> float:
183 """
184 Calculate the subtraction between two numbers (minuend - subtrahend).
185 Use this tool to subtract one number from another.
186
187 Args:
188 minuend (float): The number from which to subtract (the first number).
189 subtrahend (float): The number to subtract (the second number).
190 """
191 print(f"--- ESECUZIONE DEL TOOL 'subtract_numbers' CON INPUT: minuend={minuend}, subtrahend={subtrahend} ---")
192 return minuend - subtrahend
193
194
195@tool("divide_tool")
196def divide_tool(dividend: float, divisor: float) -> Union[float, str]:
197 """
198 Calculate the division between two numbers (dividend / divisor).
199 Also handles the case of division by zero.
200
201 Args:
202 dividend (float): The number to be divided (the numerator).
203 divisor (float): The number to divide by (the denominator).
204 """
205 print(f"--- ESECUZIONE DEL TOOL 'divide_numbers' CON INPUT: dividend={dividend}, divisor={divisor} ---")
206 if divisor == 0:
207 return "Error: Division by zero is not allowed."
208 return dividend / divisor
209
210
211@tool("modulus_tool")
212def modulus_tool(dividend: float, divisor: float) -> Union[float, str]:
213 """
214 Calculate the remainder of the division between two numbers (dividend % divisor).
215 Use this tool when asked for the 'remainder' or the 'modulus' of a division.
216
217 Args:
218 dividend (float): The number being divided (the numerator).
219 divisor (float): The number by which to divide (the denominator).
220 """
221 print(f"--- ESECUZIONE DEL TOOL 'calculate_remainder' CON INPUT: dividend={dividend}, divisor={divisor} ---")
222 if divisor == 0:
223 return "Error: The divisor cannot be zero for the modulus operation."
224 return dividend % divisor
225
226
227@tool("power_tool")
228def power_tool(base: float, exponent: float) -> Union[float, str]:
229 """
230 Calculate a number raised to a power (base^exponent).
231 Use this tool for exponentiation operations.
232
233 Args:
234 base (float): The base of the operation.
235 exponent (float): The exponent to which the base is raised.
236 """
237 print(f"--- ESECUZIONE DEL TOOL 'calculate_power' CON INPUT: base={base}, exponent={exponent} ---")
238 try:
239 # Usiamo math.pow per coerenza e una migliore gestione degli errori
240 result = math.pow(base, exponent)
241 return result
242 except ValueError:
243 # Si verifica se, ad esempio, si cerca di calcolare (-4)^(0.5), che produce un numero complesso.
244 return "Error: Invalid operation. Ensure that the base and exponent do not result in a complex number (e.g., even root of a negative number)."
245
246
247@tool("square_root_tool")
248def square_root_tool(number: float) -> Union[float, str]:
249 """
250 Calculate the square root of a non-negative number.
251 Use this tool specifically to compute the square root.
252
253 Args:
254 number (float): The number for which to calculate the square root. Must be >= 0.
255 """
256 print(f"--- ESECUZIONE DEL TOOL 'square_root' CON INPUT: number={number} ---")
257 if number < 0:
258 return "Error: Cannot calculate the square root of a negative number."
259 return math.sqrt(number)
260
261
262#-----------------------------------------------------------------------------
263# File Tools
264#-----------------------------------------------------------------------------
265@tool("tabular_tool")
266def tabular_tool(filename: str) -> Union[str, str]:
267 """
268 Analyze a local tabular data file (CSV, XLSX, XLS) and return its content
269 as a formatted string. For Excel files, each worksheet is processed individually.
270
271 Args:
272 filename (str): The filename of the CSV, XLSX, or XLS file to analyze.
273
274 Returns:
275 Union[str, str]: A formatted string containing the file's data, or an error message in case of issues.
276 """
277 print(f"--- ESECUZIONE DEL TOOL 'analyze_tabular_data' CON INPUT: filename='{filename}' ---")
278 UPLOADS_DIR = "./uploads/"
279 file_path = os.path.join(UPLOADS_DIR, filename)
280
281 # 1. Validazione dell'input: controlla se il file esiste
282 if not os.path.exists(file_path):
283 return f"Error: The file '{file_path}' was not found. Make sure it has been downloaded first."
284
285 try:
286 # 2. Determina il tipo di file e prepara la lista dei CSV da processare
287 file_extension = Path(file_path).suffix.lower()
288 csv_files_to_process = []
289
290 # --- CASO 1: Il file è un Excel ---
291 if file_extension in ['.xlsx', '.xls']:
292 print(f"Rilevato file Excel. Inizio la conversione dei fogli in CSV temporanei...")
293
294 # Legge tutti i fogli in un dizionario di DataFrame
295 excel_sheets = pd.read_excel(file_path, sheet_name=None)
296
297 if not excel_sheets:
298 return f"Error: The Excel file '{file_path}' is empty or contains no worksheets."
299
300 # Ottiene il nome base del file per i file temporanei
301 base_name = Path(file_path).stem
302 uploads_dir = Path(file_path).parent
303
304 for sheet_name, df in excel_sheets.items():
305 # Crea un nome di file sicuro per il CSV temporaneo
306 safe_sheet_name = "".join(c for c in sheet_name if c.isalnum() or c in (' ', '_')).rstrip()
307 temp_csv_path = uploads_dir / f"{base_name}_sheet_{safe_sheet_name}.csv"
308
309 # Salva il DataFrame del foglio in un file CSV
310 df.to_csv(temp_csv_path, index=False)
311 print(f" - Foglio '{sheet_name}' convertito e salvato in: {temp_csv_path}")
312 csv_files_to_process.append(str(temp_csv_path))
313
314 # --- CASO 2: Il file è già un CSV ---
315 elif file_extension == '.csv':
316 print(f"Rilevato file CSV. Verrà processato direttamente.")
317 csv_files_to_process.append(file_path)
318
319 # --- CASO 3: Formato non supportato ---
320 else:
321 return f"Error: Unsupported file format '{file_extension}'. This tool supports only CSV, XLSX, and XLS."
322
323 # 3. Usa CSVLoader su tutti i file CSV identificati (originali o convertiti)
324 if not csv_files_to_process:
325 return "Error: No file to process was found."
326
327 all_docs = []
328 for csv_path in csv_files_to_process:
329 loader = CSVLoader(file_path=csv_path)
330 docs = loader.load()
331 all_docs.extend(docs)
332
333 # 4. Formatta l'output come richiesto
334 # Aumentiamo il limite di caratteri per dare più contesto all'LLM
335 formatted_output = "\n\n---\n\n".join(
336 [
337 f'<Document source="{Path(doc.metadata["source"]).name}" page="{doc.metadata.get("page", 0)}">\n{doc.page_content[:2500]}\n</Document>'
338 for doc in all_docs
339 ]
340 )
341
342 print("Analisi completata con successo.")
343 return formatted_output
344
345 except Exception as e:
346 error_message = f"An unexpected error occurred while analyzing the file '{file_path}': {e}"
347 print(error_message)
348 return error_message
349
350
351@tool("audio_tool")
352def audio_tool(filename: str) -> Union[str, str]:
353 """
354 Transcribes a local audio file into text using OpenAI's Whisper model.
355 Use this tool when you need to extract the textual content from an audio file.
356 Supports common formats such as MP3, MP4, MPEG, MPGA, M4A, WAV, and WEBM.
357
358 Args:
359 filename (str): The filename of the audio file to transcribe.
360
361 Returns:
362 Union[str, str]: The transcribed text if successful, or an error message string in case of failure.
363 """
364 print(f"--- ESECUZIONE DEL TOOL 'transcribe_audio' CON INPUT: file_path='{filename}' ---")
365 UPLOADS_DIR = "./uploads/"
366 file_path = os.path.join(UPLOADS_DIR, filename)
367 client = OpenAI()
368
369 # 1. Controlla se il client è stato inizializzato correttamente
370 if client is None:
371 return "Error: The OpenAI client is not configured. Please check your API key."
372
373 # 2. Controlla se il file esiste prima di tentare di aprirlo
374 if not os.path.exists(file_path):
375 return f"Error: The file '{file_path}' was not found. Make sure it has been downloaded first."
376
377 try:
378 # 3. Apri il file in modalità binaria e invialo all'API di OpenAI
379 with open(file_path, "rb") as audio_file:
380 transcription = client.audio.transcriptions.create(
381 model="whisper-1",
382 file=audio_file
383 )
384
385 print("Trascrizione completata con successo.")
386 # La risposta dell'API contiene il testo nel campo 'text'
387 return transcription.text
388
389 except APIError as e:
390 # Gestisce errori specifici dell'API di OpenAI (es. file non valido, auth error)
391 error_message = f"Error from the OpenAI API during transcription: {e}"
392 print(error_message)
393 return error_message
394
395 except Exception as e:
396 # Gestisce altri errori imprevisti (es. problemi di lettura del file)
397 error_message = f"An unexpected error occurred while transcribing the file '{file_path}': {e}"
398 print(error_message)
399 return error_message
400
401
402@tool("image_tool")
403def image_tool(filename: str, user_question: str) -> Union[str, str]:
404 """
405 Reads a local image file and encodes it in base64 format, ready to be analyzed by a multimodal model (such as GPT-4o).
406 Use this tool to prepare any image (JPG, PNG, WEBP, etc.) before asking questions about its content.
407
408 Args:
409 filename (str): The filename of the image file to prepare.
410 user_question (str): The user's original question to guide the analysis.
411
412 Returns:
413 Union[str, str]: A textual analysis based on the base64 encoded image data, or an error message string.
414 """
415 print(f"--- ESECUZIONE DEL TOOL 'prepare_image_for_analysis' CON INPUT: file_path='{filename}' ---")
416 UPLOADS_DIR = "./uploads/"
417 file_path = os.path.join(UPLOADS_DIR, filename)
418 client = OpenAI()
419
420 # 1. Controlla se il file esiste
421 if not os.path.exists(file_path):
422 return f"Error: The image file '{file_path}' was not found."
423
424 try:
425 # 2. Determina il tipo MIME dell'immagine (es. 'image/jpeg', 'image/png')
426 mime_type, _ = mimetypes.guess_type(file_path)
427 if not mime_type or not mime_type.startswith('image/'):
428 return f"Error: The file '{file_path}' is not a supported image format."
429
430 # 3. Leggi il file in modalità binaria e codificalo in base64
431 with open(file_path, "rb") as image_file:
432 encoded_string = base64.b64encode(image_file.read()).decode('utf-8')
433
434 # 4. Formatta l'output come un data URI, il formato standard per passare immagini
435 # a modelli multimodali.
436 image_data = f"data:{mime_type};base64,{encoded_string}"
437 # Crea il prompt per l'analisi
438 analysis_prompt = [
439 {
440 "role": "user",
441 "content": [
442 {
443 "type": "text",
444 "text": f"""
445 You are an expert visual analyst. Your task is to describe the provided image in extreme detail to help answer the user's question.
446 Focus on the elements relevant to the question. Be objective and precise.
447
448 **User's Question:** '{user_question}'
449
450 Analyze the image and provide a detailed description.
451 """
452 },
453 {
454 "type": "image_url",
455 "image_url": {
456 "url": image_data,
457 "detail": "high" # Usa alta risoluzione per la massima precisione
458 }
459 }
460 ]
461 }
462 ]
463
464
465 # 5. Restituisce la risposta in base all'immagine analizzata.
466 # Chiama l'API di OpenAI
467 response = client.chat.completions.create(
468 model="gpt-4o-mini", # o "gpt-4-vision-preview"
469 messages=analysis_prompt,
470 max_tokens=1000,
471 temperature=0
472 )
473
474 description = response.choices[0].message.content
475 print("--- Image analysis complete. ---")
476 return description
477 except Exception as e:
478 return f"An error occurred during the visual analysis: {e}"
479
480
481#-----------------------------------------------------------------------------
482# Code Execution Tools
483#-----------------------------------------------------------------------------
484@tool("code_writer_tool")
485def code_writer_tool(code: str, task_id: str) -> str:
486 """
487 Writes a string of Python code to a local file. This is the first step
488 for any task that requires writing and then executing code. The task_id is the name for the file.
489
490 Args:
491 code (str): A string containing the complete, valid Python code to be written to the file.
492 task_id (str): The name for the file.
493
494 Returns:
495 local_filename (str): The local filename of python file to execute.
496 """
497 print(f"--- TOOL: Writing code to file: {task_id}.py ---")
498 UPLOADS_DIR = "./uploads/"
499 local_filename = f"{task_id}.py"
500 file_path = os.path.join(UPLOADS_DIR, local_filename)
501
502 try:
503 # Scrive il codice nel file
504 with open(file_path, "w", encoding="utf-8") as f:
505 f.write(code)
506
507 success_message = f"Successfully wrote code to {file_path}."
508 print(success_message)
509
510 # Restituisce il percorso del file, che servirà al tool di esecuzione
511 return local_filename
512 except Exception as e:
513 error_message = f"An error occurred while writing the file: {e}"
514 print(error_message)
515 return error_message
516
517
518@tool("code_tool")
519def code_tool(filename: str, timeout_seconds: int = 100) -> Union[str, dict]:
520 """
521 Executes a programming code file in an isolated and secure environment, capturing its standard output and errors.
522 Use this tool to run programming code when you need to analyze its behavior or output.
523
524 Args:
525 filename (str): The filename of the code file to execute.
526 timeout_seconds (int, optional): The maximum number of seconds the execution is allowed to run before forcibly terminating the process. Default is 10.
527
528 Returns:
529 Union[str, dict]: A dictionary containing 'stdout', 'stderr', and 'return_code' if successful, or an error message string if the tool itself fails.
530 """
531 print(f"--- ESECUZIONE DEL TOOL 'execute_python_file' SU: {filename} ---")
532 UPLOADS_DIR = "./uploads/"
533 file_path = os.path.join(UPLOADS_DIR, filename)
534
535 # 1. Controlla di sicurezza: il file esiste?
536 if not os.path.exists(file_path):
537 return f"Error: The code file '{file_path}' was not found."
538
539 # 2. Usa subprocess.run per eseguire il codice in modo sicuro
540 try:
541 # 'subprocess.run' è il modo moderno e raccomandato per eseguire processi
542 process = subprocess.run(
543 ['python', file_path], # Il comando da eseguire (es. 'python nomefile.py')
544 capture_output=True, # Cattura stdout e stderr
545 text=True, # Decodifica stdout/stderr come testo (UTF-8)
546 timeout=timeout_seconds # Imposta un timeout
547 )
548
549 execution_result = {
550 "return_code": process.returncode,
551 "stdout": process.stdout.strip(),
552 "stderr": process.stderr.strip()
553 }
554
555 # 3. Ritorna un dizionario strutturato con i risultati
556 return execution_result
557
558 except FileNotFoundError:
559 # Questo errore si verifica se l'interprete 'python' non è nel PATH del sistema
560 return "Error: The 'python' interpreter was not found on the system. Unable to execute the code."
561 except subprocess.TimeoutExpired as e:
562 # Gestisce il caso in cui il codice va in timeout
563 return {
564 "return_code": -1, # Codice di ritorno personalizzato per timeout
565 "stdout": e.stdout.strip() if e.stdout else "",
566 "stderr": f"Error: Execution terminated after {timeout_seconds} seconds (Timeout)."
567 }
568 except Exception as e:
569 # Cattura qualsiasi altro errore imprevisto durante l'esecuzione del tool
570 return f"An unexpected error occurred while executing the tool: {e}"
571
572
573#-----------------------------------------------------------------------------
574# Youtube Video Tools
575#-----------------------------------------------------------------------------
576@tool("youtube_info_tool")
577def youtube_info_tool(youtube_url: str) -> Union[str, dict]:
578 """
579 Collects information and resources from a YouTube video. Downloads both audio and video, and retrieves the official transcript if available.
580 This is ALWAYS the first tool to call when working with a YouTube video.
581
582 Args:
583 youtube_url (str): The full URL of the YouTube video.
584
585 Returns:
586 Union[str, dict]: A dictionary with the collected resources (transcript, audio_filename, video_filename) or an error message.
587 """
588 print(f"--- ESECUZIONE DEL TOOL 'get_youtube_video_info' CON URL: {youtube_url} ---")
589 UPLOADS_DIR = "./uploads/"
590
591 try:
592 yt = YouTube(youtube_url)
593 video_id = yt.video_id
594
595 # 1. Recupera la trascrizione ufficiale
596 transcript_text = None
597 try:
598 transcript_list = YouTubeTranscriptApi.get_transcript(video_id)
599 transcript_text = " ".join([d['text'] for d in transcript_list])
600 print("Trascrizione ufficiale trovata.")
601 except Exception:
602 print("Nessuna trascrizione ufficiale disponibile.")
603
604 # 2. Scarica l'audio
605 audio_stream = yt.streams.get_audio_only()
606 audio_path = audio_stream.download(output_path=UPLOADS_DIR, filename=f"{video_id}.m4a")
607 if transcript_text is None:
608 transcript_text = audio_tool.invoke({"filename":f"{video_id}.m4a"})
609 print(f"Audio scaricato in: {audio_path}")
610
611 # 3. Scarica il video
612 video_stream = yt.streams.get_highest_resolution()
613 video_path = video_stream.download(output_path=UPLOADS_DIR, filename=f"{video_id}.mp4")
614 print(f"Video scaricato in: {video_path}")
615
616 video_info = {
617 "title": yt.title,
618 "description": yt.description,
619 "transcript": transcript_text,
620 "audio_filename": f"{video_id}.m4a",
621 "video_filename": f"{video_id}.mp4"
622 }
623
624 return video_info
625 except Exception as e:
626 return f"Error while retrieving information from the YouTube video: {e}"
627
628
629@tool("youtube_frame_tool")
630def youtube_frame_tool(filename: str, title: str, description: str, transcript: str, user_question: str, sample_rate_seconds: int = 5) -> Union[str, str]:
631 """
632 Analyzes video content by combining visual information from frames with the provided transcript to answer a specific user question.
633 To be used as a last resort, when the transcript and audio are not sufficient, or for purely visual questions.
634
635 Args:
636 filename (str): The filename of the video file.
637 title (str): The title of the video.
638 description (str): A brief description of the video.
639 transcript (str): The full text transcript of the video (either official or from audio).
640 user_question (str): The user's original question to guide the analysis.
641 sample_rate_seconds (int): Interval in seconds between frames to analyze. Default is 3.
642
643 Returns:
644 Union[str, str]: A textual analysis based on the video frames or an error message.
645 """
646 print(f"--- ESECUZIONE DEL TOOL 'analyze_video_frames' SU: {filename} ---")
647 UPLOADS_DIR = "./uploads/"
648 video_path = os.path.join(UPLOADS_DIR, filename)
649 client = OpenAI()
650
651 if not os.path.exists(video_path):
652 return f"Error: Video file not found at '{video_path}'."
653 if client is None:
654 return "Error: The OpenAI client is not configured."
655
656 video = cv2.VideoCapture(video_path)
657 fps = video.get(cv2.CAP_PROP_FPS)
658 frame_interval = int(fps * sample_rate_seconds)
659
660 base64_frames = []
661 frame_count = 0
662
663 while video.isOpened():
664 success, frame = video.read()
665 if not success:
666 break
667
668 if frame_count % frame_interval == 0:
669 _, buffer = cv2.imencode(".jpg", frame)
670 base64_frames.append(base64.b64encode(buffer).decode("utf-8"))
671
672 frame_count += 1
673
674 video.release()
675 print(f"Campionati {len(base64_frames)} frame dal video.")
676
677 if not base64_frames:
678 return "Error: Unable to extract frames from the video."
679
680 prompt_messages = [
681 {
682 "role": "user",
683 "content": [
684 {
685 "type": "text",
686 "text": f"""
687 You are a video content analyst.
688 Your task is to answer the user's question by combining information from different sources:
689 - the title
690 - the description
691 - the transcript
692 - a series of sampled frames
693 **IMPORTANT**: Analyze the all sources of the video in great detail, because there may be important information to solve the task.
694
695 **User Question**: {user_question}
696 **Video Title**: {title}
697 **Video Description**: {description}
698 **Video Transcript**: {transcript if transcript else "No transcript available."}
699
700 Begin your rigorous analysis now. Here are the frames:
701 """
702 },
703 # Inserimento dei Frame
704 *map(lambda x: {"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{x}", "detail": "low"}}, base64_frames),
705 ],
706 }
707 ]
708
709 try:
710 response = client.chat.completions.create(
711 model="gpt-4o-mini",
712 temperature=0,
713 messages=prompt_messages,
714 max_tokens=1000,
715 )
716 analysis_summary = response.choices[0].message.content
717 return analysis_summary
718 except Exception as e:
719 return f"Error while analyzing frames with the OpenAI API: {e}"
720
721
722#-----------------------------------------------------------------------------
723# Web Search Tools
724#-----------------------------------------------------------------------------
725@tool("web_search_tool")
726def web_search_tool(task: str) -> str:
727 """
728 Delegates complex research tasks to a specialized, cyclic research agent.
729 Use this for any question that requires external, up-to-date, or detailed knowledge.
730 """
731 print(f"--- MAIN AGENT: DELEGATING RESEARCH FOR: '{task}' ---")
732
733 # Lo stato iniziale ora contiene il task e un primo messaggio umano vuoto per avviare il ciclo.
734 # L'agente di ricerca leggerà il task dallo stato e ignorerà questo messaggio.
735 initial_state = {"task": task, "context_summary": ""}
736
737 # Esegui il sub-grafo
738 final_state = web_search_graph.invoke(initial_state)
739
740 # Il risultato finale è l'ultimo messaggio nella cronologia, che sarà la risposta del writer.
741 final_answer = final_state["messages"][-1].content
742 return final_answer
743
744
745assistant_tools_list = [
746 sort_tool, download_tool, add_tool, multiply_tool, subtract_tool, divide_tool, modulus_tool, tabular_tool, audio_tool, image_tool, code_writer_tool, code_tool, youtube_info_tool, youtube_frame_tool, web_search_tool
747]
748