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.

## 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.

```