Kimi Batch API 允许一次提交大量请求，异步处理，不阻塞主流程。适合内容批量生成、数据集标注、文档批量分析等非实时场景，通常比实时 API 提供价格优惠。目前支持 kimi-k2.6 和 kimi-k2.5 模型。

Kimi Batch API：批量异步处理

适用场景

• 对数千条数据批量打标签、分类、摘要
• 批量翻译文档
• 生成大量训练数据
• 非实时的内容生成任务

不适用：需要实时响应的用户交互场景。

---

工作流程

1. 准备 JSONL 文件（每行一个请求）
2. 上传文件（purpose="batch"）
3. 创建 Batch 任务
4. 轮询状态直到 completed
5. 下载输出文件

---

准备 JSONL 输入文件

每行一个 JSON 对象：

{"custom_id": "task-001", "method": "POST", "url": "/v1/chat/completions", "body": {"model": "kimi-k2.6", "messages": [{"role": "user", "content": "用一句话总结：人工智能是什么？"}]}}
{"custom_id": "task-002", "method": "POST", "url": "/v1/chat/completions", "body": {"model": "kimi-k2.6", "messages": [{"role": "user", "content": "用一句话总结：机器学习是什么？"}]}}

注意：

• 同一个 JSONL 文件中所有请求必须使用相同的模型
• kimi-k2.6 和 kimi-k2.5 的 temperature、top_p 等参数不可修改
• 支持多模态输入（base64 图片/视频）

---

Python 完整示例

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

client = OpenAI(
    api_key="MOONSHOT_API_KEY",
    base_url="https://api.moonshot.cn/v1",
)

# 1. 上传 JSONL 文件
with open("batch_input.jsonl", "rb") as f:
    file_obj = client.files.create(file=f, purpose="batch")

print(f"文件上传成功: {file_obj.id}")

# 2. 创建 Batch 任务
batch = client.batches.create(
    input_file_id=file_obj.id,
    endpoint="/v1/chat/completions",
    completion_window="24h",
)

print(f"Batch 任务创建成功: {batch.id}, 状态: {batch.status}")

# 3. 轮询状态
while batch.status not in ("completed", "failed", "cancelled", "expired"):
    time.sleep(30)
    batch = client.batches.retrieve(batch.id)
    print(f"当前状态: {batch.status}, 完成: {batch.request_counts.completed}/{batch.request_counts.total}")

# 4. 下载结果
if batch.status == "completed":
    output_content = client.files.content(batch.output_file_id).text
    with open("batch_output.jsonl", "w", encoding="utf-8") as f:
        f.write(output_content)
    print("结果已保存到 batch_output.jsonl")

    # 如果有失败的请求
    if batch.error_file_id:
        error_content = client.files.content(batch.error_file_id).text
        print("失败请求：", error_content[:500])

解析输出文件

with open("batch_output.jsonl") as f:
    for line in f:
        result = json.loads(line)
        custom_id = result["custom_id"]
        if result["response"]["status_code"] == 200:
            content = result["response"]["body"]["choices"][0]["message"]["content"]
            print(f"{custom_id}: {content}")
        else:
            print(f"{custom_id}: 失败 - {result['error']}")

---

TypeScript 示例

import OpenAI from "openai";
import fs from "fs";

const client = new OpenAI({
  apiKey: process.env.MOONSHOT_API_KEY,
  baseURL: "https://api.moonshot.cn/v1",
});

async function runBatch() {
  // 上传文件
  const file = await client.files.create({
    file: fs.createReadStream("batch_input.jsonl"),
    purpose: "batch",
  });

  // 创建 Batch
  let batch = await client.batches.create({
    input_file_id: file.id,
    endpoint: "/v1/chat/completions",
    completion_window: "24h",
  });

  // 轮询
  while (!["completed", "failed", "cancelled"].includes(batch.status)) {
    await new Promise((r) => setTimeout(r, 30000));
    batch = await client.batches.retrieve(batch.id);
    console.log(`状态: ${batch.status}`);
  }

  if (batch.status === "completed" && batch.output_file_id) {
    const output = await client.files.content(batch.output_file_id);
    fs.writeFileSync("batch_output.jsonl", await output.text());
  }
}

runBatch();

---

常见问题

Q: Batch API 支持哪些模型？

A: 当前支持 kimi-k2.6 和 kimi-k2.5。moonshot-v1 系列不支持 Batch API。

Q: 任务最长可以运行多久？

A: completion_window 设置为 "24h"，即 24 小时内处理完成。对于超大批次可能接近此限制。

Q: 如何取消进行中的 Batch 任务？

A: 调用 client.batches.cancel(batch_id)。只有在 validating、in_progress 状态时可取消，cancelling 状态下已提交的请求仍会完成。

Kimi Batch API 允许将大量请求打包成 JSONL 文件批量提交，异步处理后下载结果。支持 kimi-k2.6 和 kimi-k2.5 两个模型，不支持 temperature 等参数修改。适合非实时任务，价格低于实时 API。

使用 Kimi Batch API

什么时候用 Batch API

场景	推荐方案
实时对话、交互式应用	普通 Chat Completions API
大批量数据处理、报告生成	Batch API
定时任务、非实时内容生成	Batch API
成本敏感的批量评估/测试	Batch API

Batch API 的主要优势：单价通常低于实时 API，适合不需要立即响应的任务。

基本流程

1. 准备 JSONL 文件（每行一个请求）
2. 上传文件（Files API，purpose="batch"）
3. 创建批处理任务
4. 轮询任务状态
5. 下载并处理结果

1. 准备 JSONL 文件

每行是一个独立的请求，格式固定：

{"custom_id": "req-001", "method": "POST", "url": "/v1/chat/completions", "body": {"model": "kimi-k2.6", "messages": [{"role": "user", "content": "请总结：人工智能的发展历史"}]}}
{"custom_id": "req-002", "method": "POST", "url": "/v1/chat/completions", "body": {"model": "kimi-k2.6", "messages": [{"role": "user", "content": "请总结：量子计算的基本原理"}]}}
{"custom_id": "req-003", "method": "POST", "url": "/v1/chat/completions", "body": {"model": "kimi-k2.6", "messages": [{"role": "user", "content": "请总结：深度学习的核心概念"}]}}

生成 JSONL 文件的 TypeScript 代码：

import fs from "fs";

const tasks = [
  { id: "req-001", prompt: "请总结：人工智能的发展历史" },
  { id: "req-002", prompt: "请总结：量子计算的基本原理" },
  { id: "req-003", prompt: "请总结：深度学习的核心概念" },
];

const lines = tasks.map(task =>
  JSON.stringify({
    custom_id: task.id,
    method: "POST",
    url: "/v1/chat/completions",
    body: {
      model: "kimi-k2.6",
      messages: [{ role: "user", content: task.prompt }],
    },
  })
);

fs.writeFileSync("batch_tasks.jsonl", lines.join("\n"));

限制：

• 文件格式必须是 .jsonl，文件大小不能超过 100MB
• 同一个文件中所有请求必须使用相同的模型
• 支持图片/视频输入（base64 或 ms://<file_id>）

2. 上传文件

import OpenAI from "openai";
import fs from "fs";

const client = new OpenAI({
  apiKey: process.env.MOONSHOT_API_KEY,
  baseURL: "https://api.moonshot.cn/v1",
});

const uploadedFile = await client.files.create({
  file: fs.createReadStream("batch_tasks.jsonl"),
  purpose: "batch", // 注意：batch 任务用 "batch"，文件提取用 "file-extract"
});

console.log("上传文件 ID：", uploadedFile.id);

3. 创建批处理任务

const batch = await (client as any).batches.create({
  input_file_id: uploadedFile.id,
  endpoint: "/v1/chat/completions",
  completion_window: "24h", // 完成窗口，24 小时内完成
});

console.log("批处理任务 ID：", batch.id);
console.log("当前状态：", batch.status);

4. 轮询状态

async function waitForBatch(batchId: string): Promise<any> {
  const terminalStates = ["completed", "failed", "expired", "cancelled"];
  
  while (true) {
    const batch = await (client as any).batches.retrieve(batchId);
    console.log(`状态：${batch.status}，完成：${batch.request_counts?.completed}/${batch.request_counts?.total}`);
    
    if (terminalStates.includes(batch.status)) {
      return batch;
    }
    
    // 每 30 秒轮询一次（大批量任务不必轮询太频繁）
    await new Promise(resolve => setTimeout(resolve, 30000));
  }
}

const completedBatch = await waitForBatch(batch.id);

任务状态说明：

状态	含义
validating	验证输入文件
in_progress	处理中
finalizing	即将完成
completed	全部完成
failed	任务失败
expired	超时未完成
cancelling	取消中
cancelled	已取消

5. 下载结果

if (completedBatch.status === "completed" && completedBatch.output_file_id) {
  const outputContent = await client.files.content(completedBatch.output_file_id);
  const text = await outputContent.text();
  
  // 每行是一个结果
  const results = text.trim().split("\n").map(line => JSON.parse(line));
  
  for (const result of results) {
    console.log(`请求 ID：${result.custom_id}`);
    console.log(`回复：${result.response.body.choices[0].message.content}`);
  }
}

取消任务

只有处于 in_progress 或 validating 状态的任务可以取消：

await (client as any).batches.cancel(batch.id);

常见问题

Q: Batch API 的价格比实时 API 低多少？

A: 具体折扣以 官方价格页面 为准。通常批量 API 提供一定折扣，适合大规模非实时任务降低成本。

Q: 一个批处理任务最多可以提交多少个请求？

A: 取决于 JSONL 文件大小限制（100MB 以内）。对于大批量任务，可以拆分为多个文件，分批提交。

Q: 批处理任务失败了怎么处理？

A: 任务 failed 时，检查 error_file_id，下载错误文件查看详情。部分请求失败时，成功的请求仍会出现在结果文件中，custom_id 可以用来对应原始请求。