Gemini Batch Processing API Guide— Process Thousands of Requests at 50% Off

import google.generativeai as genai
import json
import time
 
genai.configure(api_key="YOUR_API_KEY")
 
def create_batch_job(requests: list[dict], model: str = "gemini-2.5-flash") -> str:
    """
    Create a batch job and return the job name.
    Each request includes a custom ID and content.
    """
    batch_requests = []
    for req in requests:
        batch_requests.append(
            genai.types.BatchRequest(
                custom_id=req["id"],
                request=genai.types.GenerateContentRequest(
                    model=f"models/{model}",
                    contents=[
                        genai.types.Content(
                            parts=[genai.types.Part(text=req["prompt"])]
                        )
                    ],
                    config=genai.types.GenerateContentConfig(
                        temperature=0.3,
                        max_output_tokens=1024,
                    ),
                ),
            )
        )
 
    batch_job = genai.batches.create(
        model=f"models/{model}",
        requests=batch_requests,
        config=genai.types.CreateBatchJobConfig(
            display_name="my-batch-job",
        ),
    )
 
    print(f"Batch job created: {batch_job.name}")
    print(f"State: {batch_job.state}")
    return batch_job.name

def wait_for_batch_completion(job_name: str, poll_interval: int = 30) -> dict:
    """
    Poll until the batch job completes and return results.
    """
    while True:
        job = genai.batches.get(name=job_name)
        state = job.state.name
 
        if state == "JOB_STATE_SUCCEEDED":
            print(f"Job complete: succeeded={job.succeeded_count}, failed={job.failed_count}")
            return collect_results(job)
 
        elif state == "JOB_STATE_FAILED":
            raise RuntimeError(f"Batch job failed: {job.error}")
 
        elif state == "JOB_STATE_CANCELLED":
            raise RuntimeError("Batch job was cancelled")
 
        else:
            total = job.total_count or 0
            succeeded = job.succeeded_count or 0
            progress = (succeeded / total * 100) if total > 0 else 0
            print(f"Processing... {succeeded}/{total} ({progress:.1f}%)")
            time.sleep(poll_interval)
 
def collect_results(job) -> dict:
    """
    Collect results from a completed batch job.
    """
    results = {}
    for response in job.responses:
        custom_id = response.custom_id
        if response.response:
            text = response.response.candidates[0].content.parts[0].text
            results[custom_id] = {
                "status": "success",
                "text": text,
                "usage": {
                    "input_tokens": response.response.usage_metadata.prompt_token_count,
                    "output_tokens": response.response.usage_metadata.candidates_token_count,
                },
            }
        else:
            results[custom_id] = {
                "status": "error",
                "error": str(response.error),
            }
    return results

def batch_sentiment_analysis(texts: list[dict]) -> dict:
    """
    Run sentiment analysis on a large set of texts using batch processing.
    texts: [{"id": "review_001", "text": "This product is amazing..."}, ...]
    """
    requests = []
    for item in texts:
        prompt = f"""Analyze the sentiment of the following text and respond in JSON format.
 
Text: {item["text"]}
 
Output format:
{{"sentiment": "positive|negative|neutral", "confidence": 0.0-1.0, "summary": "one-line summary"}}"""
 
        requests.append({"id": item["id"], "prompt": prompt})
 
    # Split into batches of 100
    BATCH_SIZE = 100
    all_results = {}
 
    for i in range(0, len(requests), BATCH_SIZE):
        chunk = requests[i : i + BATCH_SIZE]
        job_name = create_batch_job(chunk)
        results = wait_for_batch_completion(job_name)
        all_results.update(results)
        print(f"Batch {i // BATCH_SIZE + 1} complete: {len(chunk)} items processed")
 
    return all_results

import { GoogleGenAI } from "@google/genai";
 
const ai = new GoogleGenAI({ apiKey: process.env.GEMINI_API_KEY! });
 
interface BatchRequest {
  id: string;
  prompt: string;
}
 
interface BatchResult {
  status: "success" | "error";
  text?: string;
  error?: string;
  usage?: { inputTokens: number; outputTokens: number };
}
 
async function createAndRunBatch(
  requests: BatchRequest[],
  model = "gemini-2.5-flash"
): Promise<Map<string, BatchResult>> {
  const batchRequests = requests.map((req) => ({
    customId: req.id,
    request: {
      model: `models/${model}`,
      contents: [{ parts: [{ text: req.prompt }] }],
      generationConfig: {
        temperature: 0.3,
        maxOutputTokens: 1024,
      },
    },
  }));
 
  const batchJob = await ai.batches.create({
    model: `models/${model}`,
    requests: batchRequests,
    displayName: "ts-batch-job",
  });
 
  console.log(`Batch job created: ${batchJob.name}`);
 
  // Poll for completion
  const results = new Map<string, BatchResult>();
  let job = batchJob;
 
  while (true) {
    job = await ai.batches.get({ name: job.name! });
 
    if (job.state === "JOB_STATE_SUCCEEDED") {
      for (const resp of job.responses ?? []) {
        if (resp.response) {
          results.set(resp.customId!, {
            status: "success",
            text: resp.response.candidates?.[0]?.content?.parts?.[0]?.text,
            usage: {
              inputTokens: resp.response.usageMetadata?.promptTokenCount ?? 0,
              outputTokens:
                resp.response.usageMetadata?.candidatesTokenCount ?? 0,
            },
          });
        } else {
          results.set(resp.customId!, {
            status: "error",
            error: String(resp.error),
          });
        }
      }
      break;
    }
 
    if (
      job.state === "JOB_STATE_FAILED" ||
      job.state === "JOB_STATE_CANCELLED"
    ) {
      throw new Error(`Batch job ${job.state}: ${job.error}`);
    }
 
    console.log(
      `Processing... ${job.succeededCount ?? 0}/${job.totalCount ?? 0}`
    );
    await new Promise((r) => setTimeout(r, 30_000));
  }
 
  return results;
}

import base64
from pathlib import Path
 
def create_multimodal_batch(image_tasks: list[dict]) -> str:
    """
    Create a multimodal batch job with image inputs.
    image_tasks: [{"id": "img_001", "image_uri": "gs://bucket/img.jpg", "prompt": "Describe this image"}]
    """
    batch_requests = []
 
    for task in image_tasks:
        parts = []
 
        # Use Cloud Storage URI
        if task["image_uri"].startswith("gs://"):
            parts.append(
                genai.types.Part(
                    file_data=genai.types.FileData(
                        file_uri=task["image_uri"],
                        mime_type="image/jpeg",
                    )
                )
            )
        # Base64-encode local files
        else:
            image_bytes = Path(task["image_uri"]).read_bytes()
            parts.append(
                genai.types.Part(
                    inline_data=genai.types.Blob(
                        data=base64.b64encode(image_bytes).decode(),
                        mime_type="image/jpeg",
                    )
                )
            )
 
        parts.append(genai.types.Part(text=task["prompt"]))
 
        batch_requests.append(
            genai.types.BatchRequest(
                custom_id=task["id"],
                request=genai.types.GenerateContentRequest(
                    model="models/gemini-2.5-flash",
                    contents=[genai.types.Content(parts=parts)],
                ),
            )
        )
 
    batch_job = genai.batches.create(
        model="models/gemini-2.5-flash",
        requests=batch_requests,
    )
 
    return batch_job.name

from dataclasses import dataclass, field
 
@dataclass
class BatchJobTracker:
    """Track batch job execution and manage retries for failed requests"""
 
    max_retries: int = 3
    failed_requests: list = field(default_factory=list)
    retry_count: dict = field(default_factory=dict)
 
    def process_results(self, results: dict, original_requests: list[dict]):
        """Process results and record failed requests for retry"""
        for req in original_requests:
            rid = req["id"]
            result = results.get(rid, {})
 
            if result.get("status") == "error":
                current_retries = self.retry_count.get(rid, 0)
                if current_retries < self.max_retries:
                    self.retry_count[rid] = current_retries + 1
                    self.failed_requests.append(req)
                    print(f"Retry queued: {rid} (attempt {current_retries + 1}/{self.max_retries})")
                else:
                    print(f"Max retries reached: {rid}")
 
    def get_retry_requests(self) -> list[dict]:
        """Return requests that need to be retried"""
        requests = self.failed_requests.copy()
        self.failed_requests.clear()
        return requests
 
def robust_batch_processing(
    requests: list[dict], model: str = "gemini-2.5-flash"
) -> dict:
    """
    Batch processing with automatic retries.
    Failed requests are retried up to 3 times.
    """
    tracker = BatchJobTracker(max_retries=3)
    all_results = {}
    pending = requests.copy()
 
    round_num = 0
    while pending:
        round_num += 1
        print(f"\n--- Round {round_num}: processing {len(pending)} requests ---")
 
        job_name = create_batch_job(pending, model=model)
        results = wait_for_batch_completion(job_name)
 
        # Store successful results
        for rid, result in results.items():
            if result["status"] == "success":
                all_results[rid] = result
 
        # Check for retries
        tracker.process_results(results, pending)
        pending = tracker.get_retry_requests()
 
        if pending:
            print(f"Retrying: {len(pending)} requests")
 
    return all_results

# ❌ Verbose prompt (wastes tokens)
bad_prompt = """
You are a professional sentiment analysis expert.
Please perform a very detailed sentiment analysis of the following text.
First, evaluate the overall tone of the text, then analyze the sentiment
of each sentence, and finally calculate a comprehensive sentiment score.
Please output in JSON format.
Text: {text}
"""
 
# ✅ Concise prompt (same accuracy, fewer tokens)
good_prompt = """Classify sentiment as JSON: {{"sentiment":"positive|negative|neutral","confidence":0.0-1.0}}
 
Text: {text}"""

def optimize_batch_size(total_requests: int) -> int:
    """
    Return the optimal batch size based on total request count.
    """
    if total_requests <= 100:
        return total_requests  # Small: single batch
    elif total_requests <= 1000:
        return 100  # Medium: 100 per batch
    else:
        return 500  # Large: 500 per batch (parallel jobs)

from flask import Flask, request, jsonify
 
app = Flask(__name__)
 
@app.route("/webhook/batch-complete", methods=["POST"])
def batch_complete():
    """Webhook endpoint for batch job completion"""
    payload = request.json
    job_name = payload["name"]
    state = payload["state"]
 
    if state == "JOB_STATE_SUCCEEDED":
        job = genai.batches.get(name=job_name)
        results = collect_results(job)
        process_completed_results(results)
 
    return jsonify({"status": "received"})

from google.cloud import scheduler_v1
 
def schedule_daily_batch():
    """Create a schedule to run batch processing every night"""
    client = scheduler_v1.CloudSchedulerClient()
 
    job = scheduler_v1.Job(
        name="projects/my-project/locations/us-central1/jobs/daily-batch",
        http_target=scheduler_v1.HttpTarget(
            uri="https://my-api.run.app/trigger-batch",
            http_method=scheduler_v1.HttpMethod.POST,
        ),
        schedule="0 2 * * *",  # Daily at 2 AM
        time_zone="Asia/Tokyo",
    )
 
    client.create_job(
        parent="projects/my-project/locations/us-central1", job=job
    )