批量推理是无需实时响应需求的离线大批量数据处理方案，适用于模型评测、批量回归等场景，支持通过 SDK 或控制台创建异步任务处理请求。相较于在线推理，批量推理具备以下优势： * **更低成本**：输入输出单价为在线推理的 50%；受益于透明前缀缓存，命中缓存的输入单价可进一步降低 60%。 * **更高配额**：默认 100亿 token/天，支持工单申请提额；受益于灵活的调度策略，高峰期仍可获得可观的任务处理速率。 * **场景灵活**：同时提供批量推理任务（处理静态存储数据）、类在线推理方式批量推理接入点（处理动态数据）两种模式。 * **改造简单**：Batch Chat 接口参数与 Chat 接口一致，只需关注超时时间和并发策略，即可快速切换至批量推理，无需修改业务逻辑。

说明

本篇教程不包含批量生成视频的相关内容。若有批量视频生成的诉求，参见离线推理。

# 支持的模型以 [控制台](https://ark.volcengine.com/region:cn-beijing/batchInference/create) 可选模型为准。 > 更多模型陆续开放中，如有其他模型的批量推理需求，请提交 [工单](https://console.volcengine.com/workorder/create?step=2&SubProductID=P00001166)。

注意

对于已默认开启 thinking summary 的模型（参见回传思考内容），批量推理不返回思考内容：结果中不包含 choices.message.reasoning_content（思考内容摘要），仅返回回答内容（choices.message.content）。如有在批量推理中输出思考内容的诉求，请提交工单。

# 选择调用方式方舟提供了两种批量推理方式，您可根据自己的业务场景选择合适的调用方式。 |方式 |批量推理任务 |批量推理接入点 | |---|---|---| |工作方式 |使用 `JSONL` （JSON Lines，一种每行一个JSON对象的文件格式）文件存储要处理的请求，上传至火山引擎对象存储（TOS）上。通过 [Batch Job API](https://ark.volcengine.com/region:cn-beijing/docs/ark/create-batch-inference-job-api) 创建批量任务，等待所有请求完成后将存储结果返回到 TOS 中。 |调用接口将处理任务推送给平台后，一段时间后（时间长短受资源影响），方舟平台会返回结果。 | |应用场景 |* 数据已静态存储在火山引擎对象存储产品 TOS 中。

* 每天要处理的数据量很大。

* 多模态模型批量推理，对网络带宽压力大。 |* 可直接处理非静态存储、动态生成的数据，无需提前整理为静态文件上传至对象存储，降低数据预处理的改造成本。

* 调用模型的上下游都是在线链路。

* 完整业务链路中调用模型其中一环，单独改造成本高。 | |核心流程 |1. 上传或分片上传数据到对象存储

2. 创建批量推理任务

3. 查询批量推理任务状态

4. 从对象存储中读取推理结果 |1. 创建批量推理接入点

2. 使用批量推理接入点接口进行批量推理

> 根据业务配置超时（大于30分钟）与并发数 | |优势 |

说明

当两种方式都可满足业务需求时，推荐优先使用批量推理任务的方式进行批量推理。

* 数据处理规模或网络带宽上限更高。

* 托管了所有请求的资源调度，支持灵活任务调度，吞吐可达最优。

* 依赖批量推理更灵活的任务调度，即使是高峰期，仍可享受一定的处理效率。

* 支持透明前缀缓存能力，命中缓存的 token 会有更低折扣。 |* 小改造成本，您的工程在使用在线推理接口的基础上进行小的调整，即可切换到批量推理。

* 依赖批量推理更灵活的任务调度，即使是高峰期，仍可享受一定的处理效率。

* 支持透明前缀缓存能力，命中缓存的 token 会有更低折扣。 | |劣势 |存在工程改造量，可能涉及数据上传、分片、切分目录等环节改造。 |* 高峰期间会收到较多`ServerOverloaded` 报错

> 需应用重试退避策略（使用方舟 SDK 在超时范围内自动处理）。

* 每日吞吐由平台资源限制与用户并发控制共同决定，每日吞吐可控性较批量推理任务的方式稍弱。 |

说明

Batch Chat API 已标记为待下线。 建议您直接使用 Chat API，并在请求中设置 service_tier=flex，以在线推理（低优）的方式完成批量数据处理。接口参数一致，改造成本低。

批量推理接入点适用于通过 Batch Chat API 处理动态数据的场景。对于动态产生的数据、后台任务或其他需要保持同步调用方式的时延不敏感业务，可以迁移至在线推理（低优）。迁移后，使用普通在线推理的预置推理接入点或自定义推理接入点，并在请求中设置 service_tier=flex。

在线推理（低优）支持 Chat API、Responses API 和 Messages API，请根据业务当前使用的 API 协议及所需能力选择迁移后的接口。

注意：doubao\-seed\-2.1 后续模型不再支持调用 Batch Chat API。

# 批量推理任务开发教程批量推理任务是业界常规的批量、离线使用模式，需配合火山引擎对象存储（TOS）使用。 > 接口文档：[Batch Job API](https://ark.volcengine.com/region:cn-beijing/docs/ark/create-batch-inference-job-api) ## 前提条件 * [开通指定模型服务](https://ark.volcengine.com/region:cn-beijing/openManagement?LLM=%7B%7D&advancedActiveKey=model)。 * [开通 TOS](https://console.volcengine.com/tos)（火山引擎对象存储产品），并已 [创建存储桶](https://console.volcengine.com/tos/bucket?projectName=default)。

说明

存储桶（bucket）的区域（region）需配置为华北2（北京）。

* 使用 SDK 方式进行批量推理任务开发，需完成以下操作： * 安装 [TOS SDK](https://www.volcengine.com/docs/6349/93479)、[火山引擎SDK](https://api.volcengine.com/api-sdk/view?serviceCode=ark&version=2024-01-01&language=Python) * [获取Access Key](https://console.volcengine.com/iam/keymanage/) ## 快速入门可以根据以下步骤快速实现批量推理任务的创建、查询、下载结果等。 ### 步骤一：准备批量推理任务文件请参考文件格式要求准备文件格式为 `JSONL` （JSON Lines，一行一个 JSON 对象）的数据文件，其中每行包含对 API 的单个请求的详细信息。 #### 文件示例下面是包含 2 个请求的输入文件的示例（`.jsonl`文件）：文本生成 ```JSON {"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}} ``` 图片理解 ```JSON {"custom_id":"request-1","body":{"messages":[{"role":"user","content":[{"type":"text","text":"支持输入是图片的模型系列是哪个？"},{"type":"image_url","image_url":{"url":"https://ark-project.tos-cn-beijing.volces.com/doc_image/ark_demo_img_1.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://ark-project.tos-cn-beijing.volces.com/doc_image/ark_demo_img_1.png"}},{"type":"image_url","image_url":{"url":"https://ark-project.tos-cn-beijing.volces.com/doc_image/ark_demo_img_2.png"}},{"type":"text","text":"图片里都有什么？"}]}],"max_tokens":1000}} ``` 视频理解 ```JSON {"custom_id":"request-1","body":{"messages":[{"role":"user","content":[{"type":"text","text":"视频里有什么？"},{"type":"video_url","video_url":{"url":"https://ark-project.tos-cn-beijing.volces.com/doc_video/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://ark-project.tos-cn-beijing.volces.com/doc_video/ark_vlm_video_input_2.mp4"}},{"type":"video_url","video_url":{"url":"https://ark-project.tos-cn-beijing.volces.com/doc_video/ark_vlm_video_input_3.mp4"}},{"type":"text","text":"视频里都有什么？"}]}],"max_tokens":1000}} ``` 文本向量化 ```JSON {"custom_id": "request-1", "body": {"input": ["天很蓝","海很深"]}} ``` 图像向量化 ```JSON {"custom_id": "request-1", "body": {"input": [{"text": "天很蓝，海很深","type": "text"},{"image_url": {"url": "https://ark-project.tos-cn-beijing.volces.com/images/view.jpeg"},"type": "image_url"}]}} ``` **文件格式校验脚本**：为避免格式错误导致任务失败，请使用校验脚本检测文件，检测前请配置对应的文件。 jsonl_linter.py #### 文件规范要求 * **文件基础要求** 默认最大文件大小 5 GB；单个文件只能包含一个批量推理接入点的请求。 * **字段规则** * 每个请求都必须包含`custom_id` 字段，**类型为字符串且在文件内唯一**，用于关联请求与对应结果。 * 每条请求独立发送、独立收到结果。若多个请求提示词相同，需在请求中都加上相同的提示词。 * 每个请求`body` 字段参数需与底层模型调用 API 的 request body 一致，为合法 JSON Object。完整配置参见对应 API 文档：[对话(Chat) API](https://ark.volcengine.com/region:cn-beijing/docs/ark/chat-api)、[文本向量化 API](https://ark.volcengine.com/region:cn-beijing/docs/ark/text-vectorization-api)、[图像向量化 API](https://ark.volcengine.com/region:cn-beijing/docs/ark/multimodal-vectorization-api)。 * **视觉理解批量任务要求** * 传入图片/视频的方式要求：批量推理任务仅支持 TOS（火山引擎对象存储）链接，不支持其他链接和 Base64 编码（批量推理接入点的传入方式与此不同，参见 [使用批量推理接入点](https://ark.volcengine.com/region:cn-beijing/docs/ark/batch-inference#7fad62bf) 章节）。 * 图片链接需要保持长期可用（推荐 7 天）。TOS 预签名链接默认有效期为 1 小时，需手动设置 expires 参数变更过期时间（建议 7 天），避免因链接过期导致任务失败。完整签名字段说明见 [TOS 预签名文档](https://www.volcengine.com/docs/6349/135725)。 ```Python ... # 1. 创建TOS客户端 client = tos.TosClientV2(AK, SK, ENDPOINT, REGION) # 2. 生成“下载图片”的预签名URL（GET方法，有效时间3600秒） download_pre_signed = client.pre_signed_url( HttpMethodType.Http_Method_Get, # 下载用GET方法（核心！） bucket_name=BUCKET_NAME, key=OBJECT_KEY, expires=604800 # 下载URL有效期，可按需调整（如86400=24小时） ) ... ``` * **配置示例** * 输出长度控制，配置 **max_tokens**。 ```JSON {"custom_id": "request-1", "body": {"messages": [{"role": "system", "content": "You are an unhelpful assistant."},{"role":"user", "content":"天空为什么这么蓝？"}],"max_tokens":1000}} ``` * 关闭深度思考，设置 **thinking** 字段（需模型支持 **thinking** 字段）。 ```JSON {"custom_id": "request-1", "body": {"messages": [{"role": "system", "content": "You are an unhelpful assistant."},{"role":"user", "content":"天空为什么这么蓝？"}],"thinking":{"type":"disabled"}}} ``` * 控制模型回复的丰富度，设置 **temperature** 、**top_p** 字段。 ```JSON {"custom_id": "request-1", "body": {"messages": [{"role": "user", "content": "天空为什么这么蓝？"}],"top_p":1,"temperature":0.7}} ``` ### 步骤二：上传任务文件至TOS 可以通过 SDK 或 TOS 控制台上传任务文件至 TOS 的 bucket 中，后续方舟平台会读取文件里的请求信息进行批量推理。 SDK **上传方式**：使用 SDK 支持的文件上传方式如下。 * 普通上传：最大上传 5 GB 文件。 * 分片上传：文件大小超过 5 GB 需要走分片上传，该方式最大支持上传 48.8 TB文件（每个分片最大 5 GB，分片数量最多 10000）。 **普通上传**： > [普通上传（Python SDK）](https://www.volcengine.com/docs/6349/92800?lang=zh) ```Python import 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) ``` **分片上传** > [分片上传（Python SDK）](https://www.volcengine.com/docs/6349/92801?lang=zh)、[断点续传（Python SDK）](https://www.volcengine.com/docs/6349/135729?lang=zh)

说明

文件大小超过 5GB 需要走分片上传，网络稳定的情况下可以直接走分片上传，网络不稳定建议走断点续传。

每个分片大小为 4MB \- 5GB，网络环境越稳定分片大小可选越大。建议分片大小为 50 MB \- 1GB。

以下是断点续传的示例代码： ```Python # 断点续传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)) ``` 控制台

说明

通过控制台分片上传最大能够上传 50 GB 文件。

登录[对象存储控制台](https://console.volcengine.com/tos)，在桶列表页面找到目标桶名称，在对应的文件路径下上传步骤一准备的任务文件。具体操作可参见[上传文件](https://www.volcengine.com/docs/6349/74830?lang=zh#.5q2l6aqk5LiJ77ya5LiK5Lyg5paH5Lu2)。 ### 步骤三：创建批量推理任务

说明

避免异常场景下重复提交任务产生额外推理费用，平台在创建批量推理任务时提供幂等控制机制：

当提交任务的用户、项目名（Project Name）、存储桶（bucket）和输入文件名（object\-key）完全相同时，平台仅保留一个处于活跃状态（运行或排队）的任务，重复提交的任务将报错，并在错误信息中包含在运行或排队中任务的ID，供用户查询。重复提交报错后，您可以直接使用错误信息中返回的任务 ID 调用查询接口，跟踪原有任务的执行状态，无需重新提交任务。

SDK 支持通过 SDK 配置并创建批量推理任务，包括读入任务文件的路径、输出结果的路径、使用模型等信息。接口参数可参见 [CreateBatchInferenceJob - 创建批量推理任务](https://ark.volcengine.com/region:cn-beijing/docs/ark/create-batch-inference-job-api)。 ```Python import 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/ark/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}") ``` 请求将返回创建的批量推理任务的 ID。 ```Plain 创建批量推理任务，任务ID：bi-20260112213907-***** ``` 控制台在[批量推理](https://ark.volcengine.com/region:cn-beijing/batchInference)页面，单击**开始批量推理**配置批量推理任务，其中调用方式选择**创建批量推理任务**，其他信息请根据控制台提示填写。 ### 步骤四：查询批量推理任务状态 SDK 通过 SDK 支持通过任务 id、模型、运行状态等检查任务状态。接口参数信息参见 [ListBatchInferenceJobs - 获取批量推理任务列表](https://ark.volcengine.com/region:cn-beijing/docs/ark/list-batch-inference-jobs-api)。 ```Python import 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}") ``` 运行脚本，返回的任务信息如下： ```JSON 批量推理任务信息: {'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} ``` 控制台支持在批量推理任务列表页、或者点击批量推理任务名称进入任务详情页，查看批量推理任务的信息，如任务基本信息、状态、进度、配置信息等。批量推理任务状态与对应描述如下： |状态 |状态码 |描述 | |---|---|---| |排队中 |Queued |任务提交成功，由于账号下并发任务数达到上限等原因需排队等候。 | |初始化中 |Initializing |任务在初始化中。 | |运行中 |Running |任务运行中。 | |完成 |Completed |任务已完成。 | |停止中 |Terminating |由于到期等系统原因或手动终止，任务当前处于取消中状态。 | |已停止 |Terminated |任务已被停止。 | |失败 |Failed |任务执行失败，原因可能是超时等原因。 | ### 步骤五：下载批量推理任务结果批量推理任务运行结束后，可以通过 SDK 或者控制台方式下载结果文件。其中结果文件包含 2 类文件： |文件名称 |TOS 存储路径 |说明 | |---|---|---| |results.jsonl |`{OutputDirectory}/{batch_job_id}/output/results.jsonl` |批量推理任务执行后，成功执行结果的集合。其中结果顺序可能与任务文件不一致，请通过`custom_id`来查询。示例如下：

```json```
```{"id": "batch-req-123", "custom_id": "request-1", "response": {"status_code": 200, "request_id": "req1", "body": {"id": "chatcmpl-9P3WtvGUhW5Ty91CdlcBJPgqh1rGG", "object": "chat.completion", "created": 1711111111, "model": "doubao-seed-2-1-pro-260628", "choices": [{"index": 0, "message": {"role": "assistant", "content": "天空呈现蓝色是因为“瑞利散射”现象。简单来说，就是当太阳光进入地球大气层时，波长较长的红光、橙光、黄光能穿透大气层，直接射到地面，而波长较短的蓝、紫、靛等色光，很容易被悬浮在空气中的微粒阻挡，从而使光线散射向四方，使天空呈现出蔚蓝色。"}, "logprobs": null, "finish_reason": "stop"}], "usage": {"prompt_tokens": 20, "completion_tokens": 9, "total_tokens": 29}}}, "error": null}```
| |errors.jsonl |`{OutputDirectory}/{batch_job_id}/error/errors.jsonl` |批量推理任务执行后，错误请求的集合。每个失败请求对应错误文件中的一个响应行，示例如下：

```json```
```{"id":"batch-req-456","custom_id":"request-2","response":null,"error":{"code":"AccessDenied","type":"Forbidden","message":"The request failed because you do not have access to the requested resource."}}```
| SDK 脚本执行完成后，可以在`local_output_dir`文件路径下，找到批量推理任务结果文件。 ```Python import 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 ``` 控制台批量推理任务运行结束后，可以在任务列表点击**查看结果**或在任务详情页的**文件信息**区域点击链接至 TOS 页面，然后下载对应的结果文件。批量推理任务结果文件内容如下： ```JSON {"id":"021768279264106000000000000000000000000000000004984d6","custom_id":"request-2","error":null,"response":{"request_id":"021768279264106000000000000000000000000000000004984d6","status_code":200,"body":{"choices":[{"finish_reason":"stop","index":0,"logprobs":null,"message":{"content":"害，天要蓝就让它蓝呗，纠结这干啥，难道你还能把它改成别的颜色不成？不如想想晚上吃啥小龙虾还是麻辣烫呢，这比啥都实在。","role":"assistant"}}],"created":1768279310,"id":"021768279264106000000000000000000000000000000004984d6","model":"doubao-seed-2-1-pro-260628","service_tier":"default","object":"chat.completion","usage":{"completion_tokens":210,"prompt_tokens":49,"total_tokens":259,"prompt_tokens_details":{"cached_tokens":0}}}}} {"id":"021768279264005000000000000000000000000000000008aeaa6","custom_id":"request-1","error":null,"response":{"request_id":"021768279264005000000000000000000000000000000008aeaa6","status_code":200,"body":{"choices":[{"finish_reason":"stop","index":0,"logprobs":null,"message":{"content":"天空之所以呈现蓝色，主要是因为**瑞利散射**这一光学现象，背后可以拆解为3个关键逻辑：\n\n### 1. 太阳光的“颜色本质”\n太阳光是一种复合光，由红、橙、黄、绿、蓝、靛、紫等不同波长的光组成。其中：\n- 红光、橙光的波长较长（约600-760纳米），穿透力强；\n- 蓝光、紫光的波长较短（约400-500纳米），更容易被散射。\n\n### 2. 大气分子的“选择性散射”\n地球大气中充满了氮气、氧气等微小分子，它们的直径远小于可见光的波长。根据瑞利散射原理：**光的散射强度与波长的4次方成反比**——波长越短，被散射的概率越高。\n因此，蓝光、紫光这类短波长光，会被大气分子向四面八方散射，布满整个天空。\n\n### 3. 人眼的“感知偏好”\n虽然紫光的波长比蓝光更短，散射强度也更高，但人眼的视觉细胞对紫光的敏感度远低于蓝光，同时太阳光中紫光的本身含量也比蓝光少。因此，我们的眼睛最终捕捉到的，是被广泛散射后的“天空蓝”。\n\n### 补充：早晚天空为什么是红橙色？\n清晨或傍晚时，太阳角度低，光线需要穿过更厚的大气层。此时短波长的蓝光、紫光几乎被完全散射殆尽，只剩下穿透力强的长波长红光、橙光抵达我们的眼睛，所以天空会呈现出暖色调的红橙色。","role":"assistant"}}],"created":1768279556,"id":"021768279264005000000000000000000000000000000008aeaa6","model":"doubao-seed-2-1-pro-260628","service_tier":"default","object":"chat.completion","usage":{"completion_tokens":808,"prompt_tokens":53,"total_tokens":861,"prompt_tokens_details":{"cached_tokens":0}}}}} ``` ### 步骤六：（可选）订阅通知可以使用 [火山引擎消息通知服务](https://console.volcengine.com/sns/region:sns+cn-beijing/topic/list)（后简称 SNS）来感知批量任务进度，包括批量推理任务完成或失败，具体订阅流程及订阅信息如下。 1. 申请开通 SNS ，请提交 [工单](https://console.volcengine.com/workorder/create?step=2&SubProductID=P90000689) 申请“SNS 开白”，并同步申请对应的消息事件： * BatchJobFinished：批量推理任务完成。 * BatchJobFailed：批量推理任务失败。 2. 在**主题**页面创建主题。 |配置信息 |操作参考 | |---|---| |* 发布者：选择指定账号`2100444922`。

* 可发布消息的服务：`ark`。 |![图片](https://p9-arcosite.byteimg.com/tos-cn-i-goo7wpa0wc/437917a96bb84c02b1a5fafe5a3f2bd1~tplv-goo7wpa0wc-image.image) | 3. 在**云服务事件订阅**页面创建事件订阅。 |配置信息 |操作参考 | |---|---| |* 事件类型：`BatchJobFinished`、`BatchJobFailed`。

* Topic TRN：选择刚刚创建的主题。 |![图片](https://p9-arcosite.byteimg.com/tos-cn-i-goo7wpa0wc/277d60abde61422381a3be34239d81d3~tplv-goo7wpa0wc-image.image) | 4. 在**订阅**页面配置订阅信息。 |配置信息 |操作参考 | |---|---| |* 订阅主题：已创建的主题。

* 推送类型：选择HTTP/HTTPS。

* 接收端地址：配置对应类型的可接收端地址。

> 也可以使用函数服务订阅信息。推送类型选择函数服务，通过函数 [自定义开发](https://www.volcengine.com/docs/6662/146364) 通知处理逻辑，与下游应用串联。 |![图片](https://p9-arcosite.byteimg.com/tos-cn-i-goo7wpa0wc/76693dd573b34176b8ca85cce0c23e30~tplv-goo7wpa0wc-image.image) | 订阅信息配置完成后，SNS 会向接收端发送对应的确认链接，需确认该链接完成订阅。确认链接 demo 如下，需要回调示例中的 SubscribeURL。 ```JSON { "Type": "SubscriptionConfirmation", "MessageId": "f11b9a8f-****", "TopicTrn": "trn:sns:cn-beijing:2100000825:topic/test", "Message": "You have chosen to subscribe to the topic trn:sns:cn-beijing:2100000825:topic/wyy_test. To confirm the subscription, visit the SubscribeURL included in this message.", "Timestamp": "2025-01-14T07:18:59Z", "SignatureVersion": "1", "Signature": "MEUCIB3NsKw***=", "SigningCertURL": "https://sns-public-cn-beijing.tos-cn-beijing.volces.com/certificates/cn-beijing-a31d91fc-0683-****.pem", "SubscribeToken": "eyJhbGciOiJIUzI1Ni****", "SubscribeURL": "https://sns.cn-beijing.volcengineapi.com?Action=ConfirmSubscription&Version=2023-01-01&Token=eyJhbGc***" } ``` 5. 回调成功后可在**订阅**页面看到对应的订阅状态为**已确认**，表示订阅已完成。 ![图片](https://p9-arcosite.byteimg.com/tos-cn-i-goo7wpa0wc/301d7375496f4ac6863e38b360cdb1e0~tplv-goo7wpa0wc-image.image) 订阅信息内容格式如下： |订阅事件 |字段 |返回示例 | |---|---|---| |批量推理任务完成 |* EventID: 通知 ID。

* Project: 项目名称。

* EventName: BatchJobFinished。

* EventTime: 批量推理任务完成时间。

* AccountID: 用户的主账号 ID。

* JobInfo

* JobID：批量推理任务 ID。

* JobName：批量推理任务名称。

* Message：任务信息。

* FailNum：失败数：此批量推理任务完成时，请求未成功执行的数量。

* TotalNum：此批量推理任务提交的任务总数。

* SuccessNum: 此批量推理任务完成时，请求成功执行的数量。

* FailFileTOSPath：失败任务 TOS 文件路径。

* BucketName：桶名。

* ObjectKey：文件名。

* SuccessFileTOSPath：成功任务TOS文件路径。

* BucketName：桶名。

* ObjectKey：文件名。 |```JSON```
```{```
``` "Type": "Notification",```
``` "MessageId": "7dfcdd57-****--****-618c4d4c6787",```
``` "TopicTrn": "trn:sns:cn-beijing:2100000825:topic/test",```
``` "Subject": "ark:BatchJobFinished",```
``` "Message": "{\"EventID\":\"bi-20250114161208-****\",\"Project\":\"default\",\"EventName\":\"BatchJobFinished\",\"EventTime\":\"2025-01-14T16:12:08+08:00\",\"AccountID\":2100000825,\"JobInfo\":{\"JobID\":\"bi-20250114161117-hjtpg\",\"JobName\":\"批量推理回归\",\"Message\":\"\",\"FailNum\":0,\"TotalNum\":1,\"SuccessNum\":1,\"FailFileTOSPath\":{\"BucketName\":\"\",\"ObjectKey\":\"\"},\"SuccessFileTOSPath\":{\"BucketName\":\"qa-data\",\"ObjectKey\":\"批量推理/result/\"}}}",```
``` "Timestamp": "2025-01-14T08:12:08Z",```
``` "SignatureVersion": "1",```
``` "Signature": "MEYCIQD****",```
``` "SigningCertURL": "https://sns-public-cn-beijing.tos-cn-beijing.volces.com/certificates/cn-beijing-****.pem",```
``` "UnsubscribeToken": "eyJhbGciO****",```
``` "UnsubscribeURL": "https://sns.cn-beijing.volcengineapi.com?Action=Unsubscribe&Version=2023-01-01&Token=eyJhbGciO****"```
```}```
| |批量推理任务失败 |* EventID: 通知 ID。

* Project: 项目名称。

* EventName: BatchJobFailed。

* EventTime: 批量推理任务完成时间。

* AccountID: 用户的主账号 ID。

* JobInfo

* JobID：批量推理任务 ID。

* JobName：批量推理任务名称。

* Message：失败原因。 |```JSON```
```{```
``` "Type": "Notification",```
``` "MessageId": "e350ff5b-****--****-41c3a44924bd",```
``` "TopicTrn": "trn:sns:cn-beijing:2100000825:topic/test",```
``` "Subject": "ark:BatchJobFailed",```
``` "Message": "{\"EventID\":\"bi-20250114154404-****\",\"Project\":\"default\",\"EventName\":\"BatchJobFailed\",\"EventTime\":\"2025-01-14T15:44:04+08:00\",\"AccountID\":2100000825,\"JobInfo\":{\"JobID\":\"bi-20250114154324-hkd5t\",\"JobName\":\"wyy\",\"Message\":\"推理任务失败，第361行非json数据, err=\\\"Syntax error at index 91: invalid char\\\\n\\\\n\\\\tssages\\\\\\\":[\\\\\\\"role\\\\\\\":\\\\\\\"user\\\\\\\",\\\\\\\"content\\\\\\\"\\\\n\\\\t................^...............\\\\n\\\"\"}}",```
``` "Timestamp": "2025-01-14T07:44:04Z",```
``` "SignatureVersion": "1",```
``` "Signature": "MEUCIQ****",```
``` "SigningCertURL": "https://sns-public-cn-beijing.tos-cn-beijing.volces.com/certificates/cn-beijing-****.pem",```
``` "UnsubscribeToken": "eyJhbGciO****",```
``` "UnsubscribeURL": "https://sns.cn-beijing.volcengineapi.com?Action=Unsubscribe&Version=2023-01-01&Token=eyJhbGciO****"```
```}```
| # 批量推理接入点开发教程 > **注意：Batch Chat API 已标记为待下线。** 新业务建议直接使用 Chat API 并在请求中设置 `service_tier=flex`（即 [在线推理（低优）](https://ark.volcengine.com/region:cn-beijing/docs/ark/online-inference-flex)）处理批量数据；存量业务请尽早规划迁移。可通过批量推理接入点的方式，用较小的成本将业务从在线推理迁移至批量推理。 > 接口文档：[Batch Chat API](https://ark.volcengine.com/region:cn-beijing/docs/ark/batch-chat-api)

说明

方舟平台的新用户？获取 API Key 及开通模型等准备工作，请参见快速入门。

## 快速入门 ### 步骤一：创建批量推理接入点 1. 打开 [批量推理创建页面](https://ark.volcengine.com/region:cn-beijing/batchInference/create)。 2. 按照页面进行配置，其中**调用方式**配置项，选择**创建批量推理接入点**。 3. 创建成功后，在批量推理页面的**接入点**页签，可以查看批量推理接入点 ID。 > 批量推理接入点ID的格式为：`ep-bi-***`。 ![图片](https://p9-arcosite.byteimg.com/tos-cn-i-goo7wpa0wc/8ef0f0bc5cdd49be81a3df8a70c023f3~tplv-goo7wpa0wc-image.image) ### 步骤二：使用批量推理接入点接口进行批量推理使用批量推理接入点接口快速发起一个批量推理请求，其中``需替换为步骤一中创建的批量推理接入点 ID。 ```Python import os from arkruntime import Ark client = Ark( api_key=os.environ.get("ARK_API_KEY"), base_url="https://ark.cn-beijing.volces.com/api/v3", # 推荐配置 24 小时超时，避免高负载场景下请求排队超时 timeout=3600*24, ) completion = client.batch.chat.completions.create( # 替换为批量推理接入点 ID model="", messages=[ {"role": "user", "content": "你好"} ] ) print(completion.choices[0].message.content) ``` ## 使用说明及建议为了充分发挥批量推理任务高配额的优势，使用批量推理接入点时，建议保持高并发推压，例如创建大量（10万+）线程/协程进行并发请求，并配置较长的超时时间，保障任务的完成率。 ### 配置超时时间为避免服务端高负载时大量请求因排队超时失败，**推荐手动配置较长的请求超时时间**。 * **超时字段及配置**：`timeout`，推荐配置 24~72 小时（取值支持 1 ~ 72 小时）。 * **注意事项**： * 配置较长的超时时间：不会影响配额使用，可放心设置。 * 配置较短的超时时间：不建议，在深度思考、长文本输出等需要较长时间回复的场景，请求容易中途超时断开连接，既浪费 token 成本，又无法输出内容。使用批量推理接入点时，**无需手动配置重试策略**，方舟会通过智能调度机制保障任务完成率。 * **客户端**：在`timeout`设定的时间范围内自动重试，或按照服务端要求暂停后重试。 * **服务端**：根据负载情况对请求做出暂不处理、排队、指定客户端若干时间后重试等操作。 ```Python ... client = AsyncArk( api_key=os.environ.get("ARK_API_KEY"), base_url="https://ark.cn-beijing.volces.com/api/v3", # 使用异步接口并配置较高的超时时间，推荐24小时~72小时 timeout=3600*24, ) ... ``` ### 配置并发策略充分使用配额方舟为每个主账户提供至少 10B token/天的配额，可通过配置请求并发数及副本数，将大量任务提交到方舟平台处理。方舟平台会根据业务负载智能调度任务执行，甚至在低负载时段提供超出基础配额的处理能力。请求并发数`workerNum`表示向方舟服务端发起请求时的最大并发数，决定同一时间内单副本业务能够向服务端发起请求的最大数量。配置`workerNum`需结合业务预期最大峰值、请求平均时延、服务器并发上限等因素综合评估，以下是配置请求并发数及副本数的简单例子。

说明

业务预期的最大峰值 QPS 为 5000，单个请求的平均延时为 5 秒，部署在相应的机器或容器上。

关键的指标

业务并发数：5000*5 = 25000

服务器限制：不考虑业务层损耗，这里假设为 500 QPS。

副本数和并发数配置

如果需要应对业务峰值时的业务量，可部署更多副本（推荐）或提升服务器规格（不推荐）：

副本数（预估的服务器数量）：业务QPS/服务器限制=5000/500=10。

请求并发数workerNum（每个副本上设置并发数）：服务器限制*请求平均延时=500*5=2500。

在代码实现上需注意： * 创建 client 后、发起服务端请求前，需要启动 client 的相关线程。其中 Java 客户端自带线程池，无需额外手动启动。不同编程语言调用方法存在差异，具体参考下面代码示例。 * 单个副本中全局使用同一个 client（单例模式），避免创建大量实例造成资源开销与浪费。例如，Go SDK 中 client 的`StartBatchWorker`方法会创建`workerNum`个 goroutine，避免创建大量 client 带来的资源开销及浪费。 **不同语言实现并发的示例代码如下**： Python 异步实现（推荐）相比多线程方式，异步协程的方式更优： * **并发更高**：当一个任务处于 I/O 等待状态（如网络请求）时，`asyncio` 可以立即切换到其他任务继续执行，避免了线程切换的开销，从而提高了整体的并发性能。 * **更轻量**：协程是轻量级的执行单元，相比于线程，占用的系统资源更少，可以创建大量的协程来处理任务。

说明

批量推理属于标准的高 I/O 低计算场景，若您的系统适配的场景未使用同步 API，建议使用异步协程的方式。在大部分场景下，用异步协程能获得更优的效果。

```Python # 导入异步编程相关库，用于处理异步任务，异步编程中协程是核心概念 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": "", "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()) ``` Python 多线程实现与在线推理接入点使用类似，需要额外进行配置的信息如下。 1. 设置超时时间`timeout`。 2. 配置的并发线程数和线程中运行的任务数。 3. 异步方法调用`client.batch.chat.completions`方法。 ```Python # 导入 queue 模块，用于创建队列 import queue import sys import os from datetime import datetime # 从 multiprocessing.pool 模块导入 ThreadPool 类，用于创建线程池 from multiprocessing.pool import ThreadPool # 导入方舟SDK from arkruntime import Ark def worker( # 线程的唯一标识 worker_id: int, # Ark 客户端实例 client: Ark, # 待处理请求的队列 requests: queue.Queue[dict], ): """ 线程函数，负责从队列中获取请求并处理。 :param worker_id: 线程的唯一标识 :param client: Ark 客户端实例 :param requests: 待处理请求的队列 """ print(f"Worker {worker_id} is starting.") while True: # 从队列中获取一个请求 request = requests.get() # 检查是否收到无更多请求的信号 if not request: # 将信号放回队列，以便其他线程也能收到 requests.put(request) return try: # 执行请求 completion = client.batch.chat.completions.create(**request) print(completion) except Exception as e: # 打印错误信息到标准错误输出 print(e, file=sys.stderr) finally: # 标记该请求已处理完成 requests.task_done() def main(): """ 主函数，负责初始化客户端、生成请求、启动线程并监控任务完成情况。 """ # 记录开始时间 start = datetime.now() # 最大并发任务数和总任务数，此处示例为避免意外消耗资源，设置为较小值，实际使用请按需调整 worker_num, task_num = 10, 100 # 创建一个队列用于存储请求 requests = queue.Queue() # 初始化 Ark 客户端，设置超时时间为 24 小时 client = Ark( # 从环境变量中获取 API 密钥 api_key=os.environ.get("ARK_API_KEY"), # 设置超时时间为24小时，建议超时时间设置尽量大些，推荐24小时~72小时 timeout=24 * 3600, # 设置 base_url base_url="https://ark.cn-beijing.volces.com/api/v3", ) # 模拟 `task_num` 个任务 for _ in range(task_num): requests.put( { # 替换为你的批量推理接入点 ID "model": "", "messages": [ {"role": "user", "content": "Hello"}, ], } ) # 放入一个请求结束的信号 requests.put(None) # 创建 `worker_num` 个线程并启动它们 with ThreadPool(worker_num) as pool: for i in range(worker_num): # 异步执行线程函数 pool.apply_async(worker, args=(i, client, requests)) # 关闭线程池，不再接受新的任务 pool.close() # 等待所有任务完成 pool.join() # 关闭客户端连接 client.close() # 记录结束时间 end = datetime.now() print(f"Total time: {end - start}, Total task: {task_num}") if __name__ == "__main__": # 运行主函数 main() ``` Go 与在线推理接入点使用类似，需要额外进行配置的信息如下。 1. 配置超时时间`timeout`。 2. 添加 `arkruntime.WithBatchMaxParallel` 选项，设置发起请求的最大并发数量。 3. 方法换成`client.CreateBatchChatCompletion`方法。 ```Go package main import ( "context" "encoding/json" "fmt" "os" "sync" "time" "github.com/volcengine/ark-runtime-go/arkruntime" "github.com/volcengine/ark-runtime-go/arkruntime/model/chat" ) func main() { // 使用 API Key创建一个新的客户端实例 client := arkruntime.NewClientWithApiKey( os.Getenv("ARK_API_KEY"), arkruntime.WithBatchMaxParallel(10), // 设置发起请求的最大并发数量 arkruntime.WithBaseUrl("https://ark.cn-beijing.volces.com/api/v3"), ) // 手动构造100个请求，在实际应用中，你可以从文件、消息队列或数据库中加载真实的请求 requests := MockRequests("YOUR_ENDPOINT_ID", 100) wg, ctx := sync.WaitGroup{}, context.Background() // 可以在这里设置全局的超时时间，如果超过这个时间，所有的请求都会被取消，这里设置为24小时 ctx, cancel := context.WithTimeout(ctx, 24*time.Hour) defer cancel() // 发起请求 for request := range requests { wg.Add(1) // 异步发起请求 go func(request chat.ChatCompletionRequest) { defer wg.Done() // 可以在这里设置每个请求的超时时间，如果超过这个时间，这个请求会被取消 ctx, cancel := context.WithTimeout(ctx, 24*time.Hour) defer cancel() // 发起批量推理请求 result, err := client.CreateBatchChatCompletion(ctx, &request) if err != nil { fmt.Fprintln(os.Stderr, err) } else { fmt.Println(MustMarshalJson(result)) } }(request) } // 等待所有协程完成任务 wg.Wait() } // MockRequests 模拟生成请求，这里只是简单地生成了 count 个相同的请求。 // 在实际应用中，你可以从文件、消息队列或数据库中加载真实的请求。 func MockRequests(endpoint string, count int) <-chan chat.ChatCompletionRequest { requests := make(chan chat.ChatCompletionRequest) go func() { defer close(requests) for i := 0; i < count; i++ { requests <- chat.ChatCompletionRequest{ Model: endpoint, Messages: []chat.ChatCompletionRequestMessage{ userMessage("Hello"), }, } } }() return requests } func MustMarshalJson(v interface{}) string { s, _ := json.Marshal(v) return string(s) } func userMessage(content string) chat.ChatCompletionRequestMessage { return chat.ChatCompletionRequestMessage{ OneOf: chat.NewChatCompletionRequestUserMessageChatCompletionRequestMessageSum( chat.ChatCompletionRequestUserMessage{ Role: chat.ChatCompletionRequestUserMessageRoleUser, Content: chat.NewOptNilChatCompletionMessageContent( chat.NewStringChatCompletionMessageContent(content), ), }, ), } } ``` Java 与在线推理接入点使用类似，需要额外进行配置的信息如下。 1. 设置超时时间`timeout`。 2. 设置最大线程数和线程任务数。 > Java SDK中自带线程池，所以无需批量线程启动的方法。 3. 调用`ChatCompletionRequest`类，启动批量推理。 ```Java package com.example; import com.volcengine.ark.runtime.models.chat.*; import com.volcengine.ark.runtime.service.ArkService; import okhttp3.ConnectionPool; import okhttp3.Dispatcher; import java.time.Duration; import java.util.ArrayList; import java.util.List; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; public class BatchChatCompletionsExample { public static void main(String[] args) { // 为batch chat设置一个较大的超时时间，推荐24小时到72小时之间 Duration timeout = Duration.ofHours(24); // 您需要根据自己的需求配置最大线程数和每个线程执行的任务数 int workerNum = 100; int taskNum = 5; // 您需要将自己的API密钥设置为环境变量 String apikey = System.getenv("ARK_API_KEY"); ConnectionPool connectionPool = new ConnectionPool(workerNum, 10, TimeUnit.MINUTES); Dispatcher dispatcher = new Dispatcher(); dispatcher.setMaxRequests(workerNum); dispatcher.setMaxRequestsPerHost(workerNum); // 请单独为batch chat单独初始化一个service实例，且多个Endpoint间也不要复用同一个service实例，避免互相影响。 // 单个service会根据最大并发数启动对应的线程池，会占用一定的资源。 ArkService service = ArkService.builder().dispatcher(dispatcher).timeout(timeout).connectionPool(connectionPool) .apiKey(apikey) .baseUrl("https://ark.cn-beijing.volces.com/api/v3") .build(); ExecutorService executorService = Executors.newFixedThreadPool(workerNum); CountDownLatch latch = new CountDownLatch(workerNum); Runnable batchChatTask = () -> { System.out.println(Thread.currentThread().getName() + ":在此线程中执行任务"); for (int i = 0; i < taskNum; i++) { // 每个线程执行的任务逻辑，您需要按照自己业务需求进行改造，这里作为示例，展示一个最简单对话的例子 try { final List messages = new ArrayList<>(); final ChatCompletionRequestMessage userMessage = ChatCompletionRequestUserMessage.builder().role(ChatCompletionRequestMessageType.USER) .content(ChatCompletionMessageContent.ofString("Hello")).build(); messages.add(userMessage); ChatCompletionRequest batchChatCompletionRequest = ChatCompletionRequest.builder() // 您需要替换您的批量推理接入点ID .model("") .messages(messages) .build(); service.createBatchChatCompletion(batchChatCompletionRequest) .getChoices() .forEach(choice -> System.out.println(Thread.currentThread().getName() + ":" + choice.getMessage().getContent())); } catch (Exception e) { System.out.println(Thread.currentThread().getName() + "：请求" + i + "失败。"); System.out.println(" 报错信息：" + e.getMessage()); } } System.out.println(Thread.currentThread().getName() + "：此线程中任务完成。"); latch.countDown(); }; for (int i = 0; i < workerNum; i++) { executorService.submit(batchChatTask); } try { latch.await(); } catch (InterruptedException ignored) { } System.out.println("所有线程已退出"); executorService.shutdown(); System.out.println("线程池已退出"); // 当所有请求完成，请关闭服务 service.shutdownExecutor(); } } ``` ### 视觉理解批量任务说明 * 传入图片/视频的方式要求：批量推理接入点支持 TOS 链接（推荐）、其他可访问的链接、Base64 编码三种传入方式。 * 图片链接需要保持长期可用（推荐 7 天）。TOS 预签名链接默认有效期为 1 小时，需手动设置 expires 参数变更过期时间（建议 7 天），避免因链接过期导致任务失败。完整签名字段说明见 [TOS 预签名文档](https://www.volcengine.com/docs/6349/135725)。 ```Python ... # 1. 创建TOS客户端 client = tos.TosClientV2(AK, SK, ENDPOINT, REGION) # 2. 生成“下载图片”的预签名URL（GET方法，有效时间3600秒） download_pre_signed = client.pre_signed_url( HttpMethodType.Http_Method_Get, # 下载用GET方法（核心！） bucket_name=BUCKET_NAME, key=OBJECT_KEY, expires=604800 # 下载URL有效期，可按需调整（如86400=24小时） ) ... ``` # 配额说明 ## TPD 配额不同模型的 TPD 配额不同，实际配额以控制台显示信息为准。 * 配额单位：是 TPD（Tokens Per Day），指 24 小时内可消耗的 token 配额。 * 配额共享：此配额是基于模型设定的，账户下使用同一模型的所有批量推理任务共享该额度。 * 不区分子账号，即账户下所有子账号均共享配额。 * 不区分版本，即同系列模型下的不同版本共享配额。 * 配额隔离：额度与在线推理的模型限流（在线推理限流请参见 [开通管理](https://ark.volcengine.com/region:cn-beijing/openManagement?LLM=%7B%7D&OpenTokenDrawer=false) 页）隔离。运行批量推理任务不会消耗模型的在线推理限流。 * 超配额任务：当用户提交超 TPD 配额任务，会在平台资源空闲时尝试继续执行。如您希望提升批量推理总配额，您可发起 [工单](https://console.volcengine.com/workorder/create?step=2&SubProductID=P00001166) 申请。 ## 其他配额 * 单任务文件大小，默认 5 GB；如有需要可以 [工单](https://console.volcengine.com/workorder/create?step=2&SubProductID=P00001166) 申请提额。 * 账户维度还有如下配额： * 提交任务数量限制：单项目下 7 天内提交的批量推理任务数量最多为 5000，超出时将暂时无法提交新的任务。 * 同时运行中的任务数量限制：单项目下同时处于「运行中」状态的批量推理任务数量最多为 20，超出时其余任务将处于「排队中」状态等待运行。 > 账号下实际同时运行的任务数量会受到平台总体资源的限制和任务调度策略的影响。 # API 文档 * [Batch Chat API](https://ark.volcengine.com/region:cn-beijing/docs/ark/batch-chat-api) * [Batch Job API](https://ark.volcengine.com/region:cn-beijing/docs/ark/create-batch-inference-job-api) # 计费说明请参见[模型价格](https://ark.volcengine.com/region:cn-beijing/docs/ark/model-pricing)。 # 常见问题 [常见问题](https://ark.volcengine.com/region:cn-beijing/docs/ark/modelark-faq)\-[批量推理](https://ark.volcengine.com/region:cn-beijing/docs/ark/modelark-faq#d3743e3b)