|
|
import os |
|
|
from llama_index.core.tools import FunctionTool |
|
|
from llama_index.tools.tavily_research import TavilyToolSpec |
|
|
from llama_index.readers.wikipedia import WikipediaReader |
|
|
from llama_index.readers.papers import ArxivReader |
|
|
from llama_index.core.tools.ondemand_loader_tool import OnDemandLoaderTool |
|
|
from huggingface_hub import list_models |
|
|
from observability import tool_span |
|
|
|
|
|
|
|
|
|
|
|
def multiply(a: int, b: int) -> int: |
|
|
"""Multiply two numbers.""" |
|
|
return a * b |
|
|
|
|
|
def add(a: int, b: int) -> int: |
|
|
"""Add two numbers.""" |
|
|
return a + b |
|
|
|
|
|
def subtract(a: int, b: int) -> int: |
|
|
"""Subtract two numbers.""" |
|
|
return a - b |
|
|
|
|
|
def divide(a: int, b: int) -> int: |
|
|
"""Divide two numbers.""" |
|
|
if b == 0: |
|
|
raise ValueError("Cannot divide by zero.") |
|
|
return a / b |
|
|
|
|
|
def modulus(a: int, b: int) -> int: |
|
|
"""Get the modulus of two numbers.""" |
|
|
return a % b |
|
|
|
|
|
|
|
|
|
|
|
def get_hub_stats(author: str) -> str: |
|
|
"""Fetches the most downloaded model from a specific author on the Hugging Face Hub.""" |
|
|
with tool_span("huggingface_hub_stats", metadata={"author": author}) as span: |
|
|
try: |
|
|
models = list(list_models(author=author, sort="downloads", direction=-1, limit=1)) |
|
|
if models: |
|
|
model = models[0] |
|
|
result = f"The most downloaded model by {author} is {model.id} with {model.downloads:,} downloads." |
|
|
if span: |
|
|
span.set_attribute("model_found", True) |
|
|
span.set_attribute("model_id", model.id) |
|
|
span.set_attribute("downloads", model.downloads) |
|
|
return result |
|
|
else: |
|
|
result = f"No models found for author {author}." |
|
|
if span: |
|
|
span.set_attribute("model_found", False) |
|
|
return result |
|
|
except Exception as e: |
|
|
error_result = f"Error fetching models for {author}: {str(e)}" |
|
|
if span: |
|
|
span.record_exception(e) |
|
|
span.set_attribute("error", True) |
|
|
return error_result |
|
|
|
|
|
|
|
|
|
|
|
def get_tavily_tool(): |
|
|
return TavilyToolSpec(api_key=os.getenv("TAVILY_API_KEY")) |
|
|
|
|
|
def get_arxiv_reader(): |
|
|
return ArxivReader() |
|
|
|
|
|
def get_wikipedia_reader(): |
|
|
return WikipediaReader() |
|
|
|
|
|
def get_wikipedia_tool(wikipedia_reader=None): |
|
|
if wikipedia_reader is None: |
|
|
wikipedia_reader = get_wikipedia_reader() |
|
|
return OnDemandLoaderTool.from_defaults( |
|
|
wikipedia_reader, |
|
|
name="Wikipedia Tool", |
|
|
description="A tool for loading data and querying articles from Wikipedia", |
|
|
) |
|
|
|
|
|
def get_arxiv_tool(arxiv_reader=None): |
|
|
if arxiv_reader is None: |
|
|
arxiv_reader = get_arxiv_reader() |
|
|
return OnDemandLoaderTool.from_defaults( |
|
|
arxiv_reader, |
|
|
name="Arxiv Tool", |
|
|
description="A tool for loading data and querying articles from Arxiv", |
|
|
) |
|
|
|
|
|
def get_search_tool(): |
|
|
return get_tavily_tool().to_tool_list() |
|
|
|
|
|
def get_calculator_tool(): |
|
|
return [ |
|
|
FunctionTool.from_defaults(multiply), |
|
|
FunctionTool.from_defaults(add), |
|
|
FunctionTool.from_defaults(subtract), |
|
|
FunctionTool.from_defaults(divide), |
|
|
FunctionTool.from_defaults(modulus), |
|
|
] |
|
|
|
|
|
def get_hub_stats_tool(): |
|
|
return FunctionTool.from_defaults(get_hub_stats) |
|
|
|