引导
在本文中，我们将展示如何将一个包含异步的api接口封装成tool并被agent在chat过程中调用执行的过程，具体包括如下几步：
-
了解Agent原理, Modelscope-Agent框架
-
测试并了解API特性
-
将API封装到一个tool里面，并测试效果
-
通过agent的方式进行调用
-
agentfabric中使用该API
本文添加的api接口即为涂鸦作画
涂鸦作画能力
这是一种图生图能力，一张草图搭配一段描述生成内容丰富的图片能力。
举例说明譬如：
选用相关接口采用 dashscope的涂鸦作画接口（https://help.aliyun.com/zh/dashscope/developer-reference/tongyi-wanxiang-api-for-doodle?spm=a2c4g.11186623.0.0.4e534393H1eB3n）
环境准备
环境：
-
系统：任何系统均可，推荐使用魔搭镜像
-
LLM: Dashscope 上的qwen-max（需要申请账号有免费token使用）
-
框架：Modelscope-Agent
其他：
可选参考图：http://synthesis-source.oss-accelerate.aliyuncs.com/lingji/datasets/QuickDraw_sketches_final/cat/4503626191994880.png
框架熟悉
-
熟悉Agent
-
熟悉Modelscope-Agent代码框架：
https://github.com/modelscope/modelscope-agent
接口实验
-
异步请求生成图片
curl --location 'https://dashscope.aliyuncs.com/api/v1/services/aigc/image2image/image-synthesis' \
--header 'X-DashScope-Async: enable' \
--header 'Authorization: Bearer <your dashscope api token>' \
--header 'Content-Type: application/json' \
--header 'X-DashScope-OssResourceResolve: enable' \
--data '{
"input": {
"prompt": "绿色的猫",
"sketch_image_url": "http://synthesis-source.oss-accelerate.aliyuncs.com/lingji/datasets/QuickDraw_sketches_final/cat/4503626191994880.png"
},
"model": "wanx-sketch-to-image-lite"
}'
返回
{"output":{"task_status":"PENDING","task_id":"76a71d5b-8fc5-4d47-8ef8-c16af80951f3"},"request_id":"1ad6a3f4-8a80-9118-b805-4515376a9404"}
-
状态查询
curl -X GET \
--header 'Authorization: Bearer <your dashscope api token>' \
https://dashscope.aliyuncs.com/api/v1/tasks/76a71d5b-8fc5-4d47-8ef8-c16af80951f3
返回
{"request_id":"5441c445-ec10-963e-9c74-8907e507d1e2","output":{"task_id":"76a71d5b-8fc5-4d47-8ef8-c16af80951f3","task_status":"SUCCEEDED","submit_time":"2024-07-02 23:07:03.292","scheduled_time":"2024-07-02 23:07:03.317","end_time":"2024-07-02 23:07:15.401","results":[{"url":"https://dashscope-result-hz.oss-cn-hangzhou.aliyuncs.com/1d/db/20240702/96f6710c/0b3c9685-1683-4843-87b6-f0ce9bfe8972-1.png?Expires=1720019235&OSSAccessKeyId=LTAI5tQZd8AEcZX6KZV4G8qL&Signature=kdVTIwCb9OTr6V0vTRnnqWqpt4Q%3D"}],"task_metrics":{"TOTAL":1,"SUCCEEDED":1,"FAILED":0}},"usage":{"image_count":1}}
新建工具到Modelscope-Agent
-
注册新工具链路解读:
-
register_tool：用于框架层面注册tool，并唯一标识名字
-
description，name 以及parameters对齐 openai的tool calling格式，方便tool args的生成
-
call 具体执行 tool的入口function
-
import os
import time
import json
import requests
from modelscope_agent.constants import BASE64_FILES, LOCAL_FILE_PATHS, ApiNames
from modelscope_agent.tools.base import BaseTool, register_tool
from modelscope_agent.utils.utils import get_api_key, get_upload_url
from requests.exceptions import RequestException, Timeout
MAX_RETRY_TIMES = 3
WORK_DIR = os.getenv('CODE_INTERPRETER_WORK_DIR', '/tmp/ci_workspace')
@register_tool('sketch_to_image')
class SketchToImage(BaseTool):
description = '调用sketch_to_image api通过图片加文本生成图片'
name = 'sketch_to_image'
parameters: list = [{
'name': 'input.sketch_image_url',
'description': '用户上传的照片的相对路径',
'required': True,
'type': 'string'
}, {
'name': 'input.prompt',
'description': '详细描述了希望生成的图像具有什么样的特点',
'required': True,
'type': 'string'
}]
def call(self, params: str, **kwargs) -> str:
pass
-
添加涂鸦作画能力核心链路到tool：
a. 解析入参，生成图片url
def _parse_input(self, *args, **kwargs):
kwargs = super()._parse_files_input(*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:
# if the key does not contain ".", directly store the key-value pair into restored_dict
restored_dict[key] = value
kwargs = restored_dict
image_path = kwargs['input'].pop('sketch_image_url', None)
if image_path and image_path.endswith(('.jpeg', '.png', '.jpg')):
# 生成 image_url，然后设置到 kwargs['input'] 中
# 复用dashscope公共oss
if LOCAL_FILE_PATHS not in kwargs:
image_path = f'file://{os.path.join(WORK_DIR,image_path)}'
else:
image_path = f'file://{kwargs["local_file_paths"][image_path]}'
image_url = get_upload_url(
model=
'wanx-sketch-to-image-lite',
file_to_upload=image_path,
api_key=os.environ.get('DASHSCOPE_API_KEY', ''))
kwargs['input']['sketch_image_url'] = image_url
else:
raise ValueError('请先上传一张正确格式的图片')
kwargs['model'] = 'wanx-sketch-to-image-lite'
print('草图生图的tool参数：', kwargs)
return kwargs
b.调用异步请求生成接口
def call(self, params: str, **kwargs) -> str:
params = self._verify_args(params)
if isinstance(params, str):
return 'Parameter Error'
if BASE64_FILES in kwargs:
params[BASE64_FILES] = kwargs[BASE64_FILES]
remote_parsed_input = self._parse_input(**params)
remote_parsed_input = json.dumps(remote_parsed_input)
url = kwargs.get(
'url',
'https://dashscope.aliyuncs.com/api/v1/services/aigc/image2image/image-synthesis'
)
try:
self.token = get_api_key(ApiNames.dashscope_api_key, **kwargs)
except AssertionError:
raise ValueError('Please set valid DASHSCOPE_API_KEY!')
retry_times = MAX_RETRY_TIMES
headers = {
'Content-Type': 'application/json',
'Authorization': f'Bearer {self.token}',
'X-DashScope-Async': 'enable'
}
# 解析oss
headers['X-DashScope-OssResourceResolve'] = 'enable'
while retry_times:
retry_times -= 1
try:
response = requests.request(
'POST', url=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.final_result = origin_result
return self._get_dashscope_image_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.'
)
c.调用同步请求结果轮训接口
def _get_dashscope_result(self):
if 'task_id' in self.final_result['output']:
task_id = self.final_result['output']['task_id']
get_url = f'https://dashscope.aliyuncs.com/api/v1/tasks/{task_id}'
get_header = {'Authorization': f'Bearer {self.token}'}
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 = origin_result
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_dashscope_image_result(self):
try:
result = self._get_dashscope_result()
while True:
result_data = result
output = result_data.get('output', {})
task_status = output.get('task_status', '')
if task_status == 'SUCCEEDED':
print('任务已完成')
# 取出result里url的部分，提高url图片展示稳定性
output_url = result['output']['results'][0]['url']
return f''
elif task_status == 'FAILED':
raise Exception(output.get('message', '任务失败，请重试'))
else:
# 继续轮询，等待一段时间后再次调用
time.sleep(0.5) # 等待 0.5 秒钟
result = self._get_dashscope_result()
print(f'Running:{result}')
except Exception as e:
raise Exception('get Remote Error:', str(e))
-
测试用例确保功能完善
import os
import pytest
from modelscope_agent.tools.dashscope_tools.sketch_to_image import SketchToImage
from modelscope_agent.agents.role_play import RolePlay # NOQA
IS_FORKED_PR = os.getenv('IS_FORKED_PR', 'false') == 'true'
@pytest.mark.skipif(IS_FORKED_PR, reason='only run modelscope-agent main repo')
def test_sketch_to_image():
# 图片默认上传到default work_dir
params = """{'input.sketch_image_url': 'sketch.png', 'input.prompt': '绿色的猫'}"""
style_repaint = SketchToImage()
res = style_repaint.call(params)
assert (res.startswith(')
@pytest.mark.skipif(IS_FORKED_PR, reason='only run modelscope-agent main repo')
def test_sketch_to_image_role():
role_template = '你扮演一个绘画家，用尽可能丰富的描述调用工具绘制各种风格的图画。'
llm_config = {'model': 'qwen-max', 'model_server': 'dashscope'}
# input tool args
function_list = ['sketch_to_image']
bot = RolePlay(
function_list=function_list, llm=llm_config, instruction=role_template)
response = bot.run('[上传文件 "sketch.png"],我想要一只绿色耳朵带耳环的猫')
text = ''
for chunk in response:
text += chunk
print(text)
assert isinstance(text, str)
-
额外添加到modelscope_agent/tools/base.py 中注册
#
register_map = {
'sketch_to_image':
'SketchToImage',
'amap_weather':
'AMAPWeather',
'storage':
'Storage',
'web_search':
'WebSearch',
'image_gen':
'TextToImageTool',
'image_gen_lite':
'TextToImageLiteTool',
-
类引用路径添加懒加载：modelscope_agent/tools/__init__.py
import sys
from ..utils import _LazyModule
from .contrib import * # noqa F403
_import_structure = {
'amap_weather': ['AMAPWeather'],
'code_interpreter': ['CodeInterpreter'],
'contrib': ['AliyunRenewInstanceTool'],
'dashscope_tools': [
'ImageEnhancement', 'TextToImageTool', 'TextToImageLiteTool',
'ParaformerAsrTool', 'QWenVL', 'SambertTtsTool', 'StyleRepaint',
'WordArtTexture', 'SketchToImage'
],
-
提交代码pull request到代码仓库
在AgentFabric中应用
Agentfabric能力介绍：
应用新增涂鸦作画能力：
-
在代码中修改tool_config.json，确保新tool能够被调用：apps/agentfabric/config/tool_config.json
-
在agentfabric中添加并使用新生成的tool
总结
-
agent的可以通过外部tool，将LLM能力快速扩展到更广泛的应用场景
-
tool的正确调用依赖LLM的指令理解和指令生成，因此需要正确的定义指令
-
tool的调用在agent内部是同步的，因此对于一些调用异步接口需要等待的场景可以有多种办法，一种是在单步调用tool的时候等待结果产生并抛出结果，还有就是通过多个tool来让agent帮助你去做结果的查询。
已为社区贡献539条内容
所有评论(0)