| 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) |
|
|