kg_agent/knowledge_v2/tools_library.py

'''
工具库模块，提供工具库中的工具调用、保存新的待接入工具的信息
分为两个函数：

1. 调用工具库中的工具
curl --location 'http://47.84.182.56:8001/tools/call/wechat_search_article' \
--header 'Content-Type: application/json' \
--data '{
    "keyword": "英雄联盟"
}'
其中的data和wechat_search_article是函数的入参，把工具名替换到wechat_search_article，把参数字典替换到data


2. 保存新的待接入工具的信息
入参为工具信息，是一个工具的文档字符串
目前默认将这个文档字符串保存到一个文件中，文件名默认是工具的名称
'''

import requests
import os
import json

TOOL_SERVER_URL = "http://47.84.182.56:8001/tools/call"

def call_tool(tool_name: str, arguments: dict):
    """
    调用工具库中的工具
    :param tool_name: 工具名称
    :param arguments: 工具参数字典
    :return: 工具调用结果
    """
    url = f"{TOOL_SERVER_URL}/{tool_name}"
    headers = {
        'Content-Type': 'application/json'
    }
    try:
        response = requests.post(url, headers=headers, json=arguments)
        response.raise_for_status()
        return response.json()
    except requests.RequestException as e:
        # 在实际生产中可能需要更复杂的错误处理或日志记录
        return {"error": f"Failed to call tool {tool_name}: {str(e)}"}

def save_tool_info(tool_name: str, tool_doc: str):
    """
    保存新的待接入工具的信息
    :param tool_name: 工具名称
    :param tool_doc: 工具文档字符串
    :return: 保存的文件路径
    """
    # 获取当前文件所在目录
    current_dir = os.path.dirname(os.path.abspath(__file__))
    # 创建 tool_infos 目录（如果不存在）
    save_dir = os.path.join(current_dir, 'tool_infos')
    if not os.path.exists(save_dir):
        os.makedirs(save_dir)
    
'''
工具库模块，提供工具库中的工具调用、保存新的待接入工具的信息
分为两个函数：

1. 调用工具库中的工具
curl --location 'http://47.84.182.56:8001/tools/call/wechat_search_article' \
--header 'Content-Type: application/json' \
--data '{
    "keyword": "英雄联盟"
}'
其中的data和wechat_search_article是函数的入参，把工具名替换到wechat_search_article，把参数字典替换到data


2. 保存新的待接入工具的信息
入参为工具信息，是一个工具的文档字符串
目前默认将这个文档字符串保存到一个文件中，文件名默认是工具的名称
'''

import requests
import os
import json

TOOL_SERVER_URL = "http://47.84.182.56:8001/tools/call"

def call_tool(tool_name: str, arguments: dict):
    """
    调用工具库中的工具
    :param tool_name: 工具名称
    :param arguments: 工具参数字典
    :return: 工具调用结果
    """
    url = f"{TOOL_SERVER_URL}/{tool_name}"
    headers = {
        'Content-Type': 'application/json'
    }
    try:
        response = requests.post(url, headers=headers, json=arguments)
        response.raise_for_status()
        return response.json()
    except requests.RequestException as e:
        # 在实际生产中可能需要更复杂的错误处理或日志记录
        return {"error": f"Failed to call tool {tool_name}: {str(e)}"}

def save_tool_info(tool_name: str, tool_doc: str):
    """
    保存新的待接入工具的信息
    :param tool_name: 工具名称
    :param tool_doc: 工具文档字符串
    :return: 保存的文件路径
    """
    # 获取当前文件所在目录
    current_dir = os.path.dirname(os.path.abspath(__file__))
    # 创建 tool_infos 目录（如果不存在）
    save_dir = os.path.join(current_dir, 'tool_infos')
    if not os.path.exists(save_dir):
        os.makedirs(save_dir)
    
    file_path = os.path.join(save_dir, f"{tool_name}.txt")
    
    try:
        with open(file_path, 'w', encoding='utf-8') as f:
            f.write(tool_doc)
        return file_path
    except IOError as e:
        return f"Error saving tool info: {str(e)}"

def get_all_tool_infos() -> str:
    """
    获取所有已保存工具的信息
    :return: 所有工具信息的拼接字符串
    """
    current_dir = os.path.dirname(os.path.abspath(__file__))
    save_dir = os.path.join(current_dir, 'tool_infos')
    
    if not os.path.exists(save_dir):
        return ""
    
    tool_infos = []
    try:
        for filename in os.listdir(save_dir):
            if filename.endswith(".txt"):
                file_path = os.path.join(save_dir, filename)
                with open(file_path, 'r', encoding='utf-8') as f:
                    content = f.read().strip()
                    tool_infos.append(f"--- Tool: {filename[:-4]} ---\n{content}\n")
    except Exception as e:
        return f"Error reading tool infos: {str(e)}"
        
    return "\n".join(tool_infos)

def get_tool_info(tool_name: str) -> str:
    """
    获取指定工具的信息
    :param tool_name: 工具名称
    :return: 工具信息文本，如果不存在返回空字符串
    """
    current_dir = os.path.dirname(os.path.abspath(__file__))
    save_dir = os.path.join(current_dir, 'tool_infos')
    
    if not os.path.exists(save_dir):
        return ""
    
    file_path = os.path.join(save_dir, f"{tool_name}.txt")
    if not os.path.exists(file_path):
        return ""
    
    try:
        with open(file_path, 'r', encoding='utf-8') as f:
            return f.read().strip()
    except Exception as e:
        return f"Error reading tool info: {str(e)}"