[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 to the Platform component, integrated directly into the existing Platform rather than as a parallel stack.

Submission goes through the regular invocation flow: Platform::invoke($model, $inputs, ['batch' => true]) returns a DeferredResult whose asBatchJob() yields a BatchJob. Polling and result fetching live in a separate BatchManager, so the BatchJob stays a plain value object (no HTTP client) and can be persisted between requests.

The ['batch' => true] routing lives inline in Provider for now, isolated so it can re-hang on the Endpoint concept (#2235) later without a BC break.

New classes

• Batch\BatchInput - a single request within a batch submission
• Batch\BatchJob - immutable, persistable snapshot of a batch job (id, status, counts)
• Batch\BatchResult - result of a single request within a batch
• Batch\BatchStatus - canonical status enum (PROCESSING, COMPLETED, FAILED, CANCELLED, EXPIRED)
• Batch\BatchJobResult - result type unwrapped by DeferredResult::asBatchJob()
• Batch\BatchManager / BatchManagerInterface - polling and result fetching: refresh(), canFetchResults(), fetchResults(), cancel()
• Batch\BatchSubmitClientInterface / BatchClientInterface - provider HTTP contracts for submit and post-submission
• Batch\BatchResultConverterInterface - marks the converter that builds a BatchJobResult
• Capability::BATCH - capability flag for models eligible to batch

Provider support

OpenAI - uploads a JSONL file via the Files API and creates the batch job (Responses endpoint); streams output and error files line by line, with partial recovery for cancelled/expired batches.

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

AiBundle integration

The submit side needs no bundle change: it goes through Factory::createProvider, so invoke(['batch' => true]) works out of the box. The BatchManager is registered per provider as ai.batch_manager.openai / ai.batch_manager.anthropic, and BatchManagerInterface is aliased for autowiring when a single batch-capable platform is configured, mirroring the platform services.

Usage

$platform = Factory::createPlatform(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?'))),
];

// Submit through the regular invocation flow.
$job = $platform->invoke('gpt-4o-mini', $inputs, ['batch' => true])->asBatchJob();

// Poll and fetch through a separate, persistable service.
$manager = new BatchManager(new OpenAi\Batch\ModelClient(http_client(), env('OPENAI_API_KEY')));

while (!$job->isTerminal()) {
    sleep(5);
    $job = $manager->refresh($job);
}

foreach ($manager->fetchResults($job) as $result) {
    echo $result->getId().': '.($result->isSuccess() ? $result->getContent() : $result->getError());
}

Open question: model eligibility

Eligibility is expressed via Capability::BATCH on catalog models and checked by the bridge clients.

Anthropic is simple: all active Claude models support the Message Batches API, so all of them are marked.

OpenAI is trickier: batch is defined per endpoint, not per model, and there is no machine-readable list (models.dev does not expose it, and the docs say "widely available across most of our models, but not all, refer to the model reference"). For now only gpt-4o-mini is marked, to keep things focused. Open to input on the cleanest way to declare the full eligible set (a curated list, deriving it from output modalities, or something else).

Note

This PR is the full-picture reference. The plan is to peel it off into small 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

[chr-hertel]

Hi! Version 0.7 has been released. This PR has changelog/upgrade entries under the 0.7 heading, which is now a released version.

Please update the following files to place your entries under the next version 0.8:

• src/platform/CHANGELOG.md

For CHANGELOG.md files: change the version heading from 0.7 to 0.8
For UPGRADE.md: place entries under UPGRADE FROM 0.7 to 0.8

Thank you!

---

_{🤖 This comment was generated automatically by an agent.}

[tacman]

Hey, @chr-hertel , I see lots of activity and I assume you're prepping another release. Do you have any thoughts on this PR? Would you consider including it in the next release, so that we can kick the tires a bit before the 1.0 release? I know not many people use batch, but it'd be great if the platform at least had the interface for it so we could write provider-specific implementations. Thanks.

[chr-hertel]

Hi @tacman, it's not lost and I still think we should support that - I'm still not convinced about the parallel infrastructure here over integrating it into the platform.
I think the rework in #2029 might also help to integrate it better

[tacman]

@camilleislasse it is easy to get your fork in sync with main or 0.9?

@chr-hertel any feedback on this? "Not planned" is okay, then we can brainstorm about another solution besides trying to keep @camilleislasse 's fork in sync.

Batch really is pretty cool.

[chr-hertel]

Hi @tacman, it's not lost and I still think we should support that - I'm still not convinced about the parallel infrastructure here over integrating it into the platform. I think the rework in #2029 might also help to integrate it better

Still the same answer, not blocking for v1.0 from my point of view

[tacman]

Thanks @chr-hertel — that's clear.

Let me play back what I'm hearing so we build the right thing this time (@camilleislasse — does this match how you'd want to approach it?):

Submit side — integrate into Platform, no parallel stack. Drop BatchPlatformInterface / BatchPlatform. Batch becomes a normal call:

$result = $platform->invoke('gpt-4o-mini', $inputs, ['batch' => true]);
$job = $result->asBatchJob();

invoke() already returns a DeferredResult, so this is asBatchJob() added to the result we already have — not new infrastructure. The bridge's existing client handles the submit request; the response (job id + status + counts) converts into that result.

Phase two — separate, and persistable. BatchJob is a plain value object (id, status, counts) with no HTTP client inside it, so it round-trips to a DB cleanly. Polling and result-fetching live in a separate service (you compared this to DeferredResult + AsyncResponse, which fits). Roughly:

$job = $batchManager->refresh($job);     // re-fetch status by id
foreach ($batchManager->fetchResults($job) as $r) { ... }

That answers the one thing we were stuck on earlier — we didn't want to embed a client in the result, and this avoids it.

On #2029: agreed it's the better substrate. It's been sitting as a draft for a while though — would it help if I carved out slice 1 (the Endpoint value object, the Model::getEndpoints()/getDefaultEndpoint()/... methods, and the AbstractModelCatalog::endpointsForModel() hook) as a small, BC-safe PR on its own? That's the piece your description marks as landable independently, and it's what batch's ['batch' => true] routing would hang off of. If you'd rather drive #2029 yourself, I'm happy to just build batch on top once the endpoint concept exists.

Either way, if you can confirm the two shapes above, I'll start lining up the rework of #1802 to match (assuming @camilleislasse is on board).

[camilleislasse]

Thanks @chr-hertel — that's clear.

Let me play back what I'm hearing so we build the right thing this time (@camilleislasse — does this match how you'd want to approach it?):

Submit side — integrate into Platform, no parallel stack. Drop BatchPlatformInterface / BatchPlatform. Batch becomes a normal call:

$result = $platform->invoke('gpt-4o-mini', $inputs, ['batch' => true]);
$job = $result->asBatchJob();

invoke() already returns a DeferredResult, so this is asBatchJob() added to the result we already have — not new infrastructure. The bridge's existing client handles the submit request; the response (job id + status + counts) converts into that result.

Phase two — separate, and persistable. BatchJob is a plain value object (id, status, counts) with no HTTP client inside it, so it round-trips to a DB cleanly. Polling and result-fetching live in a separate service (you compared this to DeferredResult + AsyncResponse, which fits). Roughly:

$job = $batchManager->refresh($job);     // re-fetch status by id
foreach ($batchManager->fetchResults($job) as $r) { ... }

That answers the one thing we were stuck on earlier — we didn't want to embed a client in the result, and this avoids it.

On #2029: agreed it's the better substrate. It's been sitting as a draft for a while though — would it help if I carved out slice 1 (the Endpoint value object, the Model::getEndpoints()/getDefaultEndpoint()/... methods, and the AbstractModelCatalog::endpointsForModel() hook) as a small, BC-safe PR on its own? That's the piece your description marks as landable independently, and it's what batch's ['batch' => true] routing would hang off of. If you'd rather drive #2029 yourself, I'm happy to just build batch on top once the endpoint concept exists.

Either way, if you can confirm the two shapes above, I'll start lining up the rework of #1802 to match (assuming @camilleislasse is on board).

Thanks @tacman, on board with both shapes.I have 2 question before I start, just to make sure I build the right thing:

#2029: would it make sense to keep batch moving independently for now? ['batch' => true] can route against the current dispatch, and we'd re-hang it on Endpoint once slice 1 lands. Happy either way (just want to avoid blocking if #2029 stays in flight for a while)
asBatchJob() since DeferredResult is generic, I wasn't sure whether the accessor belongs there or whether the converter should just return a BatchJob directly. Do you have a preference?

Once these are settled I'll rework #1802 to match

[tacman]

@camilleislasse great, let's go.

1. Independent of #2029 — yes. Build on the current dispatch now; don't wait. The public surface is just invoke($model, $inputs, ['batch' => true]) + asBatchJob(), and that's stable regardless of how routing is wired underneath — moving the ['batch' => true] decision onto Endpoint later is an internal, non-BC swap. I'll pursue the slice-1 endpoint PR in parallel, but treat it as orthogonal, not a blocker.

2. asBatchJob() on DeferredResult. Put it there, alongside asText()/asObject()/asStreamedObject() — same "typed unwrap of the deferred result" pattern, and it's the shape @chr-hertel used in his own sketch above, so it's already blessed. The converter builds it from the submit response (job id + status + counts); asBatchJob() just returns the BatchJob VO. Keep BatchJob a pure value object (no HTTP client) so it round-trips to a DB — all polling/fetching stays in the separate phase-two service.

One scoping ask: let's keep the first PR to the submit side only (['batch' => true] routing, converter, asBatchJob(), BatchJob VO) and land refresh/fetchResults() as a follow-up. Smaller PR, easier review, and it gets the interface into a release sooner — which was the original goal.

[camilleislasse]

Thanks @tacman, that's clear. Went with all three of your points and pushed the full rework here so you can see the whole shape.

asBatchJob() lives on DeferredResult next to asText()/asVectors(), as you said. The bridge converter builds a BatchJob VO from the submit response and asBatchJob() just unwraps it

BatchJob stays a pure value object (id/status/counts, no HTTP client) so it round-trips to a DB, and all polling/fetching lives in the phase-two service

One shape note: I put refresh()/fetchResults() on the BatchManager rather than $job->getResults() from the original sketch, same reason, to keep BatchJob client-free and persistable.

Routing is inline via ['batch' => true] in Provider for now, isolated so it re-hangs on Endpoint (#2235) later without a BC change

I've put the whole thing here in one piece so the overall direction is easy to see. No rush, but whenever you get a chance, a quick glance to check the direction feels right would be great, and then I'm happy to split it into smaller PRs.

[tacman]

Thanks Camille, this looks right. I went through the diff:

• BatchPlatformInterface/BatchPlatform gone, Provider::invoke() routes on ['batch' => true] — confirmed isolated, no leakage into regular convertResult().
• BatchJob is properly client-free and persistable.
• asBatchJob() on DeferredResult next to asText()/asVectors() — exactly the shape we landed on.
• BatchManager with refresh()/fetchResults()/cancel() matches the phase-two sketch.

Direction's confirmed from my side. A few things before you split it up:

1. Split plan — the 7 commits already read like a clean PR sequence (submit-side Platform → OpenAI submit → Anthropic submit → BatchManager → OpenAI results → Anthropic results → example). Does that match how you're thinking of slicing it, or would you rather group differently?
2. ai-bundle wiring — I don't see any ai-bundle changes; the new $batchSubmitClients arg is only wired in the two bridge Factory.phps. Is DI/bundle registration intentionally deferred to a later PR, or did it get missed in the rework?
3. PR body — still describes the old BatchPlatformInterface/submitBatch()/getBatch()/createBatch() API. Worth updating once we agree on the split so reviewers aren't reading a stale description against new code.
4. cancel()/canFetchResults() on BatchManager — intentional addition, or did that ride along with the rework? No objection either way, just want to make sure it's deliberate.

Once we're aligned on the split, happy to review PR #1 (submit side) in detail.

[camilleislasse]

Thanks @tacman, on your four points:

1. Split plan: yes, the slices follow the commit sequence. Submit side first, then the fetching side (BatchManager + the two result clients), then the ai-bundle wiring

2. ai-bundle: done now. The submit side already worked through Factory::createProvider, and I've added the BatchManager as a per-provider service (ai.batch_manager.openai / ai.batch_manager.anthropic) with a BatchManagerInterface alias for autowiring when a single batch-capable platform is configured, mirroring the platform services

3. PR body: updated to match the new API

4. cancel()/canFetchResults(): deliberate. canFetchResults() is your suggestion from earlier (checking availability before fetching), cancel() was in the original API. Both kept on purpose

One open thing I'd like your input on: model eligibility. Right now I express it via Capability::BATCH on catalog models. Anthropic is easy (all active Claude models support batches, so all are marked), but OpenAI is trickier: batch is per-endpoint not per-model, and there's no machine-readable list (models.dev doesn't expose it, docs just say "most models but not all"). For now I only marked gpt-4o-mini to keep it focused

I also had a look at your ai-batch-bundle and saw you let the caller pick the model directly, with no per-model check. That's simpler and might be the cleaner path here too (route by provider, let the API reject unsupported models) rather than maintaining a Capability::BATCH list. Not decided either way, curious what you'd lean towards (@chr-hertel maybe an idea ?)

[tacman]

@camilleislasse — on model eligibility, I'd keep Capability::BATCH rather than drop the gate.

Every other capability in Platform works as a pre-flight check before the HTTP call goes out — StringToMessageBagListener/PlatformSubscriber throw early on supports(), and Decart/Deepgram route on it. If batch alone skips that and just lets the provider reject unsupported models, that's a new, inconsistent failure mode: a raw OpenAI error instead of the clear exception every other unsupported-capability path gives.

The actual problem isn't the mechanism, it's that OpenAI has no canonical list to seed it from. I wouldn't try to solve that by enumerating the full eligible set up front — that's guaranteed to go stale the moment OpenAI ships a new model. Treat it like the rest of the catalog: start with what's confirmed in the docs (gpt-4o-mini today), and add models as they're verified, same incremental process as any other capability entry. A momentarily-incomplete catalog is already the norm elsewhere; batch doesn't need to solve that problem better than everything else does.

One forward note, not a blocker: you pointed out batch is per-endpoint on OpenAI, not per-model. Now that Endpoint has landed (#2235 slice 1), eligibility might belong there eventually rather than on the model directly. I don't think that should hold up this PR — worth a follow-up issue once we've used Endpoint for a bit and can see if that reshapes the question. @chr-hertel, curious if you see it differently.

Otherwise this is ready for the split into the smaller PRs we discussed.

[chr-hertel]

why is it an generator and not plain array?

Suggested change

	$questions = [
	'req-1' => 'What is the capital of France? Answer in one word.',
	'req-2' => 'What is the capital of Germany? Answer in one word.',
	'req-3' => 'What is the capital of Spain? Answer in one word.',
	];
	$inputs = (static function () use ($questions): Generator {
	foreach ($questions as $id => $question) {
	yield new BatchInput($id, new MessageBag(Message::ofUser($question)));
	}
	})();
	$inputs = [
	new BatchInput('req-1', new MessageBag(Message::ofUser('What is the capital of France? Answer in one word.'))),
	new BatchInput('req-2', new MessageBag(Message::ofUser('What is the capital of Germany? Answer in one word.'))),
	new BatchInput('req-3', new MessageBag(Message::ofUser('What is the capital of Spain? Answer in one word.'))),
	];

[chr-hertel]

can we wire this over the main ModelClient of the corresponding model instead?