Refactored code

app_for_submission.py

@@ -3,7 +3,7 @@ import gradio as gr
 import requests
 import inspect
 import pandas as pd
-from app import alfred
+from graph.graph_builder import graph
 from langfuse.callback import CallbackHandler
 from typing import Optional
 from langchain_core.messages import AnyMessage, HumanMessage, AIMessage
@@ -60,7 +60,7 @@ def run_and_submit_all( profile: Optional[gr.OAuthProfile]):
 
     # 1. Instantiate Agent ( modify this part to create your agent)
     try:
-        agent = alfred
+        agent = graph
     except Exception as e:
         print(f"Error instantiating agent: {e}")
         return f"Error initializing agent: {e}", None
@@ -106,7 +106,7 @@ def run_and_submit_all( profile: Optional[gr.OAuthProfile]):
                 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]})
+            submitted_answer = graph.invoke(input={"messages": messages}, config={"callbacks": [langfuse_handler]})
             answers_payload.append({
                 "task_id": task_id,
                 "submitted_answer": submitted_answer['messages'][-1].content[-1] 

graph/graph_builder.py

@@ -0,0 +1,23 @@
+from langgraph.graph import START, StateGraph
+from langgraph.prebuilt import tools_condition
+from langgraph.prebuilt import ToolNode
+from nodes.core import assistant, tools
+from states.state import AgentState
+
+## 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")
+graph = builder.compile()

math_tools.py

@@ -1,44 +0,0 @@
-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)."
-)

nodes/core.py

@@ -0,0 +1,54 @@
+from states.state import AgentState
+import os
+# Import the load_dotenv function from the dotenv library
+from dotenv import load_dotenv
+from langchain_google_genai import ChatGoogleGenerativeAI
+from tools.multimodal_tools import extract_text, analyze_image_tool, analyze_audio_tool
+from tools.math_tools import add, subtract, multiply, divide
+from tools.search_tools import search_tool, serpapi_search
+from tools.youtube_tools import extract_youtube_transcript
+from langfuse.callback import CallbackHandler
+
+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")
+
+# Initialize Langfuse CallbackHandler for LangGraph/Langchain (tracing)
+langfuse_handler = CallbackHandler(
+    public_key=langfuse_public_key,
+    secret_key=langfuse_secret_key,
+    host="http://localhost:3000"
+)
+
+chat = ChatGoogleGenerativeAI(
+    model= "gemini-2.5-pro-preview-05-06",
+    temperature=0,
+    max_retries=2,
+    google_api_key=api_key,
+    thinking_budget= 0
+)
+
+tools = [
+    extract_text,
+    analyze_image_tool,
+    analyze_audio_tool,
+    extract_youtube_transcript,
+    add,
+    subtract,
+    multiply,
+    divide,
+    search_tool
+]
+
+chat_with_tools = chat.bind_tools(tools)
+
+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"])]
+    }

states/state.py

@@ -0,0 +1,7 @@
+from typing import TypedDict, Annotated
+from langchain_core.messages import AnyMessage
+from langgraph.graph.message import add_messages
+
+
+class AgentState(TypedDict):
+    messages: Annotated[list[AnyMessage], add_messages]

tools.py

@@ -1,69 +0,0 @@
-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)."
-)

tools/math_tools.py

@@ -0,0 +1,57 @@
+from langchain_core.tools import tool
+import operator
+
+@tool("add_tool", parse_docstring=True)
+def add(a: float, b: float) -> float:
+    """Adds two numbers.
+
+    Args:
+        a (float): The first number.
+        b (float): The second number.
+
+    Returns:
+        float: The sum of a and b.
+    """
+    return operator.add(a, b)
+
+@tool("subtract_tool", parse_docstring=True)
+def subtract(a: float, b: float) -> float:
+    """Subtracts the second number from the first.
+
+    Args:
+        a (float): The first number (minuend).
+        b (float): The second number (subtrahend).
+
+    Returns:
+        float: The result of subtracting b from a.
+    """
+    return operator.sub(a, b)
+
+@tool("multiply_tool", parse_docstring=True)
+def multiply(a: float, b: float) -> float:
+    """Multiplies two numbers.
+
+    Args:
+        a (float): The first number.
+        b (float): The second number.
+
+    Returns:
+        float: The product of a and b.
+    """
+    return operator.mul(a, b)
+
+@tool("divide_tool", parse_docstring=True)
+def divide(a: float, b: float) -> float:
+    """Divides the first number by the second.
+
+    Args:
+        a (float): The numerator.
+        b (float): The denominator.
+
+    Returns:
+        float: The result of dividing a by b.
+               Returns an error message string if division by zero occurs.
+    """
+    if b == 0:
+        return "Error: Cannot divide by zero."
+    return operator.truediv(a, b)

multimodal_tools.py → tools/multimodal_tools.py

@@ -15,10 +15,15 @@ vision_llm = ChatGoogleGenerativeAI(
     google_api_key=api_key
 )
 
+@tool("extract_text_tool", parse_docstring=True)
 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.
+    """Extract text from an image file using a multimodal model.
+
+    Args:
+        img_path (str): The path to the image file from which to extract text.
+    
+    Returns:
+        str: The extracted text from the image, or an empty string if an error occurs.
     """
     all_text = ""
     try:
@@ -64,12 +69,14 @@ def extract_text(img_path: str) -> str:
     
 @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.
+    """Answer the question reasoning on the image.
     
     Args:
-        user_query (str): The question to be answered.
-        img_path (str): Path to the image file.
+        user_query (str): The question to be answered based on the image.
+        img_path (str): Path to the image file to be analyzed.
+
+    Returns:
+        str: The answer to the query based on image content, or an empty string if an error occurs.
     """
     all_text = ""
     try:
@@ -114,12 +121,14 @@ def analyze_image_tool(user_query: str, img_path: str) -> str:
     
 @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.
+    """Answer the question by reasoning on the provided audio file.
     
     Args:
-        user_query (str): The question to be answered.
+        user_query (str): The question to be answered based on the audio content.
         audio_path (str): Path to the audio file (e.g., .mp3, .wav, .flac, .aac, .ogg).
+
+    Returns:
+        str: The answer to the query based on audio content, or an error message/empty string if an error occurs.
     """
     try:
         # Determine MIME type from file extension
@@ -165,10 +174,4 @@ def analyze_audio_tool(user_query: str, audio_path: str) -> str:
     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."
-)
+        return ""

serpapi_tools.py → tools/search_tools.py

@@ -2,13 +2,22 @@ import os
 from langchain.tools import Tool
 from serpapi import GoogleSearch
 from dotenv import load_dotenv
+from langchain_community.tools.tavily_search import TavilySearchResults
+from langchain_core.tools import tool
 
 # 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:
+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
+)
+
+@tool("serpapi_search_tool", parse_docstring=True)
+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.
@@ -42,12 +51,4 @@ def _serpapi_search(query: str, num_results: int = 5, gl: str = "it", hl: str =
         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."
-)
+    return formatted_results

youtube_tools.py → tools/youtube_tools.py

@@ -1,10 +1,16 @@
-from langchain.tools import Tool
+from langchain_core.tools import tool
 from youtube_transcript_api import YouTubeTranscriptApi, NoTranscriptFound, TranscriptsDisabled
 
+@tool("youtube_transcript_extractor", parse_docstring=True)
 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.
+    """Extracts the transcript from a given YouTube video URL.
+
+    Args:
+        youtube_url (str): The URL of the YouTube video.
+
+    Returns:
+        str: The transcript as a single string, or an error message if the transcript
+             cannot be found or an error occurs.
     """
     try:
         video_id = youtube_url.split("v=")[1].split("&")[0]
@@ -17,9 +23,3 @@ def extract_youtube_transcript(youtube_url: str) -> str:
         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."
-)