Processamento em lote com Claude API: como lidar com centenas de documentos eficientemente

import anthropic
import json
from datetime import datetime

client = anthropic.Anthropic(api_key="sua-chave-api")

# Sua fonte de dados
avaliações = [
    "Este produto mudou minha vida. Recomendo muito!",
    "Quebrou depois de duas semanas. Total desperdício.",
    "É ok. Nada de especial mas cumpre o que promete.",
    # ... mais 497 avaliações
]

# Construir requisições
requisições = []
for i, avaliacao in enumerate(avaliações):
    requisições.append({
        "custom_id": f"avaliacao-{i:05d}",
        "params": {
            "model": "claude-3-5-sonnet-20241022",
            "max_tokens": 200,
            "messages": [
                {
                    "role": "user",
                    "content": f"""Analise esta avaliação e forneça:
1. Sentimento (positivo/negativo/neutro)
2. Problema ou benefício principal mencionado
3. Nota geral (1-5)

Avaliação: {avaliacao}"""
                }
            ]
        }
    })

# Criar conteúdo JSONL
conteudo_jsonl = "\n".join(json.dumps(req) for req in requisições)

# Enviar lote
resposta = client.beta.messages.batches.create(
    requests=requisições,
    betas=["batch-2024-09-24"]
)

id_lote = resposta.id
print(f"Lote enviado: {id_lote}")
print(f"Contagem de requisições: {resposta.request_counts.processing}")

Pronto para o envio. Agora você tem um ID de lote que pode verificar depois.

## Padrões de Processamento e Recuperação

Não espere resultados instantâneos. A maioria dos lotes processa em algumas horas, mas você deve se preparar para esperas de até 24 horas.

**Abordagem de verificação (simples, mas nem tanto):**

```python
import time

id_lote = "seu-id-de-lote"

while True:
    lote = client.beta.messages.batches.retrieve(id_lote, betas=["batch-2024-09-24"])

    if lote.processing_status == "ended":
        break

    print(f"Status: {lote.processing_status}")
    print(f"Processadas: {lote.request_counts.succeeded}/{lote.request_counts.total}")

    time.sleep(30)  # Verificar a cada 30 segundos
```

**Abordagem com webhook (pronto para produção):**

Configure um endpoint que receba notificações quando seu lote for concluído:

```python
# No seu envio de lote
resposta = client.beta.messages.batches.create(
    requests=requisições,
    betas=["batch-2024-09-24"],
    # Alguns planos da Anthropic suportam o parâmetro webhook_url
)
```

Quando o webhook dispara, recupere seus resultados imediatamente. Sem desperdício de verificação.

## Manipulando Resultados em Escala

Os resultados vêm em formato JSONL. Cada linha contém sua requisição original combinada com a resposta:

```python
def processar_resultados_lote(id_lote):
    """Processar resultados sem carregar tudo na memória"""

    # Obter o arquivo de resultado
    lote = client.beta.messages.batches.retrieve(id_lote, betas=["batch-2024-09-24"])

    # Processar resultados linha por linha
    arquivo_resultados = client.beta.messages.batches.results(id_lote, betas=["batch-2024-09-24"])

    insights = {
        "positivos": [],
        "negativos": [],
        "neutros": [],
        "erros": []
    }

    for linha in arquivo_resultados:
        resultado = json.loads(linha)

        if resultado.get("error"):
            insights["erros"].append({
                "id": resultado["custom_id"],
                "erro": resultado["error"]
            })
            continue

        # Extrair sentimento da resposta
        conteudo = resultado["message"]["content"][0]["text"]

        if "positivo" in conteudo.lower():
            insights["positivos"].append(resultado["custom_id"])
        elif "negativo" in conteudo.lower():
            insights["negativos"].append(resultado["custom_id"])
        else:
            insights["neutros"].append(resultado["custom_id"])

    return insights
```

Isso processa resultados linha por linha em vez de carregar o arquivo inteiro. Com 10.000 requisições, é a diferença entre 50MB na memória e 1MB constante.

## Táticas de Otimização no Mundo Real

**Estratégia de tamanho de lote:** Não há limite máximo, mas 10.000 a 100.000 requisições por lote é o ideal. Lotes maiores têm eficiência ligeiramente melhor. Se você tem milhões de documentos, divida em múltiplos lotes enviados simultaneamente.

**Contagem de tokens:** Use `anthropic.tokenize()` antes de enviar o lote para estimar custos. Encontre os tokens máximos que você precisa por requisição e multiplique:

```python
estimativa = tokenize(prompt_sistema + documento_maior) * contagem_requisições * 1.1
custo_com_desconto = (estimativa / 1_000_000) * 1.50  # $1.50 por 1M tokens
```

**Lógica de retentativa de erros:** A API de lote retorna erros para requisições individuais (mensagens malformadas, limites de taxa). Não reenvie o lote inteiro. Extraia os valores `custom_id` que falharam e execute novamente:

```python
ids_falhados = [r["custom_id"] for r in resultados if r.get("error")]
requisições_retry = [r for r in requisições_originais if r["custom_id"] in ids_falhados]
# Enviar como novo lote
```

**Extração paralela:** Assim que um lote é concluído, você pode enviar o próximo imediatamente. A API suporta milhares de lotes paralelos. Use isso para pipelines contínuos.

## Quando NÃO Usar Lotes

Lotes não servem para tudo:

- **Fluxos interativos:** Qualquer coisa que exija feedback do usuário. API de lote é fire-and-forget.
- **Classificação em tempo real:** Se você precisa de resultados em segundos, use a API padrão.
- **Documentos únicos:** O overhead não compensa para 5-10 requisições.
- **Testando prompts:** Use a API de Mensagens. A API de lote é menos flexível para iteração.

## Na prática

Trabalho numa varejista com milhares de SKUs ativos no Magento. Um problema recorrente: descrições de produto que eram adequadas anos atrás mas não correspondem mais ao que o cliente pesquisa. Rodar cada SKU pelo Claude de forma interativa nunca seria viável. A API de lotes tornou uma pipeline real de enriquecimento possível. Meu primeiro teste foi 300 SKUs de porcelanato. Dei ao Claude a descrição existente, as especificações técnicas, e um template do que a versão enriquecida deveria incluir. O lote rodou overnight. O custo foi cerca de 60% menor do que rodar as mesmas requisições de forma síncrona durante o horário comercial. Mais importante: o throughput. 300 itens processados enquanto eu dormia, revisados pela manhã. A limitação que encontrei é exatamente o que o artigo descreve: você não consegue debugar interativamente um lote. Um template de prompt mal formado quebrou 40 requisições silenciosamente antes de eu pegar no arquivo de resultados. Teste em 20 requisições antes de escalar.

## A Conclusão

Processamento em lote não é complicado, mas requer planejamento. Você não consegue debugar interativamente um lote de 1.000 requisições. Você precisa de entrada estruturada, paciência e código para manipular resultados.

Mas a matemática é convincente: desconto de 50% mais conformidade garantida com limite de taxa para trabalho assíncrono que você faria mesmo assim. Se está processando centenas de documentos, essa deveria ser sua abordagem padrão. A economia de custos por si só justifica o tempo de implementação.

Comece pequeno — envie um lote de 100 requisições como teste. Construa a manipulação de resultados. Depois escale para milhares.

```