switch from langgraph to smolagents

app.py

@@ -12,7 +12,7 @@ DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space"
 # ----- THIS IS WERE YOU CAN BUILD WHAT YOU WANT ------
 class BasicAgent:
     def __init__(self):
-        print("BasicAgent initialized.")
+        self.agent = create_agent()
     def __call__(self, question: str) -> str:
         print(f"Agent received question (first 50 chars): {question[:50]}...")
         fixed_answer = "This is a default answer."

geminiAgent.py

@@ -1,22 +1,73 @@
-from typing import TypedDict, Literal, List, Dict, Optional
 import time
-import requests
-from bs4 import BeautifulSoup
 import operator
-from langchain_core.prompts import ChatPromptTemplate
-from langchain_core.runnables import RunnablePassthrough
-from langchain_core.messages import HumanMessage, AIMessage
-from langchain.tools import tool
-from langchain_core.output_parsers import StrOutputParser
-from langchain_core.pydantic_v1 import BaseModel, Field
-from langgraph.graph import StateGraph, START, END
+from smolagents import CodeAgent,DuckDuckGoSearchTool, HfApiModel, load_tool, tool, WikipediaSearchTool, LiteLLMModel
+import datetime
+import requests
+import pytz
+import yaml
+from tools.final_answer import FinalAnswerTool
+from tools.web_search import DuckDuckGoSearchTool
+from tools.visit_webpage import VisitWebpageTool
+import utils
+import pandas as pd
 import pytesseract
+import openpyxl
 from PIL import Image
 
+def create_agent():
+    return CodeAgent(
+            model=LiteLLMModel(model_id="gemini_model_id", api_key=os.getenv("GEMINI_KEY")),
+            tools=[DuckDuckGoSearchTool(), WikipediaSearchTool(), VisitWebpageTool(), DownloadTaskAttachmentTool()],
+            add_base_tools=True,
+            additional_authorized_imports=['pandas','numpy','csv','subprocess', 'exec']
+        )
 @tool
 def ocr_tool(image: str) -> str:
     """
     A tool that performs OCR processing on an image.
+
+    Args: 
+        image: path for the image
     """
     image = Image.open(image)
     return pytesseract.image_to_string(image)
+
+@tool
+def read_csv(csv: str) -> str:
+    """
+    Reads a CSV file and returns its contents as a human-readable string.
+
+    Args:
+        csv: path for the csv file
+    """
+    csv_file = pd.read_csv(csv)
+    
+    if csv_file.empty:
+        return "The CSV file is empty."
+
+    # Format the DataFrame as a string with clear headers and a separator
+    output = ""
+    output += "CSV Data:\n"
+    output += "--------------------------------\n"
+    output += csv_file.to_string(index=False)
+    output += "\n--------------------------------\n"
+    
+    return output
+
+
+
+@tool
+def read_excel(excel_path: str) -> str:
+    """
+    Reads an Excel file and returns its contents as a human-readable string.
+
+    Args:
+        excel_path: The path to the Excel file.
+    """
+    try:
+        df = pd.read_excel(excel_path)
+        return df.to_string()
+    except Exception as e:
+        return f"Error reading Excel file: {e}"
+
+

tools/WebSearchTool.py

@@ -1,75 +1,75 @@
-import time
-from typing import Optional
-import requests
-from bs4 import BeautifulSoup
-from langchain.tools import tool
+# import time
+# from typing import Optional
+# import requests
+# from bs4 import BeautifulSoup
+# from langchain.tools import tool
 
-class WebSearchTool:
-    def __init__(self):
-        self.last_request_time = 0
-        self.min_request_interval = 2.0  # Minimum time between requests in seconds
-        self.max_retries = 10
+# class WebSearchTool:
+#     def __init__(self):
+#         self.last_request_time = 0
+#         self.min_request_interval = 2.0  # Minimum time between requests in seconds
+#         self.max_retries = 10
 
-    def search(self, query: str, domain: Optional[str] = None) -> str:
-        """Perform web search with rate limiting and retries."""
-        for attempt in range(self.max_retries):
-            # Implement rate limiting
-            current_time = time.time()
-            time_since_last = current_time - self.last_request_time
-            if time_since_last < self.min_request_interval:
-                time.sleep(self.min_request_interval - time_since_last)
+#     def search(self, query: str, domain: Optional[str] = None) -> str:
+#         """Perform web search with rate limiting and retries."""
+#         for attempt in range(self.max_retries):
+#             # Implement rate limiting
+#             current_time = time.time()
+#             time_since_last = current_time - self.last_request_time
+#             if time_since_last < self.min_request_interval:
+#                 time.sleep(self.min_request_interval - time_since_last)
             
-            try:
-                # Make the search request
-                results = self._do_search(query, domain)
-                self.last_request_time = time.time()
-                return results
-            except Exception as e:
-                if "202 Ratelimit" in str(e):
-                    if attempt < self.max_retries - 1:
-                        # Exponential backoff
-                        wait_time = (2 ** attempt) * self.min_request_interval
-                        time.sleep(wait_time)
-                        continue
-                return f"Search failed after {self.max_retries} attempts: {str(e)}"
+#             try:
+#                 # Make the search request
+#                 results = self._do_search(query, domain)
+#                 self.last_request_time = time.time()
+#                 return results
+#             except Exception as e:
+#                 if "202 Ratelimit" in str(e):
+#                     if attempt < self.max_retries - 1:
+#                         # Exponential backoff
+#                         wait_time = (2 ** attempt) * self.min_request_interval
+#                         time.sleep(wait_time)
+#                         continue
+#                 return f"Search failed after {self.max_retries} attempts: {str(e)}"
         
-        return "Search failed due to rate limiting"
+#         return "Search failed due to rate limiting"
 
-    def _do_search(self, query: str, domain: Optional[str] = None) -> str:
-        """Perform the actual search request."""
-        try:
-            # Construct search URL
-            base_url = "https://html.duckduckgo.com/html"
-            params = {"q": query}
-            if domain:
-                params["q"] += f" site:{domain}"
+#     def _do_search(self, query: str, domain: Optional[str] = None) -> str:
+#         """Perform the actual search request."""
+#         try:
+#             # Construct search URL
+#             base_url = "https://html.duckduckgo.com/html"
+#             params = {"q": query}
+#             if domain:
+#                 params["q"] += f" site:{domain}"
 
-            # Make request with increased timeout
-            response = requests.get(base_url, params=params, timeout=10)
-            response.raise_for_status()
+#             # Make request with increased timeout
+#             response = requests.get(base_url, params=params, timeout=10)
+#             response.raise_for_status()
 
-            if response.status_code == 202:
-                raise Exception("202 Ratelimit")
+#             if response.status_code == 202:
+#                 raise Exception("202 Ratelimit")
 
-            # Extract search results
-            results = []
-            soup = BeautifulSoup(response.text, 'html.parser')
-            for result in soup.find_all('div', {'class': 'result'}):
-                title = result.find('a', {'class': 'result__a'})
-                snippet = result.find('a', {'class': 'result__snippet'})
-                if title and snippet:
-                    results.append({
-                        'title': title.get_text(),
-                        'snippet': snippet.get_text(),
-                        'url': title.get('href')
-                    })
+#             # Extract search results
+#             results = []
+#             soup = BeautifulSoup(response.text, 'html.parser')
+#             for result in soup.find_all('div', {'class': 'result'}):
+#                 title = result.find('a', {'class': 'result__a'})
+#                 snippet = result.find('a', {'class': 'result__snippet'})
+#                 if title and snippet:
+#                     results.append({
+#                         'title': title.get_text(),
+#                         'snippet': snippet.get_text(),
+#                         'url': title.get('href')
+#                     })
 
-            # Format results
-            formatted_results = []
-            for r in results[:10]:  # Limit to top 5 results
-                formatted_results.append(f"[{r['title']}]({r['url']})\n{r['snippet']}\n")
+#             # Format results
+#             formatted_results = []
+#             for r in results[:10]:  # Limit to top 5 results
+#                 formatted_results.append(f"[{r['title']}]({r['url']})\n{r['snippet']}\n")
 
-            return "## Search Results\n\n" + "\n".join(formatted_results)
+#             return "## Search Results\n\n" + "\n".join(formatted_results)
 
-        except requests.RequestException as e:
-            raise Exception(f"Search request failed: {str(e)}")
+#         except requests.RequestException as e:
+#             raise Exception(f"Search request failed: {str(e)}")

tools/final_answer.py

@@ -0,0 +1,14 @@
+from typing import Any, Optional
+from smolagents.tools import Tool
+
+class FinalAnswerTool(Tool):
+    name = "final_answer"
+    description = "Provides a final answer to the given problem."
+    inputs = {'answer': {'type': 'any', 'description': 'The final answer to the problem'}}
+    output_type = "any"
+
+    def forward(self, answer: Any) -> Any:
+        return answer
+
+    def __init__(self, *args, **kwargs):
+        self.is_initialized = False

tools/visit_webpage.py

@@ -0,0 +1,45 @@
+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
+
+            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/web_search.py

@@ -0,0 +1,27 @@
+from typing import Any, Optional
+from smolagents.tools import Tool
+import duckduckgo_search
+
+class DuckDuckGoSearchTool(Tool):
+    name = "web_search"
+    description = "Performs a duckduckgo web search based on your query (think a Google search) then returns the top search results."
+    inputs = {'query': {'type': 'string', 'description': 'The search query to perform.'}}
+    output_type = "string"
+
+    def __init__(self, max_results=10, **kwargs):
+        super().__init__()
+        self.max_results = max_results
+        try:
+            from duckduckgo_search import DDGS
+        except ImportError as e:
+            raise ImportError(
+                "You must install package `duckduckgo_search` to run this tool: for instance run `pip install duckduckgo-search`."
+            ) from e
+        self.ddgs = DDGS(**kwargs)
+
+    def forward(self, query: str) -> str:
+        results = self.ddgs.text(query, max_results=self.max_results)
+        if len(results) == 0:
+            raise Exception("No results found! Try a less restrictive/shorter query.")
+        postprocessed_results = [f"[{result['title']}]({result['href']})\n{result['body']}" for result in results]
+        return "## Search Results\n\n" + "\n\n".join(postprocessed_results)