Alibaba Cloud Model Studio：OpenAI相容-Batch

在正式Batch任務開始之前，您可使用測試模型batch-test-model進行全鏈路閉環測試，包括驗證輸入資料、建立任務、查詢任務及下載結果檔案等流程。請注意：

• 測試檔案不僅需滿足輸入檔案要求，還需滿足單檔案大小不超過1MB，且檔案行數不超過100行的要求；

• 並發限制：最大並行任務數2個；

• 資源使用：測試模型不走推理流程，因此不產生模型推理費用。

具體步驟如下：

1. 準備測試檔案

   • 將包含請求資訊的樣本檔案test_model.jsonl下載到本地，並確保其與下方的Python指令碼置於同一目錄下。

   • 樣本內容：model參數設定為batch-test-model，url設定為/v1/chat/ds-test 。

     {"custom_id":"1","method":"POST","url":"/v1/chat/ds-test","body":{"model":"batch-test-model","messages":[{"role":"system","content":"You are a helpful assistant."},{"role":"user","content":"你好！有什麼可以協助你的嗎？"}]}}
     {"custom_id":"2","method":"POST","url":"/v1/chat/ds-test","body":{"model":"batch-test-model","messages":[{"role":"system","content":"You are a helpful assistant."},{"role":"user","content":"What is 2+2?"}]}}

2. 運行指令碼

   • 執行此Python指令碼。

     如果需要調整檔案路徑或其他參數，請根據實際情況修改代碼。

     import os
     from pathlib import Path
     from openai import OpenAI
     import time
     
     # 初始化用戶端
     client = OpenAI(
         # 若沒有配置環境變數,可用阿里雲百鍊API Key將下行替換為：api_key="sk-xxx",但不建議在生產環境中直接將API Key寫入程式碼到代碼中,以減少API Key泄露風險.
         # 新加坡和北京地區的API Key不同。擷取API Key：https://www.alibabacloud.com/help/zh/model-studio/get-api-key
         api_key=os.getenv("DASHSCOPE_API_KEY"),
         # 以下是新加坡地區的base_url，如果使用北京地區的模型，需要將base_url替換為：https://dashscope.aliyuncs.com/compatible-mode/v1
         base_url="https://dashscope-intl.aliyuncs.com/compatible-mode/v1"  # 阿里雲百鍊服務的base_url
     )
     
     def upload_file(file_path):
         print(f"正在上傳包含請求資訊的JSONL檔案...")
         file_object = client.files.create(file=Path(file_path), purpose="batch")
         print(f"檔案上傳成功。得到檔案ID: {file_object.id}\n")
         return file_object.id
     
     def create_batch_job(input_file_id):
         print(f"正在基於檔案ID，建立Batch任務...")
         # 請注意:此處endpoint參數值需和輸入檔案中的url欄位保持一致.測試模型(batch-test-model)填寫/v1/chat/ds-test,Embedding文本向量模型填寫/v1/embeddings,其他模型填寫/v1/chat/completions
         batch = client.batches.create(input_file_id=input_file_id, endpoint="/v1/chat/ds-test", completion_window="24h")
         print(f"Batch任務建立完成。 得到Batch任務ID: {batch.id}\n")
         return batch.id
     
     def check_job_status(batch_id):
         print(f"正在檢查Batch任務狀態...")
         batch = client.batches.retrieve(batch_id=batch_id)
         print(f"Batch任務狀態: {batch.status}\n")
         return batch.status
     
     def get_output_id(batch_id):
         print(f"正在擷取Batch任務中執行成功請求的輸出檔案ID...")
         batch = client.batches.retrieve(batch_id=batch_id)
         print(f"輸出檔案ID: {batch.output_file_id}\n")
         return batch.output_file_id
     
     def get_error_id(batch_id):
         print(f"正在擷取Batch任務中執行錯誤請求的輸出檔案ID...")
         batch = client.batches.retrieve(batch_id=batch_id)
         print(f"錯誤檔案ID: {batch.error_file_id}\n")
         return batch.error_file_id
     
     def download_results(output_file_id, output_file_path):
         print(f"正在列印並下載Batch任務的請求成功結果...")
         content = client.files.content(output_file_id)
         # 列印部分內容以供測試
         print(f"列印請求成功結果的前1000個字元內容: {content.text[:1000]}...\n")
         # 儲存結果檔案至本地
         content.write_to_file(output_file_path)
         print(f"完整的輸出結果已儲存至本地輸出檔案result.jsonl\n")
     
     def download_errors(error_file_id, error_file_path):
         print(f"正在列印並下載Batch任務的請求失敗資訊...")
         content = client.files.content(error_file_id)
         # 列印部分內容以供測試
         print(f"列印請求失敗資訊的前1000個字元內容: {content.text[:1000]}...\n")
         # 儲存錯誤資訊檔至本地
         content.write_to_file(error_file_path)
         print(f"完整的請求失敗資訊已儲存至本地錯誤檔案error.jsonl\n")
     
     def main():
         # 檔案路徑
         input_file_path = "test_model.jsonl"  # 可替換為您的輸入檔案路徑
         output_file_path = "result.jsonl"  # 可替換為您的輸出檔案路徑
         error_file_path = "error.jsonl"  # 可替換為您的錯誤檔案路徑
         try:
             # Step 1: 上傳包含請求資訊的JSONL檔案,得到輸入檔案ID
             input_file_id = upload_file(input_file_path)
             # Step 2: 基於輸入檔案ID,建立Batch任務
             batch_id = create_batch_job(input_file_id)
             # Step 3: 檢查Batch任務狀態直到結束
             status = ""
             while status not in ["completed", "failed", "expired", "cancelled"]:
                 status = check_job_status(batch_id)
                 print(f"等待任務完成...")
                 time.sleep(10)  # 等待10秒後重新查詢狀態
             # 如果任務失敗,則列印錯誤資訊並退出
             if status == "failed":
                 batch = client.batches.retrieve(batch_id)
                 print(f"Batch任務失敗。錯誤資訊為:{batch.errors}\n")
                 print(f"參見錯誤碼文檔: https://www.alibabacloud.com/help/zh/model-studio/developer-reference/error-code")
                 return
             # Step 4: 下載結果：如果輸出檔案ID不為空白,則列印請求成功結果的前1000個字元內容，並下載完整的請求成功結果到本地輸出檔案;
             # 如果錯誤檔案ID不為空白,則列印請求失敗資訊的前1000個字元內容,並下載完整的請求失敗資訊到本地錯誤檔案.
             output_file_id = get_output_id(batch_id)
             if output_file_id:
                 download_results(output_file_id, output_file_path)
             error_file_id = get_error_id(batch_id)
             if error_file_id:
                 download_errors(error_file_id, error_file_path)
                 print(f"參見錯誤碼文檔: https://www.alibabacloud.com/help/zh/model-studio/developer-reference/error-code")
         except Exception as e:
             print(f"An error occurred: {e}")
             print(f"參見錯誤碼文檔: https://www.alibabacloud.com/help/zh/model-studio/developer-reference/error-code")
     
     if __name__ == "__main__":
         main()

3. 驗證測試結果

   • 任務狀態顯示completed。

   • 結果檔案result.jsonl：包含固定響應{"content":"This is a test result."}。

     {"id":"a2b1ae25-21f4-4d9a-8634-99a29926486c","custom_id":"1","response":{"status_code":200,"request_id":"a2b1ae25-21f4-4d9a-8634-99a29926486c","body":{"created":1743562621,"usage":{"completion_tokens":6,"prompt_tokens":20,"total_tokens":26},"model":"batch-test-model","id":"chatcmpl-bca7295b-67c3-4b1f-8239-d78323bb669f","choices":[{"finish_reason":"stop","index":0,"message":{"content":"This is a test result."}}],"object":"chat.completion"}},"error":null}
     {"id":"39b74f09-a902-434f-b9ea-2aaaeebc59e0","custom_id":"2","response":{"status_code":200,"request_id":"39b74f09-a902-434f-b9ea-2aaaeebc59e0","body":{"created":1743562621,"usage":{"completion_tokens":6,"prompt_tokens":20,"total_tokens":26},"model":"batch-test-model","id":"chatcmpl-1e32a8ba-2b69-4dc4-be42-e2897eac9e84","choices":[{"finish_reason":"stop","index":0,"message":{"content":"This is a test result."}}],"object":"chat.completion"}},"error":null}

     如遇到錯誤，請參考錯誤資訊進行解決。

通過測實驗證後，您可以通過以下步驟來執行正式的Batch任務流程。

1. 參考輸入檔案要求準備輸入檔案，並將檔案中的model參數設定為支援的模型，url設定為：/v1/chat/completions；

2. 替換上面Python指令碼中的endpoint；

   重要

   請確保指令碼中的endpoint與輸入檔案中的url參數保持一致。

3. 運行指令碼，等待任務完成，若任務成功，將在同一目錄下產生輸出結果檔案result.jsonl。

   若任務失敗，則程式退出並列印錯誤資訊。

   如果存在錯誤檔案ID，將在同一目錄下建置錯誤檔案error.jsonl以供檢查。

   在過程中發生的異常會被捕獲，並列印錯誤資訊。

{"custom_id":"1","method":"POST","url":"/v1/chat/completions","body":{"model":"qwen-plus","messages":[{"role":"system","content":"You are a helpful assistant."},{"role":"user","content":"Hello!"}]}}
{"custom_id":"2","method":"POST","url":"/v1/chat/completions","body":{"model":"qwen-plus","messages":[{"role":"system","content":"You are a helpful assistant."},{"role":"user","content":"What is 2+2?"}]}}

JSONL 批量產生工具

使用以下工具可快速產生 JSONL 檔案。為避免卡頓，建議單次處理的資料不超過 10,000 行，如果資料量較大，請分批操作。

輸入參數配置

返回樣本

請參見建立Batch任務的返回樣本。

返回參數

請參見建立Batch任務的返回參數。

返回參數中output_file_id和error_file_id可以通過下載Batch結果檔案擷取內容。

輸入參數配置

{
  "object": "list",
  "data": [
    {
      "id": "batch_xxx",
      "object": "batch",
      "endpoint": "/v1/chat/completions",
      "errors": null,
      "input_file_id": "file-batch-xxx",
      "completion_window": "24h",
      "status": "completed",
      "output_file_id": "file-batch_output-xxx",
      "error_file_id": null,
      "created_at": 1722234109,
      "in_progress_at": 1722234109,
      "expires_at": null,
      "finalizing_at": 1722234165,
      "completed_at": 1722234165,
      "failed_at": null,
      "expired_at": null,
      "cancelling_at": null,
      "cancelled_at": null,
      "request_counts": {
        "total": 100,
        "completed": 95,
        "failed": 5
      },
      "metadata": {}
    },
    { ... }
  ],
  "first_id": "batch_xxx",
  "last_id": "batch_xxx",
  "has_more": true
}

輸入參數配置

返回樣本

請參見建立Batch任務的返回樣本。

返回參數

請參見建立Batch任務的返回參數。