Lede. Você tem uma base de documentos e quer que um chatbot responda usando esses documentos, sem inventar. Isso se chama RAG. Neste artigo você monta a versão mais simples possível, vê ela falhar de três formas, e depois conhece as sete peças que transformam isso em um sistema que dá para medir. Todo o código roda offline, sem chave de API. Onde você está na linha. Este é o artigo principal da série: o mapa. A seção 3 mostra o conceito que cada um dos 12 artigos seguintes acrescenta e por que a ordem é essa. O resto do artigo é a anatomia que todos eles usam. Se você quer só a decisão prática, vá para as seções 6 e 7.
- O problema: o modelo não conhece o seu documento
Peça para um chatbot a política de reembolso da sua empresa. Ele responde. A resposta é fluida, bem escrita e falsa: ele não tem a sua política. Ele tem um texto parecido com a sua política, escrito por outra pessoa, para outro caso.
Esse é o comportamento esperado, e não é um bug do modelo. O modelo é treinado com texto público da internet. Ele sabe falar sobre reembolso em geral. Ele não sabe do seu reembolso, porque o seu documento nunca entrou na conversa.
Existem duas saídas possíveis. A primeira é treinar o modelo com a sua base, o que é caro, lento e deixa de ser verdade no dia seguinte em que o documento muda. A segunda é a que todo mundo chama de RAG (Retrieval-Augmented Generation, "geração aumentada por recuperação"). A ideia inteira cabe numa frase:
Antes de responder, o sistema vai buscar na sua base o trecho que responde à > pergunta, cola esse trecho na conversa, e só então o modelo escreve.
Isso muda tudo. O modelo deixa de ser a fonte e passa a ser o redator: ele organiza e explica um texto que você escolheu. Se o trecho não estiver na conversa, não há o que ele possa usar como verdade.
O que vamos construir é isso:
[ pergunta do usuário ] | v 1. EMBER o que o usuário quer encontrar | v 2. BUSCA qual trecho do acervo responde | v 3. MONTAGEM junta os trechos num texto | v 4. RESPOSTA o modelo escreve usando só esse texto
Quatro etapas, no papel. Na prática são sete, e é por isso que os sistemas quebram.
2. A tentativa de dez linhas (e as três formas de ela errar)
Vamos começar errado de propósito. É mais barato entender a correção depois de ver
a falha do que ler a solução e sair procurando o problema.
# A versão ingênua: joga o documento inteiro na conversa e pergunta. def responder(pergunta: str, documento: str) -> str: """Manda o documento inteiro + a pergunta para o modelo. `documento` aqui é o texto completo de um PDF, contrato, manual ou base inteira. """ prompt = ( "Responda a pergunta usando APENAS o documento abaixo.\n" "Se a resposta não estiver no documento, diga que não encontrou.\n\n" f"--- DOCUMENTO ---\n{documento}\n--- FIM ---\n\n" f"Pergunta: {pergunta}\n" f"Resposta:" ) return chamar_modelo(prompt) # <- a chamada de API do seu provedor def chamar_modelo(prompt: str) -> str: """Placeholder: aqui entraria a chamada real do seu provedor de LLM. Nenhuma parte deste artigo depende de rede. Quando for usar de verdade, este é o único ponto que você troca. """ return f"[o modelo recebeu {len(prompt)} caracteres de prompt]"
Só isso já funciona. E só isso já está errado três vezes.
Erro 1 — o documento inteiro não cabe. Modelos têm limite de contexto (a janela
de entrada, hoje medida em milhares de palavras). Um manual de 200 páginas não cabe.
Você vai ter que escolher o que mandar. E "escolher" é um problema de engenharia, não
uma configuração.
Erro 2 — mandar tudo é mandar o que não tem a ver com a pergunta. Se você manda
o manual inteiro para responder "qual o prazo de garantia", o modelo recebe 200
páginas e uma pergunta. Ele vai encontrar o prazo, mas também vai encontrar vinte
trechos contraditórios ao lado e vai misturar. A resposta fica mais curta e mais rasa
do que se tivesse recebido só o parágrafo certo.
Erro 3 — sem filtro, você entrega documento que a pessoa não pode ler. O exemplo
acima manda o documento inteiro para qualquer usuário. Se a sua base tem um manual
por empresa, uma norma por departamento ou um contrato por cliente, isso é vazamento
de informação. E vazamento não aparece como erro: aparece como resposta, que é muito
mais difícil de perceber.
⚠️ "Joga tudo no prompt" é a origem de quase toda decepção com RAG
Não porque a técnica seja má — é o único jeito de começar. Mas o passo seguinteobrigatório é decidir o que entra. Todo o resto deste artigo é esse passo,
aplicado.
3. A linha de aprendizado: o que cada artigo seguinte ensina
Este artigo é o mapa. Ele introduz a anatomia do sistema — as sete peças e os
quatro contratos — e depois mostra, em uma linha cada, o conceito que cada artigo
seguinte acrescenta.
A linha tem duas trilas que saem daqui e depois se encontram:
Trilha A — achar o trecho certo (a parte vetorial)
02 · Chunking: como cortar um documento sem perder a resposta — O conceito
que entra: o chunk, a unidade que o sistema indexa e devolve. O artigo responde
como eu corto o documento, e como eu sei se o corte prestou?. Cobre corte por
tamanho fixo com sobreposição, corte estrutural, small-to-big, chunk semântico, e o
caso que nenhum chunker salva sozinho: a tabela.
03 · Embeddings e bancos vetoriais — Os conceitos que
entram: o embedding (o vetor que representa significado) e o índice (a
estrutura que guarda e compara). O artigo responde qual modelo eu uso, com que métrica, com que índice?. Mostra que a métrica e o índice pesam mais que o nome do
modelo, e por que um filtro sem índice é o mesmo que não ter filtro.
04 · Busca híbrida — Os conceitos que entram: o BM25 (busca por
palavra, com peso) e a fusão por rank (RRF). O artigo responde por que a busca por vetor não acha E-4021, e como unir as duas buscas sem somar escores que não estão na mesma escala?.
Trilha B — quem se relaciona com quem (o grafo)
05 · Extração de entidades com ontologia fechada — Os conceitos
que entram: a ontologia fechada (a lista de tipos que a extração pode preencher,
e que filtra o que o modelo devolve) e o grounding (exigir que a entidade exista
literalmente no texto-fonte, com posição). O artigo responde como eu extraio com LLM sem alucinar?, e por que saída estruturada resolve formato e não veracidade.
06 · Entity linking e normalização — O conceito que entra: o
linking, que casa a forma como o texto escreve ("sertralina 50mg") com o nó
canônico do grafo. O artigo responde como eu impeço que o mesmo real vire três nós?,
com quatro estágios de tolerância crescente e a normalização que faz o casamento
acontecer em português.
07 · Do texto ao grafo: comunidades — Os conceitos que
entram: a comunidade (grupo de entidades fortemente conectadas) e a
modularidade (medida de quão bem separadas elas estão). O artigo responde o que eu ganho com o grafo? e por que rodar o detector com dez seeds e ficar com a melhor
modularidade não é robustez, e sim uma admissão de que o resultado é instável.
08 · GraphRAG: busca local e global — Os conceitos que entram: o
k-hop (caminhar N arestas a partir de uma entidade) e o community report (o
resumo de uma comunidade, gerado antes da consulta). O artigo responde quando a pergunta é sobre um documento (local) e quando é sobre o acervo inteiro (global)?,
e como o roteador decide entre os dois caminhos.
Atravessam as duas trilhas
09 · Orçamento de contexto — O conceito que entra: o orçamento,
o limite de caracteres do contexto, junto com a ordem de degradação (o que sai
primeiro quando estoura). O artigo responde recuperei 10 trechos e não cabem — o que entra, e o que sai primeiro?.
10 · Agentes com LangGraph — Os conceitos que entram: o state
(o estado tipado que o grafo carrega), o reducer (a regra que decide como o
estado de dois nós se combina) e o checkpointer (o que permite pausar e retomar
uma execução). O artigo responde quando multiagente justifica o custo extra, e
inclui os três casos em que é erro.
11 · Observar e avaliar com Langfuse — Os conceitos que entram: o
trace (o registro de uma requisição inteira, com etapas aninhadas), o
groundedness (se a afirmação se apoia no trecho que ela cita) e o golden set
(o conjunto de perguntas com resposta esperada, que é o único jeito de provar que uma
mudança melhorou). O artigo responde como eu sei que meu RAG piorou?
O fechamento
12 · Qual arquitetura para qual base — O conceito que entra: a função
recomendar(), que transforma respostas sobre a sua base em uma arquitetura
recomendada com justificativa. O artigo responde as doze perguntas que escolhem entre vetorial, híbrido, grafo e agente, e em que ordem implantar cada coisa.
13 · Do protótipo à produção — Os conceitos que entram:
idempotência (reindexar não pode duplicar nada), backoff com jitter e
degradação (o que o sistema responde quando o LLM está fora). O artigo responde
o que quebra quando sai do notebook?
Leitura mínima útil: 01 → 02 → 03. Com isso você tem um RAG que funciona e que dá para medir. A trilha do grafo (05 a 08) só é necessária se a sua base tiver pergunta que depende do caminho entre entidades, e não só do texto. Por fim, 11 é o que impede você de otimizar no escuro — leia antes de mudar qualquer coisa.
4. As sete peças, uma a uma
Para resolver os três erros, o documento inteiro deixa de ser a unidade. A unidade
passa a ser o trecho — e cada trecho carrega informação de onde veio. São sete
peças; você já conhece a última (a resposta do modelo). Vamos pelas outras seis.
3.1 Entrada — pegar o arquivo e marcar de onde veio
Toda peça do sistema precisa lembrar a origem: o endereço do arquivo, o título, a
data. Sem isso você não consegue citar a fonte, e resposta sem fonte é opinião.
O que costuma dar errado aqui: o seu carregador de arquivo silenciosamente pula
arquivo. Filtro de data, permissão negada, PDF sem camada de texto, arquivo
corrompido. O sintoma é "o sistema não acha um documento que existe" — e a causa
não está em nenhum lugar que você vai olhar primeiro.
3.2 Limpeza — tirar o que não é conteúdo
Cabeçalho, rodapé, número de página, logotipo escrito como texto, "confidencial", e o
mesmo aviso repetido em 300 páginas. Isso vira embedding (a peça 4) e ocupa o lugar
de conteúdo útil na busca. O efeito é sutil: todo trecho do acervo passa a parecer um
pouco com todo outro, e a busca perde a capacidade de diferenciar.
É o tipo de coisa que não dá erro. Só piora a qualidade e faz parecer que o modelo
está ruim.
3.3 Corte — o trecho é a unidade de busca
Aqui está a decisão que mais muda o resultado de todo o sistema, e é o assunto do
Um chunk (trecho) é a unidade que o sistema indexa e a unidade que ele devolve.
A pergunta "qual o prazo de garantia" é respondida por um parágrafo, não por um
capítulo. Então o corte precisa preservar parágrafos inteiros, respeitar a estrutura
do documento (título, seção, tabela) e manter cada trecho legível sozinho.
A tentação é cortar em "500 caracteres". É o que a biblioteca faz por padrão e é quase
sempre errado: 500 caracteres no meio de uma tabela é lixo, e 500 caracteres no meio de
uma frase é pior.
3.4 Vetor — o número que representa o significado
Embedding é um vetor (uma lista de centenas ou milhares de números) que representa o
significado de um texto. Trechos com significado parecido ficam com vetores
parecidos; trechos diferentes ficam com vetores distantes.
A analogia mais honesta não é "coordenada no espaço". É etiqueta de biblioteca indexada por assunto: você não guarda o livro, guarda uma ficha que diz do que ele
fala, e a ficha permite achar livros parecidos sem ler nenhum deles.
O detalhe que importa: o vetor não guarda o texto. Ele guarda um resumo comprimido do
significado. É por isso que ele serve para achar e não serve para responder — a
resposta tem que vir sempre do trecho original, nunca do vetor.
3.5 Índice — guardar e comparar rápido
O índice é onde os vetores ficam guardados de um jeito que responde "quais são os
10 mais parecidos com este vetor?" rápido. Sem índice você compara com todos, um por
um.
Aqui entra a busca vetorial, e junto dela o filtro por metadado: a mesma busca
pode devolver só os documentos de um cliente, de um período, ou de um departamento. O
filtro tem que ser aplicado dentro da busca, e não depois dela — volto nisso na
seção 6, porque é o erro que expõe informação.
Detalhes de índice, métrica de similaridade e os quatro motores (Qdrant, Chroma,
pgvector, sqlite-vec) estão no artigo 03.
3.6 Grafo — o que se relaciona com o quê (opcional)
Um grafo de conhecimento guarda entidades (pessoas, produtos, normas, partes) e as
relações entre elas. Ele existe para responder perguntas que dependem do caminho,
não do texto: "quem assinou o contrato que substituiu este?".
É a única peça opcional da lista, e o
artigo 08 mostra o que ela resolve e o que ela só
faz você pagar mais caro. A travessia não é uma consulta vetorial: é uma
consulta recursiva (CTE) sobre tabelas de nós
e arestas.
3.7 Seleção — o que cabe na conversa
Você recuperou 10 trechos. Não cabem todos na janela do modelo. Alguém tem que
escolher, e esse alguém é você, no código.
O termo técnico é orçamento (budget): um limite de caracteres que você define e
que o código respeita. É também onde você decide o que fazer quando estoura (cai o
trecho de menor utilidade? cai a aresta do grafo? — nunca a citação), e é o assunto
inteiro do artigo 09.
5. As quatro fichas que passam de peça para peça
Cada peça precisa falar com a próxima. Se a peça 3 entrega uma lista de strings e a
peça 5 precisa saber o tipo, a origem e a permissão, alguém vai escrever essa conversão
no meio do caminho — e ela vai estar errada três meses depois.
A solução é definir quatro fichas com os campos que realmente circulam, uma para
cada entidade que atravessa o sistema. Em Python, a forma mais direta é um modelo do
Pydantic: ele valida os tipos na
entrada e rejeita campo que não existe, o que transforma erro de digitação em erro
na hora, em vez de um None perdido três peças depois.
from __future__ import annotations import hashlib from typing import Any from pydantic import BaseModel, ConfigDict, Field class Contract(BaseModel): """Base de todas as fichas. extra="forbid" = campo que não existe é erro, não é ignorado. frozen=True = ninguém muda a ficha depois que ela foi criada. """ model_config = ConfigDict(extra="forbid", frozen=True) class Document(Contract): """A peça 1: o arquivo cru, antes de qualquer transformação. Guardar `uri` e `metadata` aqui é o que permite citar a fonte e filtrar por permissão mais adiante. """ doc_id: str uri: str # de onde veio title: str text: str metadata: dict[str, Any] = Field(default_factory=dict) @property def content_hash(self) -> str: """Impressão digital do texto. Serve para uma coisa só: saber se o documento mudou desde a última indexação. Sem isso, reindexar significa reindexar tudo. """ return hashlib.blake2b( self.text.encode("utf-8"), digest_size=16 ).hexdigest() class Chunk(Contract): """A peça 3: o trecho indexado. `parent_id` existe porque quase todo documento tem hierarquia (capítulo > seção > parágrafo). Guardar o pai desde o começo permite buscar no trecho pequeno e entregar o trecho grande, sem reindexar. """ chunk_id: str doc_id: str parent_id: str | None # None = é a raiz ordinal: int # posição dentro do pai text: str token_count: int # contagem de token, não de letra metadata: dict[str, Any] = Field(default_factory=dict)
Repare no campo token_count. Ele não é len(text). Um token é o pedaço que o
modelo realmente lê, e ele não equivale a um caractere: em português, acento e
pontuação quebram em mais de um token. Se você usar len(text) no orçamento, o número
sai errado e o sistema estoura a janela de um jeito que parece aleatório.
Agora a ficha do resultado da busca, que é a mais importante das quatro:
from enum import StrEnum class ScoreSource(StrEnum): """De onde veio o número da busca. Guardar isso é obrigatório.""" VECTOR = "vector" # similaridade de significado LEXICAL = "bm25" # busca por palavra exata GRAPH = "graph" # achou por relação, não por texto COMMUNITY = "community" # resumo de um grupo do grafo # `StrEnum` (3.11+) é um enum que já serializa como texto: `VECTOR` vai para o # banco como "vector", e `VETOR` -- typo -- vira erro de validação. class Scored(Contract): """A peça 6: um trecho escolhido para entrar na conversa. `source` existe para impedir um erro caro: **somar escores que não estão na mesma escala**. A busca por vetor devolve um número de 0 a 1. A busca por palavra devolve um número que passa fácil de 10. A travessia de grafo devolve uma contagem de saltos. Somar esses três numa média não é "fusão", é ruído com aparência de método. """ ref_id: str # qual chunk/doc/nó foi escolhido score: float # o número source: ScoreSource # e de que tipo de busca ele veio payload: dict[str, Any] = Field(default_factory=dict)
A forma correta de juntar resultados de buscas diferentes é por posição no ranking, não pelo valor do score — é o que a
fusão RRF do Qdrant faz, e
o assunto do artigo 04.
E a quarta ficha, que é a que protege o orçamento:
class BudgetExhausted(RuntimeError): """O bloco não cabe. A decisão de cortar é de quem chamou, não daqui.""" class Budget: """Conta quantos caracteres o contexto pode gastar, por tipo de bloco. Por que caracteres e não tokens: dá para medir sem carregar o tokenizador para dentro do laço. É aproximação — e por isso existe o `floor`, uma reserva que nunca é gasta, para absorver a diferença entre caractere e token. """ def __init__(self, total_chars: int, floor_ratio: float = 0.15) -> None: self._total = total_chars self._floor = int(total_chars * floor_ratio) self._spent = 0 self._by_kind: dict[str, int] = {} @property def remaining(self) -> int: return self._total - self._spent def spend(self, kind: str, text: str) -> int: """Consome o texto. Levanta erro se não couber.""" cost = len(text) if cost > self.remaining - self._floor: raise BudgetExhausted( f"{kind} custa {cost}, resta {self.remaining} (piso {self._floor})" ) self._spent += cost self._by_kind[kind] = self._by_kind.get(kind, 0) + cost return cost def report(self) -> dict[str, Any]: """Para onde foi o orçamento. É o número que você olha todo dia.""" return { "total": self._total, "spent": self._spent, "remaining": self.remaining, "by_kind": dict(sorted(self._by_kind.items(), key=lambda kv: -kv[1])), }
💡 O
floornão é otimização, é margem de segurança
Sem a reserva, o último bloco entra e a mensagem estoura exatamente no limite.Aí o provedor corta por conta própria, às vezes no meio de uma citação — e
você nunca vê o erro, porque a resposta sai. A evidência de que o meio do
contexto é a pior posição possível está em
Lost in the Middle: o conteúdo que sobrevive
a um corte cego tende a ser justamente o que não responde à pergunta.
6. Onde cada peça quebra
Agora que você sabe o que cada peça é, a tabela de falhas faz sentido. Ela responde
"por que a resposta está ruim?" sem precisar abrir log.
| Peça | Como ela quebra | O que o usuário vê | Como descobrir sem depurar |
|---|---|---|---|
| 1. Entrada | arquivo não lido (data, permissão, PDF sem texto) | "não encontrei" para algo que existe | quantos documentos entraram nesta execução, comparado com a origem |
| 2. Limpeza | cabeçalho e rodapé entram no índice | todo trecho começa igual; a busca "enche" e perde o sentido | quais as 20 palavras mais frequentes do acervo |
| 3. Corte | tabela ou frase cortada no meio | resposta genérica; citação aponta para texto incompleto | coverage@k: o trecho que respondia apareceu no top-10? |
| 4. Vetor | documento indexado com modelo antigo | a busca piora sozinha depois de um deploy | qual versão do modelo está gravada no registro do trecho |
| 5. Índice | filtro não aplicado | o usuário vê material de outra área | contar os resultados com tenant diferente do esperado |
| 6. Seleção | contexto estourado e cortado pelo provedor | resposta segura e inútil; recusa frequente | budget.report() registrado por requisição |
| 7. Resposta | instrução de citar ignorada | resposta plausível sem fonte | quanto da resposta aponta para um trecho real |
Duas linhas merecem atenção, porque o sintoma aponta para o lugar errado.
A peça 3 (corte) aparece como problema de prompt. O trecho corta a tabela no meio,
os números não chegam ao modelo, e a conclusão natural é "o modelo não entende".
Trocar de modelo não conserta, ajustar o prompt tampouco: a informação nunca existiu
no contexto.
A peça 5 (índice) aparece como problema de permissão, e é a mais grave das sete. Não
é resposta ruim, é resposta que não deveria existir. E o conserto não é "filtrar depois":
from typing import Callable def buscar_errado(query: str, search_fn: Callable, tenant: str) -> list[Scored]: """ERRADO: recupera 50, filtra em Python, devolve 10. A lista final até parece certa. O problema é o que aconteceu antes: os 50 documentos de outros tenants já foram lidos pela busca, ficaram em memória e vão para o log de tracing. E qualquer caso que o filtro local esqueça vaza direto na resposta. """ hits = search_fn(query, k=50) # <- escopo ausente hits = [h for h in hits if h.payload.get("tenant") == tenant] return hits[:10] def buscar_certo(query: str, search_fn: Callable, tenant: str) -> list[Scored]: """CERTO: o filtro vai para dentro da consulta. Os 50 candidatos de outros tenants nunca chegam a existir. """ return search_fn(query, k=10, scope={"tenant": tenant})
Rode as duas com a mesma base e compare. A lista devolvida é a mesma — a
diferença é que na versão errada o processo leu 48 documentos que não
devia. E essa leitura vai para o log de observabilidade, que é o registro do que o
sistema fez em cada requisição (o que entrou, o que saiu, quanto tempo levou). É por
isso que a versão errada vaza: o dado sensível não está na resposta, está no log.
O filtro é avaliado durante a busca, no
Na prática isso é uma cláusula WHERE na mesma consulta que ordena o resultado — a
semântica de SELECT do PostgreSQL
é onde ele é avaliado. E o campo usado no filtro precisa de índice: sem índice, o
banco filtra depois de abrir os candidatos, e o top-10 volta incompleto.
O detalhe mais barato de acertar: faça o filtro ser obrigatório na assinatura da
função de busca. Não existe chamada sem escopo, e o editor de código reclama na hora
de digitar a chamada.
O jeito de fazer isso é com um Protocol (do typing): você descreve o formato
esperado — quais métodos, quais tipos — sem escrever implementação. É assim que você
troca um banco vetorial por outro sem mexer no resto do sistema: os dois cumprem o
mesmo formato.
from typing import Protocol, Sequence, runtime_checkable @runtime_checkable class Chunker(Protocol): """O que transforma documento em trecho. A peça 3.""" def split(self, doc: Document) -> list[Chunk]: ... @runtime_checkable class Embedder(Protocol): """O que transforma texto em número. A peça 4.""" dim: int def embed_documents(self, texts: Sequence[str]) -> list[list[float]]: ... def embed_query(self, text: str) -> list[float]: ... @runtime_checkable class Retriever(Protocol): """A peça 5. `scope` é obrigatório e entra na mesma chamada. Um search() sem escopo é um search() que alguém vai chamar sem escopo um dia às 23h, com a métrica de qualidade já verde no painel. """ def search( self, query: str, *, k: int, scope: dict[str, object] ) -> list[Scored]: ...
7. A ordem para montar (e o que você tem ao fim de cada passo)
A ordem importa porque cada passo é verificável sozinho. Montar na ordem inversa é
trabalhar sem saber se a peça está certa.
Passo 1 — a busca, sem modelo nenhum. Indexe os documentos, mande uma pergunta que
você sabe que tem resposta, e confira se volta o trecho certo. Sem LLM na jogada. Se
isso não funciona, nada mais importa — e é a única etapa que dá para testar em dois
segundos.
Passo 2 — a medição, antes da geração. Escreva de 20 a 50 perguntas, e em cada uma
marque qual trecho deveria ser recuperado (não a resposta: o trecho). Meça:
def coverage_at_k(hits: Sequence[Scored], trecho_certo: str) -> float: """1.0 se o trecho que respondia à pergunta apareceu entre os k.""" return 1.0 if any(h.ref_id == trecho_certo for h in hits) else 0.0 def reciprocal_rank(hits: Sequence[Scored], trecho_certo: str) -> float: """1 / posição do trecho certo. Ignora o que veio abaixo.""" for posicao, hit in enumerate(hits, start=1): if hit.ref_id == trecho_certo: return 1.0 / posicao return 0.0
Se coverage_at_k está em 0.9, a recuperação está funcionando. Se a resposta continua
ruim, o problema não é a peça 3 nem a 5 — é a 6 ou a 7. Você acabou de eliminar
metade do sistema sem ler um prompt.
Registrar essas duas por requisição é o
mínimo de instrumentação para que a
otimização tenha contra o que se medir — assunto do
Passo 3 — a montagem, e só então o modelo. Pegue os k trechos, monte o orçamento,
cole na conversa. Chame o modelo. Agora cada etapa tem um número.
Passo 4 — as ramificações. Quatro coisas que só valem a pena aqui: busca
híbrida (une a busca por significado com a busca por palavra exata), grafo de
conhecimento, reranker (um segundo modelo, mais caro e mais lento, que reordena os
candidatos recuperados antes de entrar na conversa) e agente. Cada uma entra por cima de
um sistema que já diz onde está o defeito.
⚠️ Começar pelo agente é a forma mais cara de não saber o que está errado
Um agente com busca ruim produz uma resposta ruim que pode estar em três lugaresdiferentes. Aí você otimiza o prompt, que é a única parte rápida de editar — e o
prompt nunca foi o problema. O padrão de "pensar, observar, agir" do
ReAct pressupõe que a observação seja boa: o
ciclo só raciocina sobre o que a busca devolveu. Estado explícito, checkpoint e
retomada é o que o
LangGraph formaliza — e
é assunto do artigo 10, depois desta série
inteira.
8. O sistema inteiro rodando
Juntando as peças: contratos, um vetorizador determinístico (não é modelo semântico,
serve para exercitar o encadeamento sem rede), um índice em memória com busca exata e
filtro, e a montagem com orçamento.
from __future__ import annotations import math from collections.abc import Iterable, Sequence class FakeEmbedder: """Substitui o modelo de verdade para você rodar sem chave de API. NÃO é semântico: é a contagem de palavras projetada num vetor. Serve para testar o encadeamento das peças, não a qualidade da busca. Para qualidade, troque por um modelo de embedding de verdade (artigo 03). """ def __init__(self, dim: int = 64) -> None: self.dim = dim def _vector(self, text: str) -> list[float]: vector = [0.0] * self.dim for word in text.lower().split(): vector[hash(word) % self.dim] += 1.0 norm = math.sqrt(sum(v * v for v in vector)) or 1.0 return [v / norm for v in vector] # normalizado: norma = 1 def embed_documents(self, texts: Sequence[str]) -> list[list[float]]: return [self._vector(t) for t in texts] def embed_query(self, text: str) -> list[float]: return self._vector(text) def cosine(a: Sequence[float], b: Sequence[float]) -> float: """Vetores normalizados => o produto interno já É o cosseno.""" return sum(x * y for x, y in zip(a, b, strict=True)) class InMemoryIndex: """A peça 5, na versão mais simples possível: lista e comparação. Busca exata, sem índice aproximado. Com poucos milhares de itens, isso costuma ser melhor que um índice aproximado em acerto e em velocidade. """ def __init__(self) -> None: self._vectors: dict[str, list[float]] = {} self._chunks: dict[str, Chunk] = {} def upsert(self, chunk: Chunk, vector: Sequence[float]) -> None: self._chunks[chunk.chunk_id] = chunk self._vectors[chunk.chunk_id] = list(vector) def search( self, query: str, *, k: int, scope: dict[str, object] ) -> list[Scored]: # Instancia o embedder com a mesma dimensão do índice, senão o # produto interno estoura na comparação. dim = len(next(iter(self._vectors.values()), [])) query_vector = FakeEmbedder(dim).embed_query(query) hits: list[Scored] = [] for ref_id, vector in self._vectors.items(): chunk = self._chunks[ref_id] meta = {**chunk.metadata, "doc_id": chunk.doc_id} # FILTRO ANTES de ordenar: documento fora do escopo nunca é candidato. if any(meta.get(field) != value for field, value in scope.items()): continue hits.append( Scored(ref_id=ref_id, score=cosine(query_vector, vector), source=ScoreSource.VECTOR) ) hits.sort(key=lambda r: r.score, reverse=True) return hits[:k] def ingest( documents: Iterable[Document], chunker: Chunker, embedder: Embedder, index: InMemoryIndex, ) -> dict[str, int]: """As peças 3 -> 4 -> 5, em lote. Devolve sempre as contagens: é o número mais barato que existe para descobrir que a peça 1 (entrada) parou de funcionar. """ documents = list(documents) # materializa: se vier gerador, o for consome chunks: list[Chunk] = [] for doc in documents: chunks.extend(chunker.split(doc)) vectors = embedder.embed_documents([c.text for c in chunks]) for chunk, vector in zip(chunks, vectors, strict=True): index.upsert(chunk, vector) return {"documents": len(documents), "chunks": len(chunks)}
Para testar, copie o bloco e rode:
class SentenceChunker: """Um trecho por frase. Só para exercitar o encadeamento das peças.""" def split(self, doc: Document) -> list[Chunk]: sentences = [s.strip() for s in doc.text.split(".") if s.strip()] return [ Chunk(chunk_id=f"{doc.doc_id}#{i}", doc_id=doc.doc_id, parent_id=None, ordinal=i, text=s, token_count=len(s.split()), metadata={"tenant": doc.metadata["tenant"]}) for i, s in enumerate(sentences) ] docs = [ Document(doc_id="d1", uri="urn:doc:1", title="Política de garantia", text="O prazo de garantia é de 12 meses. A cobertura começa na entrega.", metadata={"tenant": "empresa-a"}), Document(doc_id="d2", uri="urn:doc:2", title="Política de garantia B", text="O prazo de garantia é de 6 meses. A cobertura começa na assinatura.", metadata={"tenant": "empresa-b"}), ] index = InMemoryIndex() print(ingest((d for d in docs), SentenceChunker(), FakeEmbedder(), index)) # {'documents': 2, 'chunks': 4} # Busca com escopo: empresa-b NÃO pode aparecer para empresa-a. hits = index.search("prazo de garantia", k=3, scope={"tenant": "empresa-a"}) print([h.ref_id for h in hits]) # ['d1#0', 'd1#1'] print(coverage_at_k(hits, "d1#0")) # 1.0 # O orçamento para de estourar, e avisa com o motivo. budget = Budget(total_chars=100) budget.spend("trecho", "x" * 80) try: budget.spend("trecho", "y" * 40) except BudgetExhausted as erro: print(erro) # trecho custa 40, resta 20 (piso 15) print(budget.report()["spent"]) # 80
Se a lista saiu só com d1#0 e d1#1, o filtro funcionou e o sistema está íntegro.
Se d2 aparecer, o filtro não está sendo aplicado dentro da busca.
TL;DR
-
RAG é uma ideia só: antes de responder, achar o trecho certo na sua base e
mostrar ao modelo. O modelo vira redator, não fonte.
-
"Manda o documento inteiro" é o começo e o erro. São três os problemas: não
cabe, mistura o que não tem a ver com a pergunta, e entrega documento que a pessoa
não pode ler.
-
A unidade de tudo é o trecho (chunk), e ele precisa fazer sentido sozinho —
cortar em "500 caracteres" quase sempre quebra tabela e frase.
-
O vetor serve para achar, não para responder. A resposta vem sempre do texto.
-
Filtro por permissão vai dentro da busca, nunca depois: filtrar depois já
puxou o conteúdo errado para dentro da memória e do log.
-
Monte na ordem: busca → medição → montagem → modelo → ramificações. Cada
passo é verificável sozinho, e no fim de cada um você tem um número.
-
coverage_at_k é a métrica que mais economiza tempo: ela separa "não
recuperou" de "recuperou e não usou".
Referências
- Payload filtering — Qdrant — o filtro avaliado durante a busca, base da seção 6
- Hybrid Queries — Qdrant — a fusão por ranking que explica por que a ficha guarda a origem do score
- SELECT — PostgreSQL — onde o filtro é de fato avaliado
- WITH Clause (CTE) — SQLite — base da travessia multi-hop do grafo, no artigo 08
- Lost in the Middle: How Language Models Use Long Contexts — a evidência de que o meio do contexto é a pior posição, que motiva o piso do orçamento
- ReAct: Synergizing Reasoning and Acting in Language Models — o padrão pensar/observar/agir que o artigo 10 formaliza
- Graph API — LangGraph — estado explícito e retomada, assunto do artigo 10
- Langfuse Python SDK — como instrumentar as sete peças por requisição, assunto do artigo 11
- Pydantic — models — por que a ficha da seção 4 valida na entrada em vez de confiar no chamador