批量推理

1
2{"custom_id": "request-1", "body": {"messages": [{"role": "user", "content": "天空为什么这么蓝？"}],"max_tokens": 1000,"top_p":1,"temperature":0.7}}
{"custom_id": "request-2", "body": {"messages": [{"role": "system", "content": "You are an unhelpful assistant."},{"role":"user", "content":"天空为什么这么蓝？"}],"max_tokens":1000}}

1
2{"custom_id":"request-1","body":{"messages":[{"role":"user","content":[{"type":"text","text":"支持输入是图片的模型系列是哪个？"},{"type":"image_url","image_url":{"url":"https://arkdocs.tos-cn-beijing.volces.com/images/advanced-usage/ark_demo_img_1-zh.png"}}]}],"max_tokens":1000,"top_p":1,"temperature":0.7}}
{"custom_id":"request-2","body":{"messages":[{"role":"user","content":[{"type":"image_url","image_url":{"url":"https://arkdocs.tos-cn-beijing.volces.com/images/advanced-usage/ark_demo_img_1-zh.png"}},{"type":"image_url","image_url":{"url":"https://arkdocs.tos-cn-beijing.volces.com/images/advanced-usage/ark_demo_img_2.png"}},{"type":"text","text":"图片里都有什么？"}]}],"max_tokens":1000}}

1
2{"custom_id":"request-1","body":{"messages":[{"role":"user","content":[{"type":"text","text":"视频里有什么？"},{"type":"video_url","video_url":{"url":"https://arkdocs.tos-cn-beijing.volces.com/videos/advanced-usage/ark_vlm_video_input.mp4"}}]}],"max_tokens":1000,"top_p":1,"temperature":0.7}}
{"custom_id":"request-2","body":{"messages":[{"role":"user","content":[{"type":"video_url","video_url":{"url":"https://arkdocs.tos-cn-beijing.volces.com/videos/advanced-usage/video_by_sd2-zh.mp4"}},{"type":"video_url","video_url":{"url":"https://arkdocs.tos-cn-beijing.volces.com/videos/advanced-usage/video_understanding.mp4"}},{"type":"text","text":"视频里都有什么？"}]}],"max_tokens":1000}}

1{"custom_id": "request-1", "body": {"input": ["天很蓝","海很深"]}}

1{"custom_id": "request-1", "body": {"input": [{"text": "天很蓝，海很深","type": "text"},{"image_url": {"url": "https://arkdocs.tos-cn-beijing.volces.com/images/advanced-usage/view.jpeg"},"type": "image_url"}]}}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18import os
import tos

# 从环境变量中读取您的Access Key，为了保障安全，建议您配置Access Key至环境变量
ak = os.environ.get("VOLC_ACCESSKEY")
sk = os.environ.get("VOLC_SECRETKEY")
# 存储任务文件及结果文件的桶的属性，包括ENDPOINT以及区域，当前只支持北京区域的存储桶
endpoint = "tos-cn-beijing.volces.com"
region = "cn-beijing"
bucket_name = "input-bucket"
# 上传任务文件的文件路径
object_key = "input/data.jsonl"
# 任务文件本地的路径
file_name = "/usr/local/data.jsonl"
# 初始化 TOS 客户端
client = tos.TosClientV2(ak, sk, endpoint, region)
# 上传文件
client.put_object_from_file(bucket_name, object_key, file_name)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60# 断点续传demo
import tos
from tos import DataTransferType
# 替换为您的存储桶名称
bucket_name = "demo-bucket-test"
# 对象名称，例如 example_dir 下的 example_object.txt 文件，则填写为 example_dir/example_object.txt
object_key = "input/data.jsonl"
# 本地文件完整路径，例如usr/local/testfile.txt
filename = "/usr/local/text.jsonl"
ak = os.environ.get("VOLC_ACCESSKEY")
sk = os.environ.get("VOLC_SECRETKEY")
region = "cn-beijing"
endpoint = "tos-cn-beijing.volces.com"

try:
    # 创建 TosClientV2 对象，对桶和对象的操作都通过 TosClientV2 实现
    client = tos.TosClientV2(ak, sk, endpoint, region)

    def percentage(
        consumed_bytes: int,
        total_bytes: int,
        rw_once_bytes: int,
        type: DataTransferType,
    ):
        if total_bytes:
            rate = int(100 * float(consumed_bytes) / float(total_bytes))
            print(
                "rate:{}, consumed_bytes:{},total_bytes{}, rw_once_bytes:{}, type:{}".format(
                    rate, consumed_bytes, total_bytes, rw_once_bytes, type
                )
            )

    # 可通过part_size可选参数指定分片大小
    # 通过enable_checkpoint参数开启和关闭断点续传特性
    # 通过task_num设置线程数
    client.upload_file(
        bucket_name,
        object_key,
        filename,
        # 设置断点续传执行线程数，默认为1
        task_num=3,
        # 设置断点续传分片大小，默认20mb
        part_size=1024 * 1024 * 5,
        # 设置断点续传进度条回调函数
        data_transfer_listener=percentage,
    )
except tos.exceptions.TosClientError as e:
    # 操作失败，捕获客户端异常，一般情况为非法请求参数或网络异常
    print("fail with client error, message:{}, cause: {}".format(e.message, e.cause))
except tos.exceptions.TosServerError as e:
    # 操作失败，捕获服务端异常，可从返回信息中获取详细错误信息
    print("fail with server error, code: {}".format(e.code))
    # request id 可定位具体问题，强烈建议日志中保存
    print("error with request id: {}".format(e.request_id))
    print("error with message: {}".format(e.message))
    print("error with http code: {}".format(e.status_code))
    print("error with ec: {}".format(e.ec))
    print("error with request url: {}".format(e.request_url))
except Exception as e:
    print("fail with unknown error: {}".format(e))

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68import volcenginesdkcore
import volcenginesdkark
import os

# 配置项类，您需将其改为自己的配置
class Config:
    # 存储任务文件及结果文件的桶的属性，包括ENDPOINT以及地域
    REGION = "cn-beijing"
    INPUT_BUCKET = "input-bucket"  # 替换为您的输入文件所在 TOS 桶名称
    # 上传任务文件的文件路径
    INPUT_OBJECT_KEY = "input/data.jsonl"
    OUTPUT_BUCKET = "output-bucket"  # 替换为您的结果文件存储 TOS 桶名称
    # 存储结果文件的文件夹
    OUTPUT_OBJECT_KEY = "output/"
    # 任务文件本地的路径
    LOCAL_FILE_NAME = "/usr/local/data.jsonl"
    # 模型版本及名称，可在此查询 https://ark.volcengine.com/region:cn-beijing/docs/model-list
    MODEL_VERSION = "260628"
    MODEL_NAME = "doubao-seed-2-1-pro"
    # 命名您的批量推理任务
    JOB_NAME = "demo"
    # 批量推理任务所在项目的名称，默认为default
    PROJECT_NAME = "default"
    # 从环境变量中读取您的Access Key，为了保障安全，建议您配置Access Key至环境变量
    AK = os.environ.get("VOLC_ACCESSKEY")
    SK = os.environ.get("VOLC_SECRETKEY")

# 创建批量推理任务
def create_batch_job(ark_instance):
    input_file_tos_location = (
        volcenginesdkark.InputFileTosLocationForCreateBatchInferenceJobInput(
            bucket_name=Config.INPUT_BUCKET,
            object_key=Config.INPUT_OBJECT_KEY
        )
    )
    output_dir_tos_location = (
        volcenginesdkark.OutputDirTosLocationForCreateBatchInferenceJobInput(
            bucket_name=Config.OUTPUT_BUCKET,
            object_key=Config.OUTPUT_OBJECT_KEY
        )
    )
    foundation_model = volcenginesdkark.FoundationModelForCreateBatchInferenceJobInput(
        model_version=Config.MODEL_VERSION,
        name=Config.MODEL_NAME
    )
    model_reference = volcenginesdkark.ModelReferenceForCreateBatchInferenceJobInput(
        foundation_model=foundation_model
    )
    req = volcenginesdkark.CreateBatchInferenceJobRequest(
        input_file_tos_location=input_file_tos_location,
        model_reference=model_reference,
        name=Config.JOB_NAME,
        output_dir_tos_location=output_dir_tos_location,
        project_name=Config.PROJECT_NAME,
    )
    resp = ark_instance.create_batch_inference_job(req)
    return resp.id

configuration = volcenginesdkcore.Configuration()
configuration.ak = Config.AK
configuration.sk = Config.SK
configuration.region = Config.REGION
configuration.client_side_validation = True
volcenginesdkcore.Configuration.set_default(configuration)
ark_instance = volcenginesdkark.ARKApi(volcenginesdkcore.ApiClient(configuration))
# 创建批量推理任务
batch_job_id = create_batch_job(ark_instance)
print(f"创建批量推理任务，任务ID：{batch_job_id}")

1创建批量推理任务，任务ID：bi-20260112213907-*****

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29import os
import volcenginesdkcore
import volcenginesdkark

class Config:
    # 存储任务文件及结果文件的桶的属性，包括ENDPOINT以及地域
    REGION = "cn-beijing"
    # 从环境变量中读取您的Access Key，为了保障安全，建议您配置Access Key至环境变量
    AK = os.environ.get("VOLC_ACCESSKEY")
    SK = os.environ.get("VOLC_SECRETKEY")
    # 替换为您创建的批量推理任务的 ID
    BATCH_JOB_ID="bi-20260112213907-****"

# 列出指定 ID 的批量推理任务信息。
def list_batch_inference_jobs(ark_instance, batch_job_id):
    filter = volcenginesdkark.FilterForListBatchInferenceJobsInput(ids=[batch_job_id])
    req = volcenginesdkark.ListBatchInferenceJobsRequest(filter=filter)
    resp = ark_instance.list_batch_inference_jobs(req)
    return resp

configuration = volcenginesdkcore.Configuration()
configuration.ak = Config.AK
configuration.sk = Config.SK
configuration.region = Config.REGION
configuration.client_side_validation = True
volcenginesdkcore.Configuration.set_default(configuration)
ark_instance = volcenginesdkark.ARKApi(volcenginesdkcore.ApiClient(configuration))
batch_job_status = list_batch_inference_jobs(ark_instance, Config.BATCH_JOB_ID)
print(f"批量推理任务信息:\n{batch_job_status}")

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25批量推理任务信息:
{'items': [{'completion_window': '28d',
            'create_time': '2026-01-12T13:39:07Z',
            'description': '',
            'expire_time': '2026-02-09T13:39:07Z',
            'id': 'bi-20260112213907-*****',
            'input_file_tos_location': {'bucket_name': 'batch-test-0112',
                                        'object_key': 'input/data.jsonl'},
            'model_reference': {'custom_model_id': None,
                                'foundation_model': {'model_version': '260628',
                                                     'name': 'doubao-seed-2-1-pro'}},
            'name': 'batch-demo-0112',
            'output_dir_tos_location': {'bucket_name': 'batch-test-0112',
                                        'object_key': 'output/'},
            'project_name': 'default',
            'request_counts': {'completed': 0, 'failed': 0, 'total': 0},
            'status': {'message': '',
                       'phase': 'Queued',
                       'phase_time': '2026-01-12T13:39:07Z'},
            'tags': [{'key': 'sys:ark:createdBy',
                      'value': 'IAMUser/55165459/ark_doc'}],
            'update_time': '2026-01-12T13:39:07Z'}],
 'page_number': 1,
 'page_size': 10,
 'total_count': 1}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38import os
import tos

# 替换为您的批量推理任务 ID
batch_job_id="bi-20260112213907-****"
# 从环境变量中读取您的Access Key，为了保障安全，建议您配置Access Key至环境变量
ak = os.environ.get("VOLC_ACCESSKEY")
sk = os.environ.get("VOLC_SECRETKEY")
# 替换为您存储桶的属性，包括 endpoint、region、bucket_name
endpoint = "tos-cn-beijing.volces.com"
region = "cn-beijing"
bucket_name = "output-bucket"
# 替换为您配置的批量推理任务结果文件路径
output_object_key = "output/"
# 替换为您存储批量推理任务结果文件的本地路径
local_output_dir = "./output/"

results_key = f"{output_object_key}{batch_job_id}/output/results.jsonl"
errors_key = f"{output_object_key}{batch_job_id}/error/errors.jsonl"
results_file_name = f"{local_output_dir}{batch_job_id}/results.jsonl"
errors_file_name = f"{local_output_dir}{batch_job_id}/errors.jsonl"
client = tos.TosClientV2(ak, sk, endpoint, region)
try:
    client.get_object_to_file(bucket_name, results_key, results_file_name)
    print(f"成功下载结果文件 {results_file_name}")
except tos.exceptions.TosServerError as e:
    if e.status_code == 404:
        print("results.jsonl 不存在，说明本次批量推理没有成功执行的请求")
    else:
        raise
try:
    client.get_object_to_file(bucket_name, errors_key, errors_file_name)
    print(f"成功下载错误文件 {errors_file_name}")
except tos.exceptions.TosServerError as e:
    if e.status_code == 404:
        print("errors.jsonl 不存在，说明本次批量推理所有请求均成功执行")
    else:
        raise

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102# 导入异步编程相关库，用于处理异步任务，异步编程中协程是核心概念
import asyncio
# 导入系统相关库，可用于与 Python 解释器和系统进行交互，如标准错误输出
import sys
# 导入操作系统相关库，用于访问操作系统的功能，如读取环境变量
import os
# 导入日期时间处理库，用于记录和计算程序的执行时间
from datetime import datetime
# 导入火山引擎异步 Ark 客户端库，用于与火山引擎的 Ark 服务进行异步通信
from arkruntime import AsyncArk

async def worker(
    # asyncio task 的协程唯一标识，用于区分不同的工作协程
    worker_id: int,
    # 异步 Ark 客户端实例，用于调用批量聊天完成接口处理请求
    client: AsyncArk,
    # 待处理请求的队列，存储需要处理的请求
    requests: asyncio.Queue[dict],
):
    """
    异步协程函数，负责从队列中获取请求并处理。

    :param worker_id: 协程的唯一标识，用于在日志中区分不同的协程
    :param client: 异步 Ark 客户端实例，通过该实例调用服务接口处理请求
    :param requests: 待处理请求的队列，存储待处理的请求信息
    """
    # 打印协程启动信息
    print(f"Worker {worker_id} is starting.")
    while True:
        # 从队列中获取一个请求，若队列为空则会阻塞等待
        # 这里的 await 关键字用于暂停协程的执行，等待队列中有元素可供获取
        request = await requests.get()
        try:
            # 调用客户端的批量聊天完成接口处理请求，使用解包操作将请求字典作为参数传递
            # 同样使用 await 关键字暂停协程，等待接口调用完成
            completion = await client.batch.chat.completions.create(**request)
            # 打印处理结果
            print(completion)
        except Exception as e:
            # 若处理请求过程中出现异常，将错误信息打印到标准错误输出
            print(e, file=sys.stderr)
        finally:
            # 标记该请求已处理完成，通知队列该任务已结束
            requests.task_done()

async def main():
    """
    主函数，负责初始化客户端、生成请求、启动协程并监控任务完成情况。

    使用协程来实现并发处理请求，避免了使用线程带来的较大开销。
    多个协程可以在一个线程中并发执行，提高程序的性能。
    """
    # 记录程序开始执行的时间
    start = datetime.now()
    # 定义协程数量和任务数量，此处示例为避免过度消耗资源，设置为较小值，实际使用请按需调整
    worker_num, task_num = 10, 100
    # 创建一个异步队列用于存储请求，该队列支持异步操作
    requests = asyncio.Queue()
    # 初始化异步 Ark 客户端
    client = AsyncArk(
        # 从环境变量中获取 API 密钥，确保密钥的安全性
        api_key=os.environ.get("ARK_API_KEY"),
        # 设置超时时间为24小时，建议超时时间设置尽量大些，推荐24小时~72小时，避免因网络等原因导致请求超时
        timeout=24 * 3600,
        # 设置 base_url，指定与服务通信的基础 URL
        base_url="https://ark.cn-beijing.volces.com/api/v3",
    )
    # 模拟 `task_num` 个任务，将请求信息添加到队列中
    for _ in range(task_num):
        await requests.put(
            {
                # 替换为你的批量推理接入点 ID，指定要调用的服务端点
                "model": "<YOUR_ENDPOINT_ID>",
                "messages": [
                    {"role": "user","content": "hello"}
                ]
            }
        )
    # 创建 `worker_num` 个 asyncio task 的协程，并启动它们，每个协程负责处理队列中的请求
    # 这些协程会在一个线程中并发执行，通过协程的切换来提高效率
    tasks = [
        asyncio.create_task(worker(i, client, requests))
        for i in range(worker_num)
    ]
    # 等待所有请求处理完成，即队列中的所有任务都被标记为已完成
    await requests.join()
    # 停止所有协程，取消所有正在运行的任务
    for task in tasks:
        task.cancel()
    # 等待所有协程取消完成，确保所有任务都已停止
    await asyncio.gather(*tasks, return_exceptions=True)
    # 关闭客户端连接，释放资源
    await client.close()
    # 记录程序结束执行的时间
    end = datetime.now()
    # 打印程序总执行时间和处理的总任务数
    print(f"Total time: {end - start}, Total task: {task_num}")

if __name__ == "__main__":
    # 运行异步主函数，启动整个程序
    # asyncio.run() 会创建一个事件循环，并在这个事件循环中运行主协程
    asyncio.run(main())