Como implementar um sistema RAG com FastAPI, embeddings, busca vetorial e controle de acesso

RAG útil em produção combina retrieval relevante com autorização no índice. Sem ACL, você vaza informação. Sem citações e avaliação, você não sabe se a resposta está grounded.

Pipeline ponta a ponta

1. Ingerir documentos com metadados de ACL e versão
2. Chunking com overlap controlado
3. Gerar embeddings e indexar (vetorial + lexical se possível)
4. Recuperar com filtro de ACL → rerank → gerar com citações
5. Avaliar groundedness offline e amostrar em produção

Modelo de dados do chunk

from pydantic import BaseModel

class Chunk(BaseModel):
    id: str
    doc_id: str
    text: str
    acl: set[str]
    version: str
    source_uri: str

API de consulta

@app.post("/v1/rag/query")
async def rag_query(body: QueryIn, user=Depends(auth)):
    hits = await search(
        query=body.question,
        acl=user.roles,
        k=8,
    )
    answer = await generate_with_citations(body.question, hits)
    return {
        "answer": answer.text,
        "citations": [
            {"doc_id": h.doc_id, "uri": h.source_uri, "score": h.score}
            for h in hits
        ],
    }

Controle de acesso

• Filtrar no retrieval, nunca só no prompt
• Reindexar quando ACL do documento mudar
• Separar índices por tenant quando o isolamento exigir
• Não logar trechos sensíveis em texto aberto

Avaliação

Monte um conjunto de perguntas com respostas esperadas e fontes obrigatórias. Meça recall@k, precisão de citação e taxa de recusa quando não houver evidência.

Limitações

RAG não corrige corpus ruim. Chunking cego destrói tabelas e código. Para relações complexas, considere enriquecer com KAG; para FAQs estáveis, CAG pode reduzir custo.