OpenAI Batch接口兼容

前提条件

• 已开通阿里云百炼服务，并已获取API-KEY：获取API Key。

  建议您配置API Key到环境变量中以降低API-KEY的泄露风险。

• 如果您使用OpenAI Python SDK调用Batch接口，请通过以下命令安装最新版OpenAI SDK。

  pip3 install -U openai

支持的模型

• qwen-max

• qwen-plus

• qwen-turbo

• qwen-long

• qwen-vl-max

• qwen-vl-plus

• qwq-32b-preview

计费

Batch调用的费用为实时调用的50%，具体请参见模型列表。免费额度不支持抵扣Batch调用产生的费用。

快速开始

您可以通过以下步骤来执行完整的Batch任务。此Python示例代码将上传符合输入文件格式的 test.jsonl 文件，启动并监控Batch任务，待任务完成后获取结果。

1. 准备文件：将包含请求信息的示例文件test.jsonl下载到本地，并确保其与下方的Python脚本置于同一目录下。

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任务...")
       batch = client.batches.create(input_file_id=input_file_id, endpoint="/v1/chat/completions", 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.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)
               if status == "validating" or status == "in_progress":
                   print(f"等待任务完成...")
                   time.sleep(10)  # 等待10秒后再次查询状态
               else:
                   break
           # 如果任务失败，则打印错误信息并退出
           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. 获取结果：任务完成时，若任务成功，将在同一目录下生成输出结果文件result.jsonl。

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

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

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

数据文件格式说明

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

[
  {"role": "system", "content": "You are a helpful assistant."},
  {"role": "user", "content": "What is 2+2?"}
]

import csv
import json
def messages_builder_example(content):
    messages = [{"role": "system", "content": "You are a helpful assistant."}, {"role": "user", "content": content}]
    return messages

with open("input_demo.csv", "r") as fin:
    with open("input_demo.jsonl", 'w', encoding='utf-8') as fout:
        csvreader = csv.reader(fin)
        for row in csvreader:
            body = {"model": "qwen-turbo", "messages": messages_builder_example(row[1])}
            request = {"custom_id": row[0], "method": "POST", "url": "/v1/chat/completions", "body": body}
            fout.write(json.dumps(request, separators=(',', ':'), ensure_ascii=False) + "\n", )

{"id": "batch_req_wnaDys", "custom_id": "request-2", "response": {"status_code": 200, "request_id": "req_c187b3", "body": {"id": "chatcmpl-9758Iw", "object": "chat.completion", "created": 1711475054, "model": "qwen-max", "choices": [{"index": 0, "message": {"role": "assistant", "content": "2 + 2 equals 4."}, "finish_reason": "stop"}], "usage": {"prompt_tokens": 24, "completion_tokens": 15, "total_tokens": 39}, "system_fingerprint": null}}, "error": null}

import json
import csv
columns = ["custom_id",
           "model",
           "request_id",
           "status_code",
           "error_code",
           "error_message",
           "created",
           "content",
           "usage"]

def dict_get_string(dict_obj, path):
    obj = dict_obj
    try:
        for element in path:
            obj = obj[element]
        return obj
    except:
        return None

with open("result.jsonl", "r") as fin:
    with open("result.csv", 'w', encoding='utf-8') as fout:
        rows = [columns]
        for line in fin:
            request_result = json.loads(line)
            row = [dict_get_string(request_result, ["custom_id"]),
                   dict_get_string(request_result, ["response", "body", "model"]),
                   dict_get_string(request_result, ["response", "request_id"]),
                   dict_get_string(request_result, ["response", "status_code"]),
                   dict_get_string(request_result, ["error", "error_code"]),
                   dict_get_string(request_result, ["error", "error_message"]),
                   dict_get_string(request_result, ["response", "body", "created"]),
                   dict_get_string(request_result, ["response", "body", "choices", 0, "message", "content"]),
                   dict_get_string(request_result, ["response", "body", "usage"])]
            rows.append(row)
        writer = csv.writer(fout)
        writer.writerows(rows)

具体流程

import os
from pathlib import Path
from openai import OpenAI

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
)

# test.jsonl 是一个本地示例文件，purpose必须是batch
file_object = client.files.create(file=Path("test.jsonl"), purpose="batch")

print(file_object.model_dump_json())

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

{
    "id": "file-batch-xxx",
    "bytes": 231,
    "created_at": 1729065815,
    "filename": "test.jsonl",
    "object": "file",
    "purpose": "batch",
    "status": "processed",
    "status_details": null
}

curl -X POST https://dashscope.aliyuncs.com/compatible-mode/v1/files \
-H "Authorization: Bearer $DASHSCOPE_API_KEY" \
--form 'file=@"test.jsonl"' \
--form 'purpose="batch"'

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

{
    "id": "file-batch-xxx",
    "bytes": 231,
    "created_at": 1729065815,
    "filename": "test.jsonl",
    "object": "file",
    "purpose": "batch",
    "status": "processed",
    "status_details": null
}

import os
from openai import OpenAI

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
)

batch = client.batches.create(
    input_file_id="file-batch-xxx",  # 上传文件返回的 id
    endpoint="/v1/chat/completions",  # 大语言模型固定填写，/v1/chat/completions
    completion_window="24h"  
)
print(batch)

POST https://dashscope.aliyuncs.com/compatible-mode/v1/batches

curl -X POST https://dashscope.aliyuncs.com/compatible-mode/v1/batches \
  -H "Authorization: Bearer $DASHSCOPE_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "input_file_id": "file-batch-xxx",
    "endpoint": "/v1/chat/completions",
    "completion_window": "24h"
  }'

{
  "id": "batch_cf957e2b-3295-4357-83af-5fb22ac17e81",
  "object": "batch",
  "endpoint": "/v1/chat/completions",
  "errors": null,
  "input_file_id": "file-batch-ADvrUmLWW8TFuKNjGDUUig1c",
  "completion_window": "24h",
  "status": "validating",
  "output_file_id": null,
  "error_file_id": null,
  "created_at": 1722503223,
  "in_progress_at": null,
  "expires_at": null,
  "finalizing_at": null,
  "completed_at": null,
  "failed_at": null,
  "expired_at": null,
  "cancelling_at": null,
  "cancelled_at": null,
  "request_counts": {
    "total": 0,
    "completed": 0,
    "failed": 0
  },
  "metadata": {}
}

import os
from openai import OpenAI

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
)
batch = client.batches.retrieve("batch_id")  # 将batch_id替换为Batch任务的id
print(batch)

GET https://dashscope.aliyuncs.com/compatible-mode/v1/batches/batch_id

curl --request GET 'https://dashscope.aliyuncs.com/compatible-mode/v1/batches/batch_id' \
 -H "Authorization: Bearer $DASHSCOPE_API_KEY"

import os
from openai import OpenAI

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
)
batches = client.batches.list(after="batch_id", limit=20)  # 将batch_id替换为Batch任务的id
print(batches)

GET https://dashscope.aliyuncs.com/compatible-mode/v1/batches?limit=20&after=batch_id

curl --request GET  'https://dashscope.aliyuncs.com/compatible-mode/v1/batches?limit=20&after=batch_id' \
 -H "Authorization: Bearer $DASHSCOPE_API_KEY"

{
  "object": "list",
  "data": [
    {
      "id": "batch_b8b6a83e-00a3-4068-81d9-0a1705bf6b2c",
      "object": "batch",
      "endpoint": "/v1/chat/completions",
      "errors": null,
      "input_file_id": "file-batch-IvxTEDheD70YmyXfqrkE2t64",
      "completion_window": "24h",
      "status": "completed",
      "output_file_id": "file-batch_output-fUy6gxJCyHzT0WHDg2SInyTp",
      "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_b8b6a83e-00a3-4068-81d9-0a1705bf6b2c",
  "last_id": "batch_ca912d83-e9e8-4cf2-89a4-d29743e334e4",
  "has_more": true
}

import os
from openai import OpenAI

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
)
batch = client.batches.cancel("batch_id")  # 将batch_id替换为Batch任务的id
print(batch)

POST https://dashscope.aliyuncs.com/compatible-mode/v1/batches/batch_id/cancel

curl --request POST 'https://dashscope.aliyuncs.com/compatible-mode/v1/batches/batch_id/cancel' \
 -H "Authorization: Bearer $DASHSCOPE_API_KEY"

4. 下载Batch结果文件

在Batch推理任务结束后，您可以通过接口下载结果文件。

您可以通过查询Batch任务详情或通过查询Batch任务列表返回参数中的output_file_id获取下载文件的file_id。仅支持下载以file-batch_output开头的file_id对应的文件。

import os
from openai import OpenAI

client = OpenAI(
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
)
content = client.files.content(file_id="file-batch_output-xxx")
# 打印结果文件内容
print(content.text)
# 保存结果文件至本地
content.write_to_file("result.jsonl")

{"id":"xxx","custom_id":"request-1","response":{"status_code":200,"request_id":"xxx","body":{"id":"xxx","object":"chat.completion","created":1729490866,"model":"qwen-turbo","choices":[{"finish_reason":"stop","index":0,"message":{"content":"2 + 2 equals 4."}}],"usage":{"completion_tokens":8,"prompt_tokens":36,"total_tokens":44},"system_fingerprint":null}},"error":null}
{"id":"yyy","custom_id":"request-2","response":{"status_code":200,"request_id":"yyy","body":{"id":"yyy","object":"chat.completion","created":1729490866,"model":"qwen-turbo","choices":[{"finish_reason":"stop","index":0,"message":{"content":"你好！有什么我可以帮你的吗？"}}],"usage":{"completion_tokens":8,"prompt_tokens":31,"total_tokens":39},"system_fingerprint":null}},"error":null}

GET https://dashscope.aliyuncs.com/compatible-mode/v1/files/{file_id}/content

curl -X GET https://dashscope.aliyuncs.com/compatible-mode/v1/files/file-batch_output-xxx/content \
-H "Authorization: Bearer $DASHSCOPE_API_KEY" > result.jsonl

输入参数配置

返回结果

Batch任务结果的JSONL文件，格式请参考输出文件格式。

Batch支持任务完成通知

提交Batch任务之后，可以通过查询Batch任务接口获取状态和信息，Batch任务执行时间有时会比较长，持续查询Batch任务效率较低，Batch支持任务完成之后通知，减少不必要的任务查询，提高效率。Batch任务完成通知支持两种方式：Callback回调和EventBridge消息。

import os
from openai import OpenAI

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
)

batch = client.batches.create(
    input_file_id="file-batch-xxx",  # 上传文件返回的 id
    endpoint="/v1/chat/completions",  # 大语言模型固定填写，/v1/chat/completions
    completion_window="24h", 
    metadata={
            "ds_batch_finish_callback": "https://xxx/xxx"
          }
)
print(batch)

POST https://dashscope.aliyuncs.com/compatible-mode/v1/batches

curl -X POST --location "https://dashscope.aliyuncs.com/compatible-mode/v1/batches" \
    -H "Authorization: Bearer $DASHSCOPE_API_KEY" \
    -H "Content-Type: application/json" \
    -d '{
          "input_file_id": "file-batch-xxxxx",
          "endpoint": "/v1/chat/completions",
          "completion_window": "24h",
          "metadata": {
            "ds_batch_finish_callback": "https://xxx/xxx"
          }
        }'

{
  "id": "batch_cf957e2b-3295-4357-83af-5fb22ac17e81",
  "object": "batch",
  "endpoint": "/v1/chat/completions",
  "errors": null,
  "input_file_id": "file-batch-xxxx",
  "completion_window": "24h",
  "status": "completed",
  "output_file_id": "xxxx",,
  "error_file_id": null,
  "created_at": 1722503223,
  "in_progress_at": 1732085433,
  "expires_at": null,
  "finalizing_at": 1732085488,
  "completed_at": 1732085488,
  "failed_at": null,
  "expired_at": null,
  "cancelling_at": null,
  "cancelled_at": null,
  "request_counts": {
    "total": 3,
    "completed": 3,
    "failed": 0
  },
  "metadata": {
    "ds_batch_finish_callback": "https://xxxxx"
  }
}

<dependency>
  <groupId>org.apache.rocketmq</groupId>
  <artifactId>rocketmq-client-java</artifactId>
  <version>5.0.4</version>
</dependency>

import org.apache.rocketmq.client.apis.*;
import org.apache.rocketmq.client.apis.consumer.ConsumeResult;
import org.apache.rocketmq.client.apis.consumer.FilterExpression;
import org.apache.rocketmq.client.apis.consumer.FilterExpressionType;
import org.apache.rocketmq.client.apis.consumer.PushConsumer;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.util.Collections;

public class Main {

    private Main() {
    }

    public static void main(String[] args) throws ClientException, IOException, InterruptedException {
        /*
          实例接入点，从控制台实例详情页的接入点页签中获取。
          如果是在阿里云ECS内网访问，建议填写VPC接入点。
          如果是在本地公网访问，或者是线下IDC环境访问，可以使用公网接入点。使用公网接入点访问，必须开启实例的公网访问功能。
         */
        String endpoints = "xxxx";
        //指定需要订阅哪个目标Topic，Topic需要提前在控制台创建，如果不创建直接使用会返回报错。
        String topic = "xxxx";
        //为消费者指定所属的消费者分组，Group需要提前在控制台创建，如果不创建直接使用会返回报错。
        String consumerGroup = "xxxx";
        final ClientServiceProvider provider = ClientServiceProvider.loadService();
        ClientConfigurationBuilder builder = ClientConfiguration.newBuilder().setEndpoints(endpoints);
        /*
          如果是使用公网接入点访问，configuration还需要设置实例的用户名和密码。用户名和密码在控制台实例详情页获取。
          如果是在阿里云ECS内网访问，无需填写该配置，服务端会根据内网VPC信息智能获取。
         */
        builder.setCredentialProvider(new StaticSessionCredentialsProvider("xxxx", "xxxx"));
        ClientConfiguration clientConfiguration = builder.build();
        //订阅消息的过滤规则，表示订阅所有Tag的消息。
        String tag = "*";
        FilterExpression filterExpression = new FilterExpression(tag, FilterExpressionType.TAG);
        //初始化SimpleConsumer，需要绑定消费者分组ConsumerGroup、通信参数以及订阅关系。
        PushConsumer pushConsumer = provider.newPushConsumerBuilder()
                .setClientConfiguration(clientConfiguration)
                //设置消费者分组。
                .setConsumerGroup(consumerGroup)
                //设置预绑定的订阅关系。
                .setSubscriptionExpressions(Collections.singletonMap(topic, filterExpression))
                //设置消费监听器。
                .setMessageListener(messageView -> {
                    try {
                        //处理消息并返回消费结果。
                        ByteBuffer buffer = messageView.getBody().asReadOnlyBuffer();
                        buffer.rewind();
                        byte[] array = new byte[buffer.limit()];
                        buffer.get(array);
                        String result = new String(array, StandardCharsets.UTF_8);
                        System.out.println(result);
                        return ConsumeResult.SUCCESS;
                    } catch (Exception e) {
                        e.printStackTrace();
                        return ConsumeResult.FAILURE;
                    }
                })
                .build();
        Thread.sleep(Long.MAX_VALUE);

        //如果不需要再使用PushConsumer，可关闭该进程。
        pushConsumer.close();
    }
}