Como processar milhares de chamadas ao Claude com a Message Batches API no Laravel

A Message Batches API da Anthropic permite enviar até 100 mil requisições ao Claude de uma vez, receber os resultados em algumas horas (no máximo 24) e pagar metade do preço por token. No Laravel, o fluxo cabe em três peças: um comando que monta e envia o lote, um job agendado que consulta o status e um job que baixa o arquivo JSONL de resultados e aplica cada resposta pelo custom_id. Se a tarefa não precisa de resposta imediata, como categorizar o histórico, reprocessar documentos ou enriquecer cadastros, essa é a forma mais barata e mais estável de chamar um LLM em volume.

O que você vai ter funcionando no final

Um pipeline que pega transações sem categoria, envia tudo num lote para o Claude, acompanha o processamento sozinho pelo scheduler, grava a categoria de cada transação, registra o consumo de tokens e reenvia automaticamente o que expirou ou deu erro transitório.

Pré-requisitos:

1. Laravel 11 ou 12 com PHP 8.2+.
2. PostgreSQL (o exemplo usa PostgreSQL, mas nada aqui é exclusivo dele).
3. Um worker de fila rodando (Redis + Horizon, ou o driver database).
4. O scheduler ativo (php artisan schedule:work em dev ou o cron em produção).
5. Uma chave da API da Anthropic com acesso ao modelo escolhido.

Não uso SDK: tudo vai pelo Http:: do Laravel, então você vê exatamente o JSON que trafega.

Quando usar a Batch API em vez de chamadas síncronas?

Batch API é um endpoint assíncrono: você envia uma lista de requisições, recebe um ID e volta depois para buscar os resultados. A regra que sigo é simples: se existe um humano esperando a resposta, é chamada síncrona; se não existe, é lote.

Critério	Chamada síncrona (/v1/messages)	Message Batches API
Latência	Segundos	Minutos a até 24 horas
Preço	Cheio	50% de desconto em entrada e saída
Volume	Limitado pelo rate limit da sua conta	Até 100 mil requisições ou 256 MB por lote
Retry e throttling	Você implementa	A Anthropic distribui a carga
Streaming	Sim	Não
Uso típico	Agente no WhatsApp, chat, tela do usuário	Backfill, classificação noturna, relatórios

Na Granpag, o caso clássico é categorização de extrato. Em tempo real, quando chega uma transação nova, faz sentido a chamada síncrona que descrevi em como categorizar transações do extrato com Claude no Laravel. Mas quando um cliente novo importa meses de histórico, ou quando o prompt muda e você precisa recategorizar tudo, disparar milhares de chamadas síncronas é queimar rate limit e dinheiro. Para ter uma ideia da diferença no seu volume, a calculadora de custo de IA compara o custo mensal com e sem Batch API e prompt caching.

1. Configurar a chave e o cliente HTTP

No .env:

ANTHROPIC_API_KEY=sk-ant-...
ANTHROPIC_BATCH_MODEL=claude-haiku-4-5

Em config/services.php:

'anthropic' => [
    'key' => env('ANTHROPIC_API_KEY'),
    'version' => '2023-06-01',
    'batch_model' => env('ANTHROPIC_BATCH_MODEL', 'claude-haiku-4-5'),
],

Agora o cliente, em app/Ai/AnthropicBatches.php. Ele faz quatro coisas: criar o lote, consultar, cancelar e baixar os resultados.

<?php

namespace App\Ai;

use Illuminate\Http\Client\ConnectionException;
use Illuminate\Http\Client\PendingRequest;
use Illuminate\Http\Client\RequestException;
use Illuminate\Support\Facades\Http;
use Psr\Http\Message\StreamInterface;

class AnthropicBatches
{
    private const BASE = 'https://api.anthropic.com/v1/messages/batches';

    private function http(): PendingRequest
    {
        return Http::withHeaders([
                'x-api-key' => config('services.anthropic.key'),
                'anthropic-version' => config('services.anthropic.version'),
            ])
            ->acceptJson()
            ->timeout(120)
            ->retry(3, 2000, fn ($e) => $e instanceof ConnectionException
                || ($e instanceof RequestException && $e->response->serverError()));
    }

    public function create(array $requests): array
    {
        return $this->http()->post(self::BASE, ['requests' => $requests])->throw()->json();
    }

    public function retrieve(string $id): array
    {
        return $this->http()->get(self::BASE."/{$id}")->throw()->json();
    }

    public function cancel(string $id): array
    {
        return $this->http()->post(self::BASE."/{$id}/cancel")->throw()->json();
    }

    public function results(string $resultsUrl): StreamInterface
    {
        return $this->http()->get($resultsUrl)->throw()->toPsrResponse()->getBody();
    }
}

O retry só repete erro de conexão e 5xx. Um 400 de validação repetido três vezes continua sendo 400.

2. Criar as tabelas de controle do lote

Você precisa saber, a qualquer momento, quais lotes estão em voo e qual item local corresponde a cada custom_id. custom_id é o identificador que você dá a cada requisição dentro do lote; é a única forma de ligar um resultado ao registro de origem, porque os resultados não voltam na ordem de envio.

php artisan make:migration create_llm_batches_tables

public function up(): void
{
    Schema::create('llm_batches', function (Blueprint $table) {
        $table->id();
        $table->string('provider_batch_id')->nullable()->unique();
        $table->string('purpose');
        $table->string('status'); // submitting, in_progress, canceling, ended, failed
        $table->unsignedInteger('request_count');
        $table->unsignedInteger('succeeded')->default(0);
        $table->unsignedInteger('errored')->default(0);
        $table->unsignedInteger('expired')->default(0);
        $table->unsignedInteger('canceled')->default(0);
        $table->text('results_url')->nullable();
        $table->timestamp('submitted_at')->nullable();
        $table->timestamp('ended_at')->nullable();
        $table->timestamp('processed_at')->nullable();
        $table->timestamps();
    });

    Schema::create('llm_batch_items', function (Blueprint $table) {
        $table->id();
        $table->foreignId('llm_batch_id')->constrained()->cascadeOnDelete();
        $table->string('custom_id', 64);
        $table->morphs('subject');
        $table->string('status')->default('pending'); // pending, succeeded, errored, expired, canceled, invalid
        $table->text('error')->nullable();
        $table->unsignedInteger('input_tokens')->nullable();
        $table->unsignedInteger('output_tokens')->nullable();
        $table->timestamps();
        $table->unique(['llm_batch_id', 'custom_id']);
    });
}

Nos models, LlmBatch tem items(): HasMany e LlmBatchItem tem subject(): MorphTo. Em Transaction, adicione:

public function batchItems(): MorphMany
{
    return $this->morphMany(LlmBatchItem::class, 'subject');
}

3. Montar os parâmetros de cada requisição

Cada item do lote tem o mesmo formato do corpo de /v1/messages, dentro de params. Uso tool calling forçado para garantir um JSON com formato previsível: tool calling é quando você descreve uma função com JSON Schema e o modelo responde com os argumentos dela; com tool_choice apontando para a ferramenta, ele é obrigado a usá-la.

<?php

namespace App\Ai;

use App\Models\Transaction;

final class CategorizationPrompt
{
    public static function params(Transaction $t, array $categories): array
    {
        return [
            'model' => config('services.anthropic.batch_model'),
            'max_tokens' => 256,
            'tools' => [[
                'name' => 'categorizar',
                'description' => 'Registra a categoria da transação.',
                'input_schema' => [
                    'type' => 'object',
                    'properties' => [
                        'categoria' => ['type' => 'string', 'enum' => $categories],
                        'confianca' => ['type' => 'number', 'minimum' => 0, 'maximum' => 1],
                    ],
                    'required' => ['categoria', 'confianca'],
                ],
            ]],
            'tool_choice' => ['type' => 'tool', 'name' => 'categorizar'],
            'system' => [[
                'type' => 'text',
                'text' => 'Você classifica transações de conta PJ no Brasil. Use só as categorias da ferramenta. Na dúvida, use "outros" com confiança baixa.',
                'cache_control' => ['type' => 'ephemeral'],
            ]],
            'messages' => [[
                'role' => 'user',
                'content' => sprintf(
                    "Descrição: %s\nValor: %d centavos\nDireção: %s",
                    $t->description, $t->amount_cents, $t->direction
                ),
            ]],
        ];
    }
}

O cache_control no system marca tools + system como prefixo cacheável. Prompt caching funciona dentro de lotes, mas o acerto de cache é por melhor esforço, porque as requisições são processadas em paralelo, e o prefixo precisa ter um tamanho mínimo para ser cacheado. Se o seu prompt é curto como este, o ganho vem quase todo do desconto do lote; com instruções longas e exemplos, o cache soma bem.

4. Enviar o lote com um comando Artisan

php artisan make:command SubmitCategorizationBatch

<?php

namespace App\Console\Commands;

use App\Ai\AnthropicBatches;
use App\Ai\CategorizationPrompt;
use App\Models\{Category, LlmBatch, LlmBatchItem, Transaction};
use Illuminate\Console\Command;
use Throwable;

class SubmitCategorizationBatch extends Command
{
    protected $signature = 'ai:categorize-batch {--limit=10000}';
    protected $description = 'Envia transações sem categoria para a Message Batches API';

    public function handle(AnthropicBatches $api): int
    {
        $transactions = Transaction::query()
            ->whereNull('category_id')
            ->whereDoesntHave('batchItems', fn ($q) => $q->whereIn('status', ['pending', 'succeeded', 'invalid']))
            ->whereHas('batchItems', fn ($q) => $q->whereIn('status', ['errored', 'expired']), '<', 3)
            ->orderBy('id')
            ->limit((int) $this->option('limit'))
            ->get();

        if ($transactions->isEmpty()) {
            $this->info('Nada para enviar.');
            return self::SUCCESS;
        }

        $categories = Category::pluck('slug')->all();

        $batch = LlmBatch::create([
            'purpose' => 'categorize_transactions',
            'status' => 'submitting',
            'request_count' => $transactions->count(),
        ]);

        $now = now();
        $rows = [];
        $requests = [];

        foreach ($transactions as $t) {
            $customId = "txn-{$t->id}-b{$batch->id}";
            $rows[] = [
                'llm_batch_id' => $batch->id,
                'custom_id' => $customId,
                'subject_type' => $t->getMorphClass(),
                'subject_id' => $t->id,
                'status' => 'pending',
                'created_at' => $now,
                'updated_at' => $now,
            ];
            $requests[] = ['custom_id' => $customId, 'params' => CategorizationPrompt::params($t, $categories)];
        }

        collect($rows)->chunk(1000)->each(fn ($chunk) => LlmBatchItem::insert($chunk->all()));

        try {
            $response = $api->create($requests);
        } catch (Throwable $e) {
            $batch->items()->delete();
            $batch->update(['status' => 'failed']);
            throw $e;
        }

        $batch->update([
            'provider_batch_id' => $response['id'],
            'status' => $response['processing_status'],
            'submitted_at' => now(),
        ]);

        $this->info("Lote {$response['id']} enviado com {$transactions->count()} requisições.");

        return self::SUCCESS;
    }
}

O JSON que vai para POST /v1/messages/batches fica assim (resumido):

{
  "requests": [
    {
      "custom_id": "txn-981-b12",
      "params": {
        "model": "claude-haiku-4-5",
        "max_tokens": 256,
        "tool_choice": {"type": "tool", "name": "categorizar"},
        "messages": [{"role": "user", "content": "Descrição: PIX ENVIADO ..."}]
      }
    }
  ]
}

A resposta traz id (algo como msgbatch_...), processing_status: "in_progress", request_counts e results_url: null.

Duas decisões aqui. Primeiro, crio o registro local antes de chamar a API: se a chamada falhar, apago os itens e as transações voltam a ser elegíveis. O caso ruim é o timeout depois de a Anthropic aceitar o lote: você fica com um lote órfão. Aceito esse risco porque o pior que acontece é pagar duas vezes por algumas classificações; num fluxo que mexe com dinheiro eu não aceitaria. Segundo, o custom_id inclui o ID do lote, então reenviar a mesma transação gera um custom_id novo e não colide com a tentativa anterior.

5. Acompanhar o status com o scheduler

Não existe webhook de conclusão de lote, então é polling. Um job único, rodando a cada cinco minutos, basta.

<?php

namespace App\Jobs;

use App\Ai\AnthropicBatches;
use App\Models\LlmBatch;
use Illuminate\Contracts\Queue\ShouldBeUnique;
use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Foundation\Queue\Queueable;

class PollLlmBatches implements ShouldQueue, ShouldBeUnique
{
    use Queueable;

    public function handle(AnthropicBatches $api): void
    {
        LlmBatch::whereIn('status', ['in_progress', 'canceling'])->each(function (LlmBatch $batch) use ($api) {
            $data = $api->retrieve($batch->provider_batch_id);
            $counts = $data['request_counts'];

            $batch->update([
                'status' => $data['processing_status'],
                'succeeded' => $counts['succeeded'],
                'errored' => $counts['errored'],
                'expired' => $counts['expired'],
                'canceled' => $counts['canceled'],
                'results_url' => $data['results_url'],
                'ended_at' => $data['ended_at'],
            ]);

            if ($data['processing_status'] === 'ended') {
                ProcessLlmBatchResults::dispatch($batch->id);
            }
        });
    }
}

Em routes/console.php:

use App\Jobs\PollLlmBatches;
use Illuminate\Support\Facades\Schedule;

Schedule::job(new PollLlmBatches)->everyFiveMinutes();

Como o status vira ended no mesmo update que dispara o processamento, o lote não é selecionado de novo no próximo ciclo.

6. Baixar e aplicar os resultados do JSONL

O results_url devolve um arquivo JSONL, um objeto JSON por linha. Cada linha tem custom_id e um result cujo type é succeeded, errored, canceled ou expired. Leio linha a linha do stream para não carregar o arquivo inteiro num array.

<?php

namespace App\Jobs;

use App\Ai\AnthropicBatches;
use App\Models\{Category, LlmBatch, LlmBatchItem};
use GuzzleHttp\Psr7\Utils;
use Illuminate\Contracts\Queue\{ShouldBeUnique, ShouldQueue};
use Illuminate\Foundation\Queue\Queueable;
use Illuminate\Support\Facades\{DB, Validator};
use Illuminate\Validation\Rule;

class ProcessLlmBatchResults implements ShouldQueue, ShouldBeUnique
{
    use Queueable;

    public int $timeout = 900;

    public function __construct(public int $batchId) {}

    public function uniqueId(): string
    {
        return (string) $this->batchId;
    }

    public function handle(AnthropicBatches $api): void
    {
        $batch = LlmBatch::findOrFail($this->batchId);
        if ($batch->processed_at) {
            return;
        }

        $stream = $api->results($batch->results_url);

        while (! $stream->eof()) {
            $line = trim(Utils::readLine($stream));
            if ($line === '') {
                continue;
            }

            $row = json_decode($line, true, flags: JSON_THROW_ON_ERROR);
            $item = $batch->items()->where('custom_id', $row['custom_id'])->where('status', 'pending')->first();
            if (! $item) {
                continue; // já processado numa execução anterior
            }

            $result = $row['result'];

            match ($result['type']) {
                'succeeded' => $this->applySucceeded($item, $result['message']),
                'errored' => $this->markErrored($item, $result),
                default => $item->update(['status' => $result['type']]), // expired, canceled
            };
        }

        $batch->update(['processed_at' => now()]);
    }

    private function applySucceeded(LlmBatchItem $item, array $message): void
    {
        $input = collect($message['content'])->firstWhere('type', 'tool_use')['input'] ?? [];

        $validator = Validator::make($input, [
            'categoria' => ['required', 'string', Rule::exists('categories', 'slug')],
            'confianca' => ['required', 'numeric', 'between:0,1'],
        ]);

        if ($validator->fails()) {
            $item->update(['status' => 'invalid', 'error' => $validator->errors()->first()]);
            return;
        }

        DB::transaction(function () use ($item, $input, $message) {
            if ($input['confianca'] >= 0.7) {
                $item->subject->update([
                    'category_id' => Category::where('slug', $input['categoria'])->value('id'),
                ]);
            }

            $item->update([
                'status' => 'succeeded',
                'input_tokens' => $message['usage']['input_tokens'] ?? null,
                'output_tokens' => $message['usage']['output_tokens'] ?? null,
            ]);
        });
    }

    private function markErrored(LlmBatchItem $item, array $result): void
    {
        $type = data_get($result, 'error.error.type', data_get($result, 'error.type'));

        $item->update([
            'status' => $type === 'invalid_request_error' ? 'invalid' : 'errored',
            'error' => data_get($result, 'error.error.message', json_encode($result['error'] ?? null)),
        ]);
    }
}

Pontos que importam:

1. Idempotência por item: o filtro status = pending faz o job poder rodar de novo depois de uma queda no meio do arquivo sem aplicar nada duas vezes.
2. Validação mesmo com tool calling forçado: o enum do schema ajuda, mas a categoria pode ter sido removida do banco entre o envio e o resultado. O mesmo raciocínio de validar o JSON do Claude e pedir correção quando a validação falha vale aqui, só que a correção vira um item invalid para revisão em vez de uma nova chamada.
3. Confiança baixa não grava categoria: o item fica succeeded, mas a transação continua sem categoria e vai para a fila de revisão manual. Ela não é reenviada porque já tem item succeeded.
4. Erros de validação da requisição são invalid: reenviar uma requisição mal formada não adianta.

7. Reenviar o que expirou ou falhou

Não precisa de código novo. A query do passo 4 já considera elegível qualquer transação sem categoria, sem item pending, succeeded ou invalid, e com menos de três itens errored/expired. Basta agendar o envio:

Schedule::command('ai:categorize-batch --limit=10000')->hourly()->withoutOverlapping();

Com isso, o backfill inteiro anda sozinho: cada hora sai um lote novo com o que falta, inclusive as sobras dos lotes anteriores.

8. Testar o fluxo com Http::fake

Como tudo passa pelo Http::, dá para testar o pipeline completo sem chamar a Anthropic. A ordem das chaves importa: o padrão * casa com barras, então a rota de resultados precisa vir antes.

it('aplica a categoria vinda do lote', function () {
    $category = Category::create(['slug' => 'fornecedores', 'name' => 'Fornecedores']);
    $transaction = Transaction::factory()->create(['category_id' => null]);

    $batch = LlmBatch::create([
        'provider_batch_id' => 'msgbatch_test', 'purpose' => 'categorize_transactions',
        'status' => 'in_progress', 'request_count' => 1,
    ]);
    $batch->items()->create([
        'custom_id' => "txn-{$transaction->id}-b{$batch->id}",
        'subject_type' => $transaction->getMorphClass(), 'subject_id' => $transaction->id,
    ]);

    $jsonl = json_encode([
        'custom_id' => "txn-{$transaction->id}-b{$batch->id}",
        'result' => ['type' => 'succeeded', 'message' => [
            'content' => [['type' => 'tool_use', 'name' => 'categorizar',
                'input' => ['categoria' => 'fornecedores', 'confianca' => 0.92]]],
            'usage' => ['input_tokens' => 310, 'output_tokens' => 40],
        ]],
    ])."\n";

    Http::fake([
        'api.anthropic.com/v1/messages/batches/*/results' => Http::response($jsonl),
        'api.anthropic.com/v1/messages/batches/*' => Http::response([
            'id' => 'msgbatch_test', 'processing_status' => 'ended', 'ended_at' => now()->toIso8601String(),
            'request_counts' => ['processing' => 0, 'succeeded' => 1, 'errored' => 0, 'canceled' => 0, 'expired' => 0],
            'results_url' => 'https://api.anthropic.com/v1/messages/batches/msgbatch_test/results',
        ]),
    ]);

    (new PollLlmBatches)->handle(app(AnthropicBatches::class));

    expect($transaction->fresh()->category_id)->toBe($category->id)
        ->and($batch->fresh()->processed_at)->not->toBeNull();
});

Com a fila em sync no ambiente de teste, o dispatch do processamento roda na hora. Para gravar respostas reais e reaproveitar como fixture, o processo é o mesmo de testar código que chama o Claude com Http::fake e respostas gravadas.

Erros comuns

1. custom_id com caractere inválido ou longo demais: ele aceita só letras, números, - e _, com até 64 caracteres. Um UUID com prefixo cabe; um e-mail ou algo com ponto derruba a criação do lote inteiro com 400.
2. Achar que o create valida cada requisição: a validação dos params de cada item é assíncrona. Um max_tokens faltando ou um nome de modelo errado só aparece como errored no resultado, horas depois. Antes de mandar 10 mil, mande os mesmos params uma vez pelo /v1/messages síncrono.
3. Assumir a ordem dos resultados: o JSONL não vem na ordem do envio. Sempre case pelo custom_id.
4. custom_id duplicado no mesmo lote: precisa ser único dentro do lote. Se você gera o ID só a partir da transação e ela aparece duas vezes na query, o lote é rejeitado.
5. Esquecer de processar a tempo: os resultados ficam disponíveis por 29 dias depois da criação. Um job que falhou em silêncio e só foi notado depois disso significa pagar de novo.
6. Usar lote em fluxo com usuário esperando: a maioria termina em menos de uma hora, mas o limite é 24 horas, e o que não terminar até lá volta como expired. Agente no WhatsApp, tela de checkout ou qualquer coisa com SLA de segundos não é caso de lote.
7. Lote gigante num único job: o limite é 100 mil requisições ou 256 MB, mas lotes de alguns milhares são mais fáceis de monitorar, reprocessar e caber no timeout do worker. Ajuste o --limit e o $timeout juntos.
8. Polling agressivo: consultar a cada poucos segundos não acelera nada e consome rate limit da sua conta. Cinco minutos é suficiente.

Perguntas frequentes

Quanto a Batch API economiza?

O preço por token cai pela metade, tanto na entrada quanto na saída, em relação à chamada síncrona do mesmo modelo. Combinada com prompt caching em prompts longos, a economia é ainda maior.

Dá para cancelar um lote em andamento?

Sim, com POST /v1/messages/batches/{id}/cancel. O status passa para canceling e depois ended; o que já foi processado volta como succeeded e o resto como canceled, e você paga só pelo que foi processado.

Posso usar tool calling, visão e system prompt dentro do lote?

Sim. Qualquer parâmetro aceito pelo /v1/messages vai dentro de params. A exceção prática é streaming, que não existe em lote.

A OpenAI tem algo equivalente?

Tem a Batch API da OpenAI, com desconto parecido, mas o fluxo é diferente: você sobe um arquivo JSONL pela Files API e cria o lote apontando para ele. A estrutura de tabelas, polling e processamento por custom_id deste tutorial serve para os dois.

Como saber quanto cada lote custou?

Some input_tokens e output_tokens dos itens succeeded agrupando por llm_batch_id e aplique o preço do modelo com 50% de desconto. Gravar o usage por item, como no passo 6, é o que torna essa conta possível depois.

Conclusão prática

Antes de otimizar prompt ou trocar de modelo para economizar, olhe quais chamadas ao Claude no seu sistema não têm ninguém esperando a resposta. Backfill, recategorização depois de mudar o prompt, enriquecimento de cadastro e relatórios noturnos quase sempre podem virar lote. Comece por um único caso, com as duas tabelas de controle e os três componentes deste tutorial (comando de envio, polling e processamento idempotente por custom_id), rode um lote pequeno validando os params antes e só então aumente o --limit. O resto do pipeline já fica pronto para os próximos casos.