from smolagents import CodeAgent,DuckDuckGoSearchTool, HfApiModel,load_tool,Tool,tool, LiteLLMModel, TransformersModel import datetime import requests import pytz import yaml import os from tools.final_answer import FinalAnswerTool from Gradio_UI import GradioUI @tool def anime_recommender(desc: str) -> str: """ Suggest an anime whose story is based on the description. Args: desc: The story of an anime. """ if len(desc) <= 50: return 'One Punch Man' elif len(desc) <= 100: return 'Solo Leveling' else: return 'Anime Not Found' class SimilarAnimeSuggester(Tool): name = "similar_anime_name_suggester" description = """ This tool suggest a similar anime based on the recommended anime """ inputs = { 'recommendation':{ 'type': 'string', "description": 'The name of an anime', } } output_type = 'string' def forward(self, recommendation: str): if recommendation == 'One Punch Man': return 'Mob 100' elif recommendation == 'Solo Leveling': return 'The Rising Of The Shield Hero' else: return 'This anime is unique' final_answer = FinalAnswerTool() from recipe_scrapers import scrape_html from urllib.request import urlopen from Gradio_UI import GradioUI @tool def food_recipe_recommender(url: str) -> str: """This tool look at the urls provided and returns the recipe from a wroking, existing page. Args: url: list of url pages to scrape the recipe from. """ scraper = None try: html = urlopen(url).read().decode("utf-8") scraper = scrape_html(html, org_url=url) except: return f"Recipe not found for provided url: {url}." ingredients = [f" - {ing}\n" for ing in scraper.ingredients()] recipe = f""" 👩🏼‍🍳 Et voilà! The {scraper.title()}! Preparation requires {scraper.total_time()} minutes, and it is for {scraper.yields()} people. 🧂 Ingredients:\n {''.join(ingredients)} 🥘 Instructions: {scraper.instructions()} """ return recipe class VisitWebpageMarkdown(Tool): name = "visit_webpage_markdown" description = "Visits a webpage at the given url and reads its content as a markdown document via Jina. Use this to browse webpages. It skips images / medias" inputs = {'url': {'type': 'string', 'description': 'The url of the webpage to visit.'}} output_type = "string" def forward(self, url: str) -> str: import requests try: # Send a GET request to the URL response = requests.get('https://r.jina.ai/' +url) response.raise_for_status() # Raise an exception for bad status codes markdown_content =response.text.strip() return markdown_content except Exception as e: return f"An unexpected error occurred: {str(e)}" def __init__(self, *args, **kwargs): self.is_initialized = False # class TravelDistanceDuration(Tool): # name = "get_travel_duration" # description = "Gets the travel time between two places." # inputs = {"start_location":{"type":"string","description":"the place from which you start your ride"},"destination_location":{"type":"string","description":"the place of arrival"},"transportation_mode":{"type":"string","nullable":True,"description":"The transportation mode, in 'driving', 'walking', 'bicycling', or 'transit'. Defaults to 'driving'."}} # output_type = "string" # def forward(self, start_location: str, destination_location: str, transportation_mode: Optional[str] = None) -> str: # """Gets the travel time between two places. # Args: # start_location: the place from which you start your ride # destination_location: the place of arrival # transportation_mode: The transportation mode, in 'driving', 'walking', 'bicycling', or 'transit'. Defaults to 'driving'. # """ # import os # All imports are placed within the function, to allow for sharing to Hub. # import googlemaps # from datetime import datetime # gmaps = googlemaps.Client(os.getenv("GMAPS_API_KEY")) # if transportation_mode is None: # transportation_mode = "driving" # try: # directions_result = gmaps.directions( # start_location, # destination_location, # mode=transportation_mode, # departure_time=datetime(2025, 12, 6, 11, 0), # At 11, date far in the future # ) # if len(directions_result) == 0: # return "No way found between these places with the required transportation mode." # return directions_result[0]["legs"][0]["duration"]["text"] # except Exception as e: # print(e) # return e from typing import Any, Optional class WebContentAnalyzer(Tool): name = "web_content_analyzer" description = "Analyzes web content using AI models." inputs = {"url":{"type":"string","description":"The webpage URL to analyze."}} output_type = "string" def forward(self, url: str) -> str: """Analyzes web content using AI models. Args: url: The webpage URL to analyze. Returns: str: Analysis results in JSON format. """ import requests from bs4 import BeautifulSoup import re from transformers import pipeline import json try: # Fetch content headers = {'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64)'} response = requests.get(url, headers=headers, timeout=10) # Parse HTML soup = BeautifulSoup(response.text, 'html.parser') for tag in soup(['script', 'style', 'meta']): tag.decompose() # Extract basic info title = soup.title.string if soup.title else "No title found" text = re.sub(r'\s+', ' ', soup.get_text()).strip() if len(text) < 100: return json.dumps({ "error": "Not enough content to analyze" }) # Get summary summarizer = pipeline("summarization", model="facebook/bart-large-cnn") summary = summarizer(text[:1024], max_length=100, min_length=30)[0]['summary_text'] # Get sentiment classifier = pipeline("text-classification", model="nlptown/bert-base-multilingual-uncased-sentiment") sentiment = classifier(text[:512])[0] score = int(sentiment['label'][0]) mood = ["Very Negative", "Negative", "Neutral", "Positive", "Very Positive"][score-1] # Format results result = { "title": title, "summary": summary, "sentiment": f"{mood} ({score}/5)", "stats": { "words": len(text.split()), "chars": len(text) } } return json.dumps(result) except Exception as e: return json.dumps({ "error": str(e) }) """------Applied TF-IDF for better semantic search------""" import feedparser import urllib.parse import yaml from tools.final_answer import FinalAnswerTool import numpy as np from sklearn.feature_extraction.text import TfidfVectorizer from sklearn.metrics.pairwise import cosine_similarity import gradio as gr from smolagents import CodeAgent,DuckDuckGoSearchTool, HfApiModel,load_tool,tool import nltk import datetime import requests import pytz from tools.final_answer import FinalAnswerTool from Gradio_UI import GradioUI nltk.download("stopwords") from nltk.corpus import stopwords @tool # ✅ Register the function properly as a SmolAgents tool def fetch_latest_arxiv_papers(keywords: list, num_results: int = 5) -> list: """Fetches and ranks arXiv papers using TF-IDF and Cosine Similarity. Args: keywords: List of keywords for search. num_results: Number of results to return. Returns: List of the most relevant papers based on TF-IDF ranking. """ try: print(f"DEBUG: Searching arXiv papers with keywords: {keywords}") # Use a general keyword search query = "+AND+".join([f"all:{kw}" for kw in keywords]) query_encoded = urllib.parse.quote(query) url = f"http://export.arxiv.org/api/query?search_query={query_encoded}&start=0&max_results=50&sortBy=submittedDate&sortOrder=descending" print(f"DEBUG: Query URL - {url}") feed = feedparser.parse(url) papers = [] # Extract papers from arXiv for entry in feed.entries: papers.append({ "title": entry.title, "authors": ", ".join(author.name for author in entry.authors), "year": entry.published[:4], "abstract": entry.summary, "link": entry.link }) if not papers: return [{"error": "No results found. Try different keywords."}] # Prepare TF-IDF Vectorization corpus = [paper["title"] + " " + paper["abstract"] for paper in papers] vectorizer = TfidfVectorizer(stop_words=stopwords.words('english')) # Remove stopwords tfidf_matrix = vectorizer.fit_transform(corpus) # Transform Query into TF-IDF Vector query_str = " ".join(keywords) query_vec = vectorizer.transform([query_str]) #Compute Cosine Similarity similarity_scores = cosine_similarity(query_vec, tfidf_matrix).flatten() #Sort papers based on similarity score ranked_papers = sorted(zip(papers, similarity_scores), key=lambda x: x[1], reverse=True) # Return the most relevant papers return [paper[0] for paper in ranked_papers[:num_results]] except Exception as e: print(f"ERROR: {str(e)}") return [{"error": f"Error fetching research papers: {str(e)}"}] # # Create Gradio UI # with gr.Blocks() as demo: # gr.Markdown("# ScholarAgent") # keyword_input = gr.Textbox(label="Enter keywords (comma-separated)", placeholder="e.g., deep learning, reinforcement learning") # output_display = gr.Markdown() # search_button = gr.Button("Search") # search_button.click(search_papers, inputs=[keyword_input], outputs=[output_display]) # print("DEBUG: Gradio UI is running. Waiting for user input...") # # Launch Gradio App # demo.launch() class HFModelDownloadsTool(Tool): name = "model_download_counter" description = """ This is a tool that returns the most downloaded model of a given task on the Hugging Face Hub. It returns the name of the checkpoint.""" inputs = {'task': {'type': 'string', 'description': 'the task category (such as text-classification, depth-estimation, etc)'}} output_type = "string" def forward(self, task: str): from huggingface_hub import list_models model = next(iter(list_models(filter=task, sort="downloads", direction=-1))) return model.id def __init__(self, *args, **kwargs): self.is_initialized = False # hugging face is getting the hug of death so lets use litellm for now model = HfApiModel( max_tokens=2096, temperature=0.5, # model_id='Qwen/Qwen2.5-Coder-32B-Instruct', model_id='https://pflgm2locj2t89co.us-east-1.aws.endpoints.huggingface.cloud', custom_role_conversions=None, ) # model = LiteLLMModel( # model_id="gemini/gemini-2.0-flash-exp", # max_tokens=2096, # temperature=0.6, # api_key=os.getenv("LITELLM_API_KEY") # ) # ollama # model = LiteLLMModel( # model_id="ollama_chat/deepseek-r1:7b", # max_tokens=2096, # temperature=0.6, # api_base="http://localhost:11434", # num_ctx=8192 # ) # transformer # model = TransformersModel( # model_id="deepseek-ai/DeepSeek-R1-Distill-Qwen-1.5B", # device_map="auto", # torch_dtype="auto", # max_new_tokens=2096, # temperature=0.6, # ) # Import tool from Hub image_generation_tool = load_tool("agents-course/text-to-image", trust_remote_code=True) with open("prompts.yaml", 'r') as stream: prompt_templates = yaml.safe_load(stream) agent = CodeAgent( model=model, tools=[final_answer, anime_recommender, SimilarAnimeSuggester(), # TravelDistanceDuration(), food_recipe_recommender, VisitWebpageMarkdown(), fetch_latest_arxiv_papers, WebContentAnalyzer(), HFModelDownloadsTool(), ], ## add your tools here (don't remove final answer) max_steps=6, verbosity_level=1, grammar=None, planning_interval=None, name=None, description=None, prompt_templates=prompt_templates ) GradioUI(agent).launch()