Construa um pipeline de RAG

Passos

1. Prepare um corpus e faça chunking com overlap

   Objetivo

   Reúna um corpus pequeno e quebre cada documento em chunks com overlap.

   Por que fazer chunking? Modelos de embedding têm janela de entrada
   limitada e — mais importante — um vetor para um documento inteiro de 20
   páginas é uma média borrada que não casa bem com nada. Chunks menores geram
   vetores mais nítidos e recuperáveis.

   Por que overlap? Se você corta numa fronteira rígida, uma frase que
   explica um conceito pode ficar separada do próprio conceito, e nenhum dos
   chunks responde à pergunta. O overlap repete o final de um chunk no começo
   do próximo para que o contexto sobreviva ao corte.

   Corpus

   Junte 10–30 documentos curtos (Markdown ou texto puro serve): itens de FAQ,
   docs internas, posts de blog. Marque pelo menos uns dois como restritos
   (premium: true) — você vai precisar deles no Passo 5.

   Chunking

   Use ~3000 caracteres (~800 tokens) por chunk com ~400 caracteres de
   overlap. Quebre em fronteiras de parágrafo/frase quando possível, em vez de
   no meio de uma palavra.

   Documento (9000 chars)
   |--------- chunk 0: chars 0..3000 ---------|
                           |--- overlap ---|
                     |--------- chunk 1: chars 2600..5600 ---------|
                                                   |--------- chunk 2 ---------|
   

   def chunk(text, size=3000, overlap=400):
       step = size - overlap          # 2600
       chunks = []
       start = 0
       while start < len(text):
           piece = text[start:start + size]
           chunks.append(piece)
           start += step
       return chunks
   

   Cada chunk carrega a identidade do documento pai para você poder citá-lo
   depois. Modele um registro de chunk assim:

   {
     "chunk_id": "doc-42:2",
     "doc_id": "doc-42",
     "title": "Política de reset de senha",
     "premium": false,
     "text": "…os ~3000 chars do chunk…"
   }
   

   Pronto quando: rodar seu chunker sobre o corpus produz uma lista plana de
   registros de chunk, cada um com ≤ ~3000 chars, com overlap visível entre
   chunks consecutivos do mesmo documento.

2. Gere embeddings (com cache)

   Objetivo

   Transforme o texto de cada chunk em um vetor e cacheie por hash do conteúdo
   para nunca pagar para embeddar o mesmo texto duas vezes.

   Embeddar

   Chame seu modelo de embeddings no text de cada chunk. Com
   text-embedding-3-small você recebe um vetor de 1536 dimensões por chunk.
   Anote a dimensão — a coleção do vector store precisa ser criada com exatamente
   esse número.

   vector = embed(chunk["text"])   # -> [0.013, -0.220, ..., 0.008]  (len 1536)
   

   Cache por hash do conteúdo

   Embeddar custa dinheiro e tempo. Se um documento não mudou, seus chunks não
   mudaram, então os vetores deles não mudaram. Use como chave de cache um hash
   do texto exato:

   import hashlib
   
   def embed_cached(text, cache):
       key = hashlib.sha256(text.encode()).hexdigest()
       if key in cache:
           return cache[key]           # hit — sem chamada de API
       vec = embed(text)               # miss — chama o modelo
       cache[key] = vec
       return vec
   

   O cache pode ser um dict em memória para o lab, ou uma tabelinha/arquivo para
   uso real. O ponto é que a chave é o hash do texto — mude um caractere e você
   gera uma nova chave, então vetores obsoletos não sobrevivem.

   Batch

   A maioria das APIs de embedding aceita um lote de entradas numa única
   requisição. Envie os chunks em lotes (ex.: 64 por vez) para reduzir latência
   e custo.

   Pronto quando: todo registro de chunk tem um vector da dimensão esperada,
   re-rodar o passo é quase de graça (tudo cache hit), e editar um documento
   re-embeda apenas os chunks dele.

3. Suba um vector store e faça upsert com payload

   Objetivo

   Rode um vector store, crie uma coleção com distância de cosseno e faça upsert
   dos seus vetores junto com um payload de metadados.

   Rode

   # Qdrant
   docker run -p 6333:6333 qdrant/qdrant
   # ou pgvector
   docker run -p 5432:5432 -e POSTGRES_PASSWORD=pw pgvector/pgvector:pg16
   

   Crie a coleção

   A coleção precisa bater com a dimensão do seu embedding e usar distância de
   cosseno — a métrica de similaridade padrão para embeddings de texto.

   {
     "collection": "kb",
     "size": 1536,
     "distance": "Cosine"
   }
   

   Upsert dos pontos com payload

   Um "ponto" é um vetor mais um payload — os metadados nos quais você vai
   filtrar e a partir dos quais vai citar. Armazene exatamente o que a recuperação
   e a geração precisam: a identidade da fonte, o título de exibição, a flag de
   acesso e o próprio texto.

   {
     "id": "doc-42:2",
     "vector": [0.013, -0.220, "…", 0.008],
     "payload": {
       "type": "doc",
       "doc_id": "doc-42",
       "title": "Política de reset de senha",
       "premium": false,
       "text": "…o texto do chunk, para você citar e exibir…"
     }
   }
   

   Indexe os campos de payload nos quais você filtra (premium, type) se seu
   store suportar índices de payload — a busca filtrada fica muito mais rápida
   com eles.

   Reindexação idempotente

   Quando um documento muda e você reindexa, remova os pontos antigos primeiro
   e só então insira os novos. Senão os chunks obsoletos se acumulam e continuam
   sendo recuperados para sempre.

   def reindex(doc):
       store.delete(collection="kb", filter={"doc_id": doc.id})   # remove os antigos
       store.upsert(collection="kb", points=points_for(doc))      # insere os novos
   

   Pronto quando: a coleção reporta a contagem certa de vetores, e reindexar
   um documento duas vezes deixa o mesmo número de pontos (sem duplicatas).

4. Implemente a busca semântica (embed, top-k, threshold)

   Objetivo

   Dada uma pergunta em linguagem natural, recupere os chunks mais relevantes.

   O fluxo

   1. Embedde a pergunta com o mesmo modelo que você usou nos chunks.
      Isso é inegociável — vetores de dois modelos diferentes não compartilham o
      mesmo espaço e as distâncias entre eles não têm significado.
   2. Busque top-k por similaridade de cosseno — peça ao store os k pontos
      mais próximos (comece com k = 5).
   3. Aplique um threshold de score para descartar matches fracos, para que
      uma pergunta fora do tema retorne nada em vez do chunk menos irrelevante.

   def search(question, k=5, min_score=0.35):
       qvec = embed(question)                       # mesmo modelo da indexação!
       hits = store.search(
           collection="kb",
           vector=qvec,
           limit=k,
       )
       return [h for h in hits if h.score >= min_score]
   

   Exemplo de query e o que volta:

   Pergunta: "Em quanto tempo o link de reset expira?"
   
   hits:
     0.71  doc-42:2  "Política de reset de senha"
     0.63  doc-42:1  "Política de reset de senha"
     0.41  doc-17:0  "Visão geral de segurança da conta"
     0.22  doc-03:4  "FAQ de faturamento"        <- abaixo do threshold, descartado
   

   Notas de tuning

   • k baixo demais → você perde contexto que a resposta precisava. k alto
     demais → você dilui o prompt com ruído e paga por mais tokens. k = 4–8
     é um começo sensato.
   • Scores de cosseno vão de 0..1 para embeddings normalizados; um bom
     threshold depende do corpus, então meça algumas perguntas reais e escolha
     um valor que mantenha os bons hits e descarte o lixo.

   Pronto quando: uma pergunta dentro do tema retorna uma lista curta de
   chunks claramente relevantes (maior score primeiro), e uma pergunta sem sentido
   retorna uma lista vazia.

5. Aplique o filtro de acesso na query

   Objetivo

   Garanta que um usuário sem acesso nunca consiga recuperar conteúdo restrito —
   filtrando dentro da query do vector store, não depois.

   O filtro

   Empurre um filtro de payload para dentro da própria busca. Para um usuário
   gratuito, restrinja a premium = false:

   def search(question, user, k=5, min_score=0.35):
       qvec = embed(question)
       payload_filter = None if user.premium else {"premium": False}
       hits = store.search(
           collection="kb",
           vector=qvec,
           limit=k,
           filter=payload_filter,         # <- aplicado pelo store, antes do ranking
       )
       return [h for h in hits if h.score >= min_score]
   

   No Qdrant isso é uma condição must no payload; no pgvector é um
   WHERE premium = false ao lado do ORDER BY embedding <=> $q — de qualquer
   forma, o store nunca retorna os pontos restritos.

   Nota de segurança — por que este é o ponto central

   Nunca recupere tudo e filtre no código da aplicação depois. Dois jeitos
   de isso vazar:

   1. Chunks restritos ocupam suas vagas de top-k, então mesmo depois de
      descartá-los o usuário recebe uma resposta pior — uma negação sutil do
      conteúdo que ele deveria ver.
   2. Um único ramo esquecido — um caminho de erro, um log de debug, uma camada
      de cache que guarda os hits crus — e o texto restrito já está no seu
      processo, a um erro de distância da resposta.

   O filtro pertence à query para que vetores restritos nem sejam pontuados.
   O banco é sua fronteira de enforcement. Trate a flag de acesso como entrada
   não confiável vinda da sessão do usuário, não do corpo da requisição.

   Verifique que o vazamento está fechado

   Faça uma pergunta cuja melhor resposta esteja só num documento premium: true:

   • Como usuário premium → o chunk restrito é recuperado e citado.
   • Como usuário gratuito → esse chunk nunca aparece; a resposta recorre a
     conteúdo público ou diz que não sabe.

   Pronto quando: a mesma pergunta retorna resultados diferentes e corretamente
   escopados para um usuário premium vs. um gratuito, e o texto restrito está
   comprovadamente ausente dos hits recuperados do usuário gratuito.

6. Monte o prompt de contexto e gere a resposta com citações

   Objetivo

   Transforme os chunks recuperados numa resposta fundamentada e com citações
   vinda do LLM — e entregue.

   Monte o prompt de contexto

   Concatene os chunks recuperados, cada um marcado com o título da sua fonte
   para o modelo poder citá-lo. Dê ao modelo uma instrução firme para responder
   apenas a partir do contexto e admitir quando o contexto não cobre a pergunta.

   def answer(question, hits):
       context = "\n\n".join(
           f"[Fonte: {h.payload['title']}]\n{h.payload['text']}"
           for h in hits
       )
       prompt = f"""Responda à pergunta usando APENAS o contexto abaixo.
   Se o contexto não contiver a resposta, diga que não sabe.
   Cite as fontes que você usou pelos títulos.
   
   Contexto:
   {context}
   
   Pergunta: {question}
   """
       return llm(prompt)
   

   Retorne as fontes junto com a resposta

   Não dependa só do modelo para citar — você já sabe quais documentos
   alimentaram o prompt, então retorne-os também como dado estruturado:

   {
     "answer": "O link de reset expira após 30 minutos. [Política de reset de senha]",
     "sources": [
       { "doc_id": "doc-42", "title": "Política de reset de senha" }
     ]
   }
   

   Por que "responda apenas a partir do contexto"

   Essa instrução mais a evidência recuperada é o que transforma uma máquina de
   alucinação num assistente fundamentado. Se a recuperação não retornou nada
   (vazio após o threshold do Passo 4), não chame o LLM com contexto vazio e
   deixe ele improvisar — retorne "Não tenho informação sobre isso."

   Critério de submissão

   Seu pipeline responde a uma pergunta real sobre o seu corpus, fundamentado nos
   chunks recuperados e citando os títulos das fontes — e respeita o filtro de
   acesso do Passo 5 (um usuário gratuito não consegue uma resposta construída a
   partir de conteúdo restrito).

   Entregue o repositório, incluindo:

   • O corpus (ou um script que o baixe) com pelo menos um documento premium: true.
   • O pipeline completo: chunk → embed (com cache) → upsert → busca → filtro → geração.
   • Um README curto com o comando para indexar o corpus e o comando para fazer uma pergunta.
   • Um transcript (ou teste) mostrando: (a) uma boa pergunta respondida com citações,
     (b) a mesma pergunta que só tem resposta em conteúdo premium respondida para um
     usuário premium mas recusada/redirecionada para um usuário gratuito.