first submission

.gitignore

@@ -0,0 +1,120 @@
+# Byte-compiled / optimized / DLL files
+__pycache__/
+*.py[cod]
+*$py.class
+
+# C extensions
+*.so
+
+# Distribution / packaging
+.Python
+build/
+develop-eggs/
+dist/
+downloads/
+eggs/
+.eggs/
+lib/
+lib64/
+parts/
+sdist/
+var/
+wheels/
+pip-wheel-metadata/
+share/python-wheels/
+*.egg-info/
+.installed.cfg
+*.egg
+MANIFEST
+
+# PyInstaller
+#  Usually these files are written by a python script from a template
+#  before PyInstaller builds the exe, so as to inject date/other infos into it.
+*.manifest
+*.spec
+
+# Installer logs
+pip-log.txt
+pip-delete-this-directory.txt
+
+# Unit test / coverage reports
+htmlcov/
+.tox/
+.nox/
+.coverage
+.coverage.*
+.cache
+nosetests.xml
+coverage.xml
+*.cover
+*.py,cover
+.hypothesis/
+.pytest_cache/
+
+# Translations
+*.mo
+*.pot
+
+# Django stuff:
+*.log
+local_settings.py
+db.sqlite3
+db.sqlite3-journal
+
+# Flask stuff:
+instance/
+.webassets-cache
+
+# Scrapy stuff:
+.scrapy
+
+# Sphinx documentation
+docs/_build/
+
+# PyBuilder
+target/
+
+# Jupyter Notebook
+.ipynb_checkpoints
+
+# IPython
+profile_default/
+ipython_config.py
+
+# pyenv
+.python-version
+
+# PEP 582; __pypackages__
+__pypackages__/
+
+# Celery stuff
+celerybeat-schedule
+celerybeat.pid
+
+# SageMath parsed files
+*.sage.py
+
+# Environments
+.env
+.venv
+env/
+venv/
+ENV/
+env.bak/
+venv.bak/
+
+# IDE / Editor specific files
+.idea/
+.vscode/
+*.project
+*.pydevproject
+.project
+.settings/
+*.sublime-workspace
+
+# dotenv
+.env
+
+# OS specific files
+.DS_Store
+Thumbs.db

app.py

@@ -1,196 +1,110 @@
 import os
-import gradio as gr
-import requests
-import inspect
-import pandas as pd
-
-# (Keep Constants as is)
-# --- Constants ---
-DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space"
-
-# --- Basic Agent Definition ---
-# ----- THIS IS WERE YOU CAN BUILD WHAT YOU WANT ------
-class BasicAgent:
-    def __init__(self):
-        print("BasicAgent initialized.")
-    def __call__(self, question: str) -> str:
-        print(f"Agent received question (first 50 chars): {question[:50]}...")
-        fixed_answer = "This is a default answer."
-        print(f"Agent returning fixed answer: {fixed_answer}")
-        return fixed_answer
-
-def run_and_submit_all( profile: gr.OAuthProfile | None):
-    """
-    Fetches all questions, runs the BasicAgent on them, submits all answers,
-    and displays the results.
-    """
-    # --- Determine HF Space Runtime URL and Repo URL ---
-    space_id = os.getenv("SPACE_ID") # Get the SPACE_ID for sending link to the code
-
-    if profile:
-        username= f"{profile.username}"
-        print(f"User logged in: {username}")
-    else:
-        print("User not logged in.")
-        return "Please Login to Hugging Face with the button.", None
-
-    api_url = DEFAULT_API_URL
-    questions_url = f"{api_url}/questions"
-    submit_url = f"{api_url}/submit"
-
-    # 1. Instantiate Agent ( modify this part to create your agent)
-    try:
-        agent = BasicAgent()
-    except Exception as e:
-        print(f"Error instantiating agent: {e}")
-        return f"Error initializing agent: {e}", None
-    # In the case of an app running as a hugging Face space, this link points toward your codebase ( usefull for others so please keep it public)
-    agent_code = f"https://huggingface.co/spaces/{space_id}/tree/main"
-    print(agent_code)
-
-    # 2. Fetch Questions
-    print(f"Fetching questions from: {questions_url}")
-    try:
-        response = requests.get(questions_url, timeout=15)
-        response.raise_for_status()
-        questions_data = response.json()
-        if not questions_data:
-             print("Fetched questions list is empty.")
-             return "Fetched questions list is empty or invalid format.", None
-        print(f"Fetched {len(questions_data)} questions.")
-    except requests.exceptions.RequestException as e:
-        print(f"Error fetching questions: {e}")
-        return f"Error fetching questions: {e}", None
-    except requests.exceptions.JSONDecodeError as e:
-         print(f"Error decoding JSON response from questions endpoint: {e}")
-         print(f"Response text: {response.text[:500]}")
-         return f"Error decoding server response for questions: {e}", None
-    except Exception as e:
-        print(f"An unexpected error occurred fetching questions: {e}")
-        return f"An unexpected error occurred fetching questions: {e}", None
-
-    # 3. Run your Agent
-    results_log = []
-    answers_payload = []
-    print(f"Running agent on {len(questions_data)} questions...")
-    for item in questions_data:
-        task_id = item.get("task_id")
-        question_text = item.get("question")
-        if not task_id or question_text is None:
-            print(f"Skipping item with missing task_id or question: {item}")
-            continue
-        try:
-            submitted_answer = agent(question_text)
-            answers_payload.append({"task_id": task_id, "submitted_answer": submitted_answer})
-            results_log.append({"Task ID": task_id, "Question": question_text, "Submitted Answer": submitted_answer})
-        except Exception as e:
-             print(f"Error running agent on task {task_id}: {e}")
-             results_log.append({"Task ID": task_id, "Question": question_text, "Submitted Answer": f"AGENT ERROR: {e}"})
-
-    if not answers_payload:
-        print("Agent did not produce any answers to submit.")
-        return "Agent did not produce any answers to submit.", pd.DataFrame(results_log)
-
-    # 4. Prepare Submission 
-    submission_data = {"username": username.strip(), "agent_code": agent_code, "answers": answers_payload}
-    status_update = f"Agent finished. Submitting {len(answers_payload)} answers for user '{username}'..."
-    print(status_update)
-
-    # 5. Submit
-    print(f"Submitting {len(answers_payload)} answers to: {submit_url}")
-    try:
-        response = requests.post(submit_url, json=submission_data, timeout=60)
-        response.raise_for_status()
-        result_data = response.json()
-        final_status = (
-            f"Submission Successful!\n"
-            f"User: {result_data.get('username')}\n"
-            f"Overall Score: {result_data.get('score', 'N/A')}% "
-            f"({result_data.get('correct_count', '?')}/{result_data.get('total_attempted', '?')} correct)\n"
-            f"Message: {result_data.get('message', 'No message received.')}"
-        )
-        print("Submission successful.")
-        results_df = pd.DataFrame(results_log)
-        return final_status, results_df
-    except requests.exceptions.HTTPError as e:
-        error_detail = f"Server responded with status {e.response.status_code}."
-        try:
-            error_json = e.response.json()
-            error_detail += f" Detail: {error_json.get('detail', e.response.text)}"
-        except requests.exceptions.JSONDecodeError:
-            error_detail += f" Response: {e.response.text[:500]}"
-        status_message = f"Submission Failed: {error_detail}"
-        print(status_message)
-        results_df = pd.DataFrame(results_log)
-        return status_message, results_df
-    except requests.exceptions.Timeout:
-        status_message = "Submission Failed: The request timed out."
-        print(status_message)
-        results_df = pd.DataFrame(results_log)
-        return status_message, results_df
-    except requests.exceptions.RequestException as e:
-        status_message = f"Submission Failed: Network error - {e}"
-        print(status_message)
-        results_df = pd.DataFrame(results_log)
-        return status_message, results_df
-    except Exception as e:
-        status_message = f"An unexpected error occurred during submission: {e}"
-        print(status_message)
-        results_df = pd.DataFrame(results_log)
-        return status_message, results_df
-
-
-# --- Build Gradio Interface using Blocks ---
-with gr.Blocks() as demo:
-    gr.Markdown("# Basic Agent Evaluation Runner")
-    gr.Markdown(
-        """
-        **Instructions:**
-
-        1.  Please clone this space, then modify the code to define your agent's logic, the tools, the necessary packages, etc ...
-        2.  Log in to your Hugging Face account using the button below. This uses your HF username for submission.
-        3.  Click 'Run Evaluation & Submit All Answers' to fetch questions, run your agent, submit answers, and see the score.
-
-        ---
-        **Disclaimers:**
-        Once clicking on the "submit button, it can take quite some time ( this is the time for the agent to go through all the questions).
-        This space provides a basic setup and is intentionally sub-optimal to encourage you to develop your own, more robust solution. For instance for the delay process of the submit button, a solution could be to cache the answers and submit in a seperate action or even to answer the questions in async.
-        """
-    )
-
-    gr.LoginButton()
-
-    run_button = gr.Button("Run Evaluation & Submit All Answers")
-
-    status_output = gr.Textbox(label="Run Status / Submission Result", lines=5, interactive=False)
-    # Removed max_rows=10 from DataFrame constructor
-    results_table = gr.DataFrame(label="Questions and Agent Answers", wrap=True)
-
-    run_button.click(
-        fn=run_and_submit_all,
-        outputs=[status_output, results_table]
-    )
-
-if __name__ == "__main__":
-    print("\n" + "-"*30 + " App Starting " + "-"*30)
-    # Check for SPACE_HOST and SPACE_ID at startup for information
-    space_host_startup = os.getenv("SPACE_HOST")
-    space_id_startup = os.getenv("SPACE_ID") # Get SPACE_ID at startup
-
-    if space_host_startup:
-        print(f"✅ SPACE_HOST found: {space_host_startup}")
-        print(f"   Runtime URL should be: https://{space_host_startup}.hf.space")
-    else:
-        print("ℹ️  SPACE_HOST environment variable not found (running locally?).")
-
-    if space_id_startup: # Print repo URLs if SPACE_ID is found
-        print(f"✅ SPACE_ID found: {space_id_startup}")
-        print(f"   Repo URL: https://huggingface.co/spaces/{space_id_startup}")
-        print(f"   Repo Tree URL: https://huggingface.co/spaces/{space_id_startup}/tree/main")
-    else:
-        print("ℹ️  SPACE_ID environment variable not found (running locally?). Repo URL cannot be determined.")
-
-    print("-"*(60 + len(" App Starting ")) + "\n")
-
-    print("Launching Gradio Interface for Basic Agent Evaluation...")
-    demo.launch(debug=True, share=False)
+# Import the load_dotenv function from the dotenv library
+from dotenv import load_dotenv
+from langchain_google_genai import ChatGoogleGenerativeAI
+
+from multimodal_tools import extract_text_tool, analyze_image_tool, analyze_audio_tool
+
+# Load environment variables from .env file
+load_dotenv()
+# Read your API key from the environment variable or set it manually
+api_key = os.getenv("GEMINI_API_KEY")
+langfuse_secret_key = os.getenv("LANGFUSE_SECRET_KEY")
+langfuse_public_key = os.getenv("LANGFUSE_PUBLIC_KEY")
+
+from typing import TypedDict, Annotated
+from langgraph.graph.message import add_messages
+from langchain_core.messages import AnyMessage, HumanMessage, AIMessage
+from langgraph.prebuilt import ToolNode
+from langgraph.graph import START, StateGraph
+from langgraph.prebuilt import tools_condition
+from langchain_community.tools.tavily_search import TavilySearchResults # Importa Tavily
+from langchain_community.tools import DuckDuckGoSearchRun
+# from langfuse import Langfuse # Langfuse is initialized by CallbackHandler directly
+from langfuse.callback import CallbackHandler
+from youtube_tools import youtube_transcript_tool
+from math_tools import add_tool, subtract_tool, multiply_tool, divide_tool
+from serpapi_tools import serpapi_search_tool
+from IPython.display import Image, display
+# Generate thfrom langchain_community.tools.tavily_search import TavilySearchResults
+from langchain_community.tools.tavily_search import TavilySearchResults
+
+
+# Initialize Langfuse CallbackHandler for LangGraph/Langchain (tracing)
+langfuse_handler = CallbackHandler(
+    public_key=langfuse_public_key,
+    secret_key=langfuse_secret_key,
+    host="http://localhost:3000"
+)
+
+# Create LLM class
+chat = ChatGoogleGenerativeAI(
+    model= "gemini-2.5-pro-preview-05-06",
+    temperature=0,
+    max_retries=2,
+    google_api_key=api_key,
+    thinking_budget= 0
+)
+
+search_tool = TavilySearchResults(
+    name="tavily_web_search", # Puoi personalizzare il nome se vuoi
+    description="Esegue una ricerca web avanzata utilizzando Tavily per informazioni aggiornate e complete. Utile per domande complesse o che richiedono dati recenti. Può essere utile fare più ricerche modificando la query per ottenere risultati migliori.", # Descrizione per l'LLM
+    max_results=5
+)
+
+tools = [
+    extract_text_tool,
+    analyze_image_tool,
+    analyze_audio_tool,
+    youtube_transcript_tool,
+    add_tool,
+    subtract_tool,
+    multiply_tool,
+    divide_tool,
+    search_tool
+]
+chat_with_tools = chat.bind_tools(tools)
+
+
+class AgentState(TypedDict):
+    messages: Annotated[list[AnyMessage], add_messages]
+
+def assistant(state: AgentState):
+    sys_msg = "You are a helpful assistant with access to tools. Understand user requests accurately. Use your tools when needed to answer effectively. Strictly follow all user instructions and constraints." \
+    "Pay attention: your output needs to contain only the final answer without any reasoning since it will be strictly evaluated against a dataset which contains only the specific response." \
+    "Your final output needs to be just the string or integer containing the answer, not an array or technical stuff."
+    return {
+        "messages": [chat_with_tools.invoke([sys_msg] + state["messages"])]
+    }
+
+    
+## The graph
+builder = StateGraph(AgentState)
+
+# Define nodes: these do the work
+builder.add_node("assistant", assistant)
+builder.add_node("tools", ToolNode(tools))
+
+# Define edges: these determine how the control flow moves
+builder.add_edge(START, "assistant")
+builder.add_conditional_edges(
+    "assistant",
+    # If the latest message requires a tool, route to tools
+    # Otherwise, provide a direct response
+    tools_condition,
+)
+builder.add_edge("tools", "assistant")
+alfred = builder.compile()
+
+""" # Salva l'immagine del grafo su un file
+graph_image_bytes = alfred.get_graph(xray=True).draw_mermaid_png()
+with open("alfred_graph.png", "wb") as f:
+    f.write(graph_image_bytes)
+print("L'immagine del grafo è stata salvata come alfred_graph.png")
+
+messages = [HumanMessage(content="Who did the actor who played Ray in the Polish-language version of Everybody Loves Raymond play in Magda M.? Give only the first name.")]
+response = alfred.invoke(input={"messages": messages}, config={"callbacks": [langfuse_handler]})
+
+print("🎩 Alfred's Response:")
+print(response['messages'][-1].content)
+ """

app_for_submission.py

@@ -0,0 +1,227 @@
+import os
+import gradio as gr
+import requests
+import inspect
+import pandas as pd
+from app import alfred
+from langfuse.callback import CallbackHandler
+from typing import Optional
+from langchain_core.messages import AnyMessage, HumanMessage, AIMessage
+# (Keep Constants as is)
+# --- Constants ---
+DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space"
+
+
+langfuse_secret_key = os.getenv("LANGFUSE_SECRET_KEY")
+langfuse_public_key = os.getenv("LANGFUSE_PUBLIC_KEY")
+
+# Initialize Langfuse CallbackHandler for LangGraph/Langchain (tracing)
+langfuse_handler = CallbackHandler(
+    public_key=langfuse_public_key,
+    secret_key=langfuse_secret_key,
+    host="http://localhost:3000"
+)
+
+# --- Basic Agent Definition ---
+# ----- THIS IS WERE YOU CAN BUILD WHAT YOU WANT ------
+""" class BasicAgent:
+    def __init__(self):
+        print("BasicAgent initialized.")
+    def __call__(self, question: str, file_name: str | None = None) -> str:
+        print(f"Agent received question (first 50 chars): {question[:50]}...")
+        if file_name:
+            print(f"Agent received file_name: {file_name}")
+        # Qui puoi aggiungere la logica per utilizzare file_name se fornito.
+        # Per ora, lo aggiungiamo alla risposta di default per dimostrazione.
+        fixed_answer = "This is a default answer."
+        if file_name:
+            fixed_answer += f" (File to use: {file_name})"
+        print(f"Agent returning fixed answer: {fixed_answer}")
+        return fixed_answer """
+
+def run_and_submit_all( profile: Optional[gr.OAuthProfile]):
+    """
+    Fetches all questions, runs the BasicAgent on them, submits all answers,
+    and displays the results.
+    """
+    # --- Determine HF Space Runtime URL and Repo URL ---
+    space_id = os.getenv("SPACE_ID") # Get the SPACE_ID for sending link to the code
+
+    if profile:
+        username= f"{profile.username}"
+        print(f"User logged in: {username}")
+    else:
+        print("User not logged in.")
+        return "Please Login to Hugging Face with the button.", None
+
+    api_url = DEFAULT_API_URL
+    questions_url = f"{api_url}/questions"
+    submit_url = f"{api_url}/submit"
+
+    # 1. Instantiate Agent ( modify this part to create your agent)
+    try:
+        agent = alfred
+    except Exception as e:
+        print(f"Error instantiating agent: {e}")
+        return f"Error initializing agent: {e}", None
+    # In the case of an app running as a hugging Face space, this link points toward your codebase ( usefull for others so please keep it public)
+    agent_code = f"https://huggingface.co/spaces/{space_id}/tree/main"
+    print(agent_code)
+
+    # 2. Fetch Questions
+    print(f"Fetching questions from: {questions_url}")
+    try:
+        response = requests.get(questions_url, timeout=15)
+        response.raise_for_status()
+        questions_data = response.json()
+        if not questions_data:
+             print("Fetched questions list is empty.")
+             return "Fetched questions list is empty or invalid format.", None
+        print(f"Fetched {len(questions_data)} questions.")
+    except requests.exceptions.RequestException as e:
+        print(f"Error fetching questions: {e}")
+        return f"Error fetching questions: {e}", None
+    except requests.exceptions.JSONDecodeError as e:
+         print(f"Error decoding JSON response from questions endpoint: {e}")
+         print(f"Response text: {response.text[:500]}")
+         return f"Error decoding server response for questions: {e}", None
+    except Exception as e:
+        print(f"An unexpected error occurred fetching questions: {e}")
+        return f"An unexpected error occurred fetching questions: {e}", None
+
+    # 3. Run your Agent
+    results_log = []
+    answers_payload = []
+    print(f"Running agent on {len(questions_data)} questions...")
+    for item in questions_data:
+        task_id = item.get("task_id")
+        question_text = item.get("question")
+        file_name = item.get("file_name")  # Estrai file_name
+
+        if not task_id or question_text is None:
+            print(f"Skipping item with missing task_id or question: {item}")
+            continue
+        try:
+            if file_name and isinstance(file_name, str) and file_name.strip():
+                messages = HumanMessage(content=question_text + " Path: files/" + file_name)
+            else:
+                messages = HumanMessage(content=question_text)
+            submitted_answer = alfred.invoke(input={"messages": messages}, config={"callbacks": [langfuse_handler]})
+            answers_payload.append({
+                "task_id": task_id,
+                "submitted_answer": submitted_answer['messages'][-1].content[-1] 
+                    if isinstance(submitted_answer['messages'][-1].content, list) 
+                    else submitted_answer['messages'][-1].content
+            })
+            results_log.append({"Task ID": task_id, "Question": question_text, "File Name": file_name if file_name and file_name.strip() else "N/A", "Submitted Answer": submitted_answer['messages'][-1].content})
+        except Exception as e:
+             print(f"Error running agent on task {task_id}: {e}")
+             results_log.append({"Task ID": task_id, "Question": question_text, "Submitted Answer": f"AGENT ERROR: {e}"})
+
+    if not answers_payload:
+        print("Agent did not produce any answers to submit.")
+        return "Agent did not produce any answers to submit.", pd.DataFrame(results_log)
+
+    # 4. Prepare Submission 
+    submission_data = {"username": username.strip(), "agent_code": agent_code, "answers": answers_payload}
+    status_update = f"Agent finished. Submitting {len(answers_payload)} answers for user '{username}'..."
+    print(status_update)
+
+    # 5. Submit
+    print(f"Submitting {len(answers_payload)} answers to: {submit_url}")
+    try:
+        response = requests.post(submit_url, json=submission_data, timeout=60)
+        response.raise_for_status()
+        result_data = response.json()
+        final_status = (
+            f"Submission Successful!\n"
+            f"User: {result_data.get('username')}\n"
+            f"Overall Score: {result_data.get('score', 'N/A')}% "
+            f"({result_data.get('correct_count', '?')}/{result_data.get('total_attempted', '?')} correct)\n"
+            f"Message: {result_data.get('message', 'No message received.')}"
+        )
+        print("Submission successful.")
+        results_df = pd.DataFrame(results_log)
+        return final_status, results_df
+    except requests.exceptions.HTTPError as e:
+        error_detail = f"Server responded with status {e.response.status_code}."
+        try:
+            error_json = e.response.json()
+            error_detail += f" Detail: {error_json.get('detail', e.response.text)}"
+        except requests.exceptions.JSONDecodeError:
+            error_detail += f" Response: {e.response.text[:500]}"
+        status_message = f"Submission Failed: {error_detail}"
+        print(status_message)
+        results_df = pd.DataFrame(results_log)
+        return status_message, results_df
+    except requests.exceptions.Timeout:
+        status_message = "Submission Failed: The request timed out."
+        print(status_message)
+        results_df = pd.DataFrame(results_log)
+        return status_message, results_df
+    except requests.exceptions.RequestException as e:
+        status_message = f"Submission Failed: Network error - {e}"
+        print(status_message)
+        results_df = pd.DataFrame(results_log)
+        return status_message, results_df
+    except Exception as e:
+        status_message = f"An unexpected error occurred during submission: {e}"
+        print(status_message)
+        results_df = pd.DataFrame(results_log)
+        return status_message, results_df
+
+
+# --- Build Gradio Interface using Blocks ---
+with gr.Blocks() as demo:
+    gr.Markdown("# Basic Agent Evaluation Runner")
+    gr.Markdown(
+        """
+        **Instructions:**
+
+        1.  Please clone this space, then modify the code to define your agent's logic, the tools, the necessary packages, etc ...
+        2.  Log in to your Hugging Face account using the button below. This uses your HF username for submission.
+        3.  Click 'Run Evaluation & Submit All Answers' to fetch questions, run your agent, submit answers, and see the score.
+
+        ---
+        **Disclaimers:**
+        Once clicking on the "submit button, it can take quite some time ( this is the time for the agent to go through all the questions).
+        This space provides a basic setup and is intentionally sub-optimal to encourage you to develop your own, more robust solution. For instance for the delay process of the submit button, a solution could be to cache the answers and submit in a seperate action or even to answer the questions in async.
+        """
+    )
+
+    gr.LoginButton()
+
+    run_button = gr.Button("Run Evaluation & Submit All Answers")
+
+    status_output = gr.Textbox(label="Run Status / Submission Result", lines=5, interactive=False)
+    # Removed max_rows=10 from DataFrame constructor
+    results_table = gr.DataFrame(label="Questions and Agent Answers", wrap=True)
+
+    run_button.click(
+        fn=run_and_submit_all,
+        outputs=[status_output, results_table]
+    )
+
+if __name__ == "__main__":
+    print("\n" + "-"*30 + " App Starting " + "-"*30)
+    # Check for SPACE_HOST and SPACE_ID at startup for information
+    space_host_startup = os.getenv("SPACE_HOST")
+    space_id_startup = os.getenv("SPACE_ID") # Get SPACE_ID at startup
+
+    if space_host_startup:
+        print(f"✅ SPACE_HOST found: {space_host_startup}")
+        print(f"   Runtime URL should be: https://{space_host_startup}.hf.space")
+    else:
+        print("ℹ️  SPACE_HOST environment variable not found (running locally?).")
+
+    if space_id_startup: # Print repo URLs if SPACE_ID is found
+        print(f"✅ SPACE_ID found: {space_id_startup}")
+        print(f"   Repo URL: https://huggingface.co/spaces/{space_id_startup}")
+        print(f"   Repo Tree URL: https://huggingface.co/spaces/{space_id_startup}/tree/main")
+    else:
+        print("ℹ️  SPACE_ID environment variable not found (running locally?). Repo URL cannot be determined.")
+
+    print("-"*(60 + len(" App Starting ")) + "\n")
+
+    print("Launching Gradio Interface for Basic Agent Evaluation...")
+    demo.launch(debug=True, share=False)

math_tools.py

@@ -0,0 +1,44 @@
+from langchain.tools import Tool
+import operator
+
+def add(a: float, b: float) -> float:
+    """Adds two numbers."""
+    return operator.add(a, b)
+
+def subtract(a: float, b: float) -> float:
+    """Subtracts the second number from the first."""
+    return operator.sub(a, b)
+
+def multiply(a: float, b: float) -> float:
+    """Multiplies two numbers."""
+    return operator.mul(a, b)
+
+def divide(a: float, b: float) -> float:
+    """Divides the first number by the second. Returns an error message if division by zero."""
+    if b == 0:
+        return "Error: Cannot divide by zero."
+    return operator.truediv(a, b)
+
+add_tool = Tool(
+    name="calculator_add",
+    func=add,
+    description="Adds two numbers. Input should be two numbers (a, b)."
+)
+
+subtract_tool = Tool(
+    name="calculator_subtract",
+    func=subtract,
+    description="Subtracts the second number from the first. Input should be two numbers (a, b)."
+)
+
+multiply_tool = Tool(
+    name="calculator_multiply",
+    func=multiply,
+    description="Multiplies two numbers. Input should be two numbers (a, b)."
+)
+
+divide_tool = Tool(
+    name="calculator_divide",
+    func=divide,
+    description="Divides the first number by the second. Input should be two numbers (a, b)."
+)

multimodal_tools.py

@@ -0,0 +1,174 @@
+import base64
+import os
+from langchain_core.messages import AnyMessage, SystemMessage, HumanMessage
+from langchain_google_genai import ChatGoogleGenerativeAI
+from langchain.tools import Tool
+from langchain_core.tools import tool
+
+api_key = os.getenv("GEMINI_API_KEY")
+
+# Create LLM class
+vision_llm = ChatGoogleGenerativeAI(
+    model= "gemini-2.5-flash-preview-05-20",
+    temperature=0,
+    max_retries=2,
+    google_api_key=api_key
+)
+
+def extract_text(img_path: str) -> str:
+    """
+    Extract text from an image file using a multimodal model.
+    Input needs to be the path of the image.
+    """
+    all_text = ""
+    try:
+        # Read image and encode as base64
+        with open(img_path, "rb") as image_file:
+            image_bytes = image_file.read()
+
+        image_base64 = base64.b64encode(image_bytes).decode("utf-8")
+
+        # Prepare the prompt including the base64 image data
+        message = [
+            HumanMessage(
+                content=[
+                    {
+                        "type": "text",
+                        "text": (
+                            "Extract all the text from this image. "
+                            "Return only the extracted text, no explanations."
+                        ),
+                    },
+                    {
+                        "type": "image_url",
+                        "image_url": {
+                            "url": f"data:image/png;base64,{image_base64}"
+                        },
+                    },
+                ]
+            )
+        ]
+
+        # Call the vision-capable model
+        response = vision_llm.invoke(message)
+
+        # Append extracted text
+        all_text += response.content + "\n\n"
+
+        return all_text.strip()
+    except Exception as e:
+        # A butler should handle errors gracefully
+        error_msg = f"Error extracting text: {str(e)}"
+        print(error_msg)
+        return ""
+    
+@tool("analyze_image_tool", parse_docstring=True)
+def analyze_image_tool(user_query: str, img_path: str) -> str:
+    """
+    Answer the question reasoning on the image.
+    
+    Args:
+        user_query (str): The question to be answered.
+        img_path (str): Path to the image file.
+    """
+    all_text = ""
+    try:
+        # Read image and encode as base64
+        with open(img_path, "rb") as image_file:
+            image_bytes = image_file.read()
+
+        image_base64 = base64.b64encode(image_bytes).decode("utf-8")
+
+        # Prepare the prompt including the base64 image data
+        message = [
+            HumanMessage(
+                content=[
+                    {
+                        "type": "text",
+                        "text": (
+                            f"User query: {user_query}"
+                        ),
+                    },
+                    {
+                        "type": "image_url",
+                        "image_url": {
+                            "url": f"data:image/png;base64,{image_base64}"
+                        },
+                    },
+                ]
+            )
+        ]
+
+        # Call the vision-capable model
+        response = vision_llm.invoke(message)
+
+        # Append extracted text
+        all_text += response.content + "\n\n"
+
+        return all_text.strip()
+    except Exception as e:
+        # A butler should handle errors gracefully
+        error_msg = f"Error analyzing image: {str(e)}"
+        print(error_msg)
+        return ""
+    
+@tool("analyze_audio_tool", parse_docstring=True)
+def analyze_audio_tool(user_query: str, audio_path: str) -> str:
+    """
+    Answer the question by reasoning on the provided audio file.
+    
+    Args:
+        user_query (str): The question to be answered.
+        audio_path (str): Path to the audio file (e.g., .mp3, .wav, .flac, .aac, .ogg).
+    """
+    try:
+        # Determine MIME type from file extension
+        _filename, file_extension = os.path.splitext(audio_path)
+        file_extension = file_extension.lower()
+
+        supported_formats = {
+            ".mp3": "audio/mp3", ".wav": "audio/wav", ".flac": "audio/flac",
+            ".aac": "audio/aac", ".ogg": "audio/ogg"
+        }
+
+        if file_extension not in supported_formats:
+            return (f"Error: Unsupported audio file format '{file_extension}'. "
+                    f"Supported extensions: {', '.join(supported_formats.keys())}.")
+        mime_type = supported_formats[file_extension]
+
+        # Read audio file and encode as base64
+        with open(audio_path, "rb") as audio_file:
+            audio_bytes = audio_file.read()
+        audio_base64 = base64.b64encode(audio_bytes).decode("utf-8")
+
+        # Prepare the prompt including the base64 audio data
+        message = [ 
+        HumanMessage(
+            content=[
+                {
+                    "type": "text",
+                    "text": f"User query: {user_query}",
+                },
+                {
+                    "type": "audio",
+                    "source_type": "base64",
+                    "mime_type": mime_type,
+                    "data": audio_base64
+                },
+            ]
+        )
+        ]
+
+        # Call the vision-capable model
+        response = vision_llm.invoke(message)
+        return response.content.strip()
+    except Exception as e:
+        error_msg = f"Error analyzing audio: {str(e)}"
+        print(error_msg)
+        return ""
+    
+extract_text_tool = Tool(
+    name="extract_text_tool",
+    func=extract_text,
+    description="Extract text from an image file using a multimodal model."
+)

serpapi_tools.py

@@ -0,0 +1,53 @@
+import os
+from langchain.tools import Tool
+from serpapi import GoogleSearch
+from dotenv import load_dotenv
+
+# Carica le variabili d'ambiente se hai la chiave API in un file .env
+load_dotenv()
+
+SERPAPI_API_KEY = os.getenv("SERPAPI_API_KEY")
+
+def _serpapi_search(query: str, num_results: int = 5, gl: str = "it", hl: str = "it") -> str:
+    """
+    Esegue una ricerca sul web utilizzando SerpAPI con Google Search e restituisce i risultati formattati.
+    Questo tool ha un costo elevato, pertanto sono da preferire altri tool se disponibili.
+    Richiamare questo tool soltanto in caso gli altri tool non siano stati soddisfacenti.
+
+    Args:
+        query: La query di ricerca.
+        num_results: Il numero di risultati da restituire.
+        gl: Codice del paese per la geolocalizzazione dei risultati (es. "it" per Italia).
+        hl: Codice della lingua per i risultati della ricerca (es. "it" per Italiano).
+
+    Returns:
+        Una stringa formattata con i risultati della ricerca o un messaggio di errore.
+    """
+    if not SERPAPI_API_KEY:
+        return "Errore: La variabile d'ambiente SERPAPI_API_KEY non è impostata."
+
+    params = {
+        "engine": "google",
+        "q": query,
+        "api_key": SERPAPI_API_KEY,
+        "num": num_results,
+        "gl": gl,
+        "hl": hl
+    }
+    search = GoogleSearch(params)
+    results = search.get_dict()
+    organic_results = results.get("organic_results", [])
+
+    if not organic_results:
+        return f"Nessun risultato trovato per '{query}'."
+
+    formatted_results = "\n\n".join([f"Title: {res.get('title')}\nLink: {res.get('link')}\nSnippet: {res.get('snippet')}" for res in organic_results])
+    return formatted_results
+
+serpapi_search_tool = Tool(
+    name="serpapi_web_search",
+    func=_serpapi_search,
+    description="Esegue una ricerca sul web utilizzando SerpAPI (Google Search) per trovare informazioni aggiornate. L'input dovrebbe essere la query di ricerca." \
+    " Questo tool ha un costo elevato, pertanto sono da preferire altri tool se disponibili. \
+    Richiamare questo tool soltanto in caso gli altri tool non siano stati soddisfacenti."
+)

tools.py

@@ -0,0 +1,69 @@
+from langchain.tools import Tool
+from youtube_transcript_api import YouTubeTranscriptApi, NoTranscriptFound, TranscriptsDisabled
+import operator
+
+
+def extract_youtube_transcript(youtube_url: str) -> str:
+    """
+    Extracts the transcript from a given YouTube video URL.
+    Returns the transcript as a single string or an error message if not found.
+    """
+    try:
+        video_id = youtube_url.split("v=")[1].split("&")[0]
+        transcript_list = YouTubeTranscriptApi.get_transcript(video_id)
+        transcript = " ".join([item['text'] for item in transcript_list])
+        return transcript
+    except NoTranscriptFound:
+        return "Error: No transcript found for this video. It might be disabled or not available in English."
+    except TranscriptsDisabled:
+        return "Error: Transcripts are disabled for this video."
+    except Exception as e:
+        return f"Error extracting transcript: {str(e)}"
+
+youtube_transcript_tool = Tool(
+    name="youtube_transcript_extractor",
+    func=extract_youtube_transcript,
+    description="Extracts the full transcript from a YouTube video given its URL. Input should be a valid YouTube video URL."
+)
+
+def add(a: float, b: float) -> float:
+    """Adds two numbers."""
+    return operator.add(a, b)
+
+def subtract(a: float, b: float) -> float:
+    """Subtracts the second number from the first."""
+    return operator.sub(a, b)
+
+def multiply(a: float, b: float) -> float:
+    """Multiplies two numbers."""
+    return operator.mul(a, b)
+
+def divide(a: float, b: float) -> float:
+    """Divides the first number by the second. Returns an error message if division by zero."""
+    if b == 0:
+        return "Error: Cannot divide by zero."
+    return operator.truediv(a, b)
+
+add_tool = Tool(
+    name="calculator_add",
+    func=add,
+    description="Adds two numbers. Input should be two numbers (a, b)."
+)
+
+subtract_tool = Tool(
+    name="calculator_subtract",
+    func=subtract,
+    description="Subtracts the second number from the first. Input should be two numbers (a, b)."
+)
+
+multiply_tool = Tool(
+    name="calculator_multiply",
+    func=multiply,
+    description="Multiplies two numbers. Input should be two numbers (a, b)."
+)
+
+divide_tool = Tool(
+    name="calculator_divide",
+    func=divide,
+    description="Divides the first number by the second. Input should be two numbers (a, b)."
+)

youtube_tools.py

@@ -0,0 +1,25 @@
+from langchain.tools import Tool
+from youtube_transcript_api import YouTubeTranscriptApi, NoTranscriptFound, TranscriptsDisabled
+
+def extract_youtube_transcript(youtube_url: str) -> str:
+    """
+    Extracts the transcript from a given YouTube video URL.
+    Returns the transcript as a single string or an error message if not found.
+    """
+    try:
+        video_id = youtube_url.split("v=")[1].split("&")[0]
+        transcript_list = YouTubeTranscriptApi.get_transcript(video_id)
+        transcript = " ".join([item['text'] for item in transcript_list])
+        return transcript
+    except NoTranscriptFound:
+        return "Error: No transcript found for this video. It might be disabled or not available in English."
+    except TranscriptsDisabled:
+        return "Error: Transcripts are disabled for this video."
+    except Exception as e:
+        return f"Error extracting transcript: {str(e)}"
+
+youtube_transcript_tool = Tool(
+    name="youtube_transcript_extractor",
+    func=extract_youtube_transcript,
+    description="Extracts the full transcript from a YouTube video given its URL. Input should be a valid YouTube video URL."
+)