feata: my proposition

need some cleaning before publishing

.gitignore

@@ -0,0 +1,6 @@
+.venv/
+.env
+
+__pycache__/
+test/*
+logs/*

README.md

@@ -1,5 +1,5 @@
 ---
-title: Template Final Assignment
+title: Template Final Assignment - frenchtoast.dev
 emoji: 🕵🏻‍♂️
 colorFrom: indigo
 colorTo: indigo
@@ -12,4 +12,28 @@ hf_oauth: true
 hf_oauth_expiration_minutes: 480
 ---
 
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+
+# HF agent course final assigment
+
+## local running
+ 
+ Create a venv
+ Install requirements
+ create a .env file (see env.example)
+ start with `python app.py`
+
+DuckDuckGo and GoogleSearch have too many rate limit,
+This code assume that a local running instance of searxng is on http://localhost:8888
+
+On windows run it with:
+$PORT=8888
+docker run --rm `
+ -p ${PORT}:8080 `
+ -v "${PWD}/etcsearxng:/etc/searxng" `
+ -e "BASE_URL=http://localhost:$PORT/" `
+ -e "INSTANCE_NAME=my-instance" `
+ searxng/searxng
+
+ be sure to allow the json format in /etc/seraxng/settings.yml
+

app.py

@@ -3,25 +3,22 @@ import gradio as gr
 import requests
 import inspect
 import pandas as pd
+from dotenv import load_dotenv
+
+from myagent import BasicAgent  # Import your agent class from myagent.py
+from multiagents import MultiAgent
 
 # (Keep Constants as is)
 # --- Constants ---
 DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space"
+load_dotenv()
+
+max_questions = 20
+
 
-# --- 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):
+def run_and_submit_all(nb_questions: int, profile: gr.OAuthProfile | None): 
     """
-    Fetches all questions, runs the BasicAgent on them, submits all answers,
+    Fetches all questions, runs my Agent on them, submits all answers,
     and displays the results.
     """
     # --- Determine HF Space Runtime URL and Repo URL ---
@@ -36,11 +33,13 @@ def run_and_submit_all( profile: gr.OAuthProfile | None):
 
     api_url = DEFAULT_API_URL
     questions_url = f"{api_url}/questions"
+    file_url = f"{api_url}/files"    
     submit_url = f"{api_url}/submit"
 
-    # 1. Instantiate Agent ( modify this part to create your agent)
+    # 1. Instantiate Agent 
     try:
-        agent = BasicAgent()
+        # agent = BasicAgent()
+        agent = MultiAgent()
     except Exception as e:
         print(f"Error instantiating agent: {e}")
         return f"Error initializing agent: {e}", None
@@ -72,15 +71,27 @@ def run_and_submit_all( profile: gr.OAuthProfile | None):
     # 3. Run your Agent
     results_log = []
     answers_payload = []
+
+    # for testing keep only some questions
+    questions_data = questions_data[:nb_questions]
+
     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")
+        file_url = None
+        if file_name:
+            file_url = f"{file_url}/{task_id}"
         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)
+            agent_question = question_text
+            if file_url:
+                agent_question += f"\n\nFile URL: {file_url}"
+            
+            submitted_answer = agent(agent_question)
             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:
@@ -160,6 +171,8 @@ with gr.Blocks() as demo:
 
     gr.LoginButton()
 
+    nb_questions = gr.Number(value=20)
+
     run_button = gr.Button("Run Evaluation & Submit All Answers")
 
     status_output = gr.Textbox(label="Run Status / Submission Result", lines=5, interactive=False)
@@ -168,6 +181,7 @@ with gr.Blocks() as demo:
 
     run_button.click(
         fn=run_and_submit_all,
+        inputs=[nb_questions],
         outputs=[status_output, results_table]
     )
 

env.example

@@ -0,0 +1 @@
+HF_TOKEN=<<your hf token>>

multiagents.py

@@ -0,0 +1,110 @@
+
+import os
+import dotenv
+from smolagents import CodeAgent, ToolCallingAgent
+from smolagents import OpenAIServerModel
+from tools.fetch import fetch_webpage, search_web
+from smolagents import VisitWebpageTool, GoogleSearchTool, SpeechToTextTool
+from tools.yttranscript import get_youtube_transcript, get_youtube_title_description
+import myprompts
+
+dotenv.load_dotenv()
+
+gemini_model = OpenAIServerModel(
+    model_id="gemini-2.0-flash",
+    api_key=os.environ["GEMINI_API_KEY"],
+    # Google Gemini OpenAI-compatible API base URL
+    api_base="https://generativelanguage.googleapis.com/v1beta/openai/",
+)
+
+vllm_model = OpenAIServerModel(
+    model_id="Qwen/Qwen2.5-1.5B-Instruct",
+    api_base="http://192.168.1.39:18000/v1",  
+    api_key="token-abc123",  
+)
+
+openai_41nano_model = OpenAIServerModel(
+    model_id="gpt-4.1-nano",
+    api_base="https://api.openai.com/v1",
+    api_key=os.environ["OPENAI_API_KEY"],
+)
+
+
+def chek_final_answer(final_answer, agent_memory)  -> bool:
+    """
+    Check if the final answer is correct.
+    This is a placeholder function. You can implement your own logic here.
+    """
+    # For demonstration, we assume the answer is always correct
+    return True
+
+
+web_agent = CodeAgent(
+    model=openai_41nano_model,
+    tools=[
+        search_web,
+        fetch_webpage,                
+    ],
+    name="web_agent",
+    description="Browses the web to find information",
+    verbosity_level=1,
+    max_steps=7,
+)
+
+audiovideo_agent = CodeAgent(
+    model=openai_41nano_model,
+    tools=[
+        get_youtube_transcript,
+        get_youtube_title_description,
+        SpeechToTextTool()
+    ],
+    name="audiovideo_agent",
+    description="Extracts information from video or audio files",
+    verbosity_level=1,
+    max_steps=7,
+)
+
+
+
+manager_agent = CodeAgent(
+    model=openai_41nano_model,
+    tools=[],
+    managed_agents=[web_agent, audiovideo_agent],    
+    additional_authorized_imports=["pandas", "numpy","bs4"],
+    planning_interval=5,
+    verbosity_level=2,
+    final_answer_checks=[chek_final_answer],
+    max_steps=15,
+    name="manager_agent",
+    description="A manager agent that coordinates the work of other agents to answer questions.",
+)
+
+class MultiAgent:
+    def __init__(self):
+        print("BasicAgent initialized.")
+
+    def __call__(self, question: str) -> str:
+        try:
+            # log agent call in file
+            with open("logs/agent_calls.log", "a") as log_file:
+                log_file.write(f"Agent called with question: {question}\n")
+
+        try:
+            question = question + '\n' + myprompts.output_format
+            fixed_answer = ""
+
+            fixed_answer = manager_agent.run(question)
+            
+            return fixed_answer
+        except Exception as e:
+            error = f"An error occurred while processing the question: {e}"
+            print(error)
+            return error
+
+
+if __name__ == "__main__":
+    # Example usage
+    question = "How many studio albums were published by Mercedes Sosa between 2000 and 2009 (included)? You can use the latest 2022 version of english wikipedia."
+    agent = MultiAgent()
+    answer = agent(question)
+    print(f"Answer: {answer}")

myagent.py

@@ -0,0 +1,59 @@
+import os
+from smolagents import CodeAgent, ToolCallingAgent
+from smolagents import OpenAIServerModel
+from tools.fetch import fetch_webpage
+from tools.yttranscript import get_youtube_transcript, get_youtube_title_description
+import myprompts
+
+# --- Basic Agent Definition ---
+class BasicAgent:
+    def __init__(self):
+        print("BasicAgent initialized.")
+    def __call__(self, question: str) -> str:
+
+        print(f"Agent received question (first 50 chars): {question[:50]}...")
+
+        try:
+            # Use the reviewer agent to determine if the question can be answered by a model or requires code
+            print("Calling reviewer agent...")
+            reviewer_answer = reviewer_agent.run(myprompts.review_prompt + "\nThe question is:\n" + question)
+            print(f"Reviewer agent answer: {reviewer_answer}")
+
+            question = question + '\n' + myprompts.output_format
+            fixed_answer = ""
+
+            if reviewer_answer == "code":
+                fixed_answer = gaia_agent.run(question)
+                print(f"Code agent answer: {fixed_answer}")
+                
+            elif reviewer_answer == "model":    
+                # If the reviewer agent suggests using the model, we can proceed with the model agent
+                print("Using model agent to answer the question.")
+                fixed_answer = model_agent.run(model_prompt + "\nThe question is:\n" + question)
+                print(f"Model agent answer: {fixed_answer}")
+
+            return fixed_answer
+        except Exception as e:
+            error = f"An error occurred while processing the question: {e}"
+            print(error)
+            return error
+
+        
+      
+    
+model = OpenAIServerModel(
+    model_id="gpt-4.1-nano",
+    api_base="https://api.openai.com/v1",
+    api_key=os.environ["OPENAI_API_KEY"],
+)
+
+reviewer_agent= ToolCallingAgent(model=model, tools=[])
+model_agent = ToolCallingAgent(model=model,tools=[fetch_webpage])
+gaia_agent = CodeAgent(tools=[fetch_webpage,get_youtube_title_description,get_youtube_transcript ], model=model)
+
+if __name__ == "__main__":
+    # Example usage
+    question = "How many studio albums were published by Mercedes Sosa between 2000 and 2009 (included)? You can use the latest 2022 version of english wikipedia."
+    agent = BasicAgent()
+    answer = agent(question)
+    print(f"Answer: {answer}")

myprompts.py

@@ -0,0 +1,14 @@
+review_prompt = """You are a reviewer agent. You will be given a question. 
+Your task it to assert if a LLM agent with access to web content can answer the question or not, or if a coding agent and more tools is needed to answer the question.
+If the question is too complex for a LLM agent, you should return "code" as the answer else you should return "model".
+"""
+
+model_prompt = """You are a LLM agent. You will be given a question.
+Your task it to answer the question using the tools you have access to.
+take time to analyse the steps to answer the question.
+if a filename is given in the question you can infer the url 
+"""
+
+output_format = """OUTPUT FORMAT should be a number OR as few words as possible OR a comma separated list of numbers and/or strings.
+ If you are asked for a number, don't use comma to write your number neither use units such as $ or percent sign unless specified otherwise, use digits only.
+ If you are asked for a string, don't use articles, neither abbreviations (e.g. for cities), and write the digits in plain text unless specified otherwise. If you are asked for a comma separated list, apply the above rules depending of whether the element to be put in the list is a number or a string."""

requirements.txt

@@ -1,2 +1,11 @@
 gradio
-requests
+gradio[oauth]
+requests
+python-dotenv
+smolagents 
+duckduckgo-search
+smolagents[openai]
+markdownify
+beautifulsoup4
+transformers
+smolagents[transformers]

tools/fetch.py

@@ -0,0 +1,108 @@
+# this is  asmolagent too to fetch html content from a url
+from smolagents import tool
+import requests
+from markdownify import markdownify as md
+import time
+from bs4 import BeautifulSoup
+
+@tool
+def fetch_webpage(url: str, convert_to_markdown: bool = True) -> str:
+    """
+    Fetches the HTML content of a given URL. 
+    if markdown conversion is enabled, it will remove script and style and return the text content as markdown else return raw unfiltered HTML
+    Args:
+        url (str): The URL to fetch.
+        convert_to_markdown (bool): If True, convert the HTML content to Markdown format. else return the raw HTML.
+    Returns:
+        str: The HTML content of the URL.
+    """    
+    response = requests.get(url)
+    if (convert_to_markdown):
+        soup = BeautifulSoup(response.text, "html.parser")
+        # remove script and style tags
+        for script in soup(["script", "style"]):
+            script.extract()
+
+        # for wikipedia only keep the main content
+        if "wikipedia.org" in url:
+            main_content = soup.find("main",{"id":"content"})
+            if main_content:
+                content = md(str(main_content),strip=['script', 'style']).strip()
+            else:
+                content = md(response.text,strip=['script', 'style']).strip()
+    else:
+        content = response.text
+    
+    try:
+        # save content to a file in test folder before returning
+        # compute filepath with correct extension based on convert_to_markdown and add a timestamp for unicity
+        file_extension = ".md" if convert_to_markdown else ".html"
+        unicity_suffix = str(int(time.time()))
+        file_name = f"test/fetched_content_{unicity_suffix}{file_extension}"
+        with open(file_name, "w", encoding="utf-8") as f:
+            f.write(content)
+    except Exception as e:
+        print(f"Error saving content to file: {e}")
+        
+    return content
+
+@tool
+# this tool allow web search on a local SearXNG instance
+def search_web(query: str, num_results: int = 5) -> list:
+    """
+    Perform a web search using local SearXNG instance.
+    Args:
+        query (str): The search query.
+        num_results (int): The number of results to return.
+    Returns:
+        list: A list of search results sorted by score with {url, title, content, score} for each result.
+    """
+    # local metaserach engine searxng, run on localhost:8888
+    searxng_url = "http://localhost:8888/search"
+    params = {"q": query, "format": 'json'}
+    response = requests.get(searxng_url, params=params)
+    if response.status_code == 200:
+        ret = response.json()
+        # keep only the response'results' array
+        results = ret.get("results", [])
+        # keep only the first num_results   
+        results = results[:num_results]
+        # for each result keep only the url, title and content ans score
+        results = [
+            {
+                "url": result.get("url"),
+                "title": result.get("title"),
+                "content": result.get("content"),
+                "score": result.get("score"),
+            }
+            for result in results
+        ]
+        
+        return results
+        
+    else:
+        print(f"Error: {response.status_code}")
+        return []
+
+if __name__ == "__main__":
+   
+    try:
+        # Test the function
+        query = "What is the capital of France?"
+        results = search_web(query,3)
+        print(results)
+    except Exception as e:
+        print(f"An error occurred: {e}")
+
+    try:
+        # Test the function
+        video_id = "L1vXCYZAYYM"  # Replace with your YouTube video ID
+        video_url = "https://www.youtube.com/watch?v=" + video_id
+        url = "https://en.wikipedia.org/wiki/Malko_Competition"
+        # page_content = fetch_webpage(video_url)
+        page_content = fetch_webpage(url, convert_to_markdown=True)
+        print(page_content.encode("utf-8"))
+    except Exception as e:
+        print(f"An error occurred: {e}")
+ 
+   

tools/yttranscript.py

@@ -0,0 +1,72 @@
+
+from smolagents import tool
+from youtube_transcript_api import YouTubeTranscriptApi
+from bs4 import BeautifulSoup
+from tools.fetch import fetch_webpage
+
+@tool
+# a function to get youtube transcript from video id
+def get_youtube_transcript(video_id: str) -> str:
+    """
+    Fetches the transcript of a YouTube video given its video ID.    
+    Args:
+        video_id (str): The ID of the YouTube video. Pass in the video ID, NOT the video URL. For a video with the URL https://www.youtube.com/watch?v=12345 the ID is 12345.
+    Returns:
+        str: The transcript of the YouTube video. as a single string with each line separated by a newline character.
+    """
+    # Initialize the YouTubeTranscriptApi
+    ytt_api = YouTubeTranscriptApi()
+    fetched_transcript = ytt_api.fetch(video_id)
+    raw_data = fetched_transcript.to_raw_data()
+    # raw data is in the form of [{        'text': 'Hey there',        'start': 0.0,        'duration': 1.54    },    {        'text': 'how are you',,        'start': 1.54,        'duration': 4.16    },   ... ] we will return ony the text element as lines
+    transcript = "\n".join([item['text'] for item in raw_data])
+    return transcript
+
+
+@tool
+# a function to get video title and description from video url
+def get_youtube_title_description(video_url: str) -> str:
+    """
+    Fetches the title and description of a YouTube video given its video ID.    
+    Args:
+        video_url (str): The url of the YouTube video.
+    Returns:
+        str: The title and description of the YouTube video.
+    """
+    # Initialize the YouTube object
+    soup = BeautifulSoup(fetch_webpage(video_url, convert_to_markdown=False), "html.parser")
+    # Extract the title by looking at the meta tag with name="title" and getting the content
+    metatitle = soup.find("meta", {"name": "title"})
+    if metatitle is not None:
+        title = metatitle["content"]
+    else:
+        title = "No title found"
+
+    # same for description
+    metadescription = soup.find("meta", {"name": "description"})    
+    if metadescription is not None:
+        description = metadescription["content"]
+    else:
+        description = "No description found"    
+    
+    return f"Title: {title}\nDescription: {description}"
+
+
+if __name__ == "__main__":
+    from dotenv import load_dotenv
+    load_dotenv
+    # Test the function
+    video_id = "1htKBjuUWec"  # Replace with your YouTube video ID
+    video_url = "https://www.youtube.com/watch?v=" + video_id
+    # Get the title and description
+    try:
+        title_description = get_youtube_title_description(video_url)
+        print(title_description)
+    except Exception as e:
+        print(f"Error fetching title and description: {e}")
+    
+    try:
+        transcript = get_youtube_transcript(video_id)
+    except Exception as e:
+        print(f"Error fetching transcript: {e}")
+    print(transcript)

vllm_asopenai_test.py

@@ -0,0 +1,15 @@
+import os
+from smolagents import OpenAIServerModel, CodeAgent
+
+model = OpenAIServerModel(
+    model_id="Qwen/Qwen2.5-1.5B-Instruct",
+    api_base="http://192.168.1.39:18000/v1",  
+    api_key="token-abc123",  
+)
+
+myagent = CodeAgent(
+    model=model,
+    tools=[])
+
+result = myagent.run("Hello who are you?")  # Replace with your question
+print(result)

vllm_test.py

@@ -0,0 +1,14 @@
+from openai import OpenAI
+client = OpenAI(
+    base_url="http://192.168.1.39:18000/v1",
+    api_key="token-abc123",
+)
+
+completion = client.chat.completions.create(
+  model="Qwen/Qwen2.5-1.5B-Instruct",
+  messages=[
+    {"role": "user", "content": "Hello!"}
+  ]
+)
+
+print(completion.choices[0].message)