Pipeline responsável pela integração de dados de peças processuais e anexos do oracle para o elasticsearch, gerando metadados, OCR dos documentos e embeddings para uso da IA.

O primeiro passo da pipeline é monitorar quando um documento deve ser indexado, para isso, diáriamente a pipeline monitora as tabelas do Oracle (TCE_GO.PRO_SUMARIO, TCE_GO.PRO_AUTUACAO), caso exista um novo doucmento, ele é enviado para execução da Pipeline.

SELECT
  PS.PROSUM_ID AS ID,
  PS.PROAUTU_ID AS PROAUTU_ID,
  PS.ID_DOCUMENTO_N AS ID_DOCUMENTO,
  PA.CODG_PROCESSO_N AS PROCESSO,
  PS.TIPO_DOCUMENT_A AS TIPO_DOCUMENTO,
  PS.DESC_TIPODOCUMENTO_A AS DESC_TIPO_DOCUMENTO,
  TO_CHAR(PS.DATA_CRIACAO_D,'YYYY-MM-DD') AS DATA_CRIACAO,
  PA.NUMR_ANOREF_N AS ANO,
  PS.NUMR_ORDEM_N AS NUMERO,
  PA.DESC_HISTORICO_T AS HISTORICO,
  PA.DESC_EMENTA_A AS EMENTA
FROM PRO_SUMARIO PS
JOIN PRO_AUTUACAO PA ON PA.PROAUTU_ID = PS.PROAUTU_ID
LEFT JOIN IAG.IAG_CONTROLE_EXTRACAO CONTROL
    ON CONTROL.ID_PROCEDIMENTO = PS.PROSUM_ID
    AND CONTROL.DESC_BASE_DADOS = 'AUTUACAO'
    AND CONTROL.DESC_STATUS = 'OK'
WHERE CONTROL.ID_PROCEDIMENTO IS NULL
ORDER BY PS.DATA_CRIACAO_D DESC
OFFSET 0 ROWS FETCH NEXT 10000 ROWS ONLY;

Uma vez identificado quais documentos devem ser indexados, os mesmo devem ser enviados para processamento, o processamento consiste em 6 etapas:

1. Extração de Metadados
2. Extração do Documento/Texto e Geração de OCR
3. Geração de Embeddings com o Texto do Documento
4. Cálculo de Tokens do Documento
5. Geração de Resumo do Documento
6. Geração de Resumo do Processo

Processamento de Metadados

O processamento dos metadados é feito chamando a API do ETCE: https://api-etce.tce.go.gov.br/api/v1/Processo/consulta/extracao

{
  "autuacaoId": 334965,
  "dataAutuacao": "2020-02-14T17:30:24",
  "situacaoAndamento": null,
  "anoReferencia": 2020,
  "assunto": "019-01-PROJETO-RESOLUÇÃO",
  "nomeAuditor": "USUARIO 58863",
  "codigoProcesso": "202000047000376",
  "colegiado": null,
  "dataPublicacao": "0001-01-01T00:00:00",
  "dataSessao": "0001-01-01T00:00:00",
  "ementa": "Processo nº 202000047000376/01901, em que a Presidência do Tribunal de Contas do Estado de Goiás, encaminha Proposta de Resolução apresentada pelo corpo de Conselheiros Substitutos desta Corte de Contas que trata da nomenclatura do cargo por estes ocupado.",
  "historico": "Em que a Presidência do Tribunal de Contas do Estado de Goiás, encaminha Proposta de Resolução apresentada pelo corpo de Conselheiros Substitutos desta Corte de Contas que trata da nomenclatura do cargo por estes ocupado.\n",
  "tipoDocumento": null,
  "numero": 202000047000376,
  "numeroAno": "2020/202000047000376",
  "orgaoOrigem": "TRIBUNAL DE CONTAS DO ESTADO DE GOIÁS",
  "nomeProcurador": "USUARIO 52254",
  "nomeRelator": "USUARIO 213215",
  "interessados": [
    {
      "pessoaId": 46094,
      "cpfCnpj": "00091730000114",
      "nome": "PESSOA JURIDICA 46094"
    }
  ],
  "decisoes": [
    {
      "autuacaoId": 334965,
      "numeroDecisao": "7/2024",
      "tipoDocumento": "Resolução",
      "acordaoId": null,
      "resolucaoId": 23067,
      "dataDecisao": "2024-10-03T16:06:44",
      "dataPublicacao": "2024-10-08T00:00:00",
      "descricaoTipoJulgamento": "Ato Legal",
      "colegiado": "Tribunal Pleno",
      "numeroSessao": 19,
      "nomeRelatorDecisao": "USUARIO 213215",
      "dataSessao": "2024-09-30T11:00:00"
    }
  ],
  "ehProcessoSigiloso": false,
  "ehProcessoReservado": false
}

Processamento de Texto e OCR

O processamento e extração do documento PDF do texto do documento é realizado pela API do ETCE:

https://api-etce.tce.go.gov.br/api/v1/Processo/documento/extracao &
https://api-etce.tce.go.gov.br/api/v1/Processo/documento/extracao/download

{
  "documentoId": 614125,
  "autuacaoId": 371468,
  "codigoProcesso": 2024,
  "dataCriacao": "2025-01-15T17:23:27",
  "descricaoTipoDocumento": "Documento Digital",
  "descricaoTitulo": "1-Petição de Certidão Negativa - Copia - Copia.pdf",
  "texto": null,
  "tipoDocumento": "L1",
  "ehProcessoSigiloso": false
}

Com o documento PDF recuperado, o mesmo é submetido a OCR pela API do Docling: http://gpu-server01.tce.go.gov.br:11000/extract_text_from_file

OBS: Antes de indexar o texto no elasticsearch, a pipeline verifica se o texto ou a OCR possui no mínimo 90% do texto legível. Caso contrário o documento não é indexado. O mesmo é salvo como um erro na tabela do controle, onde deve ser análisado manualmente, isso é feito para garantir que documentos com ruídos entre na base.

Processamento de Embeddings

Com o texto extraido do OCR, os embeddings são gerados em chunks pelo mesmo modelo com finetunnig usado no legis arturmatos/bge-ft-legis-full

import torch
from transformers import AutoTokenizer, AutoModel

# Detecta GPU, se disponível
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")

# Carrega modelo e tokenizer
model_name = "arturmatos/bge-ft-legis-full"
tokenizer = AutoTokenizer.from_pretrained(model_name)
model = AutoModel.from_pretrained(model_name)
model.to(device)
model.eval()

def get_cuda_embedding(texto: str):
    start = time.time()
    # Tokenização + envio para GPU
    inputs = tokenizer(texto, return_tensors="pt", truncation=True, padding=True).to(device)

    with torch.no_grad():
        outputs = model(**inputs)

    # Assume-se que os embeddings são extraídos da saída da última camada (hidden_states)
    # Aqui usamos o embedding do CLS token (posição 0)
    embeddings = outputs.last_hidden_state[:, 0, :]

    # Opcional: normaliza o vetor
    embeddings = torch.nn.functional.normalize(embeddings, p=2, dim=1)

    # Move para CPU e converte para numpy array
    return embeddings.squeeze().tolist()

Cálculo de Tokens

Com o texto extraido do documento e OCR, é realizado o cálculo da quantidade de tokens presentes no texto usando o script a seguir e salvos no campo do elasticsearch metainfo.tokens:

class DocUtils:

@staticmethod
def count_tokens(text: str, model: str = "gpt-4") -> int:
    try:
        codificador = tiktoken.encoding_for_model(model)
    except KeyError:
        codificador = tiktoken.get_encoding("cl100k_base")
    tokens = codificador.encode(text)
    return len(tokens)

Ao final o resultado do processamento é carregado no elasticsearch, o link para visualizar os documentos carregados é: https://elk.tce.go.gov.br/app/discover#/view/89d03c4d-9087-4644-9300-216e1eaea51b

Essa pipeline é orquestrada e executada periodicamente pelo Airflow, o link de acesso é:

Pipeline de Dados: https://dataflow.tce.go.gov.br/dags/iago.autuacao/grid
Pipeline de Resumos: https://dataflow.tce.go.gov.br/dags/iago.autuacao.summary/grid

No oracle é usado a tabela IAG.IAG_CONTROLE_EXTRACAO para realizar o controle de quais documentos foram processados com sucesso e com erro, nos quais os documentos com erro serão reprocessados novamente em uma nova oportunidade.

Atualmente com a média de 1.6 Milhões de documentos, o tempo total para processar toda a base dura 7 dias.