大模型服务平台百炼：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任务的返回参数。