借助 Assistant API 构建具备自动规划能力的 Multi Agent 系统
更新时间：
本教程简要介绍如何将百炼RAG应用与阿里云OpenAPI集成到Agent中，并利用百炼平台的Assistants API进行Multi Agent开发，最终通过Gradio前端界面展示。
实验简介
通过本教程，您可以了解到将百炼RAG应用与阿里云的OpenAPI能力集成到Agent中的方式，以及使用百炼平台的Assistants API进行Multi Agent开发的步骤，并最终通过基于Gradio的前端界面展示出来。
您可以通过修改Agent中的提示词、细化Agent之间的交互方式、修改工具函数等方法，将Multi Agent应用到您的业务中。
实验室资源方式简介
进入实操前，请确保阿里云账号满足以下条件：
个人账号资源
使用您个人的云资源进行操作，资源归属于个人。
所有实验操作将保留至您的账号，请谨慎操作。
平台仅提供手册参考，不会对资源做任何操作。
- 说明
使用个人账户资源，在创建资源时，可能会产生一定的费用，请您及时关注相关云产品资源的计费概述。
确保已完成云工开物300元代金券领取。
已通过实名认证且账户余额≥0元。
本实验预计费用约为 xx 元（按 xx分钟/小时 且使用默认参数计费），实际费用请以实际账单为准。
本实验产生的费用优先使用优惠券，优惠券使用完毕后需您自行承担。如果您调整了资源规格、使用时长，或执行了本方案以外的操作，可能导致费用发生变化，请以控制台显示的实际价格和最终账单为准。
实操结束后，您可以选择继续付费保留资源，但这将导致持续产生费用，否则请根据实验手册释放资源。
新注册百炼的用户，通义系列模型提供模型的免费额度，不同模型的免费额度不同，请在使用前，先阅读产品文档确认模型的免费额度，或在使用前，登录百炼-模型广场-模型卡片详情，查看具体模型的免费额度。注意在使用过程中的token消耗。模型列表 只有开通百炼服务后才能体验模型的免费额度。
实操结束后，无需对百炼进行注销。
领取300元高校专属权益优惠券
领取300元高校专属权益优惠券（若已领取请跳过）
重要
实验产生的费用优先使用优惠券，优惠券使用完毕后需您自行承担。
实验步骤
详细工作流程：
用户进行提问；
PlannerAssistant接收到用户的提问，并根据用户问题对其它Agent进行选择并编排其运行顺序；
ChatAssistant、AliyunInfoAssistant与InstanceTypeDetailAssistant分别有对应的职责与能力，它们是本教程Multi Agent系统中被PlannerAssistant编排的Agent。编排完成后的Agent组合会接收用户的提问，并按照程序中设计的Multi Agent交互逻辑进行分工合作；
SummaryAssistant将各Agent的输出信息汇总，并结合用户问题对信息进行总结，其输出会作为Multi Agent系统的最终输出。
前期准备
请参考创建AccessKey，获取用户的
ACCESS_KEY_ID和ACCESS_KEY_SECRET。请开通百炼服务并获取API Key。
我们推荐您将
ALIBABA_CLOUD_ACCESS_KEY_ID、ALIBABA_CLOUD_ACCESS_KEY_SECRET和DASHSCOPE_API_KEY配置在环境变量中，以降低泄露风险。环境变量配置方法可参考：配置API Key到环境变量。在本教程中，您可以参考以下命令，根据您的操作系统选择配置环境变量的方法：方法一：
Bash
# 用您的API-KEY与阿里云AK信息进行替换 export DASHSCOPE_API_KEY="YOUR_DASHSCOPE_API_KEY" export ALIBABA_CLOUD_ACCESS_KEY_ID="YOUR_ALIBABA_CLOUD_ACCESS_KEY_ID" export ALIBABA_CLOUD_ACCESS_KEY_SECRET="YOUR_ALIBABA_CLOUD_ACCESS_KEY_SECRET"方法二：
PowerShell
# 用您的API-KEY与阿里云AK信息进行替换 export DASHSCOPE_API_KEY="YOUR_DASHSCOPE_API_KEY" export ALIBABA_CLOUD_ACCESS_KEY_ID="YOUR_ALIBABA_CLOUD_ACCESS_KEY_ID" export ALIBABA_CLOUD_ACCESS_KEY_SECRET="YOUR_ALIBABA_CLOUD_ACCESS_KEY_SECRET"如果您没有创建ECS实例，请前往ECS控制台创建一个按量付费的实例。
重要如果您在完成本实践教程后没有继续使用ECS实例的需求，请及时释放您的ECS资源，避免产生不必要的消费。
请确认您的计算环境中已安装Python。您可以新建一个
requirements.txt文件，将以下内容复制到txt文件中。alibabacloud_tea_openapi alibabacloud_tea_util alibabacloud_openapi_util alibabacloud_ecs20140526 alibabacloud_bssopenapi20171214 dashscope gradio在您保存
requirements.txt文件后，请您在requirements.txt所在目录中运行以下命令安装依赖：pip install -r requirements.txt请您参考最佳实践-基于RAG的官方文档助手文档，在百炼平台创建RAG应用，知识库文件选择PDF格式的阿里云ECS官方文档实例规格族.pdf，并获取其应用ID。
代码实现
您需要准备两个py文件，分别为
tools.py与main.py，这两个文件需要放置在同一目录中。2.1 tools.py（用于定义工具函数）
tools.py主要定义了Agent会使用到的工具函数。tools.py整体代码可见tools.py整体代码。引入依赖
from alibabacloud_tea_util import models as util_models from alibabacloud_ecs20140526.client import Client as Ecs20140526Client from alibabacloud_ecs20140526 import models as ecs_20140526_models from alibabacloud_tea_openapi import models as open_api_models from alibabacloud_bssopenapi20171214.client import Client as BssOpenApi20171214Client from dashscope import Application import os2.1.1 定义工具
tools.py中主要使用到了两个工具类，分别为ECS与Billing。ECS的类方法包含有query_source和call_agent_app，Billing的类方法包含get_balance。您可以参考以下代码，添加您需要的工具类与工具函数以适配您的业务。说明ECS：
ECS有两个类函数：
query_source与call_agent_app。query_source通过阿里云的OpenAPI服务查询指定区域的ECS实例信息。函数输入为RegionID，如cn-hangzhou、cn-beijing、cn-shanghai等；输出包含实例ID、实例规格与价格信息。call_agent_app函数通过集成阿里云ECS官方文档实例规格族.pdf知识库的百炼RAG应用查询实例规格的详细信息，包括vCPU个数、内存大小等指标数据。您需要获取您在百炼创建应用的app_id并在代码中的对应位置进行替换。函数输入参数为ECS实例规格列表，如['ecs.e-c1m1.large', 'ecs.u1-c1m4.xlarge']，输出为RAG应用的回复。需要注意的是，与Agent相似，RAG应用的背后也是基于大模型驱动，因此call_agent_app的运行时间可能较长。ECS类的定义代码如下：
（说明：请用您在百炼平台创建的RAG应用的
app_id替代代码中的app_id。）ECS类的定义代码
class ECS: @classmethod # 输入：地域ID，如cn-hangzhou，cn-beijing等 # 输出：查询ecs实例规格信息，包括实例ID，实例规格，每小时收费。（系统盘默认按照cloud_essd_entry，40GiB） def query_source(cls,RegionID): config = open_api_models.Config( # 从环境变量中获取阿里云的AK信息 access_key_id=os.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID"), access_key_secret=os.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET") ) # 杭州与北京上海等地区的endpoint不同，此处进行判断 if RegionID != 'cn-hangzhou': config.endpoint = f'ecs.{RegionID}.aliyuncs.com' else: config.endpoint = f'ecs-{RegionID}.aliyuncs.com' client = Ecs20140526Client(config) describe_instances_request = ecs_20140526_models.DescribeInstancesRequest(region_id=RegionID) runtime = util_models.RuntimeOptions() # 获得ECS实例信息 response_source = client.describe_instances_with_options(describe_instances_request, runtime).body if len(response_source.instances.instance) == 0: return "您在当前区域无ecs实例" # 初始化要返回的结果 result = "" # 可能有多个实例，因此用for循环遍历所有实例 for i in range(len(response_source.instances.instance)): # 系统盘类型与存储空间，此处默认设为cloud_essd_entry，40GiB system_disk = ecs_20140526_models.DescribePriceRequestSystemDisk( category='cloud_essd_entry', size=40 ) describe_price_request = ecs_20140526_models.DescribePriceRequest( region_id=RegionID, resource_type='instance', instance_type=response_source.instances.instance[i].instance_type, system_disk=system_disk ) response = client.describe_price_with_options(describe_price_request, runtime).body cur_result = f"""实例：{response_source.instances.instance[i].instance_id} 的规格为：{response_source.instances.instance[i].instance_type}， 每个小时的收费为{response.price_info.price.trade_price}元\n""" # 将当前实例的信息添加到返回结果中 result += cur_result return result @classmethod # RAG应用调用 def call_agent_app(cls,InstanceType): if len(InstanceType) == 0: return "您在当前区域无ecs实例" result = "" for i in range(len(InstanceType)): response = Application.call( # 此处填写RAG应用的app_id app_id='xxx', prompt=f'介绍一下{InstanceType[i]}', # 从环境变量中获取Dashscope的API Key api_key=os.getenv("DASHSCOPE_API_KEY")) result += response.output.text return result说明Billing：
Billing类中有一个函数
get_balance。get_balance通过阿里云的OpenAPI服务查询阿里云的余额信息。Billing
class Billing: # 无需输入，返回为阿里云账户余额信息 @classmethod def get_balance(cls): # 创建客户端 config = open_api_models.Config( # 必填，请确保代码运行环境设置了环境变量 ALIBABA_CLOUD_ACCESS_KEY_ID。, access_key_id=os.environ['ALIBABA_CLOUD_ACCESS_KEY_ID'], # 必填，请确保代码运行环境设置了环境变量 ALIBABA_CLOUD_ACCESS_KEY_SECRET。, access_key_secret=os.environ['ALIBABA_CLOUD_ACCESS_KEY_SECRET'] ) # Endpoint 请参考 https://api.aliyun.com/product/BssOpenApi config.endpoint = f'business.aliyuncs.com' runtime = util_models.RuntimeOptions() client = BssOpenApi20171214Client(config) balance_info = client.query_account_balance_with_options(runtime).body.data return f"""币种为：{balance_info.currency}，可用额度为{balance_info.available_amount}，信控余额为{balance_info.credit_amount}， 网商余额为{balance_info.mybank_credit_amount}，现金余额为{balance_info.available_cash_amount}，生态客户Quota限额为{balance_info.quota_limit}。"""2.1.2
tools.py整体代码from alibabacloud_tea_util import models as util_models from alibabacloud_ecs20140526.client import Client as Ecs20140526Client from alibabacloud_ecs20140526 import models as ecs_20140526_models from alibabacloud_tea_openapi import models as open_api_models from alibabacloud_bssopenapi20171214.client import Client as BssOpenApi20171214Client from dashscope import Application import os class ECS: @classmethod # 输入：地域ID，如cn-hangzhou，cn-beijing等 # 输出：查询ecs实例规格信息，包括实例ID，实例规格，每小时收费。（系统盘默认按照cloud_essd_entry，40GiB） def query_source(cls,RegionID): config = open_api_models.Config( # 从环境变量中获取阿里云的AK信息 access_key_id=os.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID"), access_key_secret=os.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET") ) # 杭州与北京上海等地区的endpoint不同，此处进行判断 if RegionID != 'cn-hangzhou': config.endpoint = f'ecs.{RegionID}.aliyuncs.com' else: config.endpoint = f'ecs-{RegionID}.aliyuncs.com' client = Ecs20140526Client(config) describe_instances_request = ecs_20140526_models.DescribeInstancesRequest(region_id=RegionID) runtime = util_models.RuntimeOptions() # 获得ECS实例信息 response_source = client.describe_instances_with_options(describe_instances_request, runtime).body if len(response_source.instances.instance) == 0: return "您在当前区域无ecs实例" # 初始化要返回的结果 result = "" # 可能有多个实例，因此用for循环遍历所有实例 for i in range(len(response_source.instances.instance)): # 系统盘类型与存储空间，此处默认设为cloud_essd_entry，40GiB system_disk = ecs_20140526_models.DescribePriceRequestSystemDisk( category='cloud_essd_entry', size=40 ) describe_price_request = ecs_20140526_models.DescribePriceRequest( region_id=RegionID, resource_type='instance', instance_type=response_source.instances.instance[i].instance_type, system_disk=system_disk ) response = client.describe_price_with_options(describe_price_request, runtime).body cur_result = f"""实例：{response_source.instances.instance[i].instance_id} 的规格为：{response_source.instances.instance[i].instance_type}， 每个小时的收费为{response.price_info.price.trade_price}元\n""" # 将当前实例的信息添加到返回结果中 result += cur_result return result @classmethod # RAG应用调用 def call_agent_app(cls,InstanceType): if len(InstanceType) == 0: return "您在当前区域无ecs实例" result = "" for i in range(len(InstanceType)): response = Application.call( # 此处填写RAG应用的app_id app_id='xxx', prompt=f'介绍一下{InstanceType[i]}', # 从环境变量中获取Dashscope的API Key api_key=os.getenv("DASHSCOPE_API_KEY")) result += response.output.text return result class Billing: # 无需输入，返回为阿里云账户余额信息 @classmethod def get_balance(cls): # 创建客户端 config = open_api_models.Config( # 必填，请确保代码运行环境设置了环境变量 ALIBABA_CLOUD_ACCESS_KEY_ID。, access_key_id=os.environ['ALIBABA_CLOUD_ACCESS_KEY_ID'], # 必填，请确保代码运行环境设置了环境变量 ALIBABA_CLOUD_ACCESS_KEY_SECRET。, access_key_secret=os.environ['ALIBABA_CLOUD_ACCESS_KEY_SECRET'] ) # Endpoint 请参考 https://api.aliyun.com/product/BssOpenApi config.endpoint = f'business.aliyuncs.com' runtime = util_models.RuntimeOptions() client = BssOpenApi20171214Client(config) balance_info = client.query_account_balance_with_options(runtime).body.data return f"""币种为：{balance_info.currency}，可用额度为{balance_info.available_amount}，信控余额为{balance_info.credit_amount}， 网商余额为{balance_info.mybank_credit_amount}，现金余额为{balance_info.available_cash_amount}，生态客户Quota限额为{balance_info.quota_limit}。""" if __name__ == '__main__': print(ECS.query_source('cn-hangzhou'))
2.2 main.py（用于创建Agent并定义交互方式）
main.py主要作用为：
其整体代码可见：main.py整体代码。
引入依赖
from dashscope import Assistants, Messages, Runs, Threads import json # 从tools.py导入工具函数 from tools import ECS,Billing # 引入前端界面展示依赖 import gradio as gr # 将列表形式的字符串解析为列表形式的数据，例如："['a','b']"-->['a','b']。 # 用于将plannerassistant的输出解析为元素为assistant的列表 import ast2.2.1 创建Agent
本教程共包含五个Agent，集成的工具函数与功能请见下表：
Agent名称
集成的工具函数
功能
PlannerAssistant
无
对Multi Agent进行编排。
ChatAssistant
无
如果无需使用工具，则使用该Agent进行回答。
AliyunInfoAssistant
query_source与get_balance查询阿里云的资源信息，包括ECS实例与阿里云余额。
InstanceTypeDetailAssistant
call_agent_app查询指定实例规格的指标数据。
SummaryAssistant
无
结合前序Agent的输出，对用户问题进行全面、完整的回复。
说明Agent中的大模型通过Assistants.create方法的
model参数定义；Agent的功能由Assistants.create方法中的tools参数定义；您可以通过Assistants.create中的instructions指引Agent使用工具的方式，以及Agent的输出格式等，例如让Agent以JSON格式输出字符串。具体实现代码请参考以下五个Agent的代码实现。PlannerAssistant
PlannerAssistant负责根据用户的输入与其它agent的功能，编排Multi Agent的工作方式，是Multi Agent的核心。在得到PlannerAssistant输出后，需要在后续程序中进行字符串解析，将其解析成列表形式的数据。代码如下： # 决策级别的agent，决定使用哪些agent，以及它们的运行顺序 PlannerAssistant = Assistants.create( # 因为该Agent作用比较重要，因此建议选择性能较强的大模型：qwen-max model="qwen-max", # 定义Agent的名称 name='流程编排机器人', # 定义Agent的功能描述 description='你是团队的leader，你的手下有很多assistant，你需要根据用户的输入，决定要以怎样的顺序去使用这些assistant', # 定义对Agent的指示语句，Agent会按照指示语句进行工具的调用并返回结果。 instructions="""你的团队中有以下assistant。AliyunInfoAssistant：可以查询用户指定区域的阿里云ecs实例信息，或者查询用户的阿里云余额；InstanceTypeDetailAssistant：可以查询指定阿里云ecs实例规格的详细信息，比如cpu核数、内存大小等，可以一次查询多个实例规格信息，因此无需多次调用； ChatAssistant：如果用户的问题无需以上两个assistant，则调用该assistant。你需要根据用户的问题，判断要以什么顺序使用这些assistant，你的返回形式是一个列表，不能返回其它信息。比如：["AliyunInfoAssistant", "AliyunInfoAssistant","InstanceTypeDetailAssistant"]或者["ChatAssistant"]，列表中的元素只能为上述的assistant""" )ChatAssistant
ChatAssistant的功能是回复日常问题，可以使用成本较低的模型（如qwen-turbo）作为agent的底座模型。 # 功能是回复日常问题。对于日常问题来说，可以使用价格较为低廉的模型作为agent的基座 ChatAssistant = Assistants.create( # 因为该Agent对大模型性能要求不高，因此使用成本较低的qwen-turbo模型 model="qwen-turbo", name='回答日常问题的机器人', description='一个智能助手，解答用户的问题', instructions='请礼貌地回答用户的问题' )AliyunInfoAssistant
AliyunInfoAssistant的功能是查询阿里云的资源信息。目前有ecs实例查询与阿里云余额查询两个功能。代码详情为： # 功能是查询阿里云的资源信息。目前有ecs实例查询与阿里云余额查询两个功能 AliyunInfoAssistant = Assistants.create( model="qwen-max", name='阿里云资源信息查询机器人', description='一个智能助手，根据用户的查询去调用工具并返回查询到的阿里云资源结果', instructions='你是一个智能助手，你有两个功能，分别是阿里云ecs实例信息查询和阿里云余额查询。请准确判断调用哪个工具，并礼貌地回答用户的问题。', # 定义Agent使用的工具，您可以根据您的业务场景在tools列表中定义一个或多个Agent可能会使用的工具。 tools=[ { 'type': 'function', 'function': { # 工具函数的名称，可通过下文代码中的function_mapper将name映射到函数本体 'name': 'ecs实例信息查询', # 工具函数的描述 'description': '当需要查询阿里云ecs实例信息时非常有用，比如实例id，实例规格，收费信息等', # 工具函数的入参 'parameters': { 'type': 'object', 'properties': { # 该工具需要用户输入地域信息 'RegionID': { 'type': 'str', # 参数的描述信息 'description': '用户想要查询实例所属的地域id，如果是杭州，则为cn-hangzhou，如果是上海，则为cn-shanghai，如果是北京，则为cn-beijing' }, }, 'required': ['RegionID']}, } }, { 'type': 'function', 'function': { 'name': '阿里云余额查询', # 工具函数的描述 'description': '当需要查询阿里云账户信息时非常有用', # 工具函数的入参，余额查询无需入参，因此为空 'parameters': {} } } ] )InstanceTypeDetailAssistant
InstanceTypeDetailAssistant接收实例规格列表信息，并通过call_agent_app函数将每一种实例规格对RAG应用进行查询，将每种实例规格返回的vCPU个数、内存大小等指标数据汇总作为输出。 # 功能是通过在百炼平台创建的RAG应用查询实例规格的详细信息 InstanceTypeDetailAssistant = Assistants.create( model="qwen-max", name='ecs实例规格介绍机器人', description='一个智能助手，可以通过用户提供的实例规格，调用已有的插件能力给用户介绍实例规格信息。', instructions='你是一个智能助手，你需要从输入中精确提取出实例规格信息，如[ecs.e-c1m1.large]，或[ecs.u1-c1m4.xlarge,ecs.e-c1m1.large]。将实例规格列表输入工具中，获得它们的详细信息', tools=[ { 'type': 'function', 'function': { 'name': 'ecs实例规格介绍', 'description': '返回客户查询指定ecs实例规格的信息', 'parameters': { 'type': 'object', 'properties': { 'InstanceType': { 'type': 'list', 'InstanceType': '用户想要查询的实例规格，有可能是一个，有可能是多个，如：[ecs.e-c1m1.large]，或[ecs.u1-c1m4.xlarge,ecs.e-c1m1.large]' }, }, 'required': ['InstanceType']}, } } ] )SummaryAssistant
每一个Agent都会有输出信息，因此需要一个用于总结的Agent将前序的Agent输出信息进行总结，从而对用户的问题进行全面、完整的回答。代码详情如下： # 在Multi Agent场景下，定义一个用于总结的Agent，该Agent会根据用户的问题与之前Agent输出的参考信息，全面、完整地回答用户问题 SummaryAssistant = Assistants.create( model="qwen-max", name='总结机器人', description='一个智能助手，根据用户的问题与参考信息，全面、完整地回答用户问题', instructions='你是一个智能助手，根据用户的问题与参考信息，全面、完整地回答用户问题' )2.2.2 定义字符串与函数、字符串与Agent本体的映射
由于大模型生成的是字符串形式的结果，因此我们在程序中需要对大模型生成的字符串进行解析、映射等操作，以达到和外界交互的功能。
代码示例
# 将工具函数的name映射到函数本体 function_mapper = { "ecs实例信息查询": ECS.query_source, "ecs实例规格介绍":ECS.call_agent_app, "阿里云余额查询":Billing.get_balance } # 将Agent的name映射到Agent本体 assistant_mapper = { "ChatAssistant": ChatAssistant, "AliyunInfoAssistant":AliyunInfoAssistant, "InstanceTypeDetailAssistant":InstanceTypeDetailAssistant }2.2.3 定义消息传递函数
消息传递函数
get_agent_response接收assistant与message两个参数，用于获得指定Agent在接收到输入message时的输出信息。代码示例
# 输入message信息，输出为指定Agent的回复 def get_agent_response(assistant, message=''): # 打印出输入Agent的信息 print(f"Query: {message}") thread = Threads.create() message = Messages.create(thread.id, content=message) run = Runs.create(thread.id, assistant_id=assistant.id) run_status = Runs.wait(run.id, thread_id=thread.id) # 如果响应失败，会打印出run failed if run_status.status == 'failed': print('run failed:') # 如果需要工具来辅助大模型输出，则进行以下流程 if run_status.required_action: f = run_status.required_action.submit_tool_outputs.tool_calls[0].function # 获得function name func_name = f['name'] # 获得function 的入参 param = json.loads(f['arguments']) # 打印出工具信息 print("function is",f) # 根据function name，通过function_mapper映射到函数，并将参数输入工具函数得到output输出 if func_name in function_mapper: output = function_mapper[func_name](**param) else: output = "" tool_outputs = [{ 'output': output }] run = Runs.submit_tool_outputs(run.id, thread_id=thread.id, tool_outputs=tool_outputs) run_status = Runs.wait(run.id, thread_id=thread.id) run_status = Runs.get(run.id, thread_id=thread.id) msgs = Messages.list(thread.id) # 将Agent的输出返回 return msgs['data'][0]['content'][0]['text']['value']2.2.4 定义Agent之间交互方式并获得回复
Agent之间的交互步骤根据PlannerAssistant进行编排，为了适配Gradio的前端界面展示，输入输出的参数需要与Gradio中的组件进行对齐。代码如下：
说明使用
yield关键字而不是return，可以迭代地生成和返回中间结果。这样，中间结果可以逐步传递给前端界面，实现实时显示，而不必等到所有结果生成后再显示。agent交互代码
# 获得Multi Agent的回复，输入与输出需要与Gradio前端展示界面中的参数对齐 def get_multi_agent_response(query,history): # 处理输入为空的情况 if len(query) == 0: return "",history+[("","")],"","" # 获取Agent的运行顺序 assistant_order = get_agent_response(PlannerAssistant,query) try: order_stk = ast.literal_eval(assistant_order) cur_query = query # 依次运行Agent for i in range(len(order_stk)): yield "----->".join(order_stk),history+[(query,"multi agent正在努力工作中...")],f"{order_stk[i]}正在处理信息...","" cur_assistant = assistant_mapper[order_stk[i]] response = get_agent_response(cur_assistant,cur_query) yield "----->".join(order_stk),history+[(query,"multi agent正在努力工作中...")],response,"" # 如果当前Agent为最后一个Agent，则将其输出作为Multi Agent的输出 if i == len(order_stk)-1: yield "----->".join(order_stk),history+[(query,response)],"assistant已处理完毕","" # 如果当前Agent不是最后一个Agent，则将上一个Agent的输出response添加到下一轮的query中，作为参考信息 else: # 在参考信息前后加上特殊标识符，可以防止大模型混淆参考信息与提问 cur_query = f"你可以参考已知的信息：\n{response}\n你要完整地回答用户的问题。问题是：{query}。" # 兜底策略，如果上述程序运行失败，则直接调用ChatAssistant except Exception as e: yield "ChatAssistant",[(query,get_agent_response(ChatAssistant,query))],"",""输入参数为query与history。其中query为用户的提问（字符串形式），history为用户与Multi Agent的对话记录（列表形式）。
第一个输出参数为Agent的编排信息（字符串形式），第二个输出参数为用户与Multi Agent的对话历史（列表形式），第三个输出参数为当前运行Agent的状态（字符串形式），为了适配Gradio组件，第四个输出参数为用户的输入框（字符串形式，设为
""以达到用户发起提问后将输入框清空的效果）。2.2.5 前端展示界面
本教程使用gradio作为前端展示工具。gradio可以快速帮助机器学习工作者创建模型效果展示界面，代码详情如下：
前端代码
# 前端界面展示 with gr.Blocks() as demo: # 在界面中央展示标题 gr.HTML('<center><h1>欢迎使用阿里云资源查询bot</h1></center>') gr.HTML('<center><h3>支持的功能有指定区域的ecs实例查询、余额查询、实例规格详情查询。您可以在tools.py中添加您需要的工具，并在main.py中配置相关的agent</h3></center>') with gr.Row(): with gr.Column(scale=10): chatbot = gr.Chatbot(value=[["hello","很高兴见到您！您想问关于阿里云资源的哪些问题呢？"]],height=600) with gr.Column(scale=4): text1 = gr.Textbox(label="assistant选择") text2 = gr.Textbox(label="当前assistant状态",lines=22) with gr.Row(): msg = gr.Textbox(label="输入",placeholder="您想了解什么呢？") # 一些示例问题 with gr.Row(): examples = gr.Examples(examples=[ '我的阿里云余额还有多少钱啊', '我在杭州有哪些ecs实例，把它的实例id，价钱以及实例规格详情告诉我', '我想了解ecs.u1-c1m4.xlarge和ecs.gn6i-c4g1.xlarge的指标'],inputs=[msg]) clear = gr.ClearButton([text1,chatbot,text2,msg]) msg.submit(get_multi_agent_response, [msg,chatbot], [text1,chatbot,text2,msg])2.2.6
main.py整体代码main.py整体代码
from dashscope import Assistants, Messages, Runs, Threads import json # 从tools.py导入工具函数 from tools import ECS,Billing # 引入前端界面展示依赖 import gradio as gr import ast # 决策级别的agent，决定使用哪些agent，以及它们的运行顺序 PlannerAssistant = Assistants.create( # 因为该Agent作用比较重要，因此建议选择性能较强的大模型：qwen-max model="qwen-max", # 定义Agent的名称 name='流程编排机器人', # 定义Agent的功能描述 description='你是团队的leader，你的手下有很多assistant，你需要根据用户的输入，决定要以怎样的顺序去使用这些assistant', # 定义对Agent的指示语句，Agent会按照指示语句进行工具的调用并返回结果。 instructions="""你的团队中有以下assistant。AliyunInfoAssistant：可以查询用户指定区域的阿里云ecs实例信息，或者查询用户的阿里云余额；InstanceTypeDetailAssistant：可以查询指定阿里云ecs实例规格的详细信息，比如cpu核数、内存大小等，可以一次查询多个实例规格信息，因此无需多次调用； ChatAssistant：如果用户的问题无需以上两个assistant，则调用该assistant。你需要根据用户的问题，判断要以什么顺序使用这些assistant，你的返回形式是一个列表，不能返回其它信息。比如：["AliyunInfoAssistant", "AliyunInfoAssistant","InstanceTypeDetailAssistant"]或者["ChatAssistant"]，列表中的元素只能为上述的assistant""" ) # 功能是回复日常问题。对于日常问题来说，可以使用价格较为低廉的模型作为agent的基座 ChatAssistant = Assistants.create( # 因为该Agent对大模型性能要求不高，因此使用成本较低的qwen-turbo模型 model="qwen-turbo", name='回答日常问题的机器人', description='一个智能助手，解答用户的问题', instructions='请礼貌地回答用户的问题' ) # 功能是查询阿里云的资源信息。目前有ECS实例查询与阿里云余额查询两个功能 AliyunInfoAssistant = Assistants.create( model="qwen-max", name='阿里云资源信息查询机器人', description='一个智能助手，根据用户的查询去调用工具并返回查询到的阿里云资源结果', instructions='你是一个智能助手，你有两个功能，分别是阿里云ecs实例信息查询和阿里云余额查询。请准确判断调用哪个工具，并礼貌地回答用户的问题。', # 定义Agent使用的工具，您可以根据您的业务场景在tools列表中定义一个或多个Agent可能会使用的工具。 tools=[ { 'type': 'function', 'function': { # 工具函数的名称，可通过下文代码中的function_mapper将name映射到函数本体 'name': 'ecs实例信息查询', # 工具函数的描述 'description': '当需要查询阿里云ecs实例信息时非常有用，比如实例id，实例规格，收费信息等', # 工具函数的入参 'parameters': { 'type': 'object', 'properties': { # 该工具需要用户输入地域信息 'RegionID': { 'type': 'str', # 参数的描述信息 'description': '用户想要查询实例所属的地域id，如果是杭州，则为cn-hangzhou，如果是上海，则为cn-shanghai，如果是北京，则为cn-beijing' }, }, 'required': ['RegionID']}, } }, { 'type': 'function', 'function': { 'name': '阿里云余额查询', # 工具函数的描述 'description': '当需要查询阿里云账户信息时非常有用', # 工具函数的入参，余额查询无需入参，因此为空 'parameters': {} } } ] ) # 功能是通过在百炼平台创建的RAG应用查询实例规格的详细信息 InstanceTypeDetailAssistant = Assistants.create( model="qwen-max", name='ecs实例规格介绍机器人', description='一个智能助手，可以通过用户提供的输入，精确识别提到的实例规格。调用已有的插件能力给用户介绍实例规格信息。', instructions='你是一个智能助手，你需要从用户的输入中精确识别提取出阿里云的实例规格信息，如[ecs.e-c1m1.large]，或[ecs.u1-c1m4.xlarge,ecs.e-c1m1.large]。将实例规格列表输入工具中，获得它们的详细信息', tools=[ { 'type': 'function', 'function': { 'name': 'ecs实例规格介绍', 'description': '返回客户查询指定ecs实例规格的信息', 'parameters': { 'type': 'object', 'properties': { 'InstanceType': { 'type': 'list', 'InstanceType': '用户想要查询的实例规格，有可能是一个，有可能是多个，如：[ecs.e-c1m1.large]，或[ecs.u1-c1m4.xlarge,ecs.e-c1m1.large]' }, }, 'required': ['InstanceType']}, } } ] ) # 在Multi Agent场景下，定义一个用于总结的Agent，该Agent会根据用户的问题与之前Agent输出的参考信息，全面、完整地回答用户问题 SummaryAssistant = Assistants.create( model="qwen-max", name='总结机器人', description='一个智能助手，根据用户的问题与参考信息，全面、完整地回答用户问题', instructions='你是一个智能助手，根据用户的问题与参考信息，全面、完整地回答用户问题' ) # 将工具函数的name映射到函数本体 function_mapper = { "ecs实例信息查询": ECS.query_source, "ecs实例规格介绍":ECS.call_agent_app, "阿里云余额查询":Billing.get_balance } # 将Agent的name映射到Agent本体 assistant_mapper = { "ChatAssistant": ChatAssistant, "AliyunInfoAssistant":AliyunInfoAssistant, "InstanceTypeDetailAssistant":InstanceTypeDetailAssistant } # 输入message信息，输出为指定Agent的回复 def get_agent_response(assistant, message=''): # 打印出输入Agent的信息 print(f"Query: {message}") thread = Threads.create() message = Messages.create(thread.id, content=message) run = Runs.create(thread.id, assistant_id=assistant.id) run_status = Runs.wait(run.id, thread_id=thread.id) # 如果响应失败，会打印出run failed if run_status.status == 'failed': print('run failed:') # 如果需要工具来辅助大模型输出，则进行以下流程 if run_status.required_action: f = run_status.required_action.submit_tool_outputs.tool_calls[0].function # 获得function name func_name = f['name'] # 获得function 的入参 param = json.loads(f['arguments']) # 打印出工具信息 print("function is",f) # 根据function name，通过function_mapper映射到函数，并将参数输入工具函数得到output输出 if func_name in function_mapper: output = function_mapper[func_name](**param) else: output = "" tool_outputs = [{ 'output': output }] run = Runs.submit_tool_outputs(run.id, thread_id=thread.id, tool_outputs=tool_outputs) run_status = Runs.wait(run.id, thread_id=thread.id) run_status = Runs.get(run.id, thread_id=thread.id) msgs = Messages.list(thread.id) # 将Agent的输出返回 return msgs['data'][0]['content'][0]['text']['value'] # 获得Multi Agent的回复，输入与输出需要与Gradio前端展示界面中的参数对齐 def get_multi_agent_response(query,history): # 处理输入为空的情况 if len(query) == 0: return "",history+[("","")],"","" # 获取Agent的运行顺序 assistant_order = get_agent_response(PlannerAssistant,query) try: order_stk = ast.literal_eval(assistant_order) cur_query = query Agent_Message = "" # 依次运行Agent for i in range(len(order_stk)): yield "----->".join(order_stk),history+[(query,"multi agent正在努力工作中...")],Agent_Message+'\n'+f"*{order_stk[i]}*正在处理中...","" cur_assistant = assistant_mapper[order_stk[i]] response = get_agent_response(cur_assistant,cur_query) Agent_Message += f"*{order_stk[i]}*的回复为：{response}\n\n" yield "----->".join(order_stk),history+[(query,"multi agent正在努力工作中...")],Agent_Message,"" # 如果当前Agent为最后一个Agent，则将其输出作为Multi Agent的输出 if i == len(order_stk)-1: prompt = f"请参考已知的信息：{Agent_Message}，回答用户的问题：{query}。" multi_agent_response = get_agent_response(SummaryAssistant,prompt) yield "----->".join(order_stk),history+[(query,multi_agent_response)],Agent_Message,"" # 如果当前Agent不是最后一个Agent，则将上一个Agent的输出response添加到下一轮的query中，作为参考信息 else: # 在参考信息前后加上特殊标识符，可以防止大模型混淆参考信息与提问 cur_query = f"你可以参考已知的信息：{response}你要完整地回答用户的问题。问题是：{query}。" # 兜底策略，如果上述程序运行失败，则直接调用ChatAssistant except Exception as e: yield "ChatAssistant",[(query,get_agent_response(ChatAssistant,query))],"","" # 前端界面展示 with gr.Blocks() as demo: # 在界面中央展示标题 gr.HTML('<center><h1>欢迎使用阿里云资源查询bot</h1></center>') gr.HTML('<center><h3>支持的功能有指定区域的ecs实例查询、余额查询、实例规格详情查询。您可以在tools.py中添加您需要的工具，并在main.py中配置相关的agent</h3></center>') with gr.Row(): with gr.Column(scale=10): chatbot = gr.Chatbot(value=[["hello","很高兴见到您！您想问关于阿里云资源的哪些问题呢？"]],height=600) with gr.Column(scale=4): text1 = gr.Textbox(label="assistant选择") text2 = gr.Textbox(label="当前assistant状态",lines=22) with gr.Row(): msg = gr.Textbox(label="输入",placeholder="您想了解什么呢？") # 一些示例问题 with gr.Row(): examples = gr.Examples(examples=[ '我的阿里云余额还有多少钱啊', '我在杭州有哪些ecs实例，把它的实例id，价钱以及实例规格详情告诉我', '我想了解ecs.u1-c1m4.xlarge和ecs.gn6i-c4g1.xlarge的指标'],inputs=[msg]) clear = gr.ClearButton([text1,chatbot,text2,msg]) msg.submit(get_multi_agent_response, [msg,chatbot], [text1,chatbot,text2,msg]) if __name__ == '__main__': demo.launch()
运行效果
请您在配置
ALIBABA_CLOUD_ACCESS_KEY_ID、ALIBABA_CLOUD_ACCESS_KEY_SECRET和DASHSCOPE_API_KEY到环境变量后，运行main.py文件。终端页面会有Running on local URL:的输出，访问对应URL，进入交互界面。输入：我想知道我在杭州的ecs实例，还有我的阿里云余额，或者单击Examples中的示例问题，将其添加到输入框中，并单击Enter，等待结果的生成。您可以观察当前assistant状态框来查看Agent的实时状态。
实验资源释放
如果您在完成本实践教程后没有继续使用ECS实例的需求，请及时释放您的ECS资源，避免产生不必要的消费。
背景知识
本场景主要涉及产品：大模型服务平台百炼。
在本实验中，需要注意您当前账号中是否有免费额度，若无免费额度或免费额度已过期，本次实验会消耗token，可能产生费用。
模型token费用：请查看模型列表：模型列表
如何查看模型免费额度：登录阿里云百炼-模型广场-对应模型卡片查看详情，查看模型的免费额度，也可以关注系统管理-系统工具中的调用统计。
关闭实验
在完成实验后，如果无需继续使用资源，选择不保留资源，单击结束实操。在结束实操对话框中，单击确定。
在完成实验后，如果需要继续使用资源，选择付费保留资源，单击结束实操。在结束实操对话框中，单击确定。请随时关注账户扣费情况，避免发生欠费。
文档内容是否对您有帮助？