引言

本月，魔搭社区推出了开源版GPTs，让所有用户都能更轻松地搭建Agent。社区也推出了 0代码快速创建发布一个专属Agent、低代码调用API创建一个炫酷功能的Agent两个教程，为大家讲解如何使用AgentFabric通过配置、对话的方式来创建自己的Agent，以及通过openapi schema来配置外部API扩展Agent的能力。但openapi schema的方式对于复杂接口的调用、以及复杂逻辑的串联会比较困难，因此我们推出了今天的文章，介绍ModelScope Agent的function call的能力，让大家通过写python代码的方式来定制自己的tool，进一步扩展Agent的能力。

API快速注册为local tool流程

  1. 初始化， 传入一个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，或者其他的自定义参数

 

  1. 工具执行逻辑定制

以下是一个简单的工具定制示例代码

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

 

也欢迎加入钉钉群交流：

 

Logo
欢迎加入ModelScope魔搭中文开源社区

ModelScope旨在打造下一代开源的模型即服务共享平台，为泛AI开发者提供灵活、易用、低成本的一站式模型服务产品，让模型应用更简单！

更多推荐

cover

AI电影从这个LoRA开始：魔搭AIGC1月赛题公布&12月赛题获奖作品新鲜出炉

avatar ModelScope魔搭社区
cover

智谱发布GLM-OS概念及Agent产品，CogAgent-9B模型开源助力GUI交互场景

avatar ModelScope魔搭社区
cover

人人都是音乐家！中科大&科大讯飞重磅开源OpenMusic：音乐生成更高质量，更有乐感

avatar ModelScope魔搭社区