[Platform] Add batch processing support

[camilleislasse]

Batch Processing

Q	A
Bug fix?	no
New feature?	yes
Docs?	no
Issues	Fix #1776
License	MIT

Summary

Adds asynchronous batch processing support to the Platform component, inspired by issue #1776.

The implementation mirrors the existing Platform / ModelClientInterface split:

• BatchPlatformInterface public API for submitting and managing batch jobs
• BatchClientInterface HTTP-level contract implemented by each provider
• BatchPlatform normalizes inputs via Contract, delegates to BatchClientInterface

New classes

• BatchInput - a single request within a batch submission
• BatchJob - immutable snapshot of a batch job status
• BatchResult - result of a single request within a completed batch
• BatchStatus - canonical status enum (PROCESSING, COMPLETED, FAILED, CANCELLED, EXPIRED)
• Capability::BATCH - new capability flag for models supporting batch processing

Provider support

OpenAI - uploads a JSONL file via the Files API, creates the batch job, streams results line by line.

Anthropic - streams the request body as a JSON array to avoid loading all requests in memory, uses the Message Batches API.

AiBundle integration

Registers BatchPlatform as a service for OpenAI and Anthropic providers. Automatically aliases BatchPlatformInterface when a single batch platform is configured.

Usage

$platform = PlatformFactory::createBatch(env('OPENAI_API_KEY'));

$inputs = [
    new BatchInput('req-1', new MessageBag(Message::ofUser('What is the capital of France?'))),
    new BatchInput('req-2', new MessageBag(Message::ofUser('What is the capital of Germany?'))),
];

$job = $platform->submitBatch('gpt-4o-mini', $inputs, ['max_tokens' => 50]);

// Poll until complete
while (!$job->isTerminal()) {
    sleep(5);
    $job = $platform->getBatch($job->getId());
}

// Stream results
foreach ($platform->fetchResults($job) as $result) {
    if ($result->isSuccess()) {
        echo $result->getId().': '.$result->getContent();
    }
}

Commits

1. [Platform] Add Batch core abstractions
2. [Platform][OpenAI] Add Batch support
3. [Platform][Anthropic] Add Batch support
4. [AIBundle] Add Batch platform registration
5. [Examples] Add OpenAI batch processing example

Draft — waiting for maintainer guidance on how to split into smaller reviewable PRs.

[OskarStark]

friendly ping @tacman

[chr-hertel]

Hi there, by the first look I don't really get why we need separate infrastructure here and cannot reuse the PlatformInterface we have. I understand that we need to deal with the input/endpoint differently - which Platform/ModelClient implementations currently do already. And I get, that we need to handle the result differently.

Can we do something like this instead?

// regular platform setup via factory
$platform = PlatformFactory::create(env('OPENAI_API_KEY'));

$inputs = [
    new BatchInput('req-1', new MessageBag(Message::ofUser('What is the capital of France?'))),
    new BatchInput('req-2', new MessageBag(Message::ofUser('What is the capital of Germany?'))),
];

$result = $platform->invoke('gpt-4o-mini', $inputs, [
    'max_tokens' => 50,
    'batch' => true, // if needed?
]);

$job = $result->asBatchJob();

// Poll until complete
while (!$job->isTerminal()) {
    sleep(5);
}

// Stream results
foreach ($job->getResults() as $result) {
    if ($result->isSuccess()) {
        echo $result->getId().': '.$result->getContent();
    }
}

[camilleislasse]

Hi there, by the first look I don't really get why we need separate infrastructure here and cannot reuse the PlatformInterface we have. I understand that we need to deal with the input/endpoint differently - which Platform/ModelClient implementations currently do already. And I get, that we need to handle the result differently.

Can we do something like this instead?

// regular platform setup via factory
$platform = PlatformFactory::create(env('OPENAI_API_KEY'));

$inputs = [
    new BatchInput('req-1', new MessageBag(Message::ofUser('What is the capital of France?'))),
    new BatchInput('req-2', new MessageBag(Message::ofUser('What is the capital of Germany?'))),
];

$result = $platform->invoke('gpt-4o-mini', $inputs, [
    'max_tokens' => 50,
    'batch' => true, // if needed?
]);

$job = $result->asBatchJob();

// Poll until complete
while (!$job->isTerminal()) {
    sleep(5);
}

// Stream results
foreach ($job->getResults() as $result) {
    if ($result->isSuccess()) {
        echo $result->getId().': '.$result->getContent();
    }
}

Hi! thanks for taking the time to look at this.

Platform/ModelClient already handle different inputs and endpoints, a BatchModelClient implementing ModelClientInterface could work for the submission side

The part we weren't sure about is the output: unlike all existing clients where results come from the same HTTP response as the request, batch results arrive from a completely separate endpoint, much later

The HTTP response only contains job metadata (id, status, counts), a BatchResultConverter could handle that part. The part we weren't sure about is what comes after: isTerminal() and getResults() both require further HTTP calls, which would mean embedding an HTTP client inside a result object, unlike any existing result type in DeferredResult

if you have something in mind for that part?

[camilleislasse]

@tacman review :

Hey Camille — quick feedback from integrating batch in a Symfony app:

Right now both OpenAI and Anthropic batch bridges only allow fetchResults() when BatchJob::isComplete() is true. That prevents retrieving output for terminal-but-not-complete jobs (e.g. cancelled / expired), even when providers can expose partial output.

Proposed behavior:

• OpenAI: allow fetch when outputFileId !== null (instead of strict isComplete()).
• Anthropic: allow fetch when batch is terminal (or at least not hard-blocked by isComplete() only), and let provider response decide.
• Keep current exception for states where no results are actually available.

Potential API tweak:

• keep fetchResults(BatchJob $job) as-is semantically ("fetch whatever is available")
• optionally add canFetchResults(BatchJob $job): bool to make command UX cleaner.

This would make partial-recovery flows much nicer (especially after cancellation/timeouts) without breaking current completed-path usage.

[chr-hertel]

The HTTP response only contains job metadata (id, status, counts), a BatchResultConverter could handle that part. The part we weren't sure about is what comes after: isTerminal() and getResults() both require further HTTP calls, which would mean embedding an HTTP client inside a result object, unlike any existing result type in DeferredResult

We have a similar issue with the DeferredResult and HttpClient's AsyncResponse already - but a difference might be, that I want to be also to persist the jobs in between.

However, I agree that there needs to be a second part of infrastructure after the initial platform call - but I would like to see the initial call as integrated in the Platform as possible