簡介
本文基於阿里雲百鍊上已經建立的RAG應用（操作方法請參考最佳實務-基於RAG的官方文檔助手（下線）），結合使用者本地Function，協助使用者實現更加定製的應用。
您可以通過本篇協助文檔實現：
-
在阿里雲百鍊官方文檔助手的基礎上，結合百鍊平台相關API的調用，實現以下功能調用：
-
返回阿里雲百鍊賬戶當前API Key；
-
當前模型訓練情況；
-
擷取阿里雲百鍊計費賬單；
-
依賴功能
-
Assistant API的基礎API參數產生能力。
-
Assistant API的Function Call能力。
-
阿里雲百鍊應用的RAG能力。
前期準備
-
已開通阿里雲百鍊服務：產品開通。
-
已建立API_KEY: 擷取API Key與API Host。
-
已安裝最新版SDK：安裝SDK。
-
已建立應用： 應用開發，並擷取到APP_ID：擷取APP ID 和 Workspace ID。
設定您的API-KEY，替換YOUR_DASHCOPE_API_KEY為您自己的API key
export DASHSCOPE_API_KEY=YOUR_DASHSCOPE_API_KEYdef call_rag_app(prompt='請問百鍊平台如何擷取API KEY？'):
#prompt= 請問百鍊平台提供的混合訓練能力是什嗎？
#請問百鍊平台如何擷取API KEY？
response = Application.call(app_id='your_app_id',
prompt=prompt)
if response.status_code != HTTPStatus.OK:
print('request_id=%s, code=%s, message=%s\n' % (response.request_id, response.status_code, response.message))
answer = response.status_code
else:
print('request_id=%s\n output=%s\n usage=%s\n' % (response.request_id, response.output, response.usage))
answer = response.output["text"]
return answer本地Function
-
使用者可以自行準備各類待使用的私人API或者Function能力，用於處理本地調用的請求。
-
本文的最佳實務採用類比的function類比真實情況。
def get_api_key(user_id):
return "sk_" + user_id+' You have incurred a charge of RMB 8.88'
#
def get_code(user_id):
return 'https://www.alibabacloud.com/help/en/model-studio/developer-reference/sdk-example'
def get_model_cost(user_id):
answer =""
# 此處以'qwen-max','qwen-turbo','qwen_plus'為例，您可按需更換模型名稱。模型列表：https://www.alibabacloud.com/help/zh/model-studio/getting-started/models
model_name=['qwen-max','qwen-turbo','qwen_plus']
for model in model_name:
answer+='\nmodel_name: '+ model+ '\ncost '+str(len(model))+' \n token usage'+str(999)
return str(answer)
def get_train_model(user_id):
answer =""
model_list=['qwen-max-sft-v1']
for model in model_list:
answer+='\nmodel_name: '+ model+ '\ntrain_acc '+str(86.44)+' \n training time'+'21 hours 40 mintues'
return str(answer)
def get_application(user_id):
# mocked api key
return str([
{'app_id':'123', 'app 名稱':'智能助理'},
{'app_id': '124', 'app 名稱': '旅遊助理'},
])
基於Assistant API，實現Function Call
實現在通過檢索文檔的基礎上，完成各類本地Function Call的能力，具體實現代碼如下：
-
call_rag_app 模組：使用之前建立的應用
-
send_message 模組：傳遞message給assistant
-
user_config ： 外部或環境變數中的部分使用者資訊
# coding: utf8
from http import HTTPStatus
from dashscope import Application, Assistants, Messages, Runs, Threads
from bailian_function_utils import *
import json
import os
import dashscope
dashscope.base_http_api_url = 'https://dashscope-intl.aliyuncs.com/api/v1'
# -*- encoding: utf-8 -*-
def create_assistant():
# create assistant with information
assistant = Assistants.create(
# 此處以qwen-max為例，您可按需更換模型名稱。模型列表：https://www.alibabacloud.com/help/zh/model-studio/getting-started/models
model="qwen-max",
name='smart helper',
description='一個智能助手，可以通過使用者訴求，調用已有的外掛程式能力協助使用者。',
instructions='你是一個資訊處理者，不會和使用者進行二次互動。首先需要總結使用者當前輸入。然後根據已有外掛程式補充使用者的語句。'
'請按優先順序依次查詢下列情況：'
'1.若提及訓練模型相關，必須調用擷取訓練模型情況外掛程式，該調用無需詢問使用者具體模型情況，擷取訓練模型情況會返回所有結果，請都展示給使用者。'
'2.若提及assistant-api，必須調用最佳實務查詢外掛程式，返回具體assistant-api的最佳實務代碼連結，該調用無需詢問使用者。'
'3.若上述情況都沒出現，還詢問api-key相關，請返回他的apikey具體值。'
'請不要主動詢問使用者，任何潛在的調用請主動發起。',
tools=[
{
'type': 'function',
'function': {
'name': '最佳實務查詢',
'description': '返回assistant-api相關的最佳實務代碼連結',
'parameters': {
'type': 'object',
'properties': {
'user_id': {
'type': 'str',
'description': '使用者id'
},
},
'required': ['user_id']},
},
},
{
'type': 'function',
'function': {
'name': '擷取訓練模型情況',
'description': '用於擷取使用者部署在百鍊平台上用資料額外訓練的模型情況。包含SFT模型的具體情況',
'parameters': {
'type': 'object',
'properties': {
'user_id': {
'type': 'str',
'description': '使用者id'
},
},
'required': ['user_id']
}
}
},
{
'type': 'function',
'function': {
'name': '應用建立情況',
'description': '可以查詢應用相關資訊。用於返回使用者當前應用的具體建立情況，即使用者路徑下的應用建立數量和對應的id。',
'parameters': {
'type': 'object',
'properties': {
'user_id': {
'type': 'str',
'description': '使用者id'
},
},
'required': ['user_id']
}
}
},
{
'type': 'function',
'function': {
'name': '模型用量查詢',
'description': '一個通用的超大規模語言模型用量查詢介面，直接查詢下所有模型（含千問）使用量。',
'parameters': {
'type': 'object',
'properties': {
'user_id': {
'type': 'str',
'description': '使用者id'
},
},
'required': ['user_id']
}
}
},
{
'type': 'function',
'function': {
'name': 'apikey查詢',
'description': '用於apikey的具體值',
'parameters': {
'type': 'object',
'properties': {
'user_id': {
'type': 'str',
'description': '使用者id'
},
'required': ['user_id']
}
}
},
},
],
)
return assistant
function_mapper = {
"apikey查詢": get_api_key,
"擷取訓練模型情況": get_train_model,
"應用建立情況": get_application,
"最佳實務查詢": get_code,
}
def send_message(assistant, message=''):
print(f"Query: {message}")
# create a thread.
thread = Threads.create()
print(thread)
# create a message.
message = Messages.create(thread.id, content=message)
print(message)
run = Runs.create(thread.id, assistant_id=assistant.id)
print(run)
# # get run statue
# run_status = Runs.get(run.id, thread_id=thread.id)
# print(run_status)
# wait for run completed or requires_action
run_status = Runs.wait(run.id, thread_id=thread.id)
print('外掛程式調用前：')
print(run_status)
if run_status.status == 'failed':
print('run failed:')
print(run_status.last_error)
# if prompt input tool result, submit tool result.
if run_status.required_action:
f = run_status.required_action.submit_tool_outputs.tool_calls[0].function
func_name = f['name']
param = json.loads(f['arguments'])
print(f)
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)
# should wait for run completed
run_status = Runs.wait(run.id, thread_id=thread.id)
print(run_status)
# verify_status_code(run_status)
run_status = Runs.get(run.id, thread_id=thread.id)
# print(run_status)
# verify_status_code(run_status)
# get the thread messages.
msgs = Messages.list(thread.id)
# print(msgs)
# print(json.dumps(msgs, default=lambda o: o.__dict__, sort_keys=True, indent=4))
print("運行結果:")
for message in msgs['data'][::-1]:
print("content: ", message['content'][0]['text']['value'])
print("\n")
if __name__ == '__main__':
assistant = create_assistant()
# answer = call_rag_app(prompt='模型訓練？') # example 1
answer = call_rag_app(prompt='請問Assistant API如何建立？') # example 2
# 'assistant-api怎麼使用'#output["text"]#call_rag_app(請問Assistant API如何建立？)
user_config = os.environ["user_config"]
send_message(assistant=assistant, message=user_config + answer)
運行結果
測試案例1
-
輸入：“請問Assistant API如何建立？”
-
answer = call_rag_app(prompt='請問Assistant API如何建立？')
-
-
運行結果：
{'name': '最佳實務查詢', 'arguments': '{"user_id": "user_123456"}', 'output': None}
運行結果:
content: 目前使用者是user_123456。建立Assistant API主要涉及幾個關鍵步驟，以下是根據提供的文檔內容整理的簡要流程：
1. **準備開發環境**：
- 確保你的系統支援Python 3.8及以上版本。
- 使用pip命令安裝Dashscope庫：`pip install dashscope>=1.17.0`
- 設定環境變數以包含你的百鍊API密鑰：`export DASHSCOPE_API_KEY=YOUR_DASHSCOPE_API_KEY`，記得將`YOUR_DASHSCOPE_API_KEY`替換為實際的API密鑰。
2. **建立Assistant**：
- 匯入必要的庫：`from http import HTTPStatus import dashscope`
- 使用`dashscope.Assistants.create`方法來建立Assistant。你需要提供模型名稱（如`qwen-max`）、Assistant的名字、描述以及指令等資訊。還可以指定使用的工具或函數，例如文本到映像轉換、搜尋引擎調用或自訂函數調用等。樣本如下：
```python
assistant = dashscope.Assistants.create(
model='qwen-max',
name='My helper',
description='A tool helper.',
instructions='You are a helpful assistant. When asked a question, use tools wherever possible.',
tools=[
{'type': 'text_to_image'}, # 映像產生外掛程式
{'type': 'quark_search'}, # 搜尋引擎外掛程式
{
'type': 'function',
'function': {
'name': 'big_add',
'description': 'Add two numbers',
'parameters': {
'type': 'object',
'properties': {
'left': {'type': 'integer', 'description': 'The left operand'},
'right': {'type': 'integer', 'description': 'The right operand'}
},
'required': ['left', 'right']
}
}
}
]
)
```
- 建立後，檢查`assistant.status_code`是否為`HTTPStatus.OK`以確認Assistant建立成功，並擷取其ID。
3. **建立Thread**：
- 為了與Assistant互動，你需要建立一個Thread。Thread是承載訊息（Message）的容器，每個Message代表一次使用者提問或Assistant回複。
```python
thread = dashscope.Threads.create(messages=[{"role": "user", "content": "Tell me something about AI."}])
```
- 同樣，檢查`thread.status_code`以確保Thread建立成功，並記錄Thread ID以便後續使用。
完成上述步驟後，你便成功建立了一個Assistant，並且為它建立了第一個對話Thread，接下來就可以向這個Thread發送訊息來擷取Assistant的響應了。
content: 為了協助您更好地實現這些步驟，這裡有關於[建立和使用Assistant API的詳細文檔](https://www.alibabacloud.com/help/zh/model-studio/developer-reference/assistantapi/)。這份文檔包含了具體的操作指南和程式碼範例，您可以參考以確保您的Assistant API開發過程順利進行。
測試案例2
-
輸入：“模型訓練？”
-
answer = call_rag_app(prompt='模型訓練？')
-
-
運行結果：
{'name': '擷取訓練模型情況', 'arguments': '{"user_id": "user_123456"}', 'output': None}
運行結果:
content: 目前使用者是user_123456。模型訓練是機器學習和深度學習中的關鍵環節，它涉及使用特定的資料集來調整模型的內部參數，以便模型能夠根據輸入資料做出準確的預測或決策。在阿里雲百鍊平台上，模型訓練具體步驟如下：
1. **準備訓練集資料**：
- 首先，你需要準備符合要求的訓練資料，這些資料應反映你希望模型學習的知識或技能。支援的檔案格式包括jsonl、xls、xlsx等，且資料集至少包含40條內容。你可以在“訓練資料”頁面上傳資料，如果需要，可下載提供的模板來格式化你的資料。
2. **選擇訓練方式**：
- 百鍊平台提供了不同的訓練方式，包括全參訓練、高效訓練和RLHF訓練。全參訓練會更新所有模型參數，適合學習新能力和全域最佳化，但訓練時間長；高效訓練利用矩陣分解加速訓練，適合局部最佳化；RLHF訓練則通過獎勵機制在強化學習架構下最佳化模型，適合特定效能提升。
3. **配置訓練參數**：
- 在“模型調優”模組，你可以選擇“訓練新模型”，然後配置訓練參數，如選擇基座模型、訓練集、驗證資料切分方式、混合訓練比例（可選）以及超參數配置。混合訓練允許你將自有的訓練資料與平台的千問基本模型資料按一定比例混合，以增強模型的泛化能力。
4. **開始訓練**：
- 預覽訓練配置無誤後，提交訓練任務。訓練任務會被排隊執行，期間你可以在“模型管理列表”中查看訓練狀態和相關指標。請注意，訓練過程會產生費用，費用計算基於訓練集token數和迴圈次數，價格為0.1元/千tokens。
5. **監控與評估**：
- 訓練過程中，可以通過平台監控訓練進度和效能指標。訓練完成後，系統會通知你，你可以進一步測試和評估模型效果，決定是否部署應用。
6. **注意系統升級資訊**：
- 特別提醒，2024年3月15日至3月16日平台有計劃的系統升級，期間訓練任務可能會受到影響，建議避開此時間段安排訓練任務。
確保在訓練前已瞭解並同意相關的費用政策，並根據模型最佳化的需求，合理選擇訓練策略和參數，以達到最佳訓練效果。
content: user_123456，您在阿里雲百鍊平台上已有一個訓練完成的模型，詳情如下：
- **模型名稱**: qwen-max-sft-v1
- **訓練準確率**: 86.44%
- **訓練時間長度**: 21小時40分鐘
此模型已經過訓練並可能準備好部署或進一步評估。請根據您的需求決定後續步驟。另外，請留意平台即將於2024年3月15日至3月16日進行的系統升級安排，以避免在此期間安排新的訓練任務。
測試案例3
-
輸入：“我的apikey在阿里雲百鍊的賬單？”
-
answer = call_rag_app(prompt='我的apikey在阿里雲百鍊的賬單？')
-
-
運行結果：
{'name': 'apikey查詢', 'arguments': '{"user_id": "user_123456"}', 'output': None}
運行結果:
content: 目前使用者是user_123456。在百鍊平台上，API-KEY的使用與賬單直接關聯，其計費依據調用的模型及相應的費率進行計算<ref>[4][5]</ref>。當你通過API或SDK調用百鍊平台的服務時，確保瞭解你的API-KEY的許可權範圍，這會影響到你能夠訪問的資源類型以及是否會產生費用<ref>[1]</ref>。具體到賬單資訊，雖然提供的文檔沒有直接展示賬單詳情，但提到不同的模型有不同的計費單價，例如`qwen-turbo`, `qwen-plus`有其特定的計費標準，並且存在免費額度以及基礎限流規則來控制使用成本和資源分派<ref>[4][5]</ref>。若要查看或管理與API-KEY相關的費用和使用方式，通常需要訪問阿里雲的計費管理介面或百鍊平台提供的相應板塊進行查詢<ref>[4][5]</ref>。
content: 您的API-KEY為`sk_user_123456`，截至目前，您已消費8.88元。 若要深入瞭解費用細節和管理您的API-KEY使用方式，請訪問阿里雲的計費管理介面或百鍊平台的相關板塊。