doc-extract

Extração de dados estruturados de documentos bagunçados com Claude API + Pydantic — tratando o modelo como um componente não-confiável, não como uma caixa mágica.

A coisa em ação

Links

Repositório:  [ PREENCHER ]
Demo:         [ PREENCHER ]/demo
Docs (ReDoc): [ PREENCHER ]/docs
Vídeo:        [ PREENCHER — opcional ]

O problema

Todo mundo que quer vaga de "IA" faz wrapper de chat — POST pra API, mostra a resposta na tela. Isso não prova nada.

A pergunta que separa engenharia de wrapper é: o que acontece quando o modelo erra? E ele erra — devolve CPF com pontuação quando o schema pede dígito, inventa item de nota fiscal, troca um valor. A versão ingênua (manda texto, espera JSON, parseia, torce) funciona na maior parte das vezes — e falha silenciosamente no resto, porque nada valida se a soma dos itens de uma nota bate com o total.

A solução (visão geral)

• Pydantic valida regra de negócio, não só tipo — o validator total_bate da nota fiscal pega alucinação aritmética (item inventado, valor trocado) que uma checagem de tipo simples deixaria passar.
• O erro do Pydantic é reenviado no prompt — não é "tenta de novo e reza", é um loop de controle: a tentativa seguinte recebe o erro exato de qual campo falhou e por quê.
• Depois de 3 tentativas, o sistema falha alto — devolve NEEDS_REVIEW com o erro e o output cru, pra um humano decidir em 10 segundos, em vez de inventar dado pra completar a resposta.

flowchart TD
    A[Texto do documento] --> B[Monta prompt + JSON schema]
    B --> C[Chama Claude API]
    C --> D{JSON válido?}
    D -->|não| E[Reenvia com erro exato]
    E --> C
    D -->|sim| F{Pydantic valida?}
    F -->|não| E
    F -->|sim| G[status: DONE]
    E -.->|3ª falha| H[status: NEEDS_REVIEW]

    style G fill:#22C55E,color:#fff
    style H fill:#EF4444,color:#fff
    style E fill:#F59E0B,color:#fff

Loading

Prints — a API rodando

A demo mostra as tentativas acontecendo de verdade (dado real da instância local — aqui, o limite de requisições da demo pública sendo respeitado):

Documentação em ReDoc (mais "aqui está o schema", menos "botão de testar" — combina mais com a vibe de API orientada a dados que Swagger UI):

Stack

Camada	Tecnologia
Linguagem	Python 3.12
Framework	FastAPI
Validação	Pydantic 2 (incluindo validators de regra de negócio)
LLM	Claude API (anthropic), modelo configurável via ANTHROPIC_MODEL
PDF	pypdf (só PDF com camada de texto — ver limitações)
Rate limit	slowapi
Jobs	Fila em memória + BackgroundTasks (sem banco — ver "o que eu faria diferente")
Testes	pytest + FakeLLM (zero chamada real no CI)
CI	GitHub Actions

Como rodar localmente

git clone <repo>
cd doc-extract
cp .env.example .env      # preencha ANTHROPIC_API_KEY
docker compose up --build

• Demo: http://localhost:8000/demo
• Docs (ReDoc): http://localhost:8000/docs
• Health check: http://localhost:8000/health

Rodar os testes (não precisa de chave de API real — usa FakeLLM):

pip install -e ".[dev]"
ruff check app tests
pytest -v

Decisões técnicas

Por que Pydantic valida regra de negócio, não só schema

total_bate na nota fiscal soma valor_total_cents de cada item e compara com o total declarado. Isso não é validação de formato — é validação de coerência aritmética. Um schema JSON comum (ou até output_config.format da própria Claude API) garante que os tipos estão certos, mas não que a matemática bate. É exatamente esse tipo de erro — item inventado, valor trocado — que passaria despercebido numa checagem ingênua de tipo.

O retry é loop de controle, não "tenta de novo"

except ValidationError as e:
    previous_error = format_validation_errors(e)  # "- campo `cpf`: ... (recebi: '123.456.789-00')"

O prompt da tentativa seguinte contém essa mensagem exata. test_prompt_da_segunda_tentativa_contem_erro_da_primeira (em test_pipeline.py) faz um assert direto no conteúdo do prompt pra provar isso — não é "deveria funcionar", é testado.

A demo expõe o pipeline rodando de verdade, não uma animação fake

POST /extract retorna 202 + job_id na hora; o trabalho real roda em background. Cada tentativa (sucesso ou falha) é gravada no job incrementalmente, e o frontend faz polling em GET /jobs/{id} — então "tentativa 1 falhou, tentativa 2 corrigiu" na tela é o estado real do backend, não uma sequência fingida no JavaScript.

Dois bugs reais que só apareceram rodando de verdade

1. Job ficava preso em PROCESSING pra sempre. run_extraction_job só capturava NeedsReviewError — qualquer erro real da API (chave inválida, rate limit, falha de rede) subia sem tratamento, o BackgroundTask engolia a exceção (só loga), e o job nunca saía de PROCESSING. Descobri isso rodando o container com uma chave da Anthropic inválida de propósito — o cliente ficaria esperando pra sempre. Corrigido capturando anthropic.APIError e Exception genérica em run_extraction_job, marcando o job como NEEDS_REVIEW com o erro real em vez de travar.
2. node:20-alpine-style trap, mas em Python: nada aqui, na real — mas o Dockerfile deste projeto usa python:3.12-slim (Debian) desde o início, evitando de propósito o mesmo tipo de armadilha de biblioteca nativa que pegou o reserva-api com Alpine + OpenSSL.

Custo e latência expostos, sempre

Toda resposta de job DONE inclui meta.cost_usd (calculado a partir de usage.input_tokens/usage.output_tokens reais da resposta, não estimado) e meta.latency_ms. Rodar 3 tentativas custa 3x a chamada — isso precisa estar visível, não escondido.

Rate limit na demo pública

10 requisições/hora por IP (slowapi), testado batendo 12 requisições seguidas contra o container rodando: as 10 primeiras retornam 202, a 11ª e a 12ª retornam 429. Existe porque o modelo padrão (Opus, ver abaixo) não é barato — sem isso, a demo pública vira uma fatura surpresa.

Qual modelo Claude usar — e por que não decidi isso "pra economizar"

ANTHROPIC_MODEL no .env.example aponta para o modelo mais capaz disponível (claude-opus-4-8) por padrão. Cogitei usar um modelo mais barato "porque é só uma demo com rate limit", mas rebaixar modelo por conta própria — sem o usuário pedir — é exatamente o tipo de decisão de custo que não é minha pra tomar sozinho. Fica configurável por env var: se o custo do Opus for um problema no seu deploy público, trocar pra claude-haiku-4-5 é uma linha de configuração, não uma mudança de código.

Testes — o que é coberto e por quê

Arquivo	Cobre	Por quê
test_pipeline.py	Os 8 casos do pipeline (JSON válido de primeira, dentro de fence markdown, retry após JSON quebrado, retry após validação de negócio, prompt contendo o erro anterior, 3 falhas seguidas → NeedsReviewError, soma de nota errada, custo acumulando)	É a prova de que o retry é loop de controle real, não decoração — roda 100% com FakeLLM, zero chamada real
test_schemas.py	Os validators de GuestForm e Invoice, incluindo total_bate e so_digitos	Garante que a regra de negócio (não só o tipo) está sendo validada
test_api.py	Os endpoints HTTP (JSON e multipart, erros 400/404/415, fluxo completo POST /extract → poll GET /jobs/{id})	Cobre a camada que test_pipeline.py não toca — parsing de request, wiring de dependências, contrato HTTP

Resultado real: 31/31 testes passando, ruff check limpo, rodado dentro do próprio container Docker (não só localmente) — inclusive o cenário de erro real de API (chave inválida), que revelou o bug do job preso descrito acima.

Privacidade e LGPD

• Zero dado real de hóspede ou cliente em qualquer lugar do repo — os exemplos da demo (public/demo.html) e as fixtures de teste são 100% sintéticos, escritos pra este projeto
• A demo pública processa o texto que você cola — não armazena nada além do resultado do job em memória (perdido no restart do processo, ver limitações)
• Nenhuma autenticação, nenhum dado de usuário coletado — é uma demo de engenharia, não um produto com contas
• O texto enviado é truncado a MAX_INPUT_CHARS (15k) antes de ir pro prompt — evita mandar documentos inteiros desnecessariamente pra API externa

Limitações conhecidas

• Só PDF com camada de texto. PDF escaneado (imagem pura) não tem OCR — retorna 422 com uma mensagem explicando isso, em vez de falhar silenciosamente.
• Jobs em memória. Reiniciar o processo apaga o histórico de jobs. Não é um banco de dados — é uma fila simples, adequada pro escopo de demo (ver IDEIAS.md).
• O retry na demo pública depende do modelo real errar. Os exemplos carregados são genuinamente bagunçados (datas de check-in/check-out invertidas, soma de nota fiscal que não bate), mas o modelo pode acertar de primeira dependendo do dia — isso é honesto, não é roteirizado. A prova determinística do mecanismo de retry está nos testes (test_pipeline.py), que usam FakeLLM com respostas programadas.

O que eu faria diferente

• Persistência real de jobs, mesmo que só SQLite — hoje um restart do processo apaga tudo, o que é aceitável pra demo mas não pra produção de verdade.
• Webhook ou WebSocket pra notificar o fim do job, em vez de só polling — o polling funciona bem pra uma demo, mas não escala pra muitos clientes simultâneos.
• Circuit breaker no cliente Anthropic — se a API estiver fora do ar, hoje cada tentativa dentro do pipeline ainda tenta chamar a API 3 vezes antes de desistir; um circuit breaker cortaria isso mais rápido.
• Testar o caminho de PDF com um arquivo real — validei a extração de texto de PDF só lendo o código do pypdf, não rodei contra um PDF de verdade nesta sessão. Ver PROGRESSO.md.

Licença

MIT.