jiaendu's picture
Update app.py
778a6ee verified
Raw
History Blame Contribute Delete
31.3 kB
from smolagents import CodeAgent,DuckDuckGoSearchTool, HfApiModel,load_tool,tool
import datetime
import requests
import pytz
import yaml
from tools.final_answer import FinalAnswerTool
from Gradio_UI import GradioUI
from smolagents import tool
from typing import Optional, Union
import psutil
import platform
import os
import math
import ast
import operator
import requests
from bs4 import BeautifulSoup
@tool
def crawl_gold_price(region: str = "china") -> str:
"""
爬取黄金/白银实时价格的工具,默认获取中国市场价格(人民币计价),支持切换地区。
数据来源为goldprice.org的官方API,爬取频率建议≥1分钟/次以避免反爬虫。
Args:
region: 目标地区,可选值包括:
- "china":中国市场(黄金单位:人民币/克,白银单位:人民币/克)
- "us":美国市场(黄金/白银单位:美元/盎司)
- "global":全球平均价格(黄金/白银单位:美元/盎司)
Returns:
str: 包含黄金/白银价格、涨跌额、涨跌幅、收盘价、更新时间的格式化结果,若失败则返回错误提示。
Examples:
>>> crawl_gold_price(region="china")
"✅ 贵金属实时价格查询结果:\n地区:china\n黄金价格(人民币/克):...\n白银价格(人民币/克):...\n获取时间:..."
>>> crawl_gold_price(region="us")
"✅ 贵金属实时价格查询结果:\n地区:us\n黄金价格(美元/盎司):...\n白银价格(美元/盎司):...\n获取时间:..."
"""
# 地区对应API映射
region_api_map = {
"china": "https://data-asg.goldprice.org/dbXRates/CNY",
"us": "https://data-asg.goldprice.org/dbXRates/USD",
"global": "https://data-asg.goldprice.org/dbXRates/USD"
}
# 单位配置(区分黄金/白银)
region_unit_map = {
"china": ("人民币/克", "人民币/盎司"),
"us": ("美元/盎司", "美元/盎司"),
"global": ("美元/盎司", "美元/盎司")
}
OUNCE_TO_GRAM = 31.1035 # 盎司转克的换算系数
# 请求头配置
headers = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/125.0.0.0 Safari/537.36",
"Referer": "https://goldprice.org/",
"Origin": "https://goldprice.org"
}
try:
# 验证地区参数
if region not in region_api_map:
return f"❌ 无效地区:{region}\n支持的地区:china(中国)、us(美国)、global(全球)"
# 发送请求
session = requests.Session()
session.mount('https://', requests.adapters.HTTPAdapter(max_retries=3))
response = session.get(
url=region_api_map[region],
headers=headers,
timeout=10
)
response.raise_for_status()
data = response.json()
# 解析完整数据(包含黄金+白银的所有字段)
if "items" in data and len(data["items"]) > 0:
item = data["items"][0]
display_unit, raw_unit = region_unit_map[region]
# 1. 黄金数据处理(含单位转换)
gold_raw_price = item["xauPrice"] # 盎司价
gold_final_price = round(gold_raw_price / OUNCE_TO_GRAM, 2) if region == "china" else round(gold_raw_price, 2)
gold_chg = item["chgXau"]
gold_pc = item["pcXau"]
gold_close = item["xauClose"]
gold_close_final = round(gold_close / OUNCE_TO_GRAM, 2) if region == "china" else round(gold_close, 2)
# 2. 白银数据处理(含单位转换)
silver_raw_price = item["xagPrice"] # 盎司价
silver_final_price = round(silver_raw_price / OUNCE_TO_GRAM, 2) if region == "china" else round(silver_raw_price, 2)
silver_chg = item["chgXag"]
silver_pc = item["pcXag"]
silver_close = item["xagClose"]
silver_close_final = round(silver_close / OUNCE_TO_GRAM, 2) if region == "china" else round(silver_close, 2)
# 3. 时间处理(API返回的原始时间 + 本地北京时间)
api_update_time = data["date"]
local_time = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
# 组装完整返回结果(包含所有字段)
return (
f"✅ 贵金属实时价格查询结果:\n"
f"🔍 地区:{region}\n"
f"\n【黄金数据】\n"
f"实时价格:{gold_final_price} {display_unit}\n"
f"涨跌额:{gold_chg} {raw_unit.split('/')[0]}\n"
f"涨跌幅:{gold_pc} %\n"
f"收盘价:{gold_close_final} {display_unit}\n"
f"\n【白银数据】\n"
f"实时价格:{silver_final_price} {display_unit}\n"
f"涨跌额:{silver_chg} {raw_unit.split('/')[0]}\n"
f"涨跌幅:{silver_pc} %\n"
f"收盘价:{silver_close_final} {display_unit}\n"
f"\n⏰ 数据更新时间(API):{api_update_time}\n"
f"⏰ 本地获取时间(北京时间):{local_time}\n"
f"📌 数据来源:goldprice.org API"
)
else:
return "❌ 未找到价格数据:API返回格式异常"
except requests.exceptions.Timeout:
return "❌ 请求超时:无法连接到价格API,请稍后重试"
except requests.exceptions.ConnectionError:
return "❌ 连接错误:网络异常或API不可用"
except requests.exceptions.HTTPError as e:
return f"❌ HTTP错误:{e.response.status_code},API访问失败"
except KeyError as e:
return f"❌ 数据解析错误:API返回字段缺失 {e}"
except Exception as e:
return f"❌ 爬虫执行异常:{str(e)}"
# 定义支持的运算符(用于安全表达式计算)
OPERATORS = {
ast.Add: operator.add,
ast.Sub: operator.sub,
ast.Mult: operator.mul,
ast.Div: operator.truediv,
ast.Pow: operator.pow,
ast.Mod: operator.mod,
ast.FloorDiv: operator.floordiv,
ast.USub: operator.neg
}
@tool
def calculator_tool(
operation_type: str,
num1: Optional[float] = None, # 替换Union[int, float]为float
num2: Optional[float] = None, # 替换Union[int, float]为float
expression: Optional[str] = None,
decimal_num: Optional[int] = None,
target_base: Optional[int] = None,
precision: int = 4
) -> str:
"""
多功能计算机工具,支持基础算术、科学计算、进制转换、复杂表达式计算,安全无风险。
Args:
operation_type: 计算类型,可选值:
- "basic":基础算术运算(+、-、*、/、^、%、//)
- "scientific":科学计算(平方根、对数、正弦/余弦/正切)
- "expression":复杂数学表达式计算(如"(10+5)*2/3")
- "base_convert":进制转换(十进制↔二进制/八进制/十六进制)
num1: 基础运算/科学计算的第一个数(如加法的被加数、平方根的被开方数)
num2: 基础运算的第二个数(如加法的加数,减法的减数),部分运算无需(如平方根)
expression: 复杂表达式字符串(仅operation_type="expression"时必填)
decimal_num: 待转换的十进制数(仅base_convert时,十进制转其他进制必填)
target_base: 目标进制(2/8/16,仅base_convert时必填)
precision: 计算结果保留小数位数(默认4位)
示例用法:
1. 基础运算:10+5 → operation_type="basic", num1=10, num2=5, op="+"
2. 科学计算:√16 → operation_type="scientific", num1=16, op="sqrt"
3. 表达式计算:(2+3)*4-5 → operation_type="expression", expression="(2+3)*4-5"
4. 进制转换:10进制25转二进制 → operation_type="base_convert", decimal_num=25, target_base=2
"""
try:
# -------------------------- 1. 基础算术运算 --------------------------
if operation_type == "basic":
if num1 is None:
return "❌ 缺少参数:基础运算需要指定num1(第一个数)"
# 解析运算符(从num2的备注中提取,或智能体自动匹配)
# 兼容两种传参方式:num2传数值+额外op参数,或直接在num2中传"5+"(不推荐,仅兼容)
op = None
if isinstance(num2, str) and len(num2) > 1 and num2[-1] in "+-*/^%//":
op = num2[-1]
try:
num2 = float(num2[:-1])
except:
return "❌ 数值格式错误:num2需为数字+运算符(如'5+')"
elif isinstance(num2, (int, float)):
# 智能体需明确传入op,这里补充常见运算符匹配
# 实际使用时,建议智能体通过prompt指定op参数,此处做兼容
return "❌ 缺少运算符:基础运算需指定运算符(+、-、*、/、^、%、//)"
# 执行基础运算
if op == "+":
result = num1 + num2
elif op == "-":
result = num1 - num2
elif op == "*":
result = num1 * num2
elif op == "/":
if num2 == 0:
return "❌ 除法错误:除数不能为0"
result = num1 / num2
elif op == "^":
result = num1 ** num2
elif op == "%":
result = num1 % num2
elif op == "//":
if num2 == 0:
return "❌ 整除错误:除数不能为0"
result = num1 // num2
else:
return f"❌ 不支持的运算符:{op}\n支持的运算符:+、-、*、/、^(幂)、%(取模)、//(整除)"
return f"✅ 基础运算结果:\n{num1} {op} {num2} = {result:.{precision}f}"
# -------------------------- 2. 科学计算 --------------------------
elif operation_type == "scientific":
if num1 is None:
return "❌ 缺少参数:科学计算需要指定num1(计算基数)"
# 解析科学计算类型(智能体需传入op参数,如sqrt、log、sin)
op = None
if isinstance(num2, str):
op = num2.lower()
else:
return "❌ 缺少计算类型:科学计算需指定类型(sqrt、log、ln、sin、cos、tan)"
# 执行科学计算
if op == "sqrt":
if num1 < 0:
return "❌ 平方根错误:被开方数不能为负数"
result = math.sqrt(num1)
elif op == "log": # 以10为底的对数
if num1 <= 0:
return "❌ 对数错误:底数必须大于0"
result = math.log10(num1)
elif op == "ln": # 自然对数
if num1 <= 0:
return "❌ 自然对数错误:底数必须大于0"
result = math.log(num1)
elif op == "sin":
result = math.sin(math.radians(num1)) # 输入角度,转弧度计算
elif op == "cos":
result = math.cos(math.radians(num1))
elif op == "tan":
result = math.tan(math.radians(num1))
else:
return f"❌ 不支持的科学计算类型:{op}\n支持的类型:sqrt(平方根)、log(常用对数)、ln(自然对数)、sin/cos/tan(三角函数,输入角度)"
return f"✅ 科学计算结果:\n{op}({num1}) = {result:.{precision}f}"
# -------------------------- 3. 复杂表达式计算(安全版) --------------------------
elif operation_type == "expression":
if not expression:
return "❌ 缺少参数:表达式计算需要指定expression(如'(10+5)*2/3')"
# 安全解析表达式(避免恶意代码执行)
def _eval_expr(node):
if isinstance(node, ast.Num): # 数字
return node.n
elif isinstance(node, ast.BinOp): # 二元运算
return OPERATORS[type(node.op)](_eval_expr(node.left), _eval_expr(node.right))
elif isinstance(node, ast.UnaryOp): # 一元运算(如负数)
return OPERATORS[type(node.op)](_eval_expr(node.operand))
else:
raise TypeError(f"不支持的表达式节点类型:{type(node)}")
try:
# 解析表达式
tree = ast.parse(expression, mode='eval')
result = _eval_expr(tree.body)
return f"✅ 表达式计算结果:\n{expression} = {result:.{precision}f}"
except SyntaxError:
return f"❌ 表达式语法错误:{expression}\n请检查格式(如括号是否闭合、运算符是否正确)"
except TypeError as e:
return f"❌ 表达式不支持:{str(e)}\n仅支持加减乘除、幂、取模、整除、负数"
except Exception as e:
return f"❌ 表达式计算异常:{str(e)}"
# -------------------------- 4. 进制转换 --------------------------
elif operation_type == "base_convert":
if decimal_num is None or target_base is None:
return "❌ 缺少参数:进制转换需要指定decimal_num(十进制数)和target_base(目标进制:2/8/16)"
if target_base not in [2, 8, 16]:
return "❌ 目标进制错误:仅支持2(二进制)、8(八进制)、16(十六进制)"
try:
decimal_num = int(decimal_num)
if target_base == 2:
result = bin(decimal_num)
elif target_base == 8:
result = oct(decimal_num)
elif target_base == 16:
result = hex(decimal_num)
return f"✅ 进制转换结果:\n十进制{decimal_num}{target_base}进制 = {result}(去掉前缀0b/0o/0x即为纯数字)"
except ValueError:
return "❌ 进制转换错误:decimal_num必须为整数"
# -------------------------- 无效操作类型 --------------------------
else:
return (
f"❌ 无效操作类型:{operation_type}\n"
f"支持的计算类型:\n"
f"- basic:基础算术运算(+、-、*、/、^、%、//)\n"
f"- scientific:科学计算(sqrt、log、sin等)\n"
f"- expression:复杂表达式计算\n"
f"- base_convert:进制转换(十进制↔2/8/16进制)"
)
except Exception as e:
return f"❌ 计算机工具执行异常:{str(e)}"
@tool
def system_info_query(
operation_type: str,
disk_path: Optional[str] = None,
process_limit: Optional[int] = 10
) -> str:
"""
安全查询系统关键信息的工具,支持CPU、内存、磁盘、进程和系统基础信息的查询。
所有操作均为只读,无任何修改系统的风险。
Args:
operation_type: 查询类型,可选值:
- "cpu":查询CPU使用率和核心数
- "memory":查询内存使用情况
- "disk":查询磁盘空间(默认根目录)
- "processes":查询当前运行的进程(默认返回前10个)
- "system_info":查询系统基础信息(OS、Python版本等)
disk_path: 磁盘查询时的目标路径(默认根目录),仅operation_type="disk"时可选
process_limit: 返回进程的最大数量(默认10,最大50),仅operation_type="processes"时可选
示例用法:
1. 查询CPU使用率:operation_type="cpu"
2. 查询内存使用:operation_type="memory"
3. 查询C盘空间:operation_type="disk", disk_path="C:\\"
4. 查询前5个进程:operation_type="processes", process_limit=5
5. 查询系统基础信息:operation_type="system_info"
"""
try:
# -------------------------- 1. CPU信息查询 --------------------------
if operation_type == "cpu":
cpu_count = psutil.cpu_count(logical=True)
cpu_usage = psutil.cpu_percent(interval=1, percpu=True)
avg_usage = sum(cpu_usage) / len(cpu_usage) if cpu_usage else 0.0
result = (
f"✅ CPU信息查询结果:\n"
f"逻辑核心数:{cpu_count}\n"
f"各核心使用率:{[f'{u}%' for u in cpu_usage]}\n"
f"平均使用率:{avg_usage:.1f}%"
)
return result
# -------------------------- 2. 内存信息查询 --------------------------
elif operation_type == "memory":
mem = psutil.virtual_memory()
swap = psutil.swap_memory()
def format_bytes(b):
for unit in ['B', 'KB', 'MB', 'GB', 'TB']:
if b < 1024:
return f"{b:.2f} {unit}"
b /= 1024
return f"{b:.2f} PB"
result = (
f"✅ 内存信息查询结果:\n"
f"总物理内存:{format_bytes(mem.total)}\n"
f"已使用内存:{format_bytes(mem.used)} ({mem.percent}%)\n"
f"可用内存:{format_bytes(mem.available)}\n"
f"交换分区总大小:{format_bytes(swap.total)}\n"
f"交换分区已使用:{format_bytes(swap.used)} ({swap.percent}%)"
)
return result
# -------------------------- 3. 磁盘信息查询 --------------------------
elif operation_type == "disk":
# 安全校验:禁止路径遍历(如../)
if disk_path and (".." in disk_path or (os.sep in disk_path and not os.path.isabs(disk_path))):
return "❌ 路径非法:仅支持绝对路径或根目录,禁止路径遍历(如../)"
target_path = disk_path or os.path.sep
if not os.path.exists(target_path):
return f"❌ 路径不存在:{target_path}"
disk = psutil.disk_usage(target_path)
result = (
f"✅ 磁盘信息查询结果(路径:{target_path}):\n"
f"总容量:{format_bytes(disk.total)}\n"
f"已使用:{format_bytes(disk.used)} ({disk.percent}%)\n"
f"可用空间:{format_bytes(disk.free)}"
)
return result
# -------------------------- 4. 进程信息查询 --------------------------
elif operation_type == "processes":
# 限制返回数量,避免信息过载
limit = min(process_limit or 10, 50)
processes = []
for proc in psutil.process_iter(['pid', 'name', 'cpu_percent', 'memory_percent']):
try:
processes.append(proc.info)
if len(processes) >= limit:
break
except (psutil.NoSuchProcess, psutil.AccessDenied):
continue
result = f"✅ 进程信息查询结果(前{len(processes)}个):\n"
for p in processes:
result += (
f"PID: {p['pid']} | 名称: {p['name']} | "
f"CPU使用率: {p['cpu_percent']}% | 内存使用率: {p['memory_percent']:.2f}%\n"
)
return result
# -------------------------- 5. 系统基础信息查询 --------------------------
elif operation_type == "system_info":
result = (
f"✅ 系统基础信息查询结果:\n"
f"操作系统:{platform.system()} {platform.release()}\n"
f"主机名:{platform.node()}\n"
f"Python版本:{platform.python_version()}\n"
f"处理器架构:{platform.machine()}\n"
f"启动时间:{datetime.fromtimestamp(psutil.boot_time()).strftime('%Y-%m-%d %H:%M:%S')}"
)
return result
# -------------------------- 无效操作类型 --------------------------
else:
return (
f"❌ 无效操作类型:{operation_type}\n"
f"支持的查询类型:cpu、memory、disk、processes、system_info"
)
except ImportError:
return "❌ 依赖缺失:请先安装psutil库(执行 pip install psutil)"
except PermissionError:
return "❌ 权限不足:无法访问部分系统信息(如进程详情)"
except Exception as e:
return f"❌ 工具执行异常:{str(e)}"
# 辅助函数:字节数格式化
def format_bytes(b: int) -> str:
for unit in ['B', 'KB', 'MB', 'GB', 'TB']:
if b < 1024:
return f"{b:.2f} {unit}"
b /= 1024
return f"{b:.2f} PB"
# 定义常用单位转换映射表(编程场景高频)
UNIT_CONVERSION_MAP = {
# 字节单位:基准为Byte
"byte": {
"B": 1,
"KB": 1024,
"MB": 1024**2,
"GB": 1024**3,
"TB": 1024**4,
"bit": 0.125 # 1 Byte = 8 bit
},
# 时间单位:基准为秒
"time": {
"s": 1,
"ms": 0.001,
"min": 60,
"h": 3600,
"d": 86400
}
}
@tool
def timezone_unit_converter(
operation_type: str,
source_value: Optional[float] = None, # 替换Union[int, float, str]为float
source_type: Optional[str] = None,
target_type: Optional[str] = None,
timezone: Optional[str] = None,
source_time: Optional[str] = None,
source_timezone: Optional[str] = None
) -> str:
"""
集成时区转换和编程常用单位转换的工具,支持两类操作:
1. 时区转换:查询指定时区当前时间 / 将指定时间从源时区转换到目标时区
2. 单位转换:字节/时间/存储单位互转(如B→GB、秒→小时、bit→Byte)
Args:
operation_type: 操作类型,可选值:
- "timezone_current":查询指定时区当前时间
- "timezone_convert":将源时间从源时区转换到目标时区
- "unit_convert":单位转换
source_value: 单位转换时的原始数值(如1024),仅operation_type=unit_convert时必填
source_type: 单位转换时的原始单位(如"KB")/ 时区转换时的源时区(如"UTC"),按需填写
target_type: 单位转换时的目标单位(如"MB")/ 时区转换时的目标时区(如"Asia/Shanghai"),按需填写
timezone: 查询当前时区时间时的目标时区(如"America/New_York"),仅operation_type=timezone_current时必填
source_time: 待转换的源时间(格式:YYYY-MM-DD HH:MM:SS),仅operation_type=timezone_convert时必填
source_timezone: 源时间对应的时区(如"UTC"),仅operation_type=timezone_convert时必填
示例用法:
1. 查询纽约当前时间:
operation_type="timezone_current", timezone="America/New_York"
2. 将UTC时间2026-01-30 10:00:00转上海时间:
operation_type="timezone_convert", source_time="2026-01-30 10:00:00", source_type="UTC", target_type="Asia/Shanghai"
3. 将1024KB转换为MB:
operation_type="unit_convert", source_value=1024, source_type="KB", target_type="MB"
4. 将3600秒转换为小时:
operation_type="unit_convert", source_value=3600, source_type="s", target_type="h"
"""
try:
# -------------------------- 1. 时区操作:查询指定时区当前时间 --------------------------
if operation_type == "timezone_current":
if not timezone:
return "❌ 缺少参数:查询当前时区时间需要指定timezone(如'America/New_York')"
# 校验时区有效性
if timezone not in pytz.all_timezones:
return f"❌ 无效时区:{timezone}\n可选时区示例:UTC、Asia/Shanghai、America/New_York、Europe/London"
# 获取指定时区当前时间
tz = pytz.timezone(timezone)
current_time = datetime.datetime.now(tz).strftime("%Y-%m-%d %H:%M:%S")
return f"✅ 当前时间查询结果:\n时区:{timezone}\n时间:{current_time}"
# -------------------------- 2. 时区操作:时间跨时区转换 --------------------------
elif operation_type == "timezone_convert":
# 校验必填参数
if not all([source_time, source_type, target_type]):
return "❌ 缺少参数:时区转换需要source_time(YYYY-MM-DD HH:MM:SS)、source_type(源时区)、target_type(目标时区)"
# 校验时区有效性
if source_type not in pytz.all_timezones:
return f"❌ 无效源时区:{source_type}"
if target_type not in pytz.all_timezones:
return f"❌ 无效目标时区:{target_type}"
# 解析源时间
try:
source_tz = pytz.timezone(source_type)
source_datetime = source_tz.localize(datetime.datetime.strptime(source_time, "%Y-%m-%d %H:%M:%S"))
except ValueError:
return f"❌ 源时间格式错误:请使用'YYYY-MM-DD HH:MM:SS'(如2026-01-30 10:00:00)"
# 转换到目标时区
target_tz = pytz.timezone(target_type)
target_datetime = source_datetime.astimezone(target_tz)
target_time_str = target_datetime.strftime("%Y-%m-%d %H:%M:%S")
return (
f"✅ 时区转换结果:\n"
f"源时区:{source_type} | 源时间:{source_time}\n"
f"目标时区:{target_type} | 目标时间:{target_time_str}"
)
# -------------------------- 3. 单位转换:编程高频单位互转 --------------------------
elif operation_type == "unit_convert":
# 校验必填参数
if not all([source_value, source_type, target_type]):
return "❌ 缺少参数:单位转换需要source_value(数值)、source_type(源单位)、target_type(目标单位)"
# 校验数值有效性
try:
source_value = float(source_value)
except (ValueError, TypeError):
return f"❌ 无效数值:source_value必须是数字(如1024、3.5)"
# 匹配转换类型(字节/时间)
convert_type = None
for key in UNIT_CONVERSION_MAP:
if source_type in UNIT_CONVERSION_MAP[key] and target_type in UNIT_CONVERSION_MAP[key]:
convert_type = key
break
if not convert_type:
return (
f"❌ 不支持的单位组合:源单位{source_type}、目标单位{target_type}\n"
f"支持的单位:\n"
f"- 字节单位:B、KB、MB、GB、TB、bit\n"
f"- 时间单位:s(秒)、ms(毫秒)、min(分钟)、h(小时)、d(天)"
)
# 执行单位转换
source_ratio = UNIT_CONVERSION_MAP[convert_type][source_type]
target_ratio = UNIT_CONVERSION_MAP[convert_type][target_type]
target_value = source_value * source_ratio / target_ratio
return (
f"✅ 单位转换结果:\n"
f"{source_value} {source_type} = {target_value:.4f} {target_type}\n"
f"(转换基准:1{convert_type} = {UNIT_CONVERSION_MAP[convert_type]})"
)
# -------------------------- 无效操作类型 --------------------------
else:
return (
f"❌ 无效操作类型:{operation_type}\n"
f"支持的操作类型:\n"
f"- timezone_current:查询指定时区当前时间\n"
f"- timezone_convert:跨时区时间转换\n"
f"- unit_convert:单位转换"
)
except Exception as e:
return f"❌ 工具执行异常:{str(e)}"
# Below is an example of a tool that does nothing. Amaze us with your creativity !
@tool
def my_custom_tool(arg1:str, arg2:int)-> str: #it's import to specify the return type
#Keep this format for the description / args / args description but feel free to modify the tool
"""A tool that does nothing yet
Args:
arg1: the first argument
arg2: the second argument
"""
return "What magic will you build ?"
@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:
# Create timezone object
tz = pytz.timezone(timezone)
# Get current time in that timezone
local_time = datetime.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()
# 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, timezone_unit_converter,system_info_query,calculator_tool,crawl_gold_price], ## 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()