引言
本月，魔搭社区推出了开源版GPTs，让所有用户都能更轻松地搭建Agent。社区也推出了 0代码快速创建发布一个专属Agent、低代码调用API创建一个炫酷功能的Agent两个教程，为大家讲解如何使用AgentFabric通过配置、对话的方式来创建自己的Agent，以及通过openapi schema来配置外部API扩展Agent的能力。但openapi schema的方式对于复杂接口的调用、以及复杂逻辑的串联会比较困难，因此我们推出了今天的文章，介绍ModelScope Agent的function call的能力，让大家通过写python代码的方式来定制自己的tool，进一步扩展Agent的能力。
API快速注册为local tool流程
- 初始化， 传入一个cfg参数，是一个python dict
class Tool: def __init__(self, cfg={}): pass
cfg是一个全局的工具配置，可以根据当前工具自身的名称获取的对应的工具信息配置，保存到自定义工具类中，其中use字段为true表示开启工具，否则表示关闭工具。
"image_gen": { "url": "https://api-inference.modelscope.cn/api-inference/v1/models/AI-ModelScope/stable-diffusion-xl-base-1.0", "use": true, "pipeline_params": { "use_safetensors": true } }, "amap_weather": { "use": false, "token": "need to be filled when you use weather" }
用户可以根据自己的开发需要进行相关参数传递，例如工具需要调用的url， 以及对应的token，或者其他的自定义参数
- 工具执行逻辑定制
以下是一个简单的工具定制示例代码
from .tool import Tool class AliyunRenewInstanceTool(Tool): description = '续费一台包年包月ECS实例' name = 'RenewInstance' parameters: list = [{ 'name': 'instance_id', 'description': 'ECS实例ID', 'required': True }, { 'name': 'period', 'description': '续费时长以月为单位', 'required': True } ] def __call__(self, remote=False, *args, **kwargs): if self.is_remote_tool or remote: return self._remote_call(*args, **kwargs) else: return self._local_call(*args, **kwargs) def _remote_call(self, *args, **kwargs): pass def _local_call(self, *args, **kwargs): instance_id = kwargs['instance_id'] period = kwargs['period'] return {'result': f'已完成ECS实例ID为{instance_id}的续费，续费时长{period}月'}
如上我们可以去重写_remote_call函数、_local_call函数，去实现工具的本地执行逻辑和远端执行逻辑，这个主要的考虑是模型本地推理和远程模型服务调用上的区分。 如果是非模型相关的工具，用户也可以直接重写__call__函数完成工具调用逻辑的实现即可。
详细的开发文档可以参考：https://github.com/modelscope/modelscope-agent/blob/master/docs/modules/tool.md
案例：艺术字纹理生成
1.在modelscope_agent/tools目录下新建一个py文件，以艺术字生成api为例，比如wordart_tool.py。
2.编辑wordart_tool.py:
import os import pandas as pd import json import requests from modelscope_agent.tools.tool import Tool, ToolSchema from pydantic import ValidationError from requests.exceptions import RequestException, Timeout import time MAX_RETRY_TIMES = 3 class WordArtTexture(Tool):# 继承基础类Tool，新建一个继承类 description = '生成艺术字纹理图片'# 对这个tool的功能描述 name = 'wordart_texture_generation'#tool name """ parameters是需要传入api tool的参数，通过api详情获取需要哪些必要入参 其中每一个参数都是一个字典，包含name，description，required三个字段 当api详情里的入参是一个嵌套object时，写成如下这种用'.'连接的格式。 """ parameters: list = [{ 'name': 'input.text.text_content', 'description': 'text that the user wants to convert to WordArt', 'required': True }, { 'name': 'input.prompt', 'description': 'Users’ style requirements for word art may be requirements in terms of shape, color, entity, etc.', 'required': True }] def __init__(self, cfg={}): self.cfg = cfg.get(self.name, {})# cfg注册见下一条说明，这里是通过name找到对应的cfg # api url self.url = 'https://dashscope.aliyuncs.com/api/v1/services/aigc/wordart/texture' # api token，可以选择注册在下面的cfg里，也可以选择将'API_TOKEN'导入环境变量 self.token = self.cfg.get('token', os.environ.get('API_TOKEN', '')) assert self.token != '', 'wordart api token must be acquired ' # 验证，转换参数格式，保持即可 try: all_param = { 'name': self.name, 'description': self.description, 'parameters': self.parameters } self.tool_schema = ToolSchema(**all_param) except ValidationError: raise ValueError(f'Error when parsing parameters of {self.name}') self._str = self.tool_schema.model_dump_json() self._function = self.parse_pydantic_model_to_openai_function( all_param) # 调用api操作函数，kwargs里是llm根据上面的parameters说明得到的对应参数 def __call__(self, *args, **kwargs): # 对入参格式调整和补充，比如解开嵌套的'.'连接的参数，还有导入你默认的一些参数， # 比如model，参考下面的_remote_parse_input函数。 remote_parsed_input = json.dumps( self._remote_parse_input(*args, **kwargs)) origin_result = None retry_times = MAX_RETRY_TIMES # 参考api详情，确定headers参数 headers = { 'Content-Type': 'application/json', 'Authorization': f'Bearer {self.token}', 'X-DashScope-Async': 'enable' } while retry_times: retry_times -= 1 try: # requests请求 response = requests.request( 'POST', url=self.url, headers=headers, data=remote_parsed_input) if response.status_code != requests.codes.ok: response.raise_for_status() origin_result = json.loads( response.content.decode('utf-8')) # self._parse_output是基础类Tool对output结果的一个格式调整，你可 # 以在这里按需调整返回格式 self.final_result = self._parse_output( origin_result, remote=True) # 下面是对异步api的额外get result操作，同步api可以直接得到结果的， # 这里返回final_result即可。 """ get_wordart_result()先对final_result解析提取出get需要的参数,然后 替换相应的参数，比如这个案例替换的是task_id，完善get请求的所有参数， 然后再去返回最后get请求的结果，这里还有一个loop去判断任务状态，返回最后 任务成功的结果，按需更改即可。 """ return self.get_wordart_result() except Timeout: continue except RequestException as e: raise ValueError( f'Remote call failed with error code: {e.response.status_code},\ error message: {e.response.content.decode("utf-8")}') raise ValueError( 'Remote call max retry times exceeded! Please try to use local call.' ) def _remote_parse_input(self, *args, **kwargs): restored_dict = {} for key, value in kwargs.items(): if '.' in key: # Split keys by "." and create nested dictionary structures keys = key.split('.') temp_dict = restored_dict for k in keys[:-1]: temp_dict = temp_dict.setdefault(k, {}) temp_dict[keys[-1]] = value else: # f the key does not contain ".", directly store the key-value pair into restored_dict restored_dict[key] = value kwargs = restored_dict kwargs['model'] = 'wordart-texture' print('传给tool的参数：', kwargs) return kwargs def get_result(self): result_data = json.loads(json.dumps(self.final_result['result'])) if 'task_id' in result_data['output']: task_id = result_data['output']['task_id'] get_url = f'https://dashscope.aliyuncs.com/api/v1/tasks/{task_id}' get_header = { 'Authorization': f'Bearer {self.token}' } origin_result = None retry_times = MAX_RETRY_TIMES while retry_times: retry_times -= 1 try: response = requests.request( 'GET', url=get_url, headers=get_header ) if response.status_code != requests.codes.ok: response.raise_for_status() origin_result = json.loads( response.content.decode('utf-8')) get_result = self._parse_output( origin_result, remote=True) return get_result except Timeout: continue except RequestException as e: raise ValueError( f'Remote call failed with error code: {e.response.status_code},\ error message: {e.response.content.decode("utf-8")}') raise ValueError( 'Remote call max retry times exceeded! Please try to use local call.' ) def get_wordart_result(self): try: result = self.get_result() print(result) while True: result_data = result.get('result', {}) output = result_data.get('output', {}) task_status = output.get('task_status', '') if task_status == 'SUCCEEDED': print('任务已完成') return result elif task_status == 'FAILED': raise("任务失败") # 继续轮询，等待一段时间后再次调用 time.sleep(1) # 等待 1 秒钟 result = self.get_result() except Exception as e: print('get request Error:', str(e))
3.注册tool init
打开modelscope_agent/tools/__init__.py
from .wordart_tool import WordArtTexture # 根据你的文件名和继承类的名字修改 TOOL_INFO_LIST = { ...... 'amap_weather': 'AMAPWeather', 'wordart_texture_generation': 'WordArtTexture' # 'wordart_texture_generation'是继承类里的tool name， # 'WordArtTexture'是继承类名 }
4.注册cfg
打开apps/agentfabric/config/tool_config.json
{ ...... "amap_weather": { "name": "高德天气", "is_active": true, "use": false }, # "wordart_texture_generation"是继承类里的tool name "wordart_texture_generation": { "name": "艺术字纹理生成", # 自定义展示在前端的工具名 "token": "xxxxx", # 可选，也可以选择将'API_TOKEN'导入环境变量 "is_active": true, "use": false } }
5.运行，create一个对应工具的Agent，比如艺术字生成小助手，然后在configure里勾选对应工具，update
在preview页面开始测试工具调用，无需编写instruction就可一步完成异步等多步骤api动作。
预告：code interpreter
除了API和tool能力之外，Agent还内置了code interpreter，也能实现二维码生成、图表可视化等高级能力，我们将在后续继续推出相关教程及案例，请关注公众号。
Agent大本营，可以看到开发者创建的有趣Agents
https://www.modelscope.cn/brand/view/agent
也欢迎加入钉钉群交流：
跳转查看Agent品牌馆：https://www.modelscope.cn/brand/view/agent
