rohanjain2312's picture
Update app.py
a1bce95 verified
Raw
History Blame Contribute Delete
5.58 kB
from smolagents import CodeAgent, HfApiModel,load_tool,tool
import datetime
import requests
import pytz
import yaml
import json
from tools.final_answer import FinalAnswerTool
from Gradio_UI import GradioUI
@tool
def crypto_market_update(
assets: str,
vs_currency: str = "USD",
provider: str = "auto",
) -> str:
"""Get real-time crypto price and 24h change for one or more assets.
Args:
assets: Comma-separated symbols, e.g. "BTC, ETH, SOL".
vs_currency: Quote currency, e.g. "USD", "USDT", "EUR".
provider: "auto" (default), "binance" or "coinbase".
Returns:
Markdown table of Symbol, Price, 24h %, Source, FetchedAt (UTC).
"""
import datetime as _dt
import json as _json
symbols = [s.strip().upper() for s in (assets or "").split(",") if s.strip()]
if not symbols:
return "No assets provided."
vs = (vs_currency or "USD").upper()
srcs = []
def _binance_symbol(sym: str, quote: str) -> str:
# Map USD to USDT for spot
q = "USDT" if quote == "USD" else quote
return f"{sym}{q}"
def _fetch_binance(sym: str, quote: str):
pair = _binance_symbol(sym, quote)
r = requests.get(
"https://api.binance.com/api/v3/ticker/24hr",
params={"symbol": pair},
timeout=10,
)
if r.status_code != 200:
raise RuntimeError(f"Binance {pair} HTTP {r.status_code}")
data = r.json()
price = float(data["lastPrice"]) if "lastPrice" in data else float(data.get("last", 0))
change = float(data.get("priceChangePercent", 0.0))
return price, change, pair, "binance"
def _fetch_coinbase(sym: str, quote: str):
pair = f"{sym}-{quote}"
r_price = requests.get(
f"https://api.coinbase.com/v2/prices/{pair}/spot",
timeout=10,
)
if r_price.status_code != 200:
raise RuntimeError(f"Coinbase {pair} HTTP {r_price.status_code}")
price = float(r_price.json()["data"]["amount"])
# Coinbase doesn't expose 24h % in simple endpoint; set None
return price, None, pair, "coinbase"
rows = []
for sym in symbols:
fetched = None
if provider in ("auto", "binance"):
try:
fetched = _fetch_binance(sym, vs)
except Exception:
fetched = None
if fetched is None and provider in ("auto", "coinbase"):
try:
fetched = _fetch_coinbase(sym, vs)
except Exception:
fetched = None
if fetched is None:
rows.append((sym, "-", "-", "unavailable"))
else:
price, change, pair, src = fetched
rows.append((sym, f"{price:,.2f} {vs}", (f"{change:.2f}%" if change is not None else "-"), src))
ts = _dt.datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S UTC")
lines = [f"Fetched at: {ts}", "", "| Symbol | Price | 24h % | Source |", "|---|---:|---:|---|"]
for sym, price_s, change_s, src in rows:
lines.append(f"| {sym} | {price_s} | {change_s} | {src} |")
return "\n".join(lines)
@tool
def get_current_time_in_timezone(timezone: str, return_format: str = "sentence") -> str:
"""Fetch the current local time in a specified timezone.
Args:
timezone: A string representing a valid timezone (e.g., 'America/New_York').
return_format: One of {'sentence','iso','dict'} controlling the output.
- 'sentence': "The current local time in <tz> is: YYYY-MM-DD HH:MM:SS"
- 'iso': ISO 8601 string with offset
- 'dict': JSON string: {timezone, iso, human, unix, utc_offset}
Returns:
A string. For return_format='dict', this is a JSON-encoded string.
"""
try:
tz = pytz.timezone(timezone)
now = datetime.datetime.now(tz)
human = now.strftime("%Y-%m-%d %H:%M:%S")
iso_str = now.isoformat()
if return_format == "iso":
return iso_str
if return_format == "dict":
payload = {
"timezone": timezone,
"iso": iso_str,
"human": human,
"unix": int(now.timestamp()),
"utc_offset": now.strftime("%z"),
}
return json.dumps(payload)
# Default to sentence
return f"The current local time in {timezone} is: {human}"
except Exception as e:
return f"Error fetching time for timezone '{timezone}': {str(e)}"
final_answer = FinalAnswerTool()
# If the agent does not answer, the model is overloaded, please use another model or the following Hugging Face Endpoint that also contains qwen2.5 coder:
# model_id='https://pflgm2locj2t89co.us-east-1.aws.endpoints.huggingface.cloud'
model = HfApiModel(
max_tokens=2096,
temperature=0.5,
model_id='Qwen/Qwen2.5-Coder-32B-Instruct',# it is possible that this model may be overloaded
custom_role_conversions=None,
)
# 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, image_generation_tool, get_current_time_in_timezone, crypto_market_update], ## 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()