Como preparar dados para aplicações com LLMs

Dados sujos, limites de token e falhas inesperadas: o que acontece quando um LLM recebe entrada mal preparada

Um engenheiro de RPA tenta gerar respostas automáticas a partir de contratos, tickets e logs, mas o pipeline explode ao chegar ao modelo: tokens excedem o limite, caracteres estranhos geram UnicodeDecodeError, e a latência dobra porque o mesmo trecho é processado múltiplas vezes. O problema não está no modelo, e sim na camada de preparação que entrega texto inconsistente, fragmentado de forma inadequada e sem controle de cache.

Arquitetura mínima de um pipeline de preparação para LLMs

Divida o fluxo em quatro estágios claros: Ingestão → Normalização → Segmentação → Serialização/Cache. Cada estágio deve ser idempotente, logar falhas e expor métricas de contagem de tokens.

Ingestão resiliente

Use pathlib e chardet para detectar encoding antes de abrir arquivos. Falhas silenciosas de leitura são a causa mais comum de dados truncados.

from pathlib import Path
import chardet
from typing import Iterable, Tuple

def read_file_safely(file_path: Path) -> str:
    """Lê texto garantindo encoding correto e captura exceções."""
    raw = file_path.read_bytes()
    result = chardet.detect(raw)
    encoding = result["encoding"] or "utf-8"
    try:
        return raw.decode(encoding)
    except UnicodeDecodeError as exc:
        raise ValueError(f"Falha ao decodificar {file_path}: {exc}") from exc

Normalização e limpeza de ruído

Remova quebras de linha excessivas, espaços duplos e caracteres de controle. Normalizar para NFC evita que o mesmo caractere seja representado de duas formas diferentes, o que inflaria o número de tokens.

import re
import unicodedata
from typing import List

def normalize_text(text: str) -> str:
    """Aplica NFC, remove controle e colapsa espaços."""
    text = unicodedata.normalize("NFC", text)
    # Remove caracteres não imprimíveis
    text = re.sub(r"[\x00-\x1F\x7F]", "", text)
    # Colapsa quebras de linha e espaços múltiplos
    text = re.sub(r"\s+", " ", text).strip()
    return text

Segmentação (chunking) consciente de limites de token

Modelos como GPT‑4 aceitam no máximo 8 192 tokens. Dividir por número de caracteres costuma gerar estouro porque tokens variam de 1 a 4 caracteres. Use a biblioteca tiktoken para contar tokens antes de criar os blocos.

import tiktoken
from typing import List

ENCODER = tiktoken.get_encoding("cl100k_base")
MAX_TOKENS = 2000  # margem para prompt + resposta

def chunk_by_tokens(text: str, max_tokens: int = MAX_TOKENS, overlap: int = 200) -> List[str]:
    """Divide texto em blocos que não ultrapassam max_tokens, com sobreposição opcional."""
    tokens = ENCODER.encode(text)
    chunks: List[str] = []
    start = 0
    while start < len(tokens):
        end = min(start + max_tokens, len(tokens))
        chunk_tokens = tokens[start:end]
        chunks.append(ENCODER.decode(chunk_tokens))
        # Avança com sobreposição para preservar contexto
        start = end - overlap if end < len(tokens) else end
    return chunks

Serialização, cache de embeddings e controle de versão

Gerar embeddings repetidamente custa tempo e dinheiro. Armazene o hash do chunk (SHA‑256) como chave e o vetor resultante como valor. Quando o texto mudar, o hash muda e o cache invalida automaticamente.

import hashlib
import json
from pathlib import Path
from typing import Dict, Any

CACHE_DIR = Path("/tmp/embedding_cache")
CACHE_DIR.mkdir(parents=True, exist_ok=True)

def chunk_hash(chunk: str) -> str:
    return hashlib.sha256(chunk.encode("utf-8")).hexdigest()

def load_embedding(chunk: str) -> Any | None:
    h = chunk_hash(chunk)
    cache_file = CACHE_DIR / f"{h}.json"
    if cache_file.is_file():
        return json.loads(cache_file.read_text())
    return None

def save_embedding(chunk: str, embedding: Any) -> None:
    h = chunk_hash(chunk)
    cache_file = CACHE_DIR / f"{h}.json"
    cache_file.write_text(json.dumps(embedding))

Trade‑offs críticos

• Sobreposição vs. custo de token: Overlap de 200 tokens garante que entidades não sejam cortadas, mas aumenta o consumo total em ~10 %.
• Cache local vs. serviço distribuído: Em ambientes de cluster, usar Redis ou DynamoDB evita duplicação, porém introduz latência de rede e necessidade de TTL para lidar com atualizações de documentos.
• Normalização agressiva vs. preservação de formato: Remover quebras de linha pode destruir tabelas ou listas numeradas; avalie a necessidade de manter marcadores de estrutura (ex.: \n- ).

Erros recorrentes em produção

• Contar caracteres ao invés de tokens – resulta em exceções InvalidRequestError: This model's maximum context length is ... tokens.
• Ignorar exceções de leitura – arquivos parcialmente corrompidos entram como strings vazias, gerando respostas vazias do modelo.
• Cache sem invalidação – documentos atualizados continuam usando embeddings antigos, produzindo respostas desatualizadas.
• Processamento síncrono de chunks grandes – bloqueia workers e causa time‑outs em APIs de orquestração.

Recomendações práticas

• Instrumente cada estágio com contadores de tokens e tamanho de chunk; alerte quando o consumo ultrapassar 90 % do limite.
• Implemente fallback: se tiktoken falhar, recorra a uma contagem aproximada baseada em regex de palavras.
• Use asyncio ou thread pool para paralelizar a geração de embeddings, mas limite a concorrência para não saturar a API do provedor.
• Versione o pipeline (ex.: pipeline_v2) e armazene a versão junto ao hash no cache; assim mudanças de tokenizador não invalidam tudo.

Armadilhas que ainda pegam equipes experientes

• Assumir que todos os documentos são texto plano; PDFs escaneados precisam de OCR pré‑processamento, caso contrário o modelo recebe apenas imagens vazias.
• Esquecer de normalizar quebras de linha em arquivos Windows (\r\n) – gera tokens de controle que aumentam o custo sem valor semântico.
• Persistir embeddings em formatos binários incompatíveis entre versões de Python ou de bibliotecas de NumPy – causa erros de deserialização silenciosa.
• Não registrar a origem do chunk (arquivo, página, linha); depurar respostas erradas torna-se impossível sem rastreabilidade.