效果展示

通過本文提供的教程，您可以實現一個能進行阿里云資源信息查詢的 Multi Agent 系統。該系統在收到用戶問題后，會自動規劃多個 Agent 之間的協作流程，按照規劃執行任務后給出最終答案。

詳細工作流程：

1. 用戶進行提問；

2. PlannerAssistant接收到用戶的提問，并根據用戶問題對其它Agent進行選擇并編排其運行順序；

3. ChatAssistant、AliyunInfoAssistant與InstanceTypeDetailAssistant分別有對應的職責與能力，它們是本教程Multi Agent系統中被PlannerAssistant編排的Agent。編排完成后的Agent組合會接收用戶的提問，并按照程序中設計的Multi Agent交互邏輯進行分工合作；

4. SummaryAssistant將各Agent的輸出信息匯總，并結合用戶問題對信息進行總結，其輸出會作為Multi Agent系統的最終輸出。

前提條件

1. 請參考創建AccessKey，獲取用戶的ACCESS_KEY_ID和ACCESS_KEY_SECRET。

2. 請開通百煉服務并獲取API-KEY。

3. 我們推薦您將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信息進行替換
   $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. 如果您沒有創建ECS實例，請前往ECS控制臺創建一個按量付費的實例。

   說明

   如果您在完成本實踐教程后沒有繼續使用ECS實例的需求，請及時釋放您的ECS資源，避免產生不必要的消費。

5. 請確認您的計算環境中已安裝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

6. 請您參考最佳實踐-基于RAG的官方文檔助手文檔，在百煉平臺創建RAG應用，知識庫文件選擇PDF格式的阿里云ECS官方文檔實例規格族.pdf，并獲取其應用ID。

代碼實現

您需要準備兩個py文件，分別為tools.py與main.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}。"""

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'))

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 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']

定義Agent之間交互方式并獲得回復

Agent之間的交互步驟根據PlannerAssistant進行編排，為了適配Gradio的前端界面展示，輸入輸出的參數需要與Gradio中的組件進行對齊。代碼如下：

# 獲得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組件，第四個輸出參數為用戶的輸入框（字符串形式，設為""以達到用戶發起提問后將輸入框清空的效果）。

前端展示界面

本教程使用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])

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的實時狀態。

總結

通過本教程，您可以了解到將百煉RAG應用與阿里云的OpenAPI能力集成到Agent中的方式，以及使用百煉平臺的Assistants API進行Multi Agent開發的步驟，并最終通過基于Gradio的前端界面展示出來。

您可以通過修改Agent中的提示詞、細化Agent之間的交互方式、修改工具函數等方法，將Multi Agent應用到您的業務中。