一、什么是批量推理（Batch Inference）

批量推理（Batch Inference）是指将大量独立的推理请求汇总成一个批次，统一提交给模型进行处理，在后台离线执行并异步返回结果的计算模式。

二、批量推理 vs 实时推理对比

三、适用场景与核心价值

3.1 典型应用场景

3.2 核心价值总结

四、阿里云百炼 BatchAPI 概览

阿里云百炼 BatchAPI 是专为大规模、非实时推理任务设计的异步处理服务，支持通过文件一次性提交海量请求，在后台离线处理。

4.1 使用前准备

1. 注册阿里云账号并完成实名认证 阿里云官网
2. 开通百炼（Model Studio）服务 在阿里云控制台搜索”百炼”或”Model Studio”，按指引开通
3. 获取 API-Key 百炼控制台 → API-Key 管理 → 创建 API-Key 妥善保存 Key，不要硬编码在代码中提交到仓库
4. 安装 Python SDK pip install alibabacloud-bailian20231229 或通过 OpenAI 兼容接口使用 openai 库

五、数据准备：JSONL 格式规范

5.1 JSONL 格式示例

{"custom_id": "req-001", "method": "POST", "url": "/v1/chat/completions", "body": {"model": "qwen-turbo", "messages": [{"role": "user", "content": "请用一句话总结：人工智能正在改变各个行业。"}]}}
{"custom_id": "req-002", "method": "POST", "url": "/v1/chat/completions", "body": {"model": "qwen-turbo", "messages": [{"role": "user", "content": "请翻译为英文：你好，世界。"}]}}
{"custom_id": "req-003", "method": "POST", "url": "/v1/chat/completions", "body": {"model": "qwen-turbo", "messages": [{"role": "user", "content": "判断情感倾向：这款产品太棒了，非常满意！"}]}}

5.2 JSONL 字段说明

5.3 数据准备代码

import json
from pathlib import Path


def prepare_batch_input(
    items: list[dict],
    output_path: str = "batch_input.jsonl",
    model: str = "qwen-turbo"
) -> str:
    """将数据列表转换为 JSONL 批量推理输入文件。

    Args:
        items: 每条包含自定义字段的数据字典
        output_path: 输出 JSONL 文件路径
        model: 使用的模型名称

    Returns:
        输出文件的绝对路径
    """
    output = Path(output_path)
    with output.open("w", encoding="utf-8") as f:
        for idx, item in enumerate(items, 1):
            record = {
                "custom_id": item.get("id", f"req-{idx:04d}"),
                "method": "POST",
                "url": "/v1/chat/completions",
                "body": {
                    "model": model,
                    "messages": [
                        {"role": "system", "content": item.get("system", "")},
                        {"role": "user", "content": item["prompt"]}
                    ] if item.get("system") else [
                        {"role": "user", "content": item["prompt"]}
                    ],
                    "max_tokens": item.get("max_tokens", 512),
                    "temperature": item.get("temperature", 0.7)
                }
            }
            f.write(json.dumps(record, ensure_ascii=False) + "\n")

    return str(output.resolve())


# 使用示例
data = [
    {"id": "sentiment-001", "prompt": "判断情感：产品质量很差，退货了。"},
    {"id": "sentiment-002", "prompt": "判断情感：物流很快，包装完好。"},
    {"id": "summary-001", "prompt": "总结：本文介绍了批量推理的优势和应用场景。"},
]

file_path = prepare_batch_input(data, model="qwen-turbo")
print(f"已生成批量推理输入文件: {file_path}")

5.4 数据校验

def validate_jsonl(file_path: str) -> tuple[int, list[str]]:
    """校验 JSONL 文件格式是否正确。

    Returns:
        (有效行数, 错误列表)
    """
    errors = []
    valid_count = 0

    with open(file_path, "r", encoding="utf-8") as f:
        for line_num, line in enumerate(f, 1):
            line = line.strip()
            if not line:
                continue
            try:
                obj = json.loads(line)
                required = ["custom_id", "method", "url", "body"]
                for field in required:
                    if field not in obj:
                        errors.append(f"第 {line_num} 行缺少字段: {field}")
                        break
                else:
                    valid_count += 1
            except json.JSONDecodeError as e:
                errors.append(f"第 {line_num} 行 JSON 解析错误: {e}")

    return valid_count, errors


count, errs = validate_jsonl("batch_input.jsonl")
print(f"有效请求: {count} 条")
if errs:
    for e in errs:
        print(f"错误: {e}")

六、提交批量推理任务（完整代码）

6.1 使用 OpenAI 兼容接口（推荐）

import os
from pathlib import Path
from openai import OpenAI


def create_batch_job(
    input_file: str,
    api_key: str | None = None,
    base_url: str = "https://dashscope.aliyuncs.com/compatible-mode/v1"
) -> dict:
    """创建阿里云百炼批量推理任务。

    Args:
        input_file: JSONL 输入文件路径
        api_key: 百炼 API-Key，默认从环境变量 DASHSCOPE_API_KEY 读取
        base_url: 百炼兼容接口地址

    Returns:
        任务信息字典，包含 batch_id
    """
    api_key = api_key or os.getenv("DASHSCOPE_API_KEY")
    if not api_key:
        raise ValueError("请设置 API-Key：环境变量 DASHSCOPE_API_KEY 或参数传入")

    client = OpenAI(api_key=api_key, base_url=base_url)

    # 1. 上传输入文件
    with open(input_file, "rb") as f:
        file_obj = client.files.create(file=f, purpose="batch")
    print(f"文件上传成功: {file_obj.id}")

    # 2. 创建批量任务
    batch = client.batches.create(
        input_file_id=file_obj.id,
        endpoint="/v1/chat/completions",
        completion_window="24h"  # 任务超时时间
    )
    print(f"批量任务创建成功: {batch.id}")
    print(f"状态: {batch.status}")

    return {
        "batch_id": batch.id,
        "input_file_id": file_obj.id,
        "status": batch.status
    }


# 使用
job = create_batch_job("batch_input.jsonl")

6.2 使用阿里云官方 SDK

import os
from alibabacloud_bailian20231229.client import Client as BailianClient
from alibabacloud_tea_openapi import models as open_api_models


def create_batch_job_sdk(
    input_file_url: str,
    access_key_id: str | None = None,
    access_key_secret: str | None = None,
    model: str = "qwen-turbo"
) -> str:
    """使用阿里云 SDK 创建批量推理任务。

    Args:
        input_file_url: 已上传到 OSS 的输入文件 URL
        access_key_id: 阿里云 AccessKey ID
        access_key_secret: 阿里云 AccessKey Secret
        model: 模型名称

    Returns:
        任务 ID
    """
    access_key_id = access_key_id or os.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID")
    access_key_secret = access_key_secret or os.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET")

    config = open_api_models.Config(
        access_key_id=access_key_id,
        access_key_secret=access_key_secret
    )
    config.endpoint = "bailian.cn-beijing.aliyuncs.com"
    client = BailianClient(config)

    # 创建批量任务
    response = client.create_batch_inference_job(
        model=model,
        input_file_url=input_file_url,
        output_file_prefix="batch-output/"
    )

    job_id = response.body.job_id
    print(f"任务创建成功: {job_id}")
    return job_id

七、任务监控与状态管理

7.1 任务状态流转

7.2 状态查询代码

import time
from openai import OpenAI


def wait_for_batch_completion(
    batch_id: str,
    api_key: str | None = None,
    poll_interval: int = 30,
    max_wait: int = 86400
) -> dict:
    """轮询等待批量任务完成。

    Args:
        batch_id: 批量任务 ID
        api_key: API-Key
        poll_interval: 轮询间隔（秒）
        max_wait: 最大等待时间（秒）

    Returns:
        最终任务状态信息
    """
    client = OpenAI(
        api_key=api_key or os.getenv("DASHSCOPE_API_KEY"),
        base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"
    )

    start_time = time.time()
    while True:
        batch = client.batches.retrieve(batch_id)
        elapsed = int(time.time() - start_time)

        print(f"[{elapsed}s] 状态: {batch.status} | "
              f"完成: {batch.request_counts.completed}/"
              f"{batch.request_counts.total} | "
              f"失败: {batch.request_counts.failed}")

        if batch.status in ("completed", "failed", "expired", "cancelled"):
            return {
                "status": batch.status,
                "batch": batch,
                "elapsed_seconds": elapsed
            }

        if elapsed > max_wait:
            raise TimeoutError(f"任务等待超时（>{max_wait}秒）")

        time.sleep(poll_interval)


# 使用
result = wait_for_batch_completion(job["batch_id"], poll_interval=30)
print(f"最终状态: {result['status']}")

7.3 任务列表面板

def list_batch_jobs(api_key: str | None = None, limit: int = 10) -> list[dict]:
    """列出最近的批量任务。"""
    client = OpenAI(
        api_key=api_key or os.getenv("DASHSCOPE_API_KEY"),
        base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"
    )

    batches = client.batches.list(limit=limit)
    jobs = []
    for batch in batches.data:
        jobs.append({
            "id": batch.id,
            "status": batch.status,
            "created_at": batch.created_at,
            "completed": batch.request_counts.completed,
            "failed": batch.request_counts.failed,
            "total": batch.request_counts.total
        })
    return jobs


# 打印任务列表
for job in list_batch_jobs():
    print(f"{job['id'][:20]:20s} | {job['status']:12s} | "
          f"{job['completed']:>4d}/{job['total']:<4d} | "
          f"失败: {job['failed']}")

八、结果获取与后处理

8.1 下载结果文件

def download_batch_results(batch_id: str, api_key: str | None = None) -> list[dict]:
    """下载并解析批量推理结果。

    Returns:
        结果记录列表，每条包含 custom_id 和模型输出
    """
    client = OpenAI(
        api_key=api_key or os.getenv("DASHSCOPE_API_KEY"),
        base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"
    )

    # 获取任务信息
    batch = client.batches.retrieve(batch_id)
    if batch.status != "completed":
        raise ValueError(f"任务未完成，当前状态: {batch.status}")

    # 下载输出文件
    output_file = client.files.content(batch.output_file_id)
    content = output_file.read().decode("utf-8")

    # 解析 JSONL
    results = []
    for line in content.strip().split("\n"):
        if not line.strip():
            continue
        obj = json.loads(line)
        results.append({
            "custom_id": obj.get("custom_id"),
            "status_code": obj.get("status_code"),
            "response": obj.get("response", {}),
            "error": obj.get("error")
        })

    return results


# 使用
results = download_batch_results(job["batch_id"])

# 查看结果
for r in results[:3]:
    cid = r["custom_id"]
    if r["error"]:
        print(f"[{cid}] 错误: {r['error']}")
    else:
        content = r["response"]["body"]["choices"][0]["message"]["content"]
        print(f"[{cid}] 结果: {content[:100]}...")

8.2 结果与输入匹配

def merge_results_with_input(
    input_items: list[dict],
    results: list[dict]
) -> list[dict]:
    """将批量推理结果与原始输入数据合并。

    Returns:
        合并后的记录列表
    """
    result_map = {r["custom_id"]: r for r in results}
    merged = []

    for item in input_items:
        cid = item.get("id", "")
        result = result_map.get(cid, {})

        merged.append({
            **item,
            "status_code": result.get("status_code"),
            "output": result.get("response", {})
                .get("body", {})
                .get("choices", [{}])[0]
                .get("message", {})
                .get("content", ""),
            "error": result.get("error")
        })

    return merged


# 保存为 CSV
import csv

merged = merge_results_with_input(data, results)
with open("batch_results.csv", "w", newline="", encoding="utf-8-sig") as f:
    writer = csv.DictWriter(f, fieldnames=["id", "prompt", "output", "error"])
    writer.writeheader()
    for row in merged:
        writer.writerow({
            "id": row["id"],
            "prompt": row["prompt"],
            "output": row["output"],
            "error": row["error"] or ""
        })

print("结果已保存到 batch_results.csv")

九、成本优化与最佳实践

9.1 成本控制策略

9.2 最佳实践清单

□ 数据准备
  - 提前校验 JSONL 格式，避免上传后校验失败
  - 单文件请求数控制在合理范围（建议 1万~10万条）
  - 单条请求不要太长，避免超时

□ 任务提交
  - 避开业务高峰期提交，减少排队时间
  - 大任务拆分为多个小批量，降低单点失败风险
  - 记录 batch_id，便于后续查询和跟踪

□ 监控管理
  - 设置合理的轮询间隔（30~60 秒），避免频繁调用
  - 任务完成后及时下载结果，释放存储
  - 定期检查失败请求，分析失败原因

□ 结果处理
  - 解析结果时做好异常处理，兼容格式变化
  - 将结果与输入通过 custom_id 精确匹配
  - 对失败请求单独记录，便于后续重试

9.3 失败重试流程

def retry_failed_requests(
    input_file: str,
    results: list[dict],
    api_key: str | None = None
) -> dict | None:
    """提取失败的请求，生成新的重试批次。

    Returns:
        新任务信息，如果没有失败则返回 None
    """
    failed_ids = {r["custom_id"] for r in results if r["error"]}
    if not failed_ids:
        print("没有失败请求，无需重试")
        return None

    # 读取原始输入
    retry_items = []
    with open(input_file, "r", encoding="utf-8") as f:
        for line in f:
            obj = json.loads(line.strip())
            if obj["custom_id"] in failed_ids:
                retry_items.append(obj)

    # 写入重试文件
    retry_file = input_file.replace(".jsonl", "_retry.jsonl")
    with open(retry_file, "w", encoding="utf-8") as f:
        for item in retry_items:
            f.write(json.dumps(item, ensure_ascii=False) + "\n")

    print(f"提取 {len(retry_items)} 条失败请求，写入 {retry_file}")
    return create_batch_job(retry_file, api_key)


# 使用
retry_job = retry_failed_requests("batch_input.jsonl", results)

十、常见问题与排查

完整工作流代码模板

"""
阿里云百炼批量推理完整工作流模板
一步完成：数据准备 → 提交任务 → 监控 → 获取结果 → 保存 CSV
"""

import os
import json
import time
import csv
from pathlib import Path
from openai import OpenAI

API_KEY = os.getenv("DASHSCOPE_API_KEY")
BASE_URL = "https://dashscope.aliyuncs.com/compatible-mode/v1"
MODEL = "qwen-turbo"


def run_batch_pipeline(input_items: list[dict], output_csv: str = "results.csv") -> None:
    """批量推理完整流水线。"""

    # Step 1: 准备 JSONL
    jsonl_path = "batch_input.jsonl"
    with open(jsonl_path, "w", encoding="utf-8") as f:
        for idx, item in enumerate(input_items, 1):
            record = {
                "custom_id": item.get("id", f"req-{idx:04d}"),
                "method": "POST",
                "url": "/v1/chat/completions",
                "body": {
                    "model": MODEL,
                    "messages": [{"role": "user", "content": item["prompt"]}],
                    "max_tokens": item.get("max_tokens", 512)
                }
            }
            f.write(json.dumps(record, ensure_ascii=False) + "\n")
    print(f"✅ 已生成输入文件: {jsonl_path} ({len(input_items)} 条)")

    # Step 2: 创建任务
    client = OpenAI(api_key=API_KEY, base_url=BASE_URL)
    with open(jsonl_path, "rb") as f:
        file_obj = client.files.create(file=f, purpose="batch")
    batch = client.batches.create(
        input_file_id=file_obj.id,
        endpoint="/v1/chat/completions",
        completion_window="24h"
    )
    print(f"✅ 任务已创建: {batch.id}")

    # Step 3: 等待完成
    while True:
        batch = client.batches.retrieve(batch.id)
        print(f"⏳ 状态: {batch.status} | "
              f"完成: {batch.request_counts.completed}/{batch.request_counts.total}")
        if batch.status in ("completed", "failed", "expired"):
            break
        time.sleep(30)

    if batch.status != "completed":
        raise RuntimeError(f"任务未成功完成: {batch.status}")

    # Step 4: 下载结果
    output_file = client.files.content(batch.output_file_id)
    content = output_file.read().decode("utf-8")

    # Step 5: 解析并保存 CSV
    results = [json.loads(line) for line in content.strip().split("\n") if line.strip()]

    with open(output_csv, "w", newline="", encoding="utf-8-sig") as f:
        writer = csv.writer(f)
        writer.writerow(["id", "prompt", "output", "error"])
        for r in results:
            cid = r.get("custom_id", "")
            error = r.get("error", "")
            output = ""
            if not error:
                try:
                    output = r["response"]["body"]["choices"][0]["message"]["content"]
                except (KeyError, IndexError):
                    output = "[解析失败]"
            writer.writerow([cid, "", output, error or ""])

    print(f"✅ 结果已保存: {output_csv}")
    print(f"📊 总计: {len(results)} 条 | "
          f"成功: {sum(1 for r in results if not r.get('error'))} | "
          f"失败: {sum(1 for r in results if r.get('error'))}")


# 运行
if __name__ == "__main__":
    test_data = [
        {"id": "test-001", "prompt": "用一句话总结人工智能的发展趋势"},
        {"id": "test-002", "prompt": "将'Hello World'翻译为中文"},
    ]
    run_batch_pipeline(test_data, "batch_results.csv")

{"custom_id": "req-001", "method": "POST", "url": "/v1/chat/completions", "body": {"model": "qwen-turbo", "messages": [{"role": "user", "content": "Summarize AI trends"}]}}
{"custom_id": "req-002", "method": "POST", "url": "/v1/chat/completions", "body": {"model": "qwen-turbo", "messages": [{"role": "user", "content": "Translate to Chinese"}]}}

import os
from openai import OpenAI

client = OpenAI(
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"
)

# Upload file
with open("batch_input.jsonl", "rb") as f:
    file_obj = client.files.create(file=f, purpose="batch")

# Create batch job
batch = client.batches.create(
    input_file_id=file_obj.id,
    endpoint="/v1/chat/completions",
    completion_window="24h"
)
print(f"Job created: {batch.id}")

def wait_for_completion(batch_id: str, poll_interval: int = 30):
    while True:
        batch = client.batches.retrieve(batch_id)
        print(f"Status: {batch.status} | "
              f"Done: {batch.request_counts.completed}/{batch.request_counts.total}")
        if batch.status in ("completed", "failed", "expired"):
            return batch
        time.sleep(poll_interval)

batch = client.batches.retrieve(batch_id)
output_file = client.files.content(batch.output_file_id)
content = output_file.read().decode("utf-8")

results = []
for line in content.strip().split("\n"):
    obj = json.loads(line)
    results.append({
        "custom_id": obj["custom_id"],
        "response": obj.get("response", {}),
        "error": obj.get("error")
    })