How to Use Vertex AI Batch Processing in LangExtract: A Complete Implementation Guide

LangExtract automatically routes large workloads to the Vertex AI Batch Prediction API when you provide a Vertex AI-enabled client and a batch configuration, handling GCS upload, job polling, and result caching automatically.

Vertex AI batch processing in LangExtract enables cost-effective extraction from large document collections by asynchronously processing prompts through Google's managed batch service. This guide explains the architecture, configuration options, and implementation patterns using the google/langextract repository's official batch processing pipeline.

Prerequisites for Vertex AI Batch Processing

Before enabling batch mode, you must instantiate a genai.Client with explicit Vertex AI support. The library validates this through _is_vertexai_client() in langextract/providers/gemini_batch.py (lines 71-80), which checks that vertexai=True is set on the client instance.

import google.genai as genai

client = genai.Client(
    vertexai=True,
    project="your-gcp-project",
    location="us-central1"
)

You also need appropriate IAM permissions for Cloud Storage (to create buckets and objects) and Vertex AI (to submit and poll batch jobs).

Understanding the Batch Processing Architecture

The integration is built around three core components defined in langextract/providers/gemini_batch.py:

Component	Purpose	Location
Vertex AI Client Detection	Guarantees batch mode only runs with vertexai=True clients	_is_vertexai_client() (lines 71-80)
Batch Configuration	Holds all knobs (enabled, threshold, poll_interval, etc.) and validates them	BatchConfig dataclass (lines 57-80)
Batch Inference Pipeline	Orchestrates request building, GCS upload, job creation, polling, and result extraction	infer_batch() (lines 688-785) and helper methods

The Batch Inference Pipeline Steps

When infer_batch() executes, it performs the following operations:

1. Request Building: _build_request() creates REST-compatible dictionaries for each prompt, including JSON schemas and generation configs.
2. File Upload: _submit_file() serializes prompts to a JSONL file and uploads it to a dedicated GCS bucket.
3. Job Creation: Calls client.batches.create() with the uploaded file as the source.
4. Polling: _poll_completion() monitors the job state until it reaches a terminal state (SUCCEEDED, PAUSED, or failure), respecting poll_interval and timeout settings.
5. Result Extraction: _extract_from_file() downloads output JSONL files and parses them with _parse_batch_line(), preserving original prompt order.
6. Caching: GCSBatchCache writes results back to GCS using SHA-256 hashes as keys, enabling instant reuse on subsequent runs.

Configuring Batch Processing in LangExtract

You can configure batch behavior using either a dictionary or the BatchConfig dataclass. Both methods validate that enable_caching and retention_days are explicitly set when enabled=True.

Using Dictionary Configuration

Pass a dictionary via language_model_params:

batch_config = {
    "enabled": True,
    "threshold": 10,          # trigger batch mode after 10 chunks

    "poll_interval": 30,      # seconds between status checks

    "timeout": 3600,          # maximum seconds to wait

    "enable_caching": True,   # store results in GCS

    "retention_days": 30,     # auto-delete cache after 30 days

    "max_prompts_per_job": 10000,  # split large workloads

    "ignore_item_errors": False,   # fail entire batch on single error

}

results = lx.extract(
    text_or_documents=large_text,
    prompt_description="Extract entities and relationships",
    model_id="gemini-2.5-flash",
    language_model_params={
        "vertexai": True,
        "project": "your-gcp-project",
        "location": "us-central1",
        "batch": batch_config,
    },
)

Using the BatchConfig Dataclass

For programmatic control, import the dataclass directly from langextract/providers/gemini_batch.py:

from langextract.providers.gemini_batch import BatchConfig

cfg = BatchConfig(
    enabled=True,
    threshold=20,
    poll_interval=15,
    timeout=1800,
    max_prompts_per_job=5000,
    ignore_item_errors=False,
    enable_caching=True,
    retention_days=None,  # keep cache indefinitely

)

results = lx.extract(
    ...,
    language_model_params={
        "vertexai": True,
        "project": "my-project",
        "location": "europe-west1",
        "batch": cfg,  # dataclass accepted directly

    },
)

The BatchConfig class is defined starting at line 57 in langextract/providers/gemini_batch.py.

Complete Working Examples

End-to-End Batch Extraction Script

This runnable example from docs/examples/batch_api_example.md demonstrates logging, text chunking, and batch configuration:

import logging
import requests
import textwrap
import langextract as lx

# Configure logging to track batch progress

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s",
    handlers=[
        logging.FileHandler("batch_process.log"),
        logging.StreamHandler(),
    ],
)

# Download sample text (Romeo & Juliet, ~60k characters)

url = "https://www.gutenberg.org/files/1513/1513-0.txt"
text = requests.get(url).text
text_subset = text[:60_000]

# Define extraction prompt and examples

prompt = textwrap.dedent("""\
    Extract characters and emotions from the text.
    Use exact text from the input for extraction_text.""")
examples = [
    lx.data.ExampleData(
        text="ROMEO. But soft! What light through yonder window breaks?",
        extractions=[
            lx.data.Extraction(extraction_class="character", extraction_text="ROMEO"),
            lx.data.Extraction(extraction_class="emotion", extraction_text="But soft!"),
        ],
    )
]

# Batch configuration

batch_config = {
    "enabled": True,
    "threshold": 10,
    "poll_interval": 30,
    "timeout": 3600,
    "enable_caching": True,
    "retention_days": 30,
}

# Execute extraction with automatic batch routing

results = lx.extract(
    text_or_documents=text_subset,
    prompt_description=prompt,
    examples=examples,
    model_id="gemini-2.5-flash",
    max_char_buffer=500,
    batch_length=1000,
    language_model_params={
        "vertexai": True,
        "project": "your-gcp-project",
        "location": "us-central1",
        "batch": batch_config,
    },
)

print(f"Extracted {len(results.extractions)} entities.")

Direct infer_batch Usage

For advanced scenarios requiring direct pipeline control, call infer_batch() from langextract/providers/gemini_batch.py (lines 688-785):

from google import genai
from langextract.providers.gemini_batch import infer_batch, BatchConfig

client = genai.Client(vertexai=True, project="my-project", location="us-central1")
prompts = ["Who is the protagonist?", "List all locations mentioned."]

cfg = BatchConfig.from_dict({
    "enabled": True,
    "threshold": 1,
    "enable_caching": False,
    "retention_days": 0,
})

results = infer_batch(
    client=client,
    model_id="gemini-1.5-pro",
    prompts=prompts,
    schema_dict=None,
    gen_config={"temperature": 0.0},
    cfg=cfg,
    system_instruction=None,
    safety_settings=None,
)

How Batch Mode Triggering Works

LangExtract's public extract() function automatically evaluates whether to use batch processing based on the threshold parameter in your BatchConfig. The default threshold is 50 prompts—if the number of chunks after text segmentation meets or exceeds this value, the library switches from real-time Gemini/Vertex AI endpoints to the asynchronous batch pipeline.

You can force batch mode regardless of prompt count by setting enabled=True with a custom threshold:

batch_config = {
    "enabled": True,
    "threshold": 5,  # Trigger batch even for small workloads

}

Caching and Cost Optimization

The batch implementation includes sophisticated caching via GCSBatchCache to minimize redundant API calls. When enable_caching=True, the system:

1. Generates SHA-256 hashes of each prompt to create unique cache keys
2. Stores results in a dedicated GCS bucket named langextract-{project}-{location}-batch
3. Checks for existing results before submitting new batch jobs, returning cached data instantly when available
4. Automatically applies lifecycle rules when retention_days is set, deleting cached objects after the specified period to prevent unbounded storage growth

For workloads exceeding max_prompts_per_job (default varies by configuration), the library automatically splits prompts into multiple batch jobs, each with unique display names like langextract-batch-{timestamp}, allowing you to process millions of documents without hitting Vertex AI request size limits.

Summary

• Vertex AI batch processing in LangExtract requires a genai.Client instantiated with vertexai=True and a valid BatchConfig object.
• The system automatically triggers batch mode when prompt counts exceed the configurable threshold (default 50), or when explicitly enabled.
• The pipeline handles the entire lifecycle: GCS upload, job submission via client.batches.create(), polling via _poll_completion(), and ordered result extraction.
• GCS caching using SHA-256 hashed keys eliminates redundant API calls, with optional lifecycle management via retention_days.
• For large datasets, the library supports chunked job submission through max_prompts_per_job to avoid Vertex AI limits.

Frequently Asked Questions

What is the default batch threshold in LangExtract?

The default threshold is 50 prompts. When the number of text chunks generated from your input meets or exceeds this value, LangExtract automatically routes the workload to the Vertex AI Batch Prediction API instead of the real-time endpoint. You can override this by setting threshold in your BatchConfig.

How does the GCS caching mechanism work?

When enable_caching=True, LangExtract creates a bucket named langextract-{project}-{location}-batch and stores each prompt-result pair using a SHA-256 hash of the prompt as the object key. Before submitting new batch jobs, the system queries this cache via GCSBatchCache and returns any existing matches instantly. Setting retention_days automatically configures lifecycle rules to delete cached objects after the specified period.

Can I use batch processing without Vertex AI?

No. The batch processing pipeline specifically requires a Vertex AI-enabled client. The _is_vertexai_client() function in langextract/providers/gemini_batch.py explicitly validates that vertexai=True is set on the genai.Client instance. Attempting to use batch configuration with a standard Gemini client will result in validation errors.

How do I handle failed items in a batch job?

By default, the batch pipeline stops on errors. However, you can set ignore_item_errors=True in your BatchConfig to continue processing when individual prompts fail. When enabled, the system returns None or partial results for failed items while preserving the order of successful extractions. The _poll_completion() function monitors job status, and _extract_from_file() handles parsing of successful outputs even when some lines contain errors.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}