Gemini Batch Images not working, working in normal mode

[alexisdrakopoulos]

Hi,

I am trying to use Gemini Batch with a single image per batch, this works perfectly fine without batch but doesn't work with batch enabled.

I get this error for every question:

['Request Bad Request: {"error": {"code": 400, "message": "Invalid value at \'contents[0].parts[0]\' (text), Starting an object on a scalar field", "status": "INVALID_ARGUMENT", "details": [{"@type": "type.googleapis.com/google.rpc.BadRequest", "fieldViolations": [{"field": "contents[0].parts[0]", "description": "Invalid value at \'contents[0].parts[0]\' (text), Starting an object on a scalar field"}]}]}}']

My class:

class PageOCR(curator.LLM):
    """A page OCR that extracts text from an image of a page."""

    def prompt(self, input: dict) -> str:
        """Generate a prompt using the image."""
        prompt_text = (
            "Please OCR the following text and return the text content in JSON format."
            "Note that this is text from a 1791 book that uses the long S character, please replace with the normal s."
        )
        image_url: str = input["image_path"]
        return prompt_text, curator.types.Image(
            url=image_url, detail="scanned page", mime_type="image/png"
        )

    def parse(self, input: dict, response: str) -> dict:
        """Parse the model response along with the input to the model into the desired output format."""
        return {
            "text": response,
        }

my call which fails:

llm = PageOCR(
        model_name="gemini-2.5-flash-preview-05-20", backend="gemini", batch=True
    )

my call which works:
llm = PageOCR(model_name="gemini-2.5-flash-preview-05-20", backend=None, batch=False)

And the execution code:

dataset_dict = {
                "image_path": [item["image_path"] for item in dataset_input_list]
            }
dataset = Dataset.from_dict(dataset_dict)
ds_result = llm(dataset)

where image_path values are GCP bucket paths (but not this works perfectly not in batch so I doubt this is the issue?)

[kartik4949]

Hello,
thanks for filling the issue, we will be reproducing the error

Thanks

[alexisdrakopoulos]

Thanks! Here is an end-to-end reproducible example, just need to fill in the env vars on a GCP project/account where Vertex AI has bucket rights.

import os
import uuid
import traceback

from bespokelabs import curator
from datasets import Dataset
from google.cloud import storage
from PIL import Image

# --- Environment Configuration ---
# Using user's provided hardcoded values.
# Ensure these are correctly set for your environment.
os.environ["GOOGLE_CLOUD_PROJECT"] = "YOUR CLOUD PROJECT"
os.environ["GEMINI_BUCKET_NAME"] = "YOUR BUCKET NAME"
os.environ["GOOGLE_CLOUD_REGION"] = "us-central1" # YOUR REGION OF COURSE


# --- GCS Helper Functions (Unchanged) ---
def upload_to_gcs(bucket_name, source_file_name, destination_blob_name):
    """Uploads a file to the bucket."""
    try:
        storage_client = storage.Client(project=os.environ["GOOGLE_CLOUD_PROJECT"])
        bucket = storage_client.bucket(bucket_name)
        blob = bucket.blob(destination_blob_name)
        blob.upload_from_filename(source_file_name)
        print(
            f"File {source_file_name} uploaded to gs://{bucket_name}/{destination_blob_name}."
        )
        return f"gs://{bucket_name}/{destination_blob_name}"
    except Exception as e:
        print(f"Error uploading {source_file_name} to GCS: {e}")
        raise


def delete_from_gcs(bucket_name, blob_name):
    """Deletes a blob from the bucket."""
    try:
        storage_client = storage.Client(project=os.environ["GOOGLE_CLOUD_PROJECT"])
        bucket = storage_client.bucket(bucket_name)
        blob = bucket.blob(blob_name)
        blob.delete()
        print(f"Successfully deleted gs://{bucket_name}/{blob_name}")
    except Exception as e:
        if "Not Found" in str(e):
            print(f"Blob gs://{bucket_name}/{blob_name} not found, skipping deletion.")
        else:
            print(f"Error deleting gs://{bucket_name}/{blob_name} from GCS: {e}")


class PageOCR(curator.LLM):
    def prompt(self, input: dict) -> list:
        prompt_text = (
            "Please OCR the following text and return the text content in JSON format."
            "Note that this is text from a 1791 book that uses the long S character, please replace with the normal s."
        )
        image_url: str = input["image_path"]
        return prompt_text, curator.types.Image(url=image_url)

    def parse(self, input: dict, response: str) -> dict:
        return {
            "text": response,
        }


if __name__ == "__main__":
    # --- Setup: Create a dummy image for a reproducible example ---
    local_image_path = "repro_image.png"
    print(f"Creating a dummy image for testing at: {local_image_path}")
    dummy_image = Image.new("RGB", (100, 50), color="blue")
    dummy_image.save(local_image_path)

    # --- Configuration ---
    llm = PageOCR(
        model_name="gemini-2.5-flash-preview-05-20", backend="gemini", batch=True
    )
    gcs_bucket_name = os.environ["GEMINI_BUCKET_NAME"]

    # Use a unique name for the GCS blob to avoid conflicts
    destination_blob_name = (
        f"temp_repro_images/{uuid.uuid4()}/{os.path.basename(local_image_path)}"
    )
    gcs_uri = None

    try:
        # --- 1. Upload the single image to GCS ---
        print(f"\n--- Step 1: Uploading {local_image_path} to GCS ---")
        gcs_uri = upload_to_gcs(
            gcs_bucket_name, local_image_path, destination_blob_name
        )
        print(f"GCS URI: {gcs_uri}")

        # --- 2. Create a dataset with the single image URI ---
        print("\n--- Step 2: Creating Hugging Face Dataset ---")
        dataset = Dataset.from_dict({"image_path": [gcs_uri]})
        print(f"Dataset created with 1 item: {dataset[0]}")

        # --- 3. Run batch prediction ---
        print(
            "\n--- Step 3: Calling the LLM with the dataset (This is where the error likely occurs) ---"
        )
        ds_result = llm(dataset)

        # --- 4. Inspect the result ---
        print("\n--- Step 4: Inspecting LLM Result ---")
        print(f"Prediction complete. Result type: {type(ds_result)}")

        if isinstance(ds_result, Dataset):
            print(f"Result has {len(ds_result)} item(s).")
            # The output dataset contains the original columns plus the 'text' column from the `parse` method.
            for i, result_item in enumerate(ds_result):
                print(f"\n--- Result Item {i + 1} ---\n{result_item}")
        else:
            print(
                "Warning: The result from the LLM call was not a datasets.Dataset object."
            )
            print("Actual Result:", ds_result)

    except Exception as e:
        print("\n" + "=" * 20 + " AN ERROR OCCURRED " + "=" * 20)
        print(f"A critical error occurred during the prediction process: {e}")
        print("Full traceback:")
        traceback.print_exc()
        print("=" * 58)

    finally:
        print("\n--- Step 5: Cleaning up resources ---")
        if gcs_uri:
            delete_from_gcs(gcs_bucket_name, destination_blob_name)
        if os.path.exists(local_image_path):
            os.remove(local_image_path)
            print(f"Successfully deleted local dummy image: {local_image_path}")
        print("\nScript finished.")

[alexisdrakopoulos]

Also I'm on the latest version if that matters.

[alexisdrakopoulos]

Let me know if there's anything I can do to help, I can't figure out the formatting issue as I'm not familiar with your repo.

[kartik4949]

This issue is being worked at #690