import os import shutil import traceback import json from modelscope_agent.tools.openapi_plugin import openapi_schema_convert from modelscope_agent.utils.logger import agent_logger as logger from modelscope.utils.config import Config DEFAULT_AGENT_DIR = '/tmp/agentfabric' DEFAULT_BUILDER_CONFIG_DIR = os.path.join(DEFAULT_AGENT_DIR, 'config') DEFAULT_BUILDER_CONFIG_FILE = os.path.join(DEFAULT_BUILDER_CONFIG_DIR, 'builder_config.json') DEFAULT_OPENAPI_PLUGIN_CONFIG_FILE = os.path.join( DEFAULT_BUILDER_CONFIG_DIR, 'openapi_plugin_config.json') DEFAULT_MODEL_CONFIG_FILE = './config/model_config.json' DEFAULT_TOOL_CONFIG_FILE = './config/tool_config.json' DEFAULT_CODE_INTERPRETER_DIR = os.getenv('CODE_INTERPRETER_WORK_DIR', '/tmp/ci_workspace') def get_user_dir(uuid_str=''): return os.path.join(DEFAULT_BUILDER_CONFIG_DIR, uuid_str) def get_ci_dir(): return DEFAULT_CODE_INTERPRETER_DIR def get_user_cfg_file(uuid_str=''): builder_cfg_file = os.getenv('BUILDER_CONFIG_FILE', DEFAULT_BUILDER_CONFIG_FILE) # convert from ./config/builder_config.json to ./config/user/builder_config.json builder_cfg_file = builder_cfg_file.replace('config/', 'config/user/') # convert from ./config/user/builder_config.json to ./config/uuid/builder_config.json if uuid_str != '': builder_cfg_file = builder_cfg_file.replace('user', uuid_str) return builder_cfg_file def get_user_openapi_plugin_cfg_file(uuid_str=''): openapi_plugin_cfg_file = os.getenv('OPENAPI_PLUGIN_CONFIG_FILE', DEFAULT_OPENAPI_PLUGIN_CONFIG_FILE) openapi_plugin_cfg_file = openapi_plugin_cfg_file.replace( 'config/', 'config/user/') if uuid_str != '': openapi_plugin_cfg_file = openapi_plugin_cfg_file.replace( 'user', uuid_str) return openapi_plugin_cfg_file def save_builder_configuration(builder_cfg, uuid_str=''): builder_cfg_file = get_user_cfg_file(uuid_str) if uuid_str != '' and not os.path.exists( os.path.dirname(builder_cfg_file)): os.makedirs(os.path.dirname(builder_cfg_file)) with open(builder_cfg_file, 'w', encoding='utf-8') as f: f.write(json.dumps(builder_cfg, indent=2, ensure_ascii=False)) def is_valid_plugin_configuration(openapi_plugin_cfg): if 'schema' in openapi_plugin_cfg: schema = openapi_plugin_cfg['schema'] if isinstance(schema, dict): return True else: return False def save_plugin_configuration(openapi_plugin_cfg, uuid_str): openapi_plugin_cfg_file = get_user_openapi_plugin_cfg_file(uuid_str) if uuid_str != '' and not os.path.exists( os.path.dirname(openapi_plugin_cfg_file)): os.makedirs(os.path.dirname(openapi_plugin_cfg_file)) with open(openapi_plugin_cfg_file, 'w', encoding='utf-8') as f: f.write(json.dumps(openapi_plugin_cfg, indent=2, ensure_ascii=False)) def get_avatar_image(bot_avatar, uuid_str=''): user_avatar_path = os.path.join( os.path.dirname(__file__), 'assets/user.jpg') bot_avatar_path = os.path.join(os.path.dirname(__file__), 'assets/bot.jpg') if len(bot_avatar) > 0: bot_avatar_path = os.path.join(DEFAULT_BUILDER_CONFIG_DIR, uuid_str, bot_avatar) if uuid_str != '': # use default if not exists if not os.path.exists(bot_avatar_path): # create parents directory os.makedirs(os.path.dirname(bot_avatar_path), exist_ok=True) # copy the template to the address temp_bot_avatar_path = os.path.join(DEFAULT_BUILDER_CONFIG_DIR, bot_avatar) if not os.path.exists(temp_bot_avatar_path): # fall back to default local avatar image temp_bot_avatar_path = os.path.join('./config', bot_avatar) if not os.path.exists(temp_bot_avatar_path): temp_bot_avatar_path = os.path.join( './config', 'custom_bot_avatar.png') shutil.copy(temp_bot_avatar_path, bot_avatar_path) return [user_avatar_path, bot_avatar_path] def save_avatar_image(image_path, uuid_str=''): bot_avatar = os.path.basename(image_path) bot_avatar_path = os.path.join(DEFAULT_BUILDER_CONFIG_DIR, uuid_str, bot_avatar) shutil.copy(image_path, bot_avatar_path) return bot_avatar, bot_avatar_path def parse_configuration(uuid_str=''): """parse configuration Args: Returns: dict: parsed configuration """ model_cfg_file = os.getenv('MODEL_CONFIG_FILE', DEFAULT_MODEL_CONFIG_FILE) builder_cfg_file = get_user_cfg_file(uuid_str) # use default if not exists if not os.path.exists(builder_cfg_file): # create parents directory os.makedirs(os.path.dirname(builder_cfg_file), exist_ok=True) # copy the template to the address builder_cfg_file_temp = './config/builder_config.json' if builder_cfg_file_temp != builder_cfg_file: shutil.copy(builder_cfg_file_temp, builder_cfg_file) tool_cfg_file = os.getenv('TOOL_CONFIG_FILE', DEFAULT_TOOL_CONFIG_FILE) builder_cfg = Config.from_file(builder_cfg_file) model_cfg = Config.from_file(model_cfg_file) tool_cfg = Config.from_file(tool_cfg_file) tools_info = builder_cfg.tools available_tool_list = [] for key, value in tools_info.items(): if value['use']: available_tool_list.append(key) tool_cfg[key]['use'] = value['use'] openapi_plugin_file = get_user_openapi_plugin_cfg_file(uuid_str) plugin_cfg = {} available_plugin_list = [] if os.path.exists(openapi_plugin_file): openapi_plugin_cfg = Config.from_file(openapi_plugin_file) try: config_dict = openapi_schema_convert( schema=openapi_plugin_cfg.schema, auth=openapi_plugin_cfg.auth.to_dict()) plugin_cfg = Config(config_dict) for name, config in config_dict.items(): available_plugin_list.append(name) except Exception as e: logger.error( uuid=uuid_str, error=str(e), content={ 'error_traceback': traceback.format_exc(), 'error_details': 'The format of the plugin config file is incorrect.' }) return builder_cfg, model_cfg, tool_cfg, available_tool_list, plugin_cfg, available_plugin_list