Process Documents in Batches

An extraction job spends most of its lifetime waiting for the API to process the document. If we submit one file, wait for it to finish, and only then submit the next, that waiting time accumulates across the whole folder. This recipe processes several documents at once. A small Python thread pool gives each document the same schema, submits it to /v2/extract, polls its job, and collects the result. One failed document returns an error without stopping the rest of the batch.

This recipe builds on the Extraction quickstart. Read that first if you want a detailed explanation of the request, polling flow, or response fields for one document.

What We’ll Build

A Python script that:

Finds every PDF in a documents/ directory.
Processes documents concurrently, using four workers by default.
Submits each PDF with one shared extraction schema.
Polls each job until it completes or fails.
Writes all results and per-file errors to results.json.

Our example uses fund fact sheets, but the batch-processing pattern is not specific to financial documents. Replace the PDFs and schema with one document family from your own workflow.

Show the Full Script

Save this as batch_extract.py, place your PDFs in documents/, and set UNSILOED_API_KEY before running it.

batch_extract.py

from concurrent.futures import ThreadPoolExecutor
import json
import os
from pathlib import Path
import time

import requests


API_KEY = os.environ["UNSILOED_API_KEY"]
BASE_URL = "https://prod.visionapi.unsiloed.ai"
DOCUMENT_DIR = Path("documents")
OUTPUT_PATH = Path("results.json")
MAX_WORKERS = 4
POLL_SECONDS = 3
MAX_POLLS = 100

SCHEMA = {
    "type": "object",
    "properties": {
        "fund_name": {
            "type": "string",
            "description": "Full name of the fund",
        },
        "ticker": {
            "type": "string",
            "description": "Primary fund ticker",
        },
        "report_date": {
            "type": "string",
            "description": "Fact sheet as-of date",
        },
        "gross_expense_ratio": {
            "type": "string",
            "description": "Expense ratio with %",
        },
    },
    "required": [
        "fund_name",
        "ticker",
        "report_date",
    ],
    "additionalProperties": False,
}


def submit_document(path):
    """Upload a document and return its job ID."""
    with path.open("rb") as document:
        response = requests.post(
            f"{BASE_URL}/v2/extract",
            headers={"api-key": API_KEY},
            files={
                "pdf_file": (
                    path.name,
                    document,
                    "application/pdf",
                )
            },
            data={
                "schema_data": json.dumps(SCHEMA),
                "model": "gamma",
                "enable_citations": "true",
            },
            timeout=60,
        )
    response.raise_for_status()
    return response.json()["job_id"]


def wait_for_result(job_id):
    """Poll one job until it completes or fails."""
    for _ in range(MAX_POLLS):
        response = requests.get(
            f"{BASE_URL}/extract/{job_id}",
            headers={"api-key": API_KEY},
            timeout=30,
        )
        response.raise_for_status()
        job = response.json()
        status = str(job.get("status", "")).lower()

        if status in ("completed", "review"):
            return job["result"]
        if status in ("failed", "cancelled"):
            message = job.get("error")
            if not message:
                message = f"job {status}"
            raise RuntimeError(message)

        time.sleep(POLL_SECONDS)

    raise TimeoutError(f"Job {job_id} timed out")


def extract_document(path):
    """Submit a document and return its result."""
    job_id = submit_document(path)
    result = wait_for_result(job_id)
    return {
        "file": path.name,
        "status": "completed",
        "job_id": job_id,
        "result": result,
    }


def process_document(path):
    """Turn an exception into a per-file error."""
    try:
        return extract_document(path)
    except Exception as error:
        return {
            "file": path.name,
            "status": "failed",
            "error": str(error),
        }


documents = sorted(
    path
    for path in DOCUMENT_DIR.iterdir()
    if path.suffix.lower() == ".pdf"
)
if not documents:
    raise RuntimeError("No PDFs in documents/")

print(
    f"Found {len(documents)} documents. "
    f"Using {MAX_WORKERS} concurrent workers..."
)
started = time.perf_counter()

with ThreadPoolExecutor(MAX_WORKERS) as pool:
    mapped = pool.map(process_document, documents)
    results = list(mapped)

output = json.dumps(results, indent=2) + "\n"
OUTPUT_PATH.write_text(output)

completed = sum(
    result["status"] == "completed"
    for result in results
)
failed = len(results) - completed
print(f"Completed: {completed}")
print(f"Failed: {failed}")
print(f"Wrote {OUTPUT_PATH}")
elapsed = time.perf_counter() - started
print(f"Elapsed: {elapsed:.1f} seconds")

Step 1: Set Up Your Documents and Dependency

We’ll create a project directory, add a group of similar PDFs, and install the only Python package the script needs.

1.1 Get an Unsiloed API Key

Create an API key in the Unsiloed dashboard, then set it in your shell:

export UNSILOED_API_KEY="your-api-key"

The script reads the key from the environment, so it never needs to appear in the source file.

1.2 Add PDFs to the Input Directory

Create a project directory with a documents/ subdirectory:

mkdir batch-extraction
cd batch-extraction
mkdir documents

Add the PDFs you want to process to documents/. The files should belong to the same document family because the script applies one schema to all of them. A batch might contain invoices from different suppliers, bank statements from different institutions, or claim forms from different customers. For this example, we used these public fund fact sheets:

The State Street document below contains the kind of repeated fields we want to collect from every fact sheet: a fund name, ticker, report date, and expense ratio.

A cropped State Street SPDR S&P 500 ETF Trust fact sheet showing the SPY ticker, report date, characteristics, and fund information table

The example links point to live issuer documents, so their contents can change. You don’t need these exact PDFs to follow the recipe.

1.3 Install Requests

Create a virtual environment and install Requests:

python3 -m venv .venv
source .venv/bin/activate
pip install requests

On Windows PowerShell, activate the environment with .venv\Scripts\Activate.ps1.

Step 2: Configure the Batch and Its Schema

Create a file named batch_extract.py in the project directory. We’ll build it in three pieces: imports, batch settings, and the schema for our example documents.

2.1 Import the Python Modules

Add these imports at the top of batch_extract.py:

batch_extract.py

from concurrent.futures import ThreadPoolExecutor
import json
import os
from pathlib import Path
import time

import requests

Python includes the thread pool, JSON, path, environment, and timing modules. Requests is the only external dependency.

2.2 Set the Batch Limits and File Locations

In batch_extract.py, add the following configuration immediately below the imports:

batch_extract.py

API_KEY = os.environ["UNSILOED_API_KEY"]
BASE_URL = "https://prod.visionapi.unsiloed.ai"
DOCUMENT_DIR = Path("documents")
OUTPUT_PATH = Path("results.json")
MAX_WORKERS = 4
POLL_SECONDS = 3
MAX_POLLS = 100

This example uses four workers as a conservative starting point so the script behaves predictably across different machines, file sizes, and API plans. Raise or lower MAX_WORKERS to suit your workload and account limits. The polling settings allow each job approximately five minutes to finish. They control how long the script waits for a result, not how many documents Unsiloed can process.

2.3 Define the Schema for Your Documents

In batch_extract.py, add a SCHEMA dictionary immediately below the configuration:

batch_extract.py

SCHEMA = {
    "type": "object",
    "properties": {
        "fund_name": {
            "type": "string",
            "description": "Full name of the fund",
        },
        "ticker": {
            "type": "string",
            "description": "Primary fund ticker",
        },
        "report_date": {
            "type": "string",
            "description": "Fact sheet as-of date",
        },
        "gross_expense_ratio": {
            "type": "string",
            "description": "Expense ratio with %",
        },
    },
    "required": [
        "fund_name",
        "ticker",
        "report_date",
    ],
    "additionalProperties": False,
}

This schema belongs to our fund-fact-sheet example. Replace the property names, descriptions, and required fields with the data shared by your own documents. If you’re processing invoices, for example, you might request invoice_number, vendor_name, invoice_date, and total_due instead. Only require a field when it should appear in every valid document. We keep gross_expense_ratio optional because some fact-sheet formats may omit it.

Step 3: Submit and Poll One Document

Before adding concurrency, we’ll define the three small functions that process one PDF from upload to completed result.

3.1 Submit a PDF for Extraction

In batch_extract.py, add submit_document() immediately below SCHEMA:

batch_extract.py

def submit_document(path):
    """Upload a document and return its job ID."""
    with path.open("rb") as document:
        response = requests.post(
            f"{BASE_URL}/v2/extract",
            headers={"api-key": API_KEY},
            files={
                "pdf_file": (
                    path.name,
                    document,
                    "application/pdf",
                )
            },
            data={
                "schema_data": json.dumps(SCHEMA),
                "model": "gamma",
                "enable_citations": "true",
            },
            timeout=60,
        )
    response.raise_for_status()
    return response.json()["job_id"]

The function opens one PDF, sends it with the shared schema, and returns the job ID from the submission response. The gamma model is the default, recommended Unsiloed extraction tier. See the model parameter for the available tiers. The enable_citations option adds a page number and bounding box to each extracted field, so a reviewer can see where every value came from. Confidence scores come back either way.

3.2 Wait for the Job to Finish

In batch_extract.py, add wait_for_result() immediately below submit_document():

batch_extract.py

def wait_for_result(job_id):
    """Poll one job until it completes or fails."""
    for _ in range(MAX_POLLS):
        response = requests.get(
            f"{BASE_URL}/extract/{job_id}",
            headers={"api-key": API_KEY},
            timeout=30,
        )
        response.raise_for_status()
        job = response.json()
        status = str(job.get("status", "")).lower()

        if status in ("completed", "review"):
            return job["result"]
        if status in ("failed", "cancelled"):
            message = job.get("error")
            if not message:
                message = f"job {status}"
            raise RuntimeError(message)

        time.sleep(POLL_SECONDS)

    raise TimeoutError(f"Job {job_id} timed out")

The function checks the result endpoint every three seconds. It returns the extraction result for a completed job and raises an error for a failed, cancelled, or timed-out job.

3.3 Combine Submission and Polling

In batch_extract.py, add extract_document() immediately below wait_for_result():

batch_extract.py

def extract_document(path):
    """Submit a document and return its result."""
    job_id = submit_document(path)
    result = wait_for_result(job_id)
    return {
        "file": path.name,
        "status": "completed",
        "job_id": job_id,
        "result": result,
    }

This function gives the thread pool one operation to run for each path. It also adds the source filename and job ID to the result, so we can connect the output to its document later.

Step 4: Run the Documents as a Batch

The remaining code isolates per-file errors, runs the worker pool, and writes one output file.

4.1 Keep One Failure From Stopping the Batch

In batch_extract.py, add process_document() immediately below extract_document():

batch_extract.py

def process_document(path):
    """Turn an exception into a per-file error."""
    try:
        return extract_document(path)
    except Exception as error:
        return {
            "file": path.name,
            "status": "failed",
            "error": str(error),
        }

Without this wrapper, an HTTP error or corrupt PDF could raise out of the thread pool and stop the script. Instead, that document becomes a failed result while the other workers continue.

4.2 Find the PDFs and Start the Worker Pool

In batch_extract.py, add the following code immediately below process_document():

batch_extract.py

documents = sorted(
    path
    for path in DOCUMENT_DIR.iterdir()
    if path.suffix.lower() == ".pdf"
)
if not documents:
    raise RuntimeError("No PDFs in documents/")

print(
    f"Found {len(documents)} documents. "
    f"Using {MAX_WORKERS} concurrent workers..."
)
started = time.perf_counter()

with ThreadPoolExecutor(MAX_WORKERS) as pool:
    mapped = pool.map(process_document, documents)
    results = list(mapped)

ThreadPoolExecutor calls process_document() once per PDF. The MAX_WORKERS setting determines how many of those calls the script runs concurrently.

Tune the worker count for your workload. More workers can improve throughput, but they also create more simultaneous uploads and polling requests, use more local memory, and may reach your account’s rate limits. Without retry logic, the script records a rate-limit response as a failed document.

4.3 Save the Results and Print a Summary

At the end of batch_extract.py, immediately below the worker-pool block, add:

batch_extract.py

output = json.dumps(results, indent=2) + "\n"
OUTPUT_PATH.write_text(output)

completed = sum(
    result["status"] == "completed"
    for result in results
)
failed = len(results) - completed
print(f"Completed: {completed}")
print(f"Failed: {failed}")
print(f"Wrote {OUTPUT_PATH}")
elapsed = time.perf_counter() - started
print(f"Elapsed: {elapsed:.1f} seconds")

The script keeps the complete API response for each successful document and the error message for each failed one. The terminal summary gives us a quick check without hiding the per-file details in results.json.

Step 5: Run the Batch and Inspect the Results

Run the completed script from the project directory:

python batch_extract.py

A successful run should print output shaped like this:

Found 3 documents. Using 4 concurrent workers...
Completed: 3
Failed: 0
Wrote results.json
Elapsed: 33.5 seconds

Timing depends on document size, complexity, model load, worker count, and account limits, so your elapsed time should differ. Open results.json to inspect every API result:

python -m json.tool results.json | less

Each successful entry keeps the complete extraction response:

{
  "file": "spy.pdf",
  "status": "completed",
  "job_id": "6464ef7b-a554-4199-a549-15b6d82d0b49",
  "result": {
    "ticker": {
      "value": "SPY",
      "score": {
        "grounding_score": 0.995,
        "extraction_score": 0.995
      },
      "citation": {
        "page": 1,
        "bbox": [502, 40, 550, 71],
        "page_width": 612.0,
        "page_height": 792.0
      }
    }
  }
}

Keeping the scores and citations lets the next stage flag uncertain values and show a reviewer where each value came from. Don’t flatten the response to values alone unless the downstream system no longer needs that evidence.

Where to Take This Next

The pattern stays the same when you replace the sample documents:

Put one document family in the input directory.
Describe its shared fields in one schema.
Tune the worker count for your throughput target, account limits, file sizes, and available memory.
Keep each file’s error separate from the rest of the batch.

This example resubmits every PDF each time it runs. For a scheduled or high-volume production pipeline, add persistent job IDs, retries with backoff, and a queue. Those solve operational problems, but they aren’t required to understand or use the core batch-processing pattern. If the input directory contains unrelated document types, classify and route them before extraction. The Sort and Extract cookbook shows how to select a schema based on document type.

Extraction Response Format

Read values, confidence scores, and citations from each completed result.

Sort and Extract a Mixed Document Pile

Route different document types to different schemas before processing them.

​What We’ll Build

​Step 1: Set Up Your Documents and Dependency

​1.1 Get an Unsiloed API Key

​1.2 Add PDFs to the Input Directory

​1.3 Install Requests

​Step 2: Configure the Batch and Its Schema

​2.1 Import the Python Modules

​2.2 Set the Batch Limits and File Locations

​2.3 Define the Schema for Your Documents

​Step 3: Submit and Poll One Document

​3.1 Submit a PDF for Extraction

​3.2 Wait for the Job to Finish

​3.3 Combine Submission and Polling

​Step 4: Run the Documents as a Batch

​4.1 Keep One Failure From Stopping the Batch

​4.2 Find the PDFs and Start the Worker Pool

​4.3 Save the Results and Print a Summary

​Step 5: Run the Batch and Inspect the Results

​Where to Take This Next

Extraction Response Format

Sort and Extract a Mixed Document Pile

What We’ll Build

Step 1: Set Up Your Documents and Dependency

1.1 Get an Unsiloed API Key

1.2 Add PDFs to the Input Directory

1.3 Install Requests

Step 2: Configure the Batch and Its Schema

2.1 Import the Python Modules

2.2 Set the Batch Limits and File Locations

2.3 Define the Schema for Your Documents

Step 3: Submit and Poll One Document

3.1 Submit a PDF for Extraction

3.2 Wait for the Job to Finish

3.3 Combine Submission and Polling

Step 4: Run the Documents as a Batch

4.1 Keep One Failure From Stopping the Batch

4.2 Find the PDFs and Start the Worker Pool

4.3 Save the Results and Print a Summary

Step 5: Run the Batch and Inspect the Results

Where to Take This Next