Gemini Batch API 사용법: JSONL 대량 요청과 결과 매칭 구현하기

주요 용어 정의

용어이 글에서의 뜻

Batch API	여러 요청을 비동기로 처리하는 인터페이스
JSONL	줄마다 JSON 객체를 담는 형식
key	요청·결과의 고유 식별자

Gemini Developer API의 파일 배치를 제출하고 결과를 원본 문서에 연결합니다. 2026-10-12 기준, 코드는 미실행 예제입니다.

 

JSONL 요청은 어떻게 만드나요?

아래 가상 요청을 requests.jsonl로 저장하세요. 한 줄에 key와 request를 넣고 배열로 감싸지 않습니다.

{"key":"doc-001","request":{"contents":[{"parts":[{"text":"배치 처리의 장점은?"}]}]}}
{"key":"doc-002","request":{"contents":[{"parts":[{"text":"JSONL의 특징은?"}]}]}}

실제 데이터는 json.dumps(행, ensure_ascii=False) + "\n"으로 직렬화합니다. 문서 ID를 키로 쓰며 빈 키·중복 키를 거부하고 원본 대응표를 저장하세요.

형식 근거는 Google Gemini Cookbook입니다.

파일 제출과 결과 회수

google-genai 설치 후 환경 변수 GEMINI_API_KEY와 GEMINI_BATCH_MODEL을 설정하세요. 모델은 계정의 배치 지원 여부를 확인해 지정해야 합니다.

import os
from pathlib import Path
from google import genai

client = genai.Client()
uploaded = client.files.upload(
    file="requests.jsonl", config={"mime_type": "jsonl"})
job = client.batches.create(
    model=os.environ["GEMINI_BATCH_MODEL"], src=uploaded.name)
Path("batch-name.txt").write_text(job.name, encoding="utf-8")

같은 세션에서는 아래 블록만 실행해 재조회하세요. 새 세션에서는 위의 Path·genai 가져오기와 client 초기화만 먼저 실행하세요. 업로드·배치 생성은 반복하지 마세요.

name = Path("batch-name.txt").read_text(encoding="utf-8")
job = client.batches.get(name=name)
print(job.state.name)
if job.state.name != "JOB_STATE_SUCCEEDED":
    raise SystemExit("대기·실행 중인지, 실패·취소·만료인지 확인")
if not job.dest or not job.dest.file_name:
    raise RuntimeError("결과 파일 없음")
Path("results.jsonl").write_bytes(
    client.files.download(file=job.dest.file_name))

Google Batch API 가이드에 따르면 재생성은 별도 작업을 만듭니다. 생성 응답이 끊기면 접수 여부부터 확인하세요.

결과를 원본에 연결하려면?

Google Batch API 참조는 입력 순서 반환을 명시합니다. 키 검증은 연결 오류를 탐지하는 설계 제안입니다.

import json

order = ["doc-001", "doc-002"]
expected, seen = set(order), set()
ok, review, invalid = {}, {}, []
with open("results.jsonl", encoding="utf-8") as f:
    for n, line in enumerate(f, 1):
        row = json.loads(line)
        key = row.get("key")
        if not isinstance(key, str) or key not in expected or key in seen:
            invalid.append(n)
            continue
        seen.add(key)
        response = row.get("response") or {}
        candidates = response.get("candidates") or []
        candidate = candidates[0] if candidates else {}
        parts = (candidate.get("content") or {}).get("parts") or []
        text = "".join(p.get("text", "") for p in parts
                       if not p.get("thought"))
        if ("error" in row or not text.strip()
                or candidate.get("finishReason") != "STOP"):
            review[key] = row
        else:
            ok[key] = text

missing = expected - seen
if invalid or missing:
    raise ValueError({"검토할_행": invalid, "누락": sorted(missing)})
ordered = [(key, ok.get(key), review.get(key)) for key in order]

review의 오류·빈 후보·중단 사유를 확인하세요. 후보가 없으면 promptFeedback도 봅니다(Google GenerateContentResponse 문서). 파싱 오류는 실행을 중단시킵니다.

예제는 결과 전체를 메모리에 받습니다. 대규모 운영에는 다운로드·저장 설계를 바꾸고 키별 결과를 저장소에 기록하세요.

재제출 전 확인

작업 성공 후에도 개별 실패와 누락을 구분하세요. 확인된 실패만 원인을 수정해 재제출합니다.

작은 입력으로 정상·오류·중복·누락, 순서 복원과 검토 목록을 검증한 뒤 요청량을 늘리세요.

참고 링크

• Google AI for Developers: Batch API
• Google Gemini Cookbook: Batch_mode