Structured Output Extractor

Uma ferramenta de linha de comando para extrair dados estruturados de texto não estruturado usando a restrição output_schema do Google ADK. Dado um texto livre como um e-mail, uma biografia ou uma descrição de fatura, a ferramenta retorna um objeto JSON validado que está em conformidade com um schema Pydantic predefinido.

Por que isso importa em produção: Texto não estruturado é o formato padrão da maioria dos dados do mundo real -- e-mails, documentos, logs, tickets de suporte. Convertê-lo em registros estruturados normalmente é feito com pipelines de regex ou anotação manual. LLMs podem realizar essa extração, mas sem restrições de schema o formato de saída é pouco confiável. O parâmetro output_schema do ADK força o modelo a emitir JSON em conformidade com um modelo Pydantic, tornando a extração baseada em LLM determinística o suficiente para consumo programático.

Arquitetura

graph TD
    Input([Free Text<br/>txt / json / csv])
    CLI[Typer CLI<br/>cli.py]
    IO[I/O Layer<br/>io.py]
    Extractor[Extractor<br/>extractor.py]
    Pipeline[Batch Pipeline<br/>pipeline.py]
    Config[Settings<br/>config.py]
    Registry[Schema Registry<br/>schemas/]
    Agent[ADK Agent<br/>with output_schema]
    Runner[ADK Runner]
    Session[InMemorySessionService]
    LLM[Gemini Model]
    Output([Structured JSON / CSV])

    Input --> CLI
    CLI --> IO
    CLI --> Config
    CLI --> Extractor
    CLI --> Pipeline

    Extractor --> Agent
    Extractor --> Runner
    Extractor --> Session
    Extractor --> Registry

    Pipeline -->|asyncio.gather| Extractor

    Agent -->|output_schema constraint| LLM
    Runner --> Agent
    Runner --> Session

    IO --> Output
    Extractor --> IO

Loading

Extração individual: O texto entra pela CLI, é encaminhado para a função extract(), que cria um ADK Agent configurado com o modelo Pydantic do schema selecionado como output_schema. O Runner executa o agente, e a resposta JSON do modelo é parseada e retornada.

Extração em lote: Múltiplos textos são processados concorrentemente via asyncio.gather(), com cada texto recebendo sua própria instância de Agent/Session/Runner. Os resultados são coletados e escritos em JSON ou CSV.

Decisões de Arquitetura

Por que output_schema do ADK ao invés de Engenharia de Prompt Manual

Solicitar manualmente que um LLM "retorne JSON com esses campos" é frágil: os modelos frequentemente adicionam campos extras, usam tipos inconsistentes ou envolvem a saída em blocos de código markdown. O parâmetro output_schema do ADK compila o modelo Pydantic em um JSON Schema e o passa como restrição de geração para a API. O modelo é forçado a obedecer no nível de decodificação, não apenas no nível de instrução. Isso elimina a necessidade de parsing de saída, limpeza com regex ou lógica de retry para JSON malformado.

Por que Pydantic para Schemas

Modelos Pydantic servem a três propósitos: definem o alvo da extração (nomes e tipos de campos), fornecem validação ao construir resultados programaticamente e serializam diretamente para JSON Schema para o parâmetro output_schema. Adicionar um novo schema é um único arquivo com uma subclasse de BaseModel -- sem templates de prompt, sem dicionários de mapeamento de campos.

Processamento em Lote com asyncio.gather

Cada extração é independente (texto separado, sessão separada), então asyncio.gather() fornece concorrência direta. Isso é mais simples que filas de tarefas ou pools de threads e naturalmente respeita os limites de taxa da API através do pool de conexões do cliente HTTP subjacente. Para lotes muito grandes, uma abordagem com semáforo seria mais apropriada.

Padrão Schema Registry

O dicionário SCHEMA_REGISTRY mapeia nomes em string para classes Pydantic, fornecendo um único ponto de consulta para a CLI, o extrator e os testes. Adicionar um novo schema requer: (1) criar uma subclasse de BaseModel em src/schemas/, (2) adicioná-la ao registro em src/schemas/__init__.py. Nenhuma outra alteração de código é necessária.

Design dos Schemas

Schemas Disponíveis

Schema	Campos	Obrigatórios
person	name, age, email, occupation, location	name
company	name, industry, founded_year, headquarters, employees_count	name
invoice	invoice_number, date, total, vendor, items	(nenhum)
event	name, date, location, organizer, description	name

Adicionando um Novo Schema

1. Crie src/schemas/product.py:

from pydantic import BaseModel

class Product(BaseModel):
    name: str
    price: float | None = None
    category: str | None = None

2. Registre em src/schemas/__init__.py:

from src.schemas.product import Product

SCHEMA_REGISTRY: dict[str, type[BaseModel]] = {
    ...
    "product": Product,
}

Isso é tudo. A CLI, o extrator e o pipeline de lote detectarão automaticamente o novo schema.

Configuração

git clone https://github.com/gabrielandrade/structured-output-extractor.git
cd structured-output-extractor
uv sync --dev
cp .env.example .env
# Add your GOOGLE_API_KEY to .env

Uso

# List available schemas
uv run extractor list-schemas

# Extract from inline text
uv run extractor extract --text "John Smith is a 35-year-old engineer from Seattle" --schema person

# Extract from file
uv run extractor extract --file input.txt --schema person --output result.json

# Batch extraction (concurrent)
uv run extractor batch --file people.txt --schema person --output results.json

# Batch to CSV
uv run extractor batch --file people.txt --schema person --output results.csv --format csv

Formatos de Entrada Suportados

Formato	Estrutura
.txt	Uma entrada de texto por linha
.json	Array JSON de strings
.csv	Primeira coluna contém as entradas de texto

Docker

docker build -t structured-output-extractor .
docker run --env-file .env structured-output-extractor list-schemas

Desenvolvimento

uv sync --dev

# Run tests with coverage
uv run pytest --cov=src --cov-report=term-missing -v

# Lint and format
uv run ruff check --fix .
uv run ruff format .

# Type check
uv run mypy src/

Trade-offs e Limitações

• A precisão do LLM na extração varia por domínio. O modelo pode alucinar valores para campos que não consegue extrair com confiança do texto. Para uso em produção, considere adicionar scores de confiança ou uma etapa de verificação.
• Limites de complexidade de schema. Schemas profundamente aninhados (ex.: faturas com itens de linha contendo sub-itens) podem produzir saída menos confiável. Schemas planos ou com um único nível de aninhamento funcionam melhor com output_schema.
• Considerações de custo para processamento em lote. Cada texto em um lote é uma chamada de API separada. Para 1000 textos a ~500 tokens por extração, os custos são não triviais. Não há API de batching para amortizar o overhead por requisição.
• Sem streaming para extração. Diferente de casos de uso conversacionais, a extração produz um único objeto JSON. Streaming é desnecessário e não foi implementado.
• Arquitetura de modelo único. Todos os schemas usam o mesmo modelo. Na prática, schemas mais simples (person, event) podem funcionar bem com modelos menores, enquanto schemas complexos (invoice com itens aninhados) podem se beneficiar de modelos maiores.

O que Aprendi

• O output_schema do ADK é notavelmente confiável para schemas planos. Ao longo dos testes com schemas de person, company e event, o modelo retornou consistentemente JSON válido com nomes e tipos de campos corretos. A restrição de schema elimina toda uma classe de bugs de parsing de saída.
• asyncio.gather é suficiente para lotes de tamanho moderado. Para lotes de 10-50 textos, gather fornece speedup quase linear sem a complexidade de filas de tarefas. Além disso, um padrão com semáforo (asyncio.Semaphore) seria necessário para evitar sobrecarregar os limites de taxa da API.
• A API model_fields do Pydantic é útil para introspecção. O comando list-schemas usa schema_class.model_fields.keys() para exibir dinamicamente os campos disponíveis, o que mantém a CLI e as definições de schema sincronizadas sem duplicação manual.
• Cada extração precisa de sua própria sessão. Diferente de agentes conversacionais onde a continuidade da sessão é o objetivo, agentes de extração devem usar sessões isoladas para evitar contaminação cruzada entre textos não relacionados. O design atual cria um InMemorySessionService novo por chamada de extração, o que é correto mas cria pressão no GC em escala.

Licença

MIT