本文将通过构建一个查询阿里云资源信息的Multi Agent系统，帮助您了解如何通过阿里云百炼平台的Assistant API构建一个无需提前定义、可自动规划编排任务流程的Multi Agent系统。
多智能体系统（Multi Agent System）是多个Agent协作完成任务的方式，在多数场景下，比单Agent完成任务准确率更高。通过给每个Agent制定明确且专业的角色名称和职责描述，不仅可以提升它们的专业性，还能帮助它们更好地理解和配合各自的工作。Multi Agent System的设计十分灵活，您可以参考本案例中的设计和示例代码，将其应用到您的业务中。
通过本文提供的教程，您可以实现一个能进行阿里云资源信息查询的 Multi Agent 系统。该系统在收到用户问题后，会自动规划多个 Agent 之间的协作流程，按照规划执行任务后给出最终答案。 详细工作流程： 用户进行提问； PlannerAssistant接收到用户的提问，并根据用户问题对其它Agent进行选择并编排其运行顺序； ChatAssistant、AliyunInfoAssistant与InstanceTypeDetailAssistant分别有对应的职责与能力，它们是本教程Multi Agent系统中被PlannerAssistant编排的Agent。编排完成后的Agent组合会接收用户的提问，并按照程序中设计的Multi Agent交互逻辑进行分工合作； SummaryAssistant将各Agent的输出信息汇总，并结合用户问题对信息进行总结，其输出会作为Multi Agent系统的最终输出。 请参考创建AccessKey，获取用户的 请开通阿里云百炼服务并获取API Key。 我们推荐您将 Bash PowerShell 如果您没有创建ECS实例，请前往ECS控制台创建一个按量付费的实例。 说明 如果您在完成本实践教程后没有继续使用ECS实例的需求，请及时释放您的ECS资源，避免产生不必要的消费。 请确认您的计算环境中已安装Python。您可以新建一个 在您保存 请您参考0代码构建问答应用文档，在阿里云百炼平台创建RAG应用，知识库文件选择PDF格式的阿里云ECS官方文档实例规格族.pdf，并获取其应用ID。 您需要准备两个py文件，分别为 请下载multi_agent.zip到本地后进行解压。 ECS Billing ECS有两个类函数： ECS类的定义代码如下： 说明 请用您在阿里云百炼平台创建的RAG应用的 main.py主要作用为： 其整体代码可见：main.py整体代码。 本教程共包含五个Agent，集成的工具函数与功能请见下表： Agent名称 集成的工具函数 功能 无 对Multi Agent进行编排。 无 如果无需使用工具，则使用该Agent进行回答。 查询阿里云的资源信息，包括ECS实例与阿里云余额。 查询指定实例规格的指标数据。 无 结合前序Agent的输出，对用户问题进行全面、完整的回复。 五个Agent的详情如下： 说明 Agent中的大模型通过Assistants.create方法的 PlannerAssistant ChatAssistant AliyunInfoAssistant InstanceTypeDetailAssistant SummaryAssistant PlannerAssistant负责根据用户的输入与其它agent的功能，编排Multi Agent的工作方式，是Multi Agent的核心。在得到PlannerAssistant输出后，需要在后续程序中进行字符串解析，将其解析成列表形式的数据。代码如下： 由于大模型生成的是字符串形式的结果，因此我们在程序中需要对大模型生成的字符串进行解析、映射等操作，以达到和外界交互的功能。 消息传递函数 Agent之间的交互步骤根据PlannerAssistant进行编排，为了适配Gradio的前端界面展示，输入输出的参数需要与Gradio中的组件进行对齐。代码如下： 说明 使用 输入参数为query与history。其中query为用户的提问（字符串形式），history为用户与Multi Agent的对话记录（列表形式）。 第一个输出参数为Agent的编排信息（字符串形式），第二个输出参数为用户与Multi Agent的对话历史（列表形式），第三个输出参数为当前运行Agent的状态（字符串形式），为了适配Gradio组件，第四个输出参数为用户的输入框（字符串形式，设为 本教程使用gradio作为前端展示工具。gradio可以快速帮助机器学习工作者创建模型效果展示界面，代码详情如下： 请您在配置 通过本教程，您可以了解到将阿里云百炼RAG应用与阿里云的OpenAPI能力集成到Agent中的方式，以及使用阿里云百炼平台的Assistants API进行Multi Agent开发的步骤，并最终通过基于Gradio的前端界面展示出来。 您可以通过修改Agent中的提示词、细化Agent之间的交互方式、修改工具函数等方法，将Multi Agent应用到您的业务中。效果展示
前提条件
ACCESS_KEY_ID和ACCESS_KEY_SECRET。ALIBABA_CLOUD_ACCESS_KEY_ID、ALIBABA_CLOUD_ACCESS_KEY_SECRET和DASHSCOPE_API_KEY配置在环境变量中，以降低泄露风险。环境变量配置方法可参考：配置API Key到环境变量。在本教程中，您可以参考以下命令，根据您的操作系统选择配置环境变量的方法：# 用您的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"
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
代码实现
tools.py与main.py，这两个文件需要放置在同一目录中。工程文件
tools.py（用于定义工具函数）
tools.py主要定义了Agent会使用到的工具函数。tools.py整体代码可见tools.py整体代码。引入依赖
from alibabacloud_tea_util import models as util_modelsfrom alibabacloud_ecs20140526.client import Client as Ecs20140526Clientfrom alibabacloud_ecs20140526 import models as ecs_20140526_modelsfrom alibabacloud_tea_openapi import models as open_api_modelsfrom alibabacloud_bssopenapi20171214.client import Client as BssOpenApi20171214Clientfrom dashscope import Applicationimport os
定义工具
tools.py中主要使用到了两个工具类，分别为ECS与Billing。ECS的类方法包含有query_source和call_agent_app，Billing的类方法包含get_balance。您可以参考以下代码，添加您需要的工具类与工具函数以适配您的业务。query_source与call_agent_app。query_source通过阿里云的OpenAPI服务查询指定区域的ECS实例信息。函数输入为RegionID，如cn-hangzhou、cn-beijing、cn-shanghai、ap-southeast-1等；输出包含实例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的运行时间可能较长。app_id替代代码中的app_id。class ECS: @classmethod
# 输入：地域ID，如cn-hangzhou，cn-beijing等
# 输出：查询ecs实例规格信息，包括实例ID，实例规格，每小时收费。（系统盘默认按照cloud_auto，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_auto，40GiB
system_disk = ecs_20140526_models.DescribePriceRequestSystemDisk(
category='cloud_auto',
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 resulttools.py整体代码tools.py整体代码如下：from alibabacloud_tea_util import models as util_modelsfrom alibabacloud_ecs20140526.client import Client as Ecs20140526Clientfrom alibabacloud_ecs20140526 import models as ecs_20140526_modelsfrom alibabacloud_tea_openapi import models as open_api_modelsfrom alibabacloud_bssopenapi20171214.client import Client as BssOpenApi20171214Clientfrom dashscope import Applicationimport osclass ECS: @classmethod
# 输入：地域ID，如cn-hangzhou，cn-beijing等
# 输出：查询ecs实例规格信息，包括实例ID，实例规格，每小时收费。（系统盘默认按照cloud_auto，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_auto，40GiB
system_disk = ecs_20140526_models.DescribePriceRequestSystemDisk(
category='cloud_auto',
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 resultclass 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'))main.py（用于创建Agent并定义交互方式）
引入依赖
from dashscope import Assistants, Messages, Runs, Threadsimport json# 从tools.py导入工具函数from tools import ECS,Billing# 引入前端界面展示依赖import gradio as gr# 将列表形式的字符串解析为列表形式的数据，例如："['a','b']"-->['a','b']。# 用于将plannerassistant的输出解析为元素为assistant的列表import ast
创建Agent
query_source与get_balancecall_agent_appmodel参数定义；Agent的功能由Assistants.create方法中的tools参数定义；您可以通过Assistants.create中的instructions指引Agent使用工具的方式，以及Agent的输出格式等，例如让Agent以JSON格式输出字符串。具体实现代码请参考以下五个Agent的代码实现。# 决策级别的agent，决定使用哪些agent，以及它们的运行顺序PlannerAssistant = Assistants.create( # 因为该Agent作用比较重要，因此建议选择性能较强的大模型：qwen-plus。模型列表：https://help.aliyun.com/zh/model-studio/getting-started/models
model="qwen-plus", # 定义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本体的映射
# 将工具函数的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
}定义消息传递函数
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']定义Agent之间交互方式并获得回复
yield关键字而不是return，可以迭代地生成和返回中间结果。这样，中间结果可以逐步传递给前端界面，实现实时显示，而不必等到所有结果生成后再显示。# 获得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))],"",""""以达到用户发起提问后将输入框清空的效果）。前端展示界面
# 前端界面展示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])main.py整体代码from dashscope import Assistants, Messages, Runs, Threadsimport json# 从tools.py导入工具函数from tools import ECS,Billing# 引入前端界面展示依赖import gradio as grimport ast# 决策级别的agent，决定使用哪些agent，以及它们的运行顺序PlannerAssistant = Assistants.create( # 因为该Agent作用比较重要，因此建议选择性能较强的大模型：qwen-plus。模型列表：https://help.aliyun.com/zh/model-studio/getting-started/models
model="qwen-plus", # 定义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模型。模型列表：https://help.aliyun.com/zh/model-studio/getting-started/models
model="qwen-turbo",
name='回答日常问题的机器人',
description='一个智能助手，解答用户的问题',
instructions='请礼貌地回答用户的问题')# 功能是查询阿里云的资源信息。目前有ECS实例查询与阿里云余额查询两个功能AliyunInfoAssistant = Assistants.create(
model="qwen-plus", # 模型列表：https://help.aliyun.com/zh/model-studio/getting-started/models
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-plus", # 模型列表：https://help.aliyun.com/zh/model-studio/getting-started/models
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-plus", # 模型列表：https://help.aliyun.com/zh/model-studio/getting-started/models
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的实时状态。
总结
