Spaces:
Sleeping
Sleeping
| 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 | |
| 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 | |
| } | |
| 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)}" | |
| 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 | |
| } | |
| } | |
| 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 ! | |
| 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 ?" | |
| 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() |