Astro final challenge agent

README.md

@@ -1,5 +1,5 @@
 ---
-title: Template Final Assignment
+title: Astro Agent
 emoji: 🕵🏻‍♂️
 colorFrom: indigo
 colorTo: indigo

agents/__init__.py

@@ -0,0 +1,5 @@
+from .hf_code_agent import hf_code_agent
+from .hf_tool_calling_agent import hf_tool_calling_agent
+
+
+__all__ = ["hf_tool_calling_agent", "hf_code_agent"]

agents/hf_code_agent.py

@@ -0,0 +1,69 @@
+import sys
+import os
+
+# Add the Project Root to sys.path
+sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
+
+from smolagents import (
+    HfApiModel,
+    CodeAgent,
+    load_tool,
+    Tool,
+    InferenceClientModel,
+    ToolCallingAgent,
+    FinalAnswerTool,
+    DuckDuckGoSearchTool,
+    VisitWebpageTool,
+    GoogleSearchTool,
+    PythonInterpreterTool,
+)
+import os
+from huggingface_hub import login
+from dotenv import load_dotenv
+from data.sample_questions import QUESTIONS
+
+# from tools.visit_website import VisitWebpageTool
+
+load_dotenv()
+login(os.environ["HF_API_KEY"])
+
+# Tools
+
+# wikipedia = Tool.from_langchain(load_tool("wikipedia", trust_remote_code=True))
+
+tools = [
+    # DuckDuckGoSearchTool(),
+    # VisitWebpageTool(),
+    PythonInterpreterTool(),
+    FinalAnswerTool(),
+    # wikipedia
+]
+
+question = QUESTIONS[0]
+
+# LLM Model
+model = HfApiModel(
+    "deepseek-ai/DeepSeek-R1",
+    provider="together",
+    # max_tokens=40096,
+    # temperature=0.1,
+    # token=get_huggingface_token(),
+)
+
+# Code Agent
+hf_code_agent = CodeAgent(
+    model=model,
+    tools=tools,
+    max_steps=20,
+    additional_authorized_imports=["pandas", "numpy", "time", "bs4", "time"],
+    verbosity_level=2,
+    name="python_interpreter_agent",
+    description="Can run and execute python code."
+)
+
+hf_code_agent.logger.console.width = 66
+
+
+if __name__ == "__main__":
+    answer = hf_code_agent.run(question)
+    print(answer)

agents/hf_tool_calling_agent.py

@@ -0,0 +1,67 @@
+from smolagents import (
+    HfApiModel,
+    CodeAgent,
+    load_tool,
+    Tool,
+    InferenceClientModel,
+    ToolCallingAgent,
+    FinalAnswerTool,
+    DuckDuckGoSearchTool,
+    VisitWebpageTool,
+    GoogleSearchTool,
+    PythonInterpreterTool,
+)
+import os
+from huggingface_hub import login
+from dotenv import load_dotenv
+# from langchain.agents import load_tools
+from langchain_community.agent_toolkits.load_tools import load_tools
+from tools.transcriber import transcribe_audio
+load_dotenv()
+login(os.environ["HF_API_KEY"])
+from data.sample_questions import QUESTIONS
+
+
+# Tools
+
+wikipedia = Tool.from_langchain(load_tools(["wikipedia"])[0])
+
+tools = [
+    # GoogleSearchTool(),
+    DuckDuckGoSearchTool(),
+    VisitWebpageTool(),
+    PythonInterpreterTool(),
+    FinalAnswerTool(),
+    wikipedia,
+    transcribe_audio
+]
+
+# Model
+# LLM Model
+model = HfApiModel(
+    "Qwen/Qwen2.5-72B-Instruct",
+    # "deepseek-ai/DeepSeek-R1",
+    provider="together",
+    # max_tokens=40096,
+    temperature=0.1,
+    # token=get_huggingface_token(),
+)
+
+# Tool Calling Agent
+llm = HfApiModel("meta-llama/Llama-3.2-3B-Instruct", temperature=0)
+
+hf_tool_calling_agent = ToolCallingAgent(
+    model=model,
+    tools=tools,
+    max_steps=10,
+    name="web_search_tool_calling_agent",
+    description="Can perform web searches and can visit the websites",
+    verbosity_level=2,
+)
+
+hf_tool_calling_agent.logger.console.width = 66
+
+if __name__ == "__main__":
+    question = QUESTIONS[0]
+    answer = hf_tool_calling_agent.run(question)
+    print(answer)

agents/orchestrator_agent.py

@@ -0,0 +1,56 @@
+from smolagents import (
+    HfApiModel,
+    CodeAgent,
+    Tool,
+    InferenceClientModel,
+    ToolCallingAgent,
+    FinalAnswerTool,
+    DuckDuckGoSearchTool,
+    VisitWebpageTool,
+    GoogleSearchTool,
+    PythonInterpreterTool,
+)
+import os
+from huggingface_hub import login
+from dotenv import load_dotenv
+
+load_dotenv()
+login(os.environ["HF_API_KEY"])
+
+tools = [
+    # DuckDuckGoSearchTool(),
+    # VisitWebpageTool(),
+    # PythonInterpreterTool(),
+    FinalAnswerTool(),
+    # wikipedia
+]
+
+# LLM Model
+model = HfApiModel(
+    "deepseek-ai/DeepSeek-R1",
+    provider="together",
+    max_tokens=8096,
+    # temperature=0.1,
+    # token=get_huggingface_token(),
+)
+
+# Agent
+from agents.hf_code_agent import hf_code_agent
+from agents.hf_tool_calling_agent import hf_tool_calling_agent
+
+orchestrator_agent = CodeAgent(
+    model=model,
+    tools=tools,
+    managed_agents=[hf_tool_calling_agent],
+    additional_authorized_imports=['pandas', 'numpy', 'time'],
+    planning_interval=5,
+    verbosity_level=2,
+    # final_answer_checks=[FinalAnswerTool()],
+    max_steps=10,
+    # name="Orchestrator Agent",
+)
+
+orchestrator_agent.logger.console.width = 66
+
+if __name__ == "__main__":
+    question =  ""

app.py

@@ -3,6 +3,7 @@ import gradio as gr
 import requests
 import inspect
 import pandas as pd
+from agents.orchestrator_agent import orchestrator_agent
 
 # (Keep Constants as is)
 # --- Constants ---
@@ -14,10 +15,11 @@ 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
+        return orchestrator_agent.run(question)
+        # 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):
     """

data/__init__.py

@@ -0,0 +1,6 @@
+from .sample_questions import QUESTIONS
+
+
+__all__ = [
+    "QUESTIONS"
+]

data/sample_questions.py

@@ -0,0 +1,6 @@
+QUESTIONS = [
+    """
+ I'm making a grocery list for my mom, but she's a professor of botany and she's a real stickler when it comes to categorizing things. I need to add different foods to different categories on the grocery list, but if I make a mistake, she won't buy anything inserted in the wrong category. Here's the list I have so far:\n\nmilk, eggs, flour, whole bean coffee, Oreos, sweet potatoes, fresh basil, plums, green beans, rice, corn, bell pepper, whole allspice, acorns, broccoli, celery, zucchini, lettuce, peanuts\n\nI need to make headings for the fruits and vegetables. Could you please create a list of just the vegetables from my list? If you could do that, then I can figure out how to categorize the rest of the list into the appropriate categories. But remember that my mom is a real stickler, so make sure that no botanical fruits end up on the vegetable list, or she won't get them when she's at the store. Please alphabetize the list of vegetables, and place each item in a comma separated list.
+""",
+    "Who are the pitchers with the number before and after Taishō Tamai's number as of July 2023? Give them to me in the form Pitcher Before, Pitcher After, use their last names only, in Roman characters.",
+]

requirements.txt

@@ -1,2 +1,13 @@
 gradio
-requests
+gradio[oauth]
+requests
+smolagents
+transformers
+dotenv
+duckduckgo-search
+markdownify
+pandas
+numpy
+langchain
+langchain-community
+wikipedia

tools/__init__.py

@@ -0,0 +1,6 @@
+from .transcriber import transcribe_audio
+
+
+__all__ = [
+    "transcribe_audio",
+]

tools/transcriber.py

@@ -0,0 +1,30 @@
+# Use a pipeline as a high-level helper
+from transformers import pipeline
+from smolagents import tool
+import os
+# print(os.getcwd() + "/audio/interview.mp3")
+transcriber_pipeline = pipeline(
+    "automatic-speech-recognition", model="facebook/wav2vec2-base-960h"
+)
+
+
+@tool
+def transcribe_audio(audio_file_path: str) -> str:
+    """Transcribe an audio file into text.
+    Args:
+        audio_file_path: The path to the audio file to transcribe.
+    Returns:
+        The transcribed text.
+    """
+    try:
+        if os.path.isfile(audio_file_path):
+            return transcriber_pipeline(audio_file_path)["text"]
+        else:
+            raise FileNotFoundError(f"Audio file not found: {audio_file_path}")
+    except FileNotFoundError as e:
+        return f"Error: {str(e)}"
+
+# file = os.getcwd() + "/audio/interview.mp3"
+# result = transcribe_audio(file)
+# print(result)
+# transcribe_audio_tool = transcribe_audio.push_to_hub()

tools/visit_website.py

@@ -0,0 +1,46 @@
+from typing import Any, Optional
+from smolagents.tools import Tool
+import requests
+import markdownify
+import smolagents
+
+class VisitWebpageTool(Tool):
+    name = "visit_webpage"
+    description = "Visits a webpage at the given url and reads its content as a markdown string. Use this to browse webpages."
+    inputs = {'url': {'type': 'string', 'description': 'The url of the webpage to visit.'}}
+    output_type = "string"
+
+    def forward(self, url: str) -> str:
+        try:
+            import requests
+            from markdownify import markdownify
+            from requests.exceptions import RequestException
+            import re
+
+            from smolagents.utils import truncate_content
+        except ImportError as e:
+            raise ImportError(
+                "You must install packages `markdownify` and `requests` to run this tool: for instance run `pip install markdownify requests`."
+            ) from e
+        try:
+            # Send a GET request to the URL with a 20-second timeout
+            response = requests.get(url, timeout=20)
+            response.raise_for_status()  # Raise an exception for bad status codes
+
+            # Convert the HTML content to Markdown
+            markdown_content = markdownify(response.text).strip()
+
+            # Remove multiple line breaks
+            markdown_content = re.sub(r"\n{3,}", "\n\n", markdown_content)
+
+            return truncate_content(markdown_content, 10000)
+
+        except requests.exceptions.Timeout:
+            return "The request timed out. Please try again later or check the URL."
+        except RequestException as e:
+            return f"Error fetching the webpage: {str(e)}"
+        except Exception as e:
+            return f"An unexpected error occurred: {str(e)}"
+
+    def __init__(self, *args, **kwargs):
+        self.is_initialized = False

tools/wikipedia_summarize.py

@@ -0,0 +1,21 @@
+from smolagents import tool
+import requests
+
+@tool
+def wiki_summarize(topic: str) -> str:
+    """Get the first paragraph summary for a Wikipedia topic.
+    rgs:
+        topic: The Wikipedia page title to summarize.
+    Returns:
+        A short summary of the page.
+
+    """
+    url = f"https://en.wikipedia.org/api/rest_v1/page/summary/{topic}"
+    resp = requests.get(url)
+    if resp.status_code == 200:
+        data = resp.json()
+        return data.get("extract", "No summary available.")
+    else:
+        return f"Error fetching Wikipedia summary (status {resp.status_code})."
+
+# wiki_summarize_tool = wiki_summarize.push_to_hub()