multitool_agent / app.py
Quivering's picture
Update app.py
8237908 verified
from smolagents import CodeAgent, DuckDuckGoSearchTool, HfApiModel, load_tool, tool
import datetime
import requests
import pytz
import yaml
import os
from tools.final_answer import FinalAnswerTool
from Gradio_UI import GradioUI
from typing import Optional, List
from datetime import datetime, timedelta
# Set up OpenWeather API key
OPENWEATHER_API_KEY = os.getenv('OPENWEATHER_API_KEY', 'your_api_key_here')
@tool
def get_weather_info(city: str) -> str:
"""A tool that fetches current weather information for a given city
Args:
city: The name of the city to get weather for
"""
try:
base_url = "http://api.openweathermap.org/data/2.5/weather"
params = {
'q': city,
'appid': OPENWEATHER_API_KEY,
'units': 'metric'
}
response = requests.get(base_url, params=params)
response.raise_for_status()
weather_data = response.json()
temperature = weather_data['main']['temp']
conditions = weather_data['weather'][0]['description']
humidity = weather_data['main']['humidity']
wind_speed = weather_data['wind']['speed']
return (f"Current weather in {city}:\n"
f"Temperature: {temperature}°C\n"
f"Conditions: {conditions}\n"
f"Humidity: {humidity}%\n"
f"Wind Speed: {wind_speed} m/s")
except Exception as e:
return f"Error fetching weather for city '{city}': {str(e)}"
@tool
def crypto_tracker(action: str, crypto_id: Optional[str] = None) -> str:
"""Tracks cryptocurrency prices and changes using CoinGecko API
Args:
action: The action to perform ('price', 'top', 'detail', 'summary')
crypto_id: The cryptocurrency ID (e.g., 'bitcoin', 'ethereum') - required for 'price' and 'detail' actions
"""
try:
base_url = "https://api.coingecko.com/api/v3"
def format_price_change(change: float) -> str:
"""Format price change with color indicator"""
if change > 0:
return f"↑ {change:.2f}%"
elif change < 0:
return f"↓ {change:.2f}%"
return f"{change:.2f}%"
if action == "top":
# Get top 10 cryptocurrencies by market cap
response = requests.get(
f"{base_url}/coins/markets",
params={
"vs_currency": "usd",
"order": "market_cap_desc",
"per_page": 10,
"page": 1,
"sparkline": False,
"price_change_percentage": "24h,7d"
}
)
response.raise_for_status()
data = response.json()
result = "Top 10 Cryptocurrencies:\n\n"
for coin in data:
result += (
f"📊 {coin['name']} ({coin['symbol'].upper()})\n"
f" Price: ${coin['current_price']:,.2f}\n"
f" 24h Change: {format_price_change(coin['price_change_percentage_24h'])}\n"
f" 7d Change: {format_price_change(coin['price_change_percentage_7d'])}\n"
f" Market Cap: ${coin['market_cap']:,.0f}\n\n"
)
return result
elif action == "price" and crypto_id:
# Get simple price data for a specific cryptocurrency
response = requests.get(
f"{base_url}/simple/price",
params={
"ids": crypto_id,
"vs_currencies": "usd",
"include_24hr_change": True,
"include_market_cap": True
}
)
response.raise_for_status()
data = response.json()
if crypto_id not in data:
return f"Cryptocurrency '{crypto_id}' not found"
coin_data = data[crypto_id]
return (
f"Current {crypto_id.title()} Price:\n"
f"USD: ${coin_data['usd']:,.2f}\n"
f"24h Change: {format_price_change(coin_data['usd_24h_change'])}\n"
f"Market Cap: ${coin_data['usd_market_cap']:,.0f}"
)
elif action == "detail" and crypto_id:
# Get detailed information about a specific cryptocurrency
response = requests.get(f"{base_url}/coins/{crypto_id}")
response.raise_for_status()
data = response.json()
market_data = data['market_data']
return (
f"📊 {data['name']} ({data['symbol'].upper()}) Detailed Report:\n\n"
f"Current Price: ${market_data['current_price']['usd']:,.2f}\n"
f"24h Change: {format_price_change(market_data['price_change_percentage_24h'])}\n"
f"7d Change: {format_price_change(market_data['price_change_percentage_7d'])}\n"
f"30d Change: {format_price_change(market_data['price_change_percentage_30d'])}\n"
f"Market Cap Rank: #{data['market_cap_rank']}\n"
f"Market Cap: ${market_data['market_cap']['usd']:,.0f}\n"
f"24h Trading Volume: ${market_data['total_volume']['usd']:,.0f}\n"
f"Circulating Supply: {market_data['circulating_supply']:,.0f} {data['symbol'].upper()}\n"
f"All-Time High: ${market_data['ath']['usd']:,.2f} ({market_data['ath_date']['usd'][:10]})\n"
f"All-Time Low: ${market_data['atl']['usd']:,.2f} ({market_data['atl_date']['usd'][:10]})\n\n"
f"Description: {data['description']['en'][:300]}..."
)
elif action == "summary":
# Get 24-hour summary for top 5 cryptocurrencies
response = requests.get(
f"{base_url}/coins/markets",
params={
"vs_currency": "usd",
"order": "market_cap_desc",
"per_page": 5,
"page": 1,
"sparkline": False,
"price_change_percentage": "24h",
"include_24hr_vol": True
}
)
response.raise_for_status()
data = response.json()
total_market_change = sum(coin['price_change_percentage_24h'] for coin in data) / len(data)
total_volume = sum(coin['total_volume'] for coin in data)
summary = "🚀 24-Hour Crypto Market Summary\n\n"
summary += f"Overall Market Trend: {format_price_change(total_market_change)} (Top 5 Average)\n"
summary += f"Total Trading Volume: ${total_volume:,.0f}\n\n"
summary += "Top 5 Cryptocurrency Performance:\n\n"
for coin in data:
price_change = coin['price_change_percentage_24h']
volume_change = (coin['total_volume'] / coin['market_cap']) * 100
# Determine market sentiment
if price_change > 5:
sentiment = "🔥 Strong Bullish"
elif price_change > 2:
sentiment = "📈 Bullish"
elif price_change < -5:
sentiment = "💧 Strong Bearish"
elif price_change < -2:
sentiment = "📉 Bearish"
else:
sentiment = "➡️ Stable"
summary += (
f"{coin['name']} ({coin['symbol'].upper()}):\n"
f" Current Price: ${coin['current_price']:,.2f}\n"
f" 24h Change: {format_price_change(price_change)}\n"
f" 24h Volume: ${coin['total_volume']:,.0f}\n"
f" Volume/Market Cap: {volume_change:.2f}%\n"
f" Market Sentiment: {sentiment}\n\n"
)
# Add significant events
significant_events = []
for coin in data:
if abs(coin['price_change_percentage_24h']) > 10:
significant_events.append(
f"⚠️ {coin['name']} moved {format_price_change(coin['price_change_percentage_24h'])} in 24h"
)
if coin['current_price'] > coin.get('ath', float('inf')):
significant_events.append(f"🎉 {coin['name']} reached a new all-time high!")
if significant_events:
summary += "Significant Events:\n"
summary += "\n".join(f"• {event}" for event in significant_events)
return summary
else:
return ("Invalid action. Use:\n"
"'top' to see top 10 cryptocurrencies\n"
"'price <crypto_id>' to get current price (e.g., 'price bitcoin')\n"
"'detail <crypto_id>' to get detailed information\n"
"'summary' to get 24-hour summary of top 5 cryptocurrencies")
except requests.exceptions.RequestException as e:
return f"Error fetching cryptocurrency data: {str(e)}"
except KeyError as e:
return f"Error processing cryptocurrency data: {str(e)}"
except Exception as e:
return f"Unexpected error: {str(e)}"
@tool
def get_current_time_in_timezone(timezone: str) -> str:
"""A tool that fetches the current local time in a specified timezone.
Args:
timezone: A string representing a valid timezone (e.g., 'America/New_York').
"""
try:
tz = pytz.timezone(timezone)
local_time = datetime.now(tz).strftime("%Y-%m-%d %H:%M:%S")
return f"The current local time in {timezone} is: {local_time}"
except Exception as e:
return f"Error fetching time for timezone '{timezone}': {str(e)}"
final_answer = FinalAnswerTool()
model = HfApiModel(
max_tokens=2096,
temperature=0.5,
model_id='Qwen/Qwen2.5-Coder-32B-Instruct',
custom_role_conversions=None,
)
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)
search_tool = DuckDuckGoSearchTool()
agent = CodeAgent(
model=model,
tools=[
final_answer,
search_tool,
image_generation_tool,
get_current_time_in_timezone,
get_weather_info,
crypto_tracker # Fixed the name here
],
max_steps=6,
verbosity_level=1,
grammar=None,
planning_interval=None,
name="Multi-tool Assistant",
description="An agent capable of searching, generating images, checking time zones, checking crypto prices and getting weather information.",
prompt_templates=prompt_templates
)
GradioUI(agent).launch()