Alibaba Cloud Model Studio:Build a multi-agent system with automatic planning capabilities using the Assistant API

Performance showcase

By following this guide, you can create a multi-agent system that can query information about Alibaba Cloud resources. When a user poses a question, the system automatically plans the workflow among agents. After executing the planned tasks, the system delivers the response to the user.

Detailed workflow:

1. The user submits a question.

2. PlannerAssistant receives the question and orchestrates the other agents accordingly.

3. PlannerAssistant orchestrates ChatAssistant, AliyunInfoAssistant, and InstanceTypeDetailAssistant to receive the question and collaborate based on the designed interaction logic. Each orchestrated agent has specific responsibilities and capabilities.

4. SummaryAssistant consolidates the outputs from each agent and the user's original question to provide a comprehensive summary, which serves as the final response.

Prerequisites

1. You have obtained an AccessKey ID and an AccessKey Secret. For more information, see Create an AccessKey pair.

2. You have activated Alibaba Cloud Model Studio and obtained an API key.

3. To minimize the risk of information leakage, we recommend that you set ALIBABA_CLOUD_ACCESS_KEY_ID, ALIBABA_CLOUD_ACCESS_KEY_SECRET, and DASHSCOPE_API_KEY as environment variables. For more information about how to configure environment variables, see Set an API key as an environment variable. In this topic, you can use the following commands to set environment variables based on your operating system:

   Bash

   # Replace the placeholders with your API key and AccessKey information.
   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

   # Replace the placeholders with your API key and AccessKey information.
   $env:DASHSCOPE_API_KEY = "YOUR_DASHSCOPE_API_KEY"
   $env:ALIBABA_CLOUD_ACCESS_KEY_ID="YOUR_ALIBABA_CLOUD_ACCESS_KEY_ID"
   $env:ALIBABA_CLOUD_ACCESS_KEY_SECRET="YOUR_ALIBABA_CLOUD_ACCESS_KEY_SECRET"

4. If you have not created an ECS instance, go to the ECS console to create a pay-as-you-go instance.

   Note

   If you no longer need the ECS instance after you complete this tutorial, release the ECS resource in time to avoid unnecessary charges.

5. Ensure that Python is installed in your computing environment. You can create a requirements.txt file and copy the following content into it.

   alibabacloud_tea_openapi
   alibabacloud_tea_util
   alibabacloud_openapi_util
   alibabacloud_ecs20140526
   alibabacloud_bssopenapi20171214
   dashscope
   gradio

   After you save the requirements.txt file, run the following command in the directory that contains the requirements.txt file to install the dependencies:

   pip install -r requirements.txt

6. Create a Retrieval-Augmented Generation (RAG) application in the Model Studio console. For the knowledge base file, use this documentation of ECS in PDF format: instance_family.pdf. Then, obtain the application ID.

   

Code implementation

You need to prepare two Python files, tools.py and main.py, and place them in the same directory.

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
import dashscope

class ECS:
    @classmethod
    # Parameters: RegionID - Region identifier (e.g., cn-hangzhou, ap-southeast-1)
    # Output: ECS instance type information, including instance ID, instance type, and hourly charge. (The system disk is cloud_auto by default, with a size of 40 GiB.)
    def query_source(cls,RegionID):
        config = open_api_models.Config(
                # Obtain the AccessKey information of Alibaba Cloud from an environment variable.
                access_key_id=os.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID"),
                access_key_secret=os.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET")
            )
        # The endpoints for regions such as China (Hangzhou), China (Beijing), and Singapore are different. This is determined here.
        config.endpoint = f'ecs.{RegionID}.aliyuncs.com'
        client = Ecs20140526Client(config)
        describe_instances_request = ecs_20140526_models.DescribeInstancesRequest(region_id=RegionID)
        runtime = util_models.RuntimeOptions()
        # Obtain ECS instance information.
        response_source = client.describe_instances_with_options(describe_instances_request, runtime).body
        if len(response_source.instances.instance) == 0:
            return "You have no ECS instances in the current region."
        # Initialize the result to be returned.
        result = ""
        # There may be multiple instances, so a for loop is used to traverse all instances.
        for i in range(len(response_source.instances.instance)):
            # System disk type and storage space. Here, the default values are cloud_auto and 40 GiB.
            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"""The instance type of instance {response_source.instances.instance[i].instance_id} is {response_source.instances.instance[i].instance_type}.
    The hourly charge is {response.price_info.price.trade_price} USD.\n"""
            # Add the information of the current instance to the result.
            result += cur_result
        return result
    @classmethod
    # RAG application call
    def call_agent_app(cls,InstanceType):
        if len(InstanceType) == 0:
            return "You have no ECS instances in the current region."
        result = ""
        for i in range(len(InstanceType)):
            response = Application.call(
                # Enter the app_id of the RAG application here.
                app_id='xxx',
                prompt=f'Introduce {InstanceType[i]}',
                # Obtain the API Key for Dashscope from an environment variable.
                api_key=os.getenv("DASHSCOPE_API_KEY"))
            result += response.output.text
        return result

class Billing:
    # No input is required. The Alibaba Cloud account balance information is returned.
    @classmethod
    def get_balance(cls):
        # Create a client.
        config = open_api_models.Config(
            # Required. Make sure that the ALIBABA_CLOUD_ACCESS_KEY_ID environment variable is set in the code execution environment.
            access_key_id=os.environ['ALIBABA_CLOUD_ACCESS_KEY_ID'],
            # Required. Make sure that the ALIBABA_CLOUD_ACCESS_KEY_SECRET environment variable is set in the code execution environment.
            access_key_secret=os.environ['ALIBABA_CLOUD_ACCESS_KEY_SECRET']
        )
        # For the endpoint, see 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"""The currency is {balance_info.currency}, the available amount is {balance_info.available_amount}, the credit balance is {balance_info.credit_amount},
        the Mybank balance is {balance_info.mybank_credit_amount}, the cash balance is {balance_info.available_cash_amount}, and the quota limit for ecological customers is {balance_info.quota_limit}."""

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
import dashscope

dashscope.base_http_api_url = 'https://dashscope-intl.aliyuncs.com/api/v1'
class ECS:
    @classmethod
    # Parameters: RegionID - Region identifier (e.g., cn-hangzhou, ap-southeast-1)
    # Output: ECS instance type information, including instance ID, instance type, and hourly charge. (The system disk is cloud_auto by default, with a size of 40 GiB.)
    def query_source(cls,RegionID):
        config = open_api_models.Config(
                # Obtain the AccessKey information of Alibaba Cloud from an environment variable.
                access_key_id=os.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID"),
                access_key_secret=os.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET")
            )
        # The endpoints for regions such as China (Hangzhou), China (Beijing), and Shanghai are different. This is determined here.
        config.endpoint = f'ecs.{RegionID}.aliyuncs.com'
        client = Ecs20140526Client(config)
        describe_instances_request = ecs_20140526_models.DescribeInstancesRequest(region_id=RegionID)
        runtime = util_models.RuntimeOptions()
        # Obtain ECS instance information.
        response_source = client.describe_instances_with_options(describe_instances_request, runtime).body
        if len(response_source.instances.instance) == 0:
            return "You have no ECS instances in the current region."
        # Initialize the result to be returned.
        result = ""
        # There may be multiple instances, so a for loop is used to traverse all instances.
        for i in range(len(response_source.instances.instance)):
            # System disk type and storage space. Here, the default values are cloud_auto and 40 GiB.
            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"""The instance type of instance {response_source.instances.instance[i].instance_id} is {response_source.instances.instance[i].instance_type}.
    The hourly charge is {response.price_info.price.trade_price} USD.\n"""
            # Add the information of the current instance to the result.
            result += cur_result
        return result
    @classmethod
    # RAG application call
    def call_agent_app(cls,InstanceType):
        if len(InstanceType) == 0:
            return "You have no ECS instances in the current region."
        result = ""
        for i in range(len(InstanceType)):
            response = Application.call(
                # Enter the app_id of the RAG application here.
                app_id='xxx',
                prompt=f'Introduce {InstanceType[i]}',
                # Obtain the API Key for Dashscope from an environment variable.
                api_key=os.getenv("DASHSCOPE_API_KEY"))
            result += response.output.text
        return result

class Billing:
    # No input is required. The Alibaba Cloud account balance information is returned.
    @classmethod
    def get_balance(cls):
        # Create a client.
        config = open_api_models.Config(
            # Required. Make sure that the ALIBABA_CLOUD_ACCESS_KEY_ID environment variable is set in the code execution environment.
            access_key_id=os.environ['ALIBABA_CLOUD_ACCESS_KEY_ID'],
            # Required. Make sure that the ALIBABA_CLOUD_ACCESS_KEY_SECRET environment variable is set in the code execution environment.
            access_key_secret=os.environ['ALIBABA_CLOUD_ACCESS_KEY_SECRET']
        )
        # For the endpoint, see https://api.aliyun.com/product/BssOpenApi.
        config.endpoint = f'business.ap-southeast-1.aliyuncs.com'
        runtime = util_models.RuntimeOptions()
        client = BssOpenApi20171214Client(config)
        balance_info = client.query_account_balance_with_options(runtime).body.data
        return f"""The currency is {balance_info.currency}, the available amount is {balance_info.available_amount}, the credit balance is {balance_info.credit_amount},
        the Mybank balance is {balance_info.mybank_credit_amount}, the cash balance is {balance_info.available_cash_amount}, and the quota limit for ecological customers is {balance_info.quota_limit}."""

if __name__ == '__main__':
    print(Billing.get_balance())
    print(ECS.query_source('ap-southeast-1'))

from dashscope import Assistants, Messages, Runs, Threads
import json
# Import tools from tools.py.
from tools import ECS,Billing
# Import frontend display dependencies.
import gradio as gr

# Parse a string in list format into data in list format, for example, "['a','b']"-->['a','b'].
# This is used to parse the output of plannerassistant into a list whose elements are assistants.
import ast
import dashscope

# A decision-making agent that decides which agents to use and their running order.
PlannerAssistant = Assistants.create(
    # Because this agent plays a relatively important role, we recommend that you choose a model with strong performance: qwen-plus. For a list of models, see https://www.alibabacloud.com/help/en/model-studio/getting-started/models.
    model="qwen-plus",
    # Define the name of the agent.
    name='Flow Orchestration Bot',
    # Define the function description of the agent.
    description='You are the leader of a team of assistants. Based on user input, you need to decide the order in which to use these assistants.',
    # Define the instruction statement for the agent. The agent calls the tools and returns the results according to the instruction statement.
    instructions="""Your team has the following assistants. AliyunInfoAssistant: provides two capabilities: retrieving ECS instance information and checking account balance. InstanceTypeDetailAssistant: can query detailed information about specified Alibaba Cloud ECS instance types, such as the number of vCPUs and memory size. It can query multiple instance types at once, so multiple calls are not required.
    ChatAssistant: is called if the user's question does not require the two assistants above. You need to determine the order in which to use these assistants based on the user's question. Your response must be in a list format, and no other information can be returned. For example: ["AliyunInfoAssistant", "AliyunInfoAssistant","InstanceTypeDetailAssistant"] or ["ChatAssistant"]. The elements in the list can only be the assistants mentioned above."""
)

# The function is to answer daily questions. For daily questions, you can use a model with a lower price as the base of the agent.
ChatAssistant = Assistants.create(
    # Because this agent does not have high requirements for the performance of the model, the lower-cost qwen-flash model is used.
    model="qwen-flash",
    name='Daily Q&A Bot',
    description='An intelligent assistant that answers user questions.',
    instructions='Please answer the user's questions politely.'
)

# The function is to query the resource information of Alibaba Cloud. Currently, it has two methods: ECS instance query and Alibaba Cloud account balance query.
AliyunInfoAssistant = Assistants.create(
    model="qwen-plus", # For a list of models, see https://www.alibabacloud.com/help/en/model-studio/getting-started/models.
    name='Alibaba Cloud Resource Information Query Bot',
    description='An intelligent assistant that calls tools based on user queries and returns the queried Alibaba Cloud resource results.',
    instructions='You are an intelligent assistant. You currently provide two capabilities: retrieving ECS instance information and checking account balance. Please accurately determine which tool to call and answer the user's questions politely.',
    # Define the tools used by the agent. You can define one or more tools that the agent may use in the tools list based on your business scenario.
    tools=[
        {
            'type': 'function',
            'function': {
                # The name of the tool function. You can map the name to the function body through the function_mapper in the code below.
                'name': 'query_ecs_instance_info',
                # The description of the tool function.
                'description': 'This is very useful when you need to query Alibaba Cloud ECS instance information, such as instance ID, instance type, and fee information.',
                # The input parameters of the tool function.
                'parameters': {
                    'type': 'object',
                    'properties': {
                        # This tool requires the user to enter region information.
                        'RegionID': {
                            'type': 'str',
                            # The description of the parameter.
                            'description': 'The region ID of the instance that the user wants to query. The value is cn-hangzhou for the China (Hangzhou) region, cn-shanghai for the Shanghai region, or cn-beijing for the China (Beijing) region.'
                        },
                    },
                    'required': ['RegionID']},
            }
        },
        {
            'type': 'function',
            'function': {
                'name': 'query_aliyun_balance',
                # The description of the tool function.
                'description': 'This is very useful when you need to query Alibaba Cloud account information.',
                # The input parameters of the tool function. The balance query does not require input parameters, so it is empty.
                'parameters': {}
            }
        }
    ]
)

# The function is to query the detailed information of instance types through the RAG application created on the Alibaba Cloud Model Studio platform.
InstanceTypeDetailAssistant = Assistants.create(
    model="qwen-plus", # For a list of models, see https://www.alibabacloud.com/help/en/model-studio/getting-started/models.
    name='ecs_instance_type_introduction_bot',
    description='An intelligent assistant that can introduce instance type information to users by calling existing plug-in capabilities based on the instance types provided by the user.',
    instructions='You are an intelligent assistant. You need to accurately extract instance type information from the input, such as [ecs.e-c1m1.large] or [ecs.u1-c1m4.xlarge,ecs.e-c1m1.large]. Input the instance type list into the tool to obtain their detailed information.',
    tools=[
        {
            'type': 'function',
            'function': {
                'name': 'introduce_ecs_instance_type',
                'description': 'Return the information of the specified ECS instance type queried by the customer.',
                'parameters': {
                    'type': 'object',
                    'properties': {
                        'InstanceType': {
                            'type': 'list',
                            'InstanceType': 'The instance type that the user wants to query. It can be a single instance type or multiple instance types, such as [ecs.e-c1m1.large] or [ecs.u1-c1m4.xlarge,ecs.e-c1m1.large].'
                        },
                    },
                    'required': ['InstanceType']},
            }
        }
    ]
)

# In a multi-agent scenario, define an agent for summarization. This agent provides a comprehensive and complete answer to the user's question based on the user's question and the reference information output by the previous agent.
SummaryAssistant = Assistants.create(
    model="qwen-plus", # For a list of models, see https://www.alibabacloud.com/help/en/model-studio/getting-started/models.
    name='Summary Bot',
    description='An intelligent assistant that provides a comprehensive and complete answer to the user's question based on the user's question and reference information.',
    instructions='You are an intelligent assistant that provides a comprehensive and complete answer to the user's question based on the user's question and reference information.'
)

# Map the name of the tool function to the function body.
function_mapper = {
    "query_ecs_instance_info": ECS.query_source,
    "introduce_ecs_instance_type":ECS.call_agent_app,
    "query_aliyun_balance":Billing.get_balance
}
# Map the name of the agent to the agent body.
assistant_mapper = {
    "ChatAssistant": ChatAssistant,
    "AliyunInfoAssistant":AliyunInfoAssistant,
    "InstanceTypeDetailAssistant":InstanceTypeDetailAssistant
}

# Input message information and output the reply of the specified agent.
def get_agent_response(assistant, message=''):
    # Print the information of the input 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)
    # If the response fails, "run failed" is printed.
    if run_status.status == 'failed':
        print('run failed:')
    # If a tool is needed to assist the model in outputting, perform the following process.
    if run_status.required_action:
        f = run_status.required_action.submit_tool_outputs.tool_calls[0].function
        # Obtain the function name.
        func_name = f['name']
        # Obtain the input parameters of the function.
        param = json.loads(f['arguments'])
        # Print the tool information.
        print("function is",f)
        # Map the function name to the function through function_mapper, and input the parameters into the tool function to obtain the 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)
    # Return the output of the agent.
    return msgs['data'][0]['content'][0]['text']['value']

Define interaction methods between agents and obtain responses

The interaction steps between agents are orchestrated by PlannerAssistant. To adapt to the Gradio frontend display, the input and output parameters must be aligned with the Gradio components. The following is the sample code:

# Obtain the response of the multi-agent system. The input and output parameters must be aligned with the parameters in the Gradio frontend display interface.
def get_multi_agent_response(query,history):
    # Handle the case where the input is empty.
    if len(query) == 0:
        return "",history+[("","")],"",""
    # Obtain the running order of the agents.
    assistant_order = get_agent_response(PlannerAssistant,query)
    try:
        order_stk = ast.literal_eval(assistant_order)
        cur_query = query
        # Run the agents in sequence.
        for i in range(len(order_stk)):
            yield "----->".join(order_stk),history+[(query,"The multi-agent system is working hard...")],f"{order_stk[i]} is processing information...",""
            cur_assistant = assistant_mapper[order_stk[i]]
            response = get_agent_response(cur_assistant,cur_query)
            yield "----->".join(order_stk),history+[(query,"The multi-agent system is working hard...")],response,""
            # If the current agent is the last agent, use its output as the output of the multi-agent system.
            if i == len(order_stk)-1:
                yield "----->".join(order_stk),history+[(query,response)],"The assistant has finished processing.",""
            # If the current agent is not the last agent, add the output response of the previous agent to the query of the next round as reference information.
            else:
                # Add special identifiers before and after the reference information to prevent the model from confusing the reference information with the question.
                cur_query = f"You can refer to the known information:\n{response}\nYou need to fully answer the user's question. The question is: {query}."
    # Fallback policy: If the preceding program fails to run, call ChatAssistant.
    except Exception as e:
        yield "ChatAssistant",[(query,get_agent_response(ChatAssistant,query))],"",""

The input parameters are `query` and `history`. `query` is the user's question (a string), and `history` is the conversation record between the user and the multi-agent system (a list).

The first output parameter is the agent orchestration information (a string). The second output parameter is the conversation history between the user and the multi-agent system (a list). The third output parameter is the status of the currently running agent (a string). To adapt to the Gradio component, the fourth output parameter is the user's input box (a string, set to "" to clear the input box after the user submits a question).

Frontend display interface

This topic uses Gradio as a frontend display tool. Gradio helps machine learning developers quickly create interfaces to display model performance. The following code provides the details:

# Frontend display interface
with gr.Blocks() as demo:
    # Display the title in the center of the interface.
    gr.HTML('<center><h1>Welcome to Alibaba Cloud Resource Query Bot</h1></center>')
    gr.HTML('<center><h3>Supported features include ECS instance query, balance query, and instance type detail query in specified regions. You can add the tools you need in tools.py and configure related agents in main.py.</h3></center>')
    with gr.Row():
        with gr.Column(scale=10):
            chatbot = gr.Chatbot(value=[["hello","Nice to meet you! What questions do you have about Alibaba Cloud resources?"]],height=600)
        with gr.Column(scale=4):
            text1 = gr.Textbox(label="Assistant Selection")
            text2 = gr.Textbox(label="Current Assistant Status",lines=22)
    with gr.Row():
        msg = gr.Textbox(label="Input",placeholder="What would you like to know?")
    # Some example questions
    with gr.Row():
        examples = gr.Examples(examples=[
            'How much is my Alibaba Cloud account balance?',
            'What ECS instances do I have in the Singapore region? Tell me their instance IDs, prices, and instance type details.',
            'I want to know the metrics of ecs.u1-c1m4.xlarge and 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])

Complete code of main.py

from dashscope import Assistants, Messages, Runs, Threads
import json
# Import tools from tools.py.
from tools import ECS,Billing
# Import frontend display dependencies.
import gradio as gr
import ast
import dashscope

dashscope.base_http_api_url = 'https://dashscope-intl.aliyuncs.com/api/v1'
# A decision-making agent that decides which agents to use and their execution order.
PlannerAssistant = Assistants.create(
    # Because this agent plays an important role, choose a model with strong performance, such as qwen-plus. For a list of models, see https://www.alibabacloud.com/help/en/model-studio/getting-started/models.
    model="qwen-plus",  
    # Define the name of the agent.
    name='Flow Orchestration Bot',
    # Define the function description of the agent.
    description='You are the leader of a team of assistants. Based on user input, you need to decide the order in which to use these assistants.',
    # Define the instruction statement for the agent. The agent calls the tools and returns the results according to the instruction statement.
    instructions="""Your team has the following assistants. AliyunInfoAssistant: provides two capabilities: retrieving ECS instance information and checking account balance. InstanceTypeDetailAssistant: can query detailed information about specified Alibaba Cloud ECS instance types, such as the number of vCPUs and memory size. It can query multiple instance types at once, so multiple calls are not required.
    ChatAssistant: is called if the user's question does not require the two assistants above. You need to determine the order in which to use these assistants based on the user's question. Your response must be in a list format, and no other information can be returned. For example: ["AliyunInfoAssistant", "AliyunInfoAssistant","InstanceTypeDetailAssistant"] or ["ChatAssistant"]. The elements in the list can only be the assistants mentioned above."""
)

# The function is to answer daily questions. For daily questions, you can use a model with a lower price as the base of the agent.
ChatAssistant = Assistants.create(
    # Because this agent does not have high requirements for the performance of the model, the lower-cost qwen-flash model is used. For a list of models, see https://www.alibabacloud.com/help/en/model-studio/getting-started/models.
    model="qwen-flash",
    name='Daily Q&A Bot',
    description='An intelligent assistant that answers user questions.',
    instructions='Please answer the user's questions politely.'
)

# The function is to query the resource information of Alibaba Cloud. Currently, it has two methods: ECS instance query and Alibaba Cloud account balance query.
AliyunInfoAssistant = Assistants.create(
    model="qwen-plus", # For a list of models, see https://www.alibabacloud.com/help/en/model-studio/getting-started/models.
    name='Alibaba Cloud Resource Query Bot',
    description='An intelligent assistant that calls tools based on user queries and returns the queried Alibaba Cloud resource results.',
    instructions='You are an intelligent assistant. You currently provide two capabilities: retrieving ECS instance information and checking account balance. Please accurately determine which tool to call and answer the user's questions politely.',
    # Define the tools used by the agent. You can define one or more tools that the agent may use in the tools list based on your business scenario.
    tools=[
        {
            'type': 'function',
            'function': {
                # The name of the tool function. You can map the name to the function body through the function_mapper in the code below.
                'name': 'query_ecs_instance_info',
                # The description of the tool function.
                'description': 'This is very useful when you need to query Alibaba Cloud ECS instance information, such as instance ID, instance type, and fee information.',
                # The input parameters of the tool function.
                'parameters': {
                    'type': 'object',
                    'properties': {
                        # This tool requires the user to enter region information.
                        'RegionID': {
                            'type': 'str',
                            # The description of the parameter.
                            'description': 'The ID of the region where the instance resides. For example, use cn-hangzhou for China (Hangzhou), ap-southeast-1 for Singapore, or cn-beijing for China (Beijing).'
                        },
                    },
                    'required': ['RegionID']},
            }
        },
        {
            'type': 'function',
            'function': {
                'name': 'query_account_balance',
                # The description of the tool function.
                'description': 'This is very useful when you need to query Alibaba Cloud account information.',
                # The input parameters of the tool function. The balance query does not require input parameters, so it is empty.
                'parameters': {}
            }
        }
    ]
)

# The function is to query the detailed information of instance types through the RAG application created on the Alibaba Cloud Model Studio platform.
InstanceTypeDetailAssistant = Assistants.create(
    model="qwen-plus",  # For a list of models, see https://www.alibabacloud.com/help/en/model-studio/getting-started/models.
    name='ECS Instance Type Details Bot',
    description='An intelligent assistant that can accurately identify the mentioned instance types based on the user's input. It calls existing plug-in capabilities to introduce instance type information to users.',
    instructions='You are an intelligent assistant. You need to accurately identify and extract Alibaba Cloud instance type information from the user's input, such as [ecs.e-c1m1.large] or [ecs.u1-c1m4.xlarge,ecs.e-c1m1.large]. Input the instance type list into the tool to obtain their detailed information.',
    tools=[
        {
            'type': 'function',
            'function': {
                'name': 'get_ecs_instance_type_details',
                'description': 'Return the information of the specified ECS instance type queried by the customer.',
                'parameters': {
                    'type': 'object',
                    'properties': {
                        'InstanceType': {
                            'type': 'list',
                            'InstanceType': 'The instance type that the user wants to query. It can be a single instance type or multiple instance types, such as [ecs.e-c1m1.large] or [ecs.u1-c1m4.xlarge,ecs.e-c1m1.large].'
                        },
                    },
                    'required': ['InstanceType']},
            }
        }
    ]
)

# In a multi-agent scenario, define an agent for summarization. This agent provides a comprehensive and complete answer to the user's question based on the user's question and the reference information output by the previous agent.
SummaryAssistant = Assistants.create(
    model="qwen-plus",  # For a list of models, see https://www.alibabacloud.com/help/en/model-studio/getting-started/models.
    name='Summary Bot',
    description='An intelligent assistant that provides a comprehensive and complete answer to the user's question based on the user's question and reference information.',
    instructions='You are an intelligent assistant that provides a comprehensive and complete answer to the user's question based on the user's question and reference information.'
)

# Map the name of the tool function to the function body.
function_mapper = {
    "query_ecs_instance_info": ECS.query_source,
    "get_ecs_instance_type_details":ECS.call_agent_app,
    "query_account_balance":Billing.get_balance
}
# Map the name of the agent to the agent body.
assistant_mapper = {
    "ChatAssistant": ChatAssistant,
    "AliyunInfoAssistant":AliyunInfoAssistant,
    "InstanceTypeDetailAssistant":InstanceTypeDetailAssistant
}

# Get the response from a specified agent for a given message.
def get_agent_response(assistant, message=''):
    # Print the input agent information.
    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)
    # If the response fails, "run failed" is printed.
    if run_status.status == 'failed':
        print('run failed:')
    # If a tool is needed to assist the model in outputting, perform the following process.
    if run_status.required_action:
        f = run_status.required_action.submit_tool_outputs.tool_calls[0].function
        # Get the function name.
        func_name = f['name']
        # Get the input parameters of the function.
        param = json.loads(f['arguments'])
        # Print the tool information.
        print("function is",f)
        # Map the function name to the function through function_mapper, and input the parameters into the tool function to get the 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)
    # Return the output of the agent.
    return msgs['data'][0]['content'][0]['text']['value']

# Get the response from the multi-agent system. The input and output parameters must align with the Gradio frontend interface.
def get_multi_agent_response(query,history):
    # Handle empty input.
    if len(query) == 0:
        return "",history+[("","")],"",""
    # Get the agent execution order.
    assistant_order = get_agent_response(PlannerAssistant,query)
    try:
        order_stk = ast.literal_eval(assistant_order)
        cur_query = query
        Agent_Message = ""
        # Run the agents in sequence.
        for i in range(len(order_stk)):
            yield "----->".join(order_stk),history+[(query,"The multi-agent system is working...")],Agent_Message+'\n'+f"*{order_stk[i]}* is processing...",""
            cur_assistant = assistant_mapper[order_stk[i]]
            response = get_agent_response(cur_assistant,cur_query)
            Agent_Message += f"Response from *{order_stk[i]}*: {response}\n\n"
            yield "----->".join(order_stk),history+[(query,"The multi-agent system is working...")],Agent_Message,""
            # If the current agent is the last one in the sequence, generate the final response.
            if i == len(order_stk)-1:
                prompt = f"Based on the following information: {Agent_Message}, answer the user's question: {query}."
                multi_agent_response = get_agent_response(SummaryAssistant,prompt)
                yield "----->".join(order_stk),history+[(query,multi_agent_response)],Agent_Message,""
            # If not the last agent, pass the response to the next agent as context.
            else:
                # Add special identifiers to prevent the model from confusing context with the original question.
                cur_query = f"You can refer to the following information: {response}. You must provide a complete answer to the user's question. The question is: {query}."
    # Fallback policy: If the program fails, call ChatAssistant directly.
    except Exception as e:
        yield "ChatAssistant",[(query,get_agent_response(ChatAssistant,query))],"",""


# Frontend display interface
with gr.Blocks() as demo:
    # Display the title in the center of the interface.
    gr.HTML('<center><h1>Welcome to Alibaba Cloud Resource Query Bot</h1></center>')
    gr.HTML('<center><h3>Supported features include ECS instance query, balance query, and instance type detail query in specified regions. You can add the tools you need in tools.py and configure related agents in main.py.</h3></center>')
    with gr.Row():
        with gr.Column(scale=10):
            chatbot = gr.Chatbot(value=[["hello","Hello! I'm happy to help. What would you like to know about your Alibaba Cloud resources?"]],height=600)
        with gr.Column(scale=4):
            text1 = gr.Textbox(label="Assistant selection")
            text2 = gr.Textbox(label="Current assistant status",lines=22)
    with gr.Row():
        msg = gr.Textbox(label="Input",placeholder="What would you like to know?")
    # Some example questions
    with gr.Row():
        examples = gr.Examples(examples=[
            'What is my Alibaba Cloud account balance?',
            'Tell me about my ECS instances in the Singapore region, including their IDs, prices, and instance type details.',
            'I want to know the metrics for ecs.u1-c1m4.xlarge and 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()

Result

After you set the ALIBABA_CLOUD_ACCESS_KEY_ID, ALIBABA_CLOUD_ACCESS_KEY_SECRET, and DASHSCOPE_API_KEY environment variables, run the main.py file. The terminal displays the Running on local URL: output. Open the URL to access the interactive interface. Enter a question, such as What ECS instances do I have in the Singapore region, and what is my Alibaba Cloud account balance?, or click an example question in Examples to add it to the input box. Then, press Enter to generate the result. You can monitor the agent's real-time status in the Current Assistant Status box.

Conclusion

This topic describes how to integrate a Model Studio RAG application with Alibaba Cloud OpenAPI capabilities into an agent. It also explains the steps to use the Assistants API for multi-agent development and how to display the results using a Gradio-based frontend interface.

You can adapt the multi-agent system for your business by modifying the system prompts, refining the interaction methods between agents, and modifying the tools.