Kimi Batch API

상위 문서: ← Kimi Debugging and operations 출처: Kimi API Platform 공식 문서 — Using Batch API for Bulk Processing 한글화 (2026-07-22) 공급사: Moonshot AI (月之暗面)

대규모 작업을 실시간성이 낮게 처리해야 할 때 Batch API가 이상적입니다. 파일로 작업을 일괄 제출할 수 있어 실시간 API 호출 대비 추론 비용 40% 절감 효과가 있습니다.

Batch API는 kimi-k2.6과 kimi-k2.5 모델을 지원합니다. kimi-k3는 미지원. 이 모델들은 temperature, top_p 등 파라미터를 수정할 수 없으며, 요청 본문에 포함하지 마세요.

1. 워크플로우

이 가이드는 Batch API를 사용한 완전한 텍스트 분류 예시를 안내합니다.

1-1. 입력 파일 작성

JSONL 파일의 각 줄은 단일 추론 요청을 나타내는 독립 JSON 객체:

{"custom_id": "request-1", "method": "POST", "url": "/v1/chat/completions", "body": {"model": "kimi-k2.6", "messages": [{"role": "system", "content": "You are a text classification assistant."}, {"role": "user", "content": "Classify this text: AI is transforming the world"}]}}

필드	필수	설명
custom_id	Yes	결과 추적용 사용자 정의 식별자. 파일 내 유일해야 함
method	Yes	요청 메서드. POST 필수
url	Yes	요청 엔드포인트. /v1/chat/completions 필수
body	Yes	요청 본문. Chat Completions API와 동일 파라미터

body의 model은 kimi-k2.6 또는 kimi-k2.5만 허용. temperature, top_p, n, presence_penalty, frequency_penalty 파라미터는 수정 불가. 본문에 포함하지 마세요.

입력 파일 요구사항:

• .jsonl 포맷, 비어있지 않음, 100MB 이하
• 각 줄은 custom_id, method, url, body 필드를 포함한 유효 JSON 객체
• custom_id는 파일 내 유일
• 모든 줄은 같은 model 사용 — 배치당 모델 1개만 허용
• method는 POST, url은 /v1/chat/completions
• 지정 모델이 존재하고 사용자에게 접근 권한이 있어야 함

1-2. 파일 업로드

JSONL 파일을 Upload File 엔드포인트로 purpose="batch"로 업로드.

Python:

import os
from openai import OpenAI
from openai.types import FileObject

client = OpenAI(
    api_key=os.environ.get("MOONSHOT_API_KEY"),
    base_url=os.environ.get("MOONSHOT_BASE_URL", "https://api.moonshot.ai/v1"),
)

file_object: FileObject = client.files.create(
    file=open("batch_requests.jsonl", "rb"),
    purpose="batch",
)
print(file_object.id)  # 다음 단계용 file_id 저장

cURL:

curl ${MOONSHOT_BASE_URL:-https://api.moonshot.ai/v1}/files \
  -H "Authorization: Bearer $MOONSHOT_API_KEY" \
  -F purpose="batch" \
  -F file="@batch_requests.jsonl"

Node.js:

const OpenAI = require("openai");
const fs = require("fs");

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

async function main() {
    const fileObject = await client.files.create({
        file: fs.createReadStream("batch_requests.jsonl"),
        purpose: "batch"
    });
    console.log(fileObject.id);  // 다음 단계용 file_id 저장
}

main();

1-3. 작업 생성

Create Batch 엔드포인트를 input_file_id와 completion_window로 호출. 큰 데이터셋에는 넉넉한 시간 창 권장.

Python:

import os
from openai import OpenAI
from openai.types import Batch

client = OpenAI(
    api_key=os.environ.get("MOONSHOT_API_KEY"),
    base_url=os.environ.get("MOONSHOT_BASE_URL", "https://api.moonshot.ai/v1"),
)

batch: Batch = client.batches.create(
    input_file_id="your_file_id",
    endpoint="/v1/chat/completions",
    completion_window="24h",
)
print(batch.id)  # 폴링용 batch_id 저장

cURL / Node.js 등 자세한 예제는 공식 문서 참고. 기본 흐름은 동일.

1-4. 완료 대기

생성 후 작업은 입력 검증 단계인 validating 상태로 진입. 검증 통과 후 in_progress로 이동. Retrieve Batch 엔드포인트로 폴링.

import os
import time
from openai import OpenAI
from openai.types import Batch

client = OpenAI(
    api_key=os.environ.get("MOONSHOT_API_KEY"),
    base_url=os.environ.get("MOONSHOT_BASE_URL", "https://api.moonshot.ai/v1"),
)

while True:
    batch: Batch = client.batches.retrieve("your_batch_id")
    completed: int = batch.request_counts.completed if batch.request_counts else 0
    total: int = batch.request_counts.total if batch.request_counts else 0
    print(f"Status: {batch.status} ({completed}/{total})")

    if batch.status == "completed":
        break
    elif batch.status in ("failed", "expired", "cancelled"):
        print(f"Task terminated: {batch.status}")
        break

    time.sleep(10)

1-5. 결과 처리

완료 시 output_file_id에 결과 파일 ID가 포함됨. Get File Content 엔드포인트로 다운로드. 일부 요청이 실패했다면 error_file_id에 오류 상세 포함.

import json
import os
from openai import OpenAI

client = OpenAI(
    api_key=os.environ.get("MOONSHOT_API_KEY"),
    base_url=os.environ.get("MOONSHOT_BASE_URL", "https://api.moonshot.ai/v1"),
)

output = client.files.content("your_output_file_id")
for line in output.text.strip().split("\n"):
    result: dict = json.loads(line)
    custom_id: str = result["custom_id"]
    content: str = result["response"]["body"]["choices"][0]["message"]["content"]
    print(f"{custom_id}: {content}")

각 줄은 처리된 요청에 대응:

{
  "id": "request-1",
  "custom_id": "request-1",
  "response": {
    "status_code": 200,
    "request_id": "",
    "body": {
      "id": "chatcmpl-xxx",
      "object": "chat.completion",
      "created": 1711475054,
      "model": "kimi-k2.6",
      "choices": [
        {
          "index": 0,
          "message": {
            "role": "assistant",
            "content": "This text belongs to the Technology category."
          },
          "finish_reason": "stop"
        }
      ],
      "usage": {
        "prompt_tokens": 30,
        "completion_tokens": 10,
        "total_tokens": 40
      }
    }
  },
  "error": null
}

2. 완전한 코드 예제 (Python)

import json
import os
import time
from pathlib import Path

from openai import OpenAI

MODEL = "kimi-k2.6"

client = OpenAI(
    api_key=os.environ.get("MOONSHOT_API_KEY"),
    base_url=os.environ.get("MOONSHOT_BASE_URL", "https://api.moonshot.ai/v1"),
)


def create_input_jsonl() -> Path:
    """분류 요청을 포함한 JSONL 입력 파일 생성"""
    texts: list[str] = [
        "Hamlet is one of Shakespeare's most famous tragedies",
        "Scientists discover new potentially habitable planet",
        "2024 Artificial Intelligence Development Report",
        "How to make a delicious braised pork dish",
        "Latest iPhone launch event details",
    ]

    requests: list[dict] = []
    for i, text in enumerate(texts):
        requests.append({
            "custom_id": f"text_{i}",
            "method": "POST",
            "url": "/v1/chat/completions",
            "body": {
                "model": MODEL,
                "messages": [
                    {"role": "system", "content": "You are a text classification expert. Classify texts into: Literature/News/Academic/Technology/Lifestyle"},
                    {"role": "user", "content": f"Please classify the following text: {text}"},
                ],
            },
        })

    output_path = Path("classification_requests.jsonl")
    with output_path.open("w", encoding="utf-8") as f:
        for req in requests:
            f.write(json.dumps(req, ensure_ascii=False) + "\n")
    return output_path


# 1. 입력 파일 생성
input_file: Path = create_input_jsonl()

# 2. 파일 업로드
file_object = client.files.create(file=input_file, purpose="batch")
print(f"File uploaded: {file_object.id}")

# 3. 배치 작업 생성
batch = client.batches.create(
    input_file_id=file_object.id,
    endpoint="/v1/chat/completions",
    completion_window="24h",
)
print(f"Batch created: {batch.id}")

# 4. 완료 폴링
while True:
    batch = client.batches.retrieve(batch.id)
    print(f"Status: {batch.status} ({batch.request_counts.completed}/{batch.request_counts.total})")
    if batch.status == "completed":
        break
    elif batch.status in ("failed", "expired", "cancelled"):
        print(f"Task terminated: {batch.status}")
        exit(1)
    time.sleep(10)

# 5. 결과 처리
output = client.files.content(batch.output_file_id)
for line in output.text.strip().split("\n"):
    data: dict = json.loads(line)
    print(f"{data['custom_id']}: {data['response']['body']['choices'][0]['message']['content']}")

3. 배치 상태 레퍼런스

상태	설명
validating	생성됨, 입력 데이터 검증 중
failed	데이터 검증 실패, 배치 종료
in_progress	검증 통과, 실행 중
finalizing	실행 완료, 결과 준비 중
completed	결과 준비 완료, 배치 완료
expired	completion_window 내 완료 실패
cancelling	취소 요청됨, 대기 중
cancelled	취소 완료, 배치 종료

4. 작업 관리

배치 목록

List Batches 엔드포인트로 조직의 모든 배치 조회.

배치 취소

Cancel Batch로 진행 중 작업 취소. validating/in_progress/finalizing 상태에서만 취소 가능. 취소 후 cancelling → cancelled로 변경.

5. 멀티모달 배치 작업

Batch API는 입력 파일에서 이미지·비디오 콘텐츠 지원. 텍스트와의 차이는 입력 파일 작성뿐 — 나머지(업로드·생성·폴링·결과 처리)는 동일.

이미지 배치 처리

이미지 포함 방법 두 가지:

• Base64 인라인 — 이미지를 base64로 JSONL에 직접 인코딩. 작은 이미지에 적합. base64는 파일 크기 ~33% 증가, 100MB 제한 고려
• 파일 참조 — Files API (purpose="image")로 먼저 업로드, JSONL에서 ms://<file_id> 참조. 큰 이미지·재사용에 적합

비디오 배치 처리

비디오도 같은 두 가지 방법 (Files API purpose="video" 사용) 지원.

자세한 코드 예제는 공식 Batch API 문서 참고.

6. 모범 사례

• 데이터 양에 따라 completion_window 설정 — 큰 데이터셋에는 3d 또는 7d
• 과도한 요청을 피하기 위해 10-60초 간격으로 폴링
• 필요에 따라 결과를 DB나 리포트로 처리
• 매우 큰 파일은 여러 배치로 분할 고려

---

변경 이력

날짜	변경
2026-07-22	초판 작성 (공식 Batch API 가이드 한글화)
2026-07-22	제목에 "Kimi" 접두사 추가, 본문 H1 중복 제거