From 8415cb50e8320fba85bbee8051b50f67756fa029 Mon Sep 17 00:00:00 2001 From: yzy_dev Date: Tue, 24 Dec 2024 10:54:08 +0800 Subject: [PATCH] =?UTF-8?q?add=20function=20calling:rpm=5Fall=EF=BC=8Crepo?= =?UTF-8?q?s=EF=BC=8Cdisk=5Fuse?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../llm_tools/function_call/tools/os_tools.py | 44 +++++++- .../function_call/tools/rpm_tools.py | 103 ++++++++++++++++++ 2 files changed, 144 insertions(+), 3 deletions(-) diff --git a/llmops/llm_tools/function_call/tools/os_tools.py b/llmops/llm_tools/function_call/tools/os_tools.py index 6d6c3ee9..0aeb96ac 100644 --- a/llmops/llm_tools/function_call/tools/os_tools.py +++ b/llmops/llm_tools/function_call/tools/os_tools.py @@ -106,9 +106,10 @@ class OsFunctionTools: print(f"Failed to get cpu info. Status code: {response.status_code}") return None + @staticmethod - def memory_info(UUID: str): - url = f"{OsFunctionTools.BASE_URL}/api/v1/api/memory_info" + def disk_use(UUID: str): + url = f"{OsFunctionTools.BASE_URL}/api/v1/disk_use" params = { "uuid": UUID } @@ -117,8 +118,24 @@ class OsFunctionTools: data = response.json() return data else: - print(f"Failed to get memory info. Status code: {response.status_code}") + print(f"Failed to get disk use. Status code: {response.status_code}") return None + + # @staticmethod + # def disk_info(UUID: str): + # url = f"{OsFunctionTools.BASE_URL}/api/v1/disk_info" + # params = { + # "uuid": UUID + # } + # response = requests.get(url, params=params) + # if response.status_code == 200: + # data = response.json() + # return data + # else: + # print(f"Failed to get disk info. Status code: {response.status_code}") + # return None + + # @staticmethod 可以用一个方法调用以上方法，但未知是否需要处理不同接口返回的数据 # def fetch_data(endpoint: str, params=None): # # 定义目标URL @@ -236,6 +253,27 @@ os_tools_info = { ) }, "func": OsFunctionTools.memory_info + }, + "DiskUse": { + "tool_name": "getDiskUse", + "desc": ( + "此方法用于获取指定 UUID 客户端的磁盘使用情况。\n" + "返回的数据包含多个磁盘分区的使用情况，每个分区的信息包含以下字段：\n" + "1. device：设备名称，例如 '/dev/dm-0'。\n" + "2. path：挂载路径，例如 '/'。\n" + "3. fstype：文件系统类型，例如 'ext2/ext3'。\n" + "4. total：磁盘总容量，例如 '33G'。\n" + "5. used：已使用的磁盘空间，例如 '8G'。\n" + "6. usedPercent：已使用空间的百分比，例如 '26%'。\n" + "如果方法调用失败，请告知用户错误原因。" + ), + "args": { + "UUID": ( + "str", + "[*Required] UUID string identifies the client whose disk usage information is being queried." + ) + }, + "func": OsFunctionTools.disk_use } } diff --git a/llmops/llm_tools/function_call/tools/rpm_tools.py b/llmops/llm_tools/function_call/tools/rpm_tools.py index 3921495c..00d5dc4d 100644 --- a/llmops/llm_tools/function_call/tools/rpm_tools.py +++ b/llmops/llm_tools/function_call/tools/rpm_tools.py @@ -1,12 +1,115 @@ +import requests + from llmops.config.config import init_config +from llmops.llm_tools.function_call.tools.os_tools import OsFunctionTools class RpmFunctionTools: config = init_config() BASE_URL = config.app_conf.tool_baseurl + @staticmethod + def rpm_all(UUID: str): + url = f"{OsFunctionTools.BASE_URL}/api/v1/rpm_all" + params = { + "uuid": UUID + } + response = requests.get(url, params=params) + if response.status_code == 200: + data = response.json() + return data + else: + print(f"Failed to get rpm all. Status code: {response.status_code}") + return None + + + # @staticmethod + # def rpm_source(UUID: str): + # url = f"{OsFunctionTools.BASE_URL}/api/v1/rpm_source" + # params = { + # "uuid": UUID + # } + # response = requests.get(url, params=params) + # if response.status_code == 200: + # data = response.json() + # return data + # else: + # print(f"Failed to get rpm source. Status code: {response.status_code}") + # return None + + @staticmethod + def repos(UUID: str): + url = f"{OsFunctionTools.BASE_URL}/api/v1/repos" + params = { + "uuid": UUID + } + response = requests.get(url, params=params) + if response.status_code == 200: + data = response.json() + return data + else: + print(f"Failed to get repos. Status code: {response.status_code}") + return None + + # @staticmethod + # def rpm_info(UUID: str, package_name: str): + # url = f"{OsFunctionTools.BASE_URL}/api/v1/rpm_info" + # params = { + # "uuid": UUID, + # "package_name": package_name + # } + # response = requests.get(url, params=params) + # if response.status_code == 200: + # data = response.json() + # return data + # else: + # print(f"Failed to get rpm info. Status code: {response.status_code}") + # return None + rpm_tools_info = { + "RpmAll": { + "tool_name": "getRpmAll", + "desc": ( + "此方法用于获取指定 UUID 客户端安装的所有 RPM 包列表。\n" + "返回一个包含所有已安装 RPM 包的字符串列表，每个包的格式为：\n" + "'包名-版本号-架构'，例如 'tar-1.35-2.oe2403.aarch64'。\n" + "如果方法调用失败，请告知用户错误原因。" + ), + "args": { + "UUID": ( + "str", + "[*Required] UUID string identifies the client whose RPM package list is being queried." + ) + }, + "func": RpmFunctionTools.rpm_all + }, + +" Repos": { + "tool_name": "getRepos", + "desc": ( + "此方法用于获取指定 UUID 客户端的仓库（repos）信息，包括多个仓库的配置。\n" + "返回的数据包含多个仓库的信息，每个仓库包含以下字段：\n" + "1. File：仓库文件名。\n" + "2. ID：仓库 ID。\n" + "3. Name：仓库名称。\n" + "4. MirrorList：镜像列表（如果有）。\n" + "5. BaseURL：仓库的基本 URL。\n" + "6. MetaLink：仓库的元数据链接。\n" + "7. MetadataExpire：元数据过期时间。\n" + "8. GPGCheck：是否启用 GPG 检查。\n" + "9. GPGKey：仓库的 GPG 密钥 URL。\n" + "10. Enabled：仓库是否启用（1 表示启用，0 表示禁用）。\n" + "如果方法调用失败，请告知用户错误原因。" + ), + "args": { + "UUID": ( + "str", + "[*Required] UUID string identifies the client whose repository information is being queried." + ) + }, + "func": RpmFunctionTools.repos + } } -- Gitee