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=os.getenv("DASHSCOPE_API_KEY"),
         base_url="https://dashscope.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,大语言模型填写/v1/chat/completions，embedding文本向量模型填写/v1/embeddings
         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,如果您需要输入OSS文件,可将下行替换为：input_file_id = "实际的OSS文件URL或资源标识符"
             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://help.aliyun.com/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://help.aliyun.com/zh/model-studio/developer-reference/error-code")
         except Exception as e:
             print(f"An error occurred: {e}")
             print(f"参见错误码文档: https://help.aliyun.com/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设置为：

   1. 大语言模型填写/v1/chat/completions

   2. embedding文本向量模型填写/v1/embeddings

2. 替换上面Python脚本中的endpoint，与输入文件中的url参数保持一致

3. 运行脚本，等待任务完成，若任务成功，将在同一目录下生成输出结果文件result.jsonl

   若任务失败，则程序退出并打印错误信息。

   如果存在错误文件ID，将在同一目录下生成错误文件error.jsonl以供检查。

   在过程中发生的异常会被捕获，并打印错误信息。

一个单行请求示例：

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

一个多行请求示例：

{"custom_id":"1","method":"POST","url":"/v1/chat/completions","body":{"model":"qwen-max","messages":[{"role":"system","content":"You are a helpful assistant."},{"role":"user","content":"你好！有什么可以帮助你的吗？"}]}}
{"custom_id":"2","method":"POST","url":"/v1/chat/completions","body":{"model":"qwen-max","messages":[{"role":"system","content":"You are a helpful assistant."},{"role":"user","content":"What is 2+2?"}]}}

请注意：

• Embedding文本向量模型目前仅支持两种输入：字符串和字符串列表

• 选择Embedding文本向量模型进行调用时，请求中url的值需填写为"/v1/embeddings"

字符串示例：

{"custom_id": "request-1", "method": "POST", "url": "/v1/embeddings", "body": {"model": "text-embedding-v3", "input": "衣服的质量杠杠的，很漂亮，不枉我等了这么久啊，喜欢，以后还来这里买", "encoding_format": "float"}}

字符串列表示例：

{"custom_id": "request-1", "method": "POST", "url": "/v1/embeddings", "body": {"model": "text-embedding-v2", "input": "['风急天高猿啸哀','渚清沙白鸟飞回','无边落木萧萧下','不尽长江滚滚来']", "encoding_format": "float"}}

输入参数配置

返回示例

请参见创建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任务的返回参数。

方式一：使用OSS文件URL

1. 获取文件URL：

   1. OSS管理控制台方式：

      1. 进入Bucket列表页面，找到目标Bucket并单击名称；

      2. 在文件列表中定位目标文件，单击右侧详情按钮；

         您可在文件列表中新建子文件夹如Batch/20250317并上传文件，示例test.jsonl。

      3. 在弹出面板中，单击复制文件URL。

   2. SDK方式：生成OSS文件URL，参考使用预签名URL下载文件。

2. 参数配置：参数input_file_id填写OSS文件URL，示例：

   input_file_id="https://testbucket.oss-cn-beijing.aliyuncs.com/xxx/xxx/Batch/20250317/test.jsonl?Expires=xxx"

方式二：使用OSS文件资源标识符

1. 授权百炼访问当前账号下的OSS Bucket：

   参阅OSS服务关联角色授权步骤完成授权并对目标Bucket添加标签。

2. 参数配置：

   1. 参数input_file_id填写OSS文件的资源标识符，格式为oss:{region}:{bucket}/{file_path}，示例：

      input_file_id="oss:cn-beijing:bucketTest/Batch/20250313/test.jsonl"

   2. 参数说明表：

      字段	类型	必选	描述

      字段	类型	必选	描述
      region	String	是	OSS的bucket所在地域，如cn-beijing。
      bucket	String	是	OSS的bucket名称，如bucketTest。
      file_path	String	是	OSS文件路径，如Batch/20250313/test.jsonl。