Sistema de Geração Aumentada por Recuperação (RAG) corporativo com agente autônomo profundo, guardrails tridimensionais, métricas estatísticas de qualidade e interface analítica completa. Construído sobre o framework Deep Agents (LangChain/LangGraph), com armazenamento vetorial no Pinecone, reranking LLM-based, System Prompt Store com guardrails, modo KB-only, temperature configurável, descoberta e crawl de URLs, filtro de cosine mínimo, gerenciamento granular de fontes, prompts em PT-BR, Langfuse tracing com rastreio por usuário, métricas RAG, avaliação de documentos, interpretação de imagens, autenticação por API key (SHA256+salt), autenticação por sessão com gestão de usuários, skills por namespace e interface web completa.
Stack: FastAPI · Pinecone · OpenAI · LangChain · LangGraph · deepagents · Langfuse · SQLite · Tailwind CSS
O ciclo de vida de uma requisição percorre nove camadas sequenciais:
- Autenticação de sessão — middleware verifica cookie
superag_sessionou headerX-API-Key - Interceptação de rede — endpoint FastAPI com roteamento por operação
- Identificação de usuário — extração do
user_loginda sessão para rastreio Langfuse - Transição de estado — consolidação de parâmetros em
_ns_stateglobal - Mutação termodinâmica — override de temperatura conforme
kb_only - Guardrail de entrada — classificação neural da consulta
- Injeção de skills — carregamento de skills do namespace para evaluate/interpret
- Raciocínio do agente — orquestração de ferramentas com deepagents/LangGraph
- Guardrails de saída — dupla barreira (output + KB-only compliance)
- Métricas e serialização — cálculo estatístico e empacotamento JSON/Pydantic
git clone superag
cd superag
python -m venv .venv
source .venv/bin/activate # Linux/Mac
ou
.venv\Scripts\activate # Windows
pip install -r requirements.txtEdite o arquivo .env:
OPENAI_API_KEY=sk-...
OPENAI_MODEL=gpt-4o
PINECONE_API_KEY=pcsk_...
PINECONE_INDEX_NAME=superag
PINECONE_ENVIRONMENT=us-east-1
EMBEDDING_MODEL=text-embedding-3-small
EMBEDDING_DIMENSION=1536
CHUNK_SIZE=1000
CHUNK_OVERLAP=200
# Autenticação API (opcional — se vazio, acesso livre)
SUPERAG_API_KEY_HASH=
SUPERAG_API_KEY_SALT=
# Langfuse (opcional)
LANGFUSE_PUBLIC_KEY=pk-lf-...
LANGFUSE_SECRET_KEY=sk-lf-...
LANGFUSE_HOST=https://cloud.langfuse.comO sistema usa SHA256 + salt para armazenar a chave. O servidor nunca armazena a chave em texto plano.
# Gerar chave aleatória + hash + salt
python -m app.auth
# Ou fornecer uma chave específica
python -m app.auth minha-chave-secretalaunch.json que deve ser usado no VSCode para modo python.
{
"version": "0.2.0",
"configurations": [
{
"name": "Python: Debug Módulo App",
"type": "debugpy",
"request": "launch",
"module": "app.main",
"cwd": "${workspaceFolder}",
"env": {
"PYTHONPATH": "${workspaceFolder}"
}
}
]
}python -m app.mainAcesse http://localhost:8000
O agente autônomo opera no padrão ReAct (Raciocínio → Ação → Observação). Recebe um system prompt completo, ferramentas de busca vetorial e decide autonomamente quantas buscas executar, quais termos usar e quando parar.
O sistema tenta instanciar via deepagents SDK (create_deep_agent). Se o pacote estiver indisponível ou ocorrer incompatibilidade de tipos, o fallback automático constrói um agente equivalente via LangGraph (create_react_agent). Ambos recebem paridade total de restrições: temperatura, ferramentas e system prompt.
_build_agent(system_prompt, temperature)
├─ try: _build_deep_agent() → deepagents.create_deep_agent()
└─ except: _build_langgraph_agent() → langgraph.prebuilt.create_react_agent()
Ferramentas disponíveis ao agente:
| Ferramenta | Descrição |
|---|---|
search_knowledge_base |
Busca vetorial primária no Pinecone via _retrieve() → _filter_above_threshold() |
refine_search |
Retentativa com sinônimos quando resultados iniciais são insuficientes |
O agente suspende seu raciocínio, invoca a ferramenta, recebe o retorno textual formatado com citações [1], [2]... e integra no percurso dedutivo. Pode encadear múltiplas chamadas antes de formular a resposta final.
Pipeline direto sem raciocínio iterativo: recuperação → filtragem → prompt template → LLM. Usa ChatPromptTemplate do LangChain com contexto injetado no system message. Mais rápido, menor custo de tokens, adequado para consultas diretas.
O deepagents é o framework preferencial para instanciação do agente autônomo. Ele encapsula:
- Orquestração de raciocínio — ciclo ReAct com planejamento implícito de etapas
- Tool calling nativo — o modelo decide autonomamente quando invocar ferramentas, com quais argumentos, e como integrar os resultados
- Tolerância a falhas — o sistema implementa try/except na instanciação; se
create_deep_agentfalhar por qualquer razão (TypeError, ImportError, incompatibilidade de versão), o fallback para LangGraph é transparente - Integração com callbacks — aceita Langfuse callback handlers para rastreabilidade completa de cada etapa do raciocínio
A assinatura de criação suporta duas variantes (detectadas automaticamente):
create_deep_agent(model=model, tools=tools, system_prompt=system_prompt)
create_deep_agent(tools=tools, instructions=system_prompt, model=model)Quando use_rerank=false, o sistema compensa a ausência de reordenação semântica com super-recuperação. Em vez de buscar exatamente top_k documentos, calcula:
retrieve_k = max(top_k * 2, top_k + 5)Para top_k=5: max(10, 10) = 10 documentos recuperados do Pinecone via ANN (Approximate Nearest Neighbors).
Quando use_rerank=true, o multiplicador é 4x (top_k * 4), criando uma matriz profunda para filtragem fina via LLM-as-a-judge no módulo reranker.
| top_k | rerank | Equação | Documentos recuperados |
|---|---|---|---|
| 5 | false | max(5*2, 5+5) |
10 |
| 5 | true | 5*4 (overretrieve) |
20 |
| 10 | false | max(10*2, 10+5) |
20 |
Quando ativado, o módulo app.services.reranker itera fragmento por fragmento, submetendo cada um ao LLM para pontuação de relevância contextual (0-10). Os top_k melhores são mantidos na ordem de relevância rerankeada, os demais são preservados para exibição na UI.
Cada documento recuperado carrega similarity_score nos metadados. O sistema compara contra o limiar configurável similarity_threshold (default: 0.60, ajustável pela UI de 0% a 95%).
- Acima do limiar → entra no contexto do LLM + exibido normalmente na UI
- Abaixo do limiar → marcado com
below_threshold=True, excluído do contexto LLM, mas preservado na UI para transparência
A filtragem ocorre em _filter_above_threshold(). Quando include_below_threshold=true (re-query automático), todos os documentos entram no contexto.
Documentos que perpassaram a ingestão com encoding Latin-1 disfarçado de UTF-8 são reparados automaticamente por _fix_mojibake():
repaired = text.encode("latin-1").decode("utf-8")Caracteres como ã → ã, ç → ç, ê → ê são corrigidos antes de qualquer processamento.
O SupeRAG implementa uma blindagem tridimensional contra alucinações e violações de compliance:
Classificador neural (temperatura=0) que opera como fiscal de aduana semântica. Avalia a consulta do usuário contra regras arbitrárias definidas pelo administrador do namespace.
- Consulta lícita → replica a consulta original (pode reformular para clareza)
- Consulta ilícita → emite
BLOQUEADO:+ justificativa → exceçãoGuardrailBlocked→ aborta todo processamento posterior
Inspeção pós-geração que avalia a resposta produzida pelo agente contra regras de saída do namespace (ex: ocultar dados sensíveis, telefones pessoais).
- Conforme → retorna inalterada
- Violação corrigível → retorna versão sanitizada
- Violação incorrigível → emite
BLOQUEADO:→ bloqueia visualização
Exclusivo do modo kb_only=true. Terceiro vetor de bloqueio que opera como LLM-as-a-judge em passagem isolada (temperatura=0). Recebe o contexto vetorial sobrevivente do threshold + a resposta gerada e julga:
- Reestruturação/síntese de dados do KB → permitido, retorna inalterado
- Inserção de dados externos (nomes, números, fatos do treinamento) → remove cirurgicamente as partes não fundamentadas
- Resposta 100% alucinatória → substitui integralmente por: "Esta informação não está disponível na base de conhecimento."
Quando kb_only=true, três mecanismos convergem:
- Temperatura forçada a 0.0 — independente do valor submetido pelo usuário, override hardcode transforma a inferência de amostragem estocástica para argmax decoding (busca gulosa determinística)
- KB_ONLY_DIRECTIVE injetada no system prompt — 7 regras peremptórias de contenção (proibição de conhecimento externo, supressão de treinamento prévio, falha intencional padronizada, interdição de fabulação)
- KB-Only Guardrail pós-geração — auditoria de conformidade por LLM-as-a-judge independente
Calculadas após cada consulta bem-sucedida (excluídas quando blocked=True). Operam exclusivamente sobre fontes acima do threshold coseno. Dividem-se em determinísticas (CPU) e probabilísticas (LLM-as-a-judge).
Calculadas por intersecção léxica direta no interpretador Python, sem chamadas à API.
Fração dos documentos recuperados que foram efetivamente utilizados na resposta. Um documento é considerado "utilizado" quando ≥15% de seu conteúdo textual aparece na saída do LLM (limiar: _RELEVANCE_THRESHOLD = 0.15).
Recall@K = (documentos com overlap ≥ 15%) / (total de documentos recuperados acima do threshold)
Interpretação: Recall@K = 0.8 significa que 80% dos documentos recuperados contribuíram materialmente para a resposta. Valores baixos indicam que o retrieval trouxe muitos documentos irrelevantes — considerar ativar rerank ou reduzir top_k.
Mede a posição do primeiro documento útil na lista ordenada pelo banco vetorial. Usa o mesmo limiar de 15% para determinar utilidade.
MRR = 1 / posição_do_primeiro_documento_útil
Interpretação: MRR = 1.0 significa que o documento mais relevante estava na posição #1. MRR = 0.5 indica que o primeiro documento útil estava na posição #2. Valores baixos sugerem que o ranking vetorial nativo não está priorizando os fragmentos corretos — rerank pode melhorar significativamente este indicador.
Profundidade expandida de citação. Exige que ≥25% do conteúdo de um documento apareça na resposta para considerá-lo "citado" (limiar elevado: 0.25 vs 0.15 do Recall).
Coverage = (documentos com overlap ≥ 25%) / (total de documentos recuperados acima do threshold)
Interpretação: Coverage mede o aproveitamento denso dos documentos, não apenas menção superficial. Coverage = 0.6 com Recall@K = 0.8 indica que embora 80% dos documentos foram mencionados, apenas 60% foram explorados em profundidade. Valores baixos em conjunto com Recall alto sugerem respostas que citam muitas fontes superficialmente sem aprofundar.
Calculadas por inferência neural em passagem única, temperatura=0, formato JSON restrito. Retornam floats no intervalo [0.0, 1.0].
Proporção de afirmações na resposta que são rastreáveis ao contexto recuperado. Penaliza severamente inserções não fundamentadas.
Faithfulness = (afirmações fundamentadas no contexto) / (total de afirmações na resposta)
Interpretação: Faithfulness = 1.0 significa que toda informação na resposta tem origem comprovável nos documentos. Valores < 0.7 indicam alucinação parcial — o agente está inserindo conhecimento de treinamento. No modo kb_only, o KB-Only Guardrail atua como segunda barreira complementar a esta métrica.
Grau de alinhamento entre a resposta e a pergunta original. Avalia se a resposta endereça diretamente a intenção interrogativa ou se é evasiva/tangencial.
Answer Relevance = grau de cobertura da intenção da pergunta pela resposta [0.0-1.0]
Interpretação: Relevance = 0.9 indica resposta altamente focada. Valores < 0.5 indicam que a resposta divagou ou endereçou aspectos periféricos da pergunta. Pode ocorrer quando o retrieval retorna documentos vagamente relacionados e o agente tenta compor uma resposta coerente a partir de material tangencial.
Se o LLM retornar JSON corrompido ou valores fora do intervalo, o sistema intercepta via try/except e anula silenciosamente os valores para zero, mantendo a resposta intacta e sem prejuízo ao usuário.
| Controle | Parâmetro | Range | Default |
|---|---|---|---|
| Namespace | namespace |
lista dinâmica | default |
| Modo | use_agent |
Deep Agent / Simple RAG | Deep Agent |
| Top-K | top_k |
1-20 | 5 |
| Rerank | use_rerank |
checkbox | desativado |
| Temperature | temperature |
0.0-1.0 (slider) | 0.7 |
| Cosine Threshold | similarity_threshold |
0%-95% (slider, step 5%) | 60% |
Cada fonte exibe: nome do documento, preview do conteúdo, barra visual de similaridade coseno (verde ≥85%, amarelo ≥70%, vermelho <70%), score de rerank quando disponível, e tag "abaixo do limiar" quando aplicável.
Toggle "Exibir fontes abaixo do limiar": mostra/oculta documentos filtrados pelo cosine threshold. Badge dinâmico indica contagem (ex: "3 de 8").
Quando a resposta é "Esta informação não está disponível na base de conhecimento" e o usuário marca o checkbox de fontes abaixo do limiar, o sistema automaticamente:
- Re-executa a consulta com
include_below_threshold=true - Remove a mensagem "não disponível" anterior
- Exibe nova resposta com contexto expandido (incluindo documentos sub-threshold)
- Previne loop desativando
_lastAnswerEmpty
O label do checkbox muda dinamicamente para "Refazer pesquisa incluindo fontes abaixo do limiar (< X%)".
Botão ativo quando há arquivo anexado. Submete o documento para avaliação estruturada contra o KB + system prompt do namespace. Retorna obrigatoriamente 6 seções:
- Score de Conformidade (0-100) + justificativa
- Alinhado — afirmações consistentes com o KB (com citações)
- Divergente — contradições ou distorções (com citações)
- Não Coberto — afirmações sem dados correspondentes no KB
- Ausente — tópicos do KB que o documento não aborda
- Recomendação — veredito detalhado + ações sugeridas
A avaliação usa chamada LLM com system/user messages separadas para evitar que o system prompt do namespace domine as instruções de avaliação.
Aceita imagens coladas/anexadas. Usa GPT-4 Vision com contexto do KB para:
- Descrição detalhada do conteúdo visual
- Referência cruzada com a base de conhecimento
- Extração de texto, tabelas, gráficos e diagramas
- Identificação de alinhamentos e divergências com o KB
Botões de ação nas respostas de AVALIAR e INTERPRETAR:
- Copiar — copia o conteúdo para a área de transferência
- Exportar .eml — gera arquivo de e-mail Outlook com a avaliação/interpretação como corpo HTML + documento original como anexo base64
Seletor de profundidade de crawl:
- Somente esta URL (
depth=0) — indexação direta da página, sem descoberta de sub-páginas. Botão: "Indexar" - Incluir sub-páginas (
depth=2, default) — descoberta via/api/discover/url, seleção em árvore, ingestão em lote. Botão: "Explorar"
Cada namespace possui configuração independente armazenada em JSON:
| Campo | Descrição |
|---|---|
system_prompt |
Personalidade e regras de comportamento do agente |
kb_only |
Flag de confinamento factual (força temp=0 + KB_ONLY_DIRECTIVE + guardrail triplo) |
input_guardrail |
Regras de filtragem de entrada (texto livre) |
output_guardrail |
Regras de sanitização de saída (texto livre) |
Quando kb_only=true, três efeitos cascateiam: temperatura forçada a 0.0, KB_ONLY_DIRECTIVE concatenada ao system prompt (7 regras), e guardrail pós-geração por LLM-as-a-judge.
Sistema de habilidades especializadas que ampliam o comportamento do LLM nas operações de avaliação de documentos e interpretação de imagens, com extensibilidade para novos tipos de operação. Persistido em SQLite (skills.db).
Um skill é um bloco de instruções em Markdown que é injetado no prompt do LLM durante uma operação específica em um namespace específico. Permite que cada namespace tenha comportamento especializado para avaliar e interpretar, sem alterar o system prompt geral.
| Tipo | Operação | Ponto de injeção |
|---|---|---|
evaluate |
Avaliação de documentos | Concatenado ao eval_system prompt, entre o prompt base e o documento |
interpret |
Interpretação de imagens | Inserido entre os critérios de conformidade e as instruções obrigatórias |
Além de evaluate e interpret, é possível criar skills com tipos arbitrários (ex: summarize, translate, audit). Ficam armazenados e acessíveis via API para integrações programáticas futuras ou extensões do agente.
evaluate_document()
├─ eval_system (prompt base fixo)
├─ ── SKILL DE AVALIAÇÃO (namespace: financeiro) ──
│ [conteúdo do skill_store para namespace/evaluate]
│ ── FIM DO SKILL DE AVALIAÇÃO ──
└─ eval_user (documento + KB + formato obrigatório)
interpret_image()
├─ sys_msg parte 1 (critérios de conformidade)
├─ ── SKILL DE INTERPRETAÇÃO (namespace: financeiro) ──
│ [conteúdo do skill_store para namespace/interpret]
│ ── FIM DO SKILL DE INTERPRETAÇÃO ──
└─ sys_msg parte 2 (instruções obrigatórias + KB)
Sem skill configurado, _get_skill_directive() retorna string vazia — zero impacto no comportamento existente.
Tabela skills em skills.db:
| Coluna | Tipo | Descrição |
|---|---|---|
| id | INTEGER PK | Auto-incremento |
| namespace | TEXT | Namespace do KB |
| skill_type | TEXT | Tipo: evaluate, interpret, ou customizado |
| skill_name | TEXT | Nome descritivo |
| skill_content | TEXT | Conteúdo Markdown (instruções) |
| created_at | TEXT | ISO timestamp |
| updated_at | TEXT | ISO timestamp |
Constraint UNIQUE(namespace, skill_type) — um skill por tipo por namespace.
# Avaliação Financeira Especializada
Ao avaliar documentos neste namespace, aplique os seguintes critérios adicionais:
## Indicadores Obrigatórios
- EBITDA e margem EBITDA
- Margem líquida
- ROE e ROIC
- Endividamento (Dívida Líquida / EBITDA)
## Conformidade Regulatória
- Verificar aderência às normas IFRS/CPC
- Identificar contingências não provisionadas
- Sinalizar operações com partes relacionadas
## Formato Adicional
Incluir seção extra na avaliação:
**7. Análise Financeira Quantitativa**
- Tabela com indicadores extraídos
- Comparação com benchmarks setoriais quando disponíveis# Interpretação de Gráficos Financeiros
Ao interpretar imagens neste namespace:
## Prioridades de Extração
1. Identificar tipo de gráfico (barras, linhas, pizza, waterfall)
2. Extrair todos os valores numéricos visíveis
3. Identificar tendências e outliers
4. Comparar com indicadores do KB
## Formato de Saída
- Tabela com dados extraídos do gráfico
- Análise de tendência (crescimento, queda, estabilidade)
- Alertas quando valores divergirem do KB em mais de 10%- Seletor de namespace (lista dinâmica dos namespaces existentes)
- Seletor de tipo: Avaliação de Documentos, Interpretação de Imagens, Customizado
- Campo de tipo customizado (visível quando "Customizado" selecionado)
- Campo nome do skill
- Textarea com conteúdo Markdown + contador de caracteres em tempo real
- Botões: Salvar, Remover, Limpar
- Lista de skills configurados com badge de tipo (AVALIAÇÃO/INTERPRETAÇÃO) e preview do conteúdo
- Painel explicativo dos tipos built-in
- Clique no card de um skill para carregar no formulário de edição
Toda interação LLM é rastreada via Langfuse callbacks injetados em cada chamada:
- Rastreamento de raciocínio do agente (cada step/tool call)
- Consumo de tokens por operação
- Latência por fase
- Metadados transacionais (query, namespace, modo, rerank)
- Rastreio por usuário — login do usuário autenticado propagado em cada trace
O login do usuário autenticado (via sessão web) é propagado para o Langfuse em três níveis:
user_idnativo no CallbackHandler — campo nativo do Langfuse, permite filtrar traces por usuário no dashboard- Tag
user:<login>— aparece na lista de tags de cada trace, filtrável por busca - Campo
user_idnos metadados — presente no JSON de metadata para queries avançadas
Cadeia de propagação:
routes.py → request.state.user.login
→ agent.query_with_agent(..., user_login="sergio")
→ _get_lf_config(..., user_login="sergio")
→ tracing.get_handler(..., user_id="sergio")
→ Langfuse CallbackHandler(user_id="sergio")
+ tags: ["ns:default", "mode:agent", "user:sergio"]
+ metadata: {"user_id": "sergio", ...}
Requests via API key (sem sessão) registram user_id vazio — o Langfuse aceita normalmente.
Design tolerante a falhas: se credenciais ausentes ou servidor inacessível, o módulo desativa silenciosamente sem interromper a aplicação.
Autenticação por chave criptográfica:
- Cliente envia
X-API-Keyno header HTTP - Sistema recupera hash + salt pré-computados do ambiente
- Chave é concatenada com salt e submetida a SHA-256
- Digest comparado com hash armazenado via função timing-safe (mitigação de timing attacks)
- Exceção: requisições do próprio frontend (header de origem = host do servidor) são isentas
Sistema de login com gestão completa de usuários, persistido em SQLite (users.db).
Se o banco de dados users.db estiver vazio na primeira tentativa de login, o sistema cria automaticamente o usuário com as credenciais informadas como administrador com perfil "Super Usuário". Nenhuma seed ou configuração prévia é necessária.
| Tipo | Acesso |
|---|---|
admin |
Acesso total ao sistema, incluindo gestão de usuários (tab "Usuarios" visível) |
user |
Acesso normal ao sistema, sem gestão de usuários |
- Browser acessa
/→SessionAuthMiddlewareverifica cookiesuperag_session - Se ausente/expirado → redirect 302 para
/login - Login
POST /auth/login→ cria sessão no SQLite → set cookie httpOnly (24h TTL) - Requests
/api/*do frontend usam o cookie automaticamente - Requests
/api/*programáticos continuam usandoX-API-Key(sem mudança)
Tabela users: login (PK), password_hash, password_salt, user_type, profile_description, created_at.
Tabela sessions: token (PK), user_login (FK), created_at, expires_at.
Senhas armazenadas como SHA-256 + salt (mesmo padrão do app/auth.py). Sessões expiram em 24 horas. Sessões expiradas são removidas automaticamente a cada autenticação.
- Não é possível excluir a si mesmo
- Não é possível remover ou rebaixar o último administrador
- Cookie httpOnly (inacessível via JavaScript)
- Sessões com TTL de 24h
Visível apenas para usuários com user_type=admin. Permite:
- Lista de usuários com avatar (inicial do login), tipo (badge ADMIN/USUARIO), descrição do perfil
- Botão "Novo Usuario" com formulário inline (login, senha, tipo, descrição)
- Edição: alterar tipo, descrição e senha (campo opcional — vazio mantém a atual)
- Exclusão com confirmação
- Badge "(voce)" no card do próprio usuário
- Nome do usuário logado exibido no header + badge
[admin]quando aplicável - Botão de logout (ícone seta) com redirect para
/login
- Detecção automática de primeiro acesso (aviso "Nenhum usuario cadastrado")
- Toggle de visibilidade da senha
- Animação shake em erro de credenciais
- Redirect automático se já autenticado
Integração direta via importação Python (sem HTTP, elimina 403/CORS).
| Comando | Ação |
|---|---|
/start |
Inicialização + instruções |
/namespace <nome> |
Altera namespace ativo |
/agent |
Alterna Deep Agent ↔ Simple RAG |
/threshold <0-95> |
Ajusta limiar de similaridade coseno |
/status |
Exibe configuração atual |
| (texto livre) | Consulta à base de conhecimento |
Execução: python supeRAG_telegram.py (mesmo nível que app/).
Características: mensagens divididas em chunks de 4096 chars (limite Telegram), fallback Markdown → texto puro, estado persistido por chat_id, execução assíncrona via run_in_executor.
superag/
├── app/
│ ├── config.py # Settings (OpenAI, Pinecone, Langfuse)
│ ├── auth.py # Autenticação API key (SHA256+salt)
│ ├── main.py # Entry point + SessionAuthMiddleware
│ ├── api/
│ │ ├── routes.py # Endpoints FastAPI (chat, ingest, evaluate, interpret)
│ │ ├── auth_routes.py # Endpoints de autenticação (/auth/*)
│ │ └── skill_routes.py # Endpoints de skills (/api/skills/*)
│ ├── models/
│ │ └── schemas.py # Pydantic schemas (request/response)
│ ├── services/
│ │ ├── agent.py # Core: agente, guardrails, retrieval, métricas, injeção de skills
│ │ ├── metrics.py # compute_metrics (Recall, MRR, Coverage, Faithfulness, Relevance)
│ │ ├── vectorstore.py # Pinecone: similarity_search, similarity_search_overretrieve
│ │ ├── reranker.py # LLM-as-a-judge reranking
│ │ ├── prompt_store.py # CRUD de configuração por namespace (JSON)
│ │ ├── user_store.py # Gestão de usuários e sessões (SQLite)
│ │ ├── skill_store.py # Skills por namespace (SQLite)
│ │ ├── tracing.py # Langfuse callback handlers (com user_id)
│ │ ├── ingestion.py # Pipeline de ingestão (chunk + embed + store)
│ │ └── loaders.py # Loaders multi-formato (PDF, DOCX, PPTX, URL, etc.)
│ └── static/
│ ├── default.html # Frontend completo (Tailwind CSS)
│ └── login.html # Tela de login
├── users.db # SQLite: usuários e sessões (criado automaticamente)
├── skills.db # SQLite: skills por namespace (criado automaticamente)
├── prompt_store.json # Configurações de prompts por namespace
├── supeRAG_telegram.py # Bot Telegram (importação direta)
└── README.md
| Variável | Descrição |
|---|---|
OPENAI_API_KEY |
Chave da API OpenAI |
OPENAI_MODEL |
Modelo (ex: gpt-4.1) |
PINECONE_API_KEY |
Chave do Pinecone |
PINECONE_INDEX |
Nome do índice vetorial |
LANGFUSE_PUBLIC_KEY |
Chave pública Langfuse (opcional) |
LANGFUSE_SECRET_KEY |
Chave secreta Langfuse (opcional) |
LANGFUSE_HOST |
URL do servidor Langfuse (opcional) |
API_KEY_HASH |
Hash SHA-256 da API key do sistema |
API_KEY_SALT |
Salt para computação do hash |
TELEGRAM_BOT_TOKEN |
Token do bot Telegram (opcional) |
Suporte a PDF, PPTX, DOCX, TXT, Markdown, HTML, CSV e URLs web com chunk_size e chunk_overlap configuráveis por ingestão. Implementação com bibliotecas raw (pypdf, python-pptx, docx2txt, BeautifulSoup). Sanitização automática de encoding (reparo de mojibake UTF-8/latin-1). Para URLs, fluxo de descoberta em 2 etapas com seleção granular de sub-páginas.
Ao informar uma URL na aba de ingestão, o botão Explorar dispara um crawl breadth-first (até 3 níveis de profundidade, máximo 200 páginas) que descobre todos os links do mesmo domínio. O resultado é apresentado como uma árvore agrupada por diretório, com checkboxes individuais e por pasta, controles "Marcar todas" / "Desmarcar" e contador em tempo real de selecionadas. O botão Indexar Selecionadas ingere sequencialmente cada URL marcada. Filtro automático de extensões binárias (.pdf, .jpg, .css, .js, etc.) e timeout de 15s por página. Resolve o problema de ingerir apenas páginas de listagem/blog ao permitir selecionar os artigos individuais.
Chunks recuperados com cosine similarity abaixo de 0.60 (60%) são descartados antes de chegar ao LLM. Evita que fragmentos irrelevantes poluam o contexto e causem respostas de baixa qualidade. Threshold configurável via constante MIN_COSINE_SCORE em agent.py.
Deep Agent com planning, ferramentas de busca na knowledge base e raciocínio. Fallback automático para ReAct agent (LangGraph) se deepagents não instalado.
Retrieval → prompt → LLM, sem overhead de planejamento.
Slider de temperatura (0.0–1.0) na interface do chat, enviado para todos os endpoints (chat, evaluate, interpret). Controla a criatividade das respostas: 0.0 = preciso/determinístico, 1.0 = criativo/variado. Propagado para Deep Agent, Simple RAG e LLM de avaliação. Modo KB-only sobrescreve para 0.0 automaticamente.
Toggle na aba Prompts que restringe respostas ao conteúdo da KB, proibindo uso de conhecimento de treinamento do LLM ou fontes externas, mas permitindo síntese e combinação dos trechos recuperados. Enforcement em 3 camadas:
- System Prompt Directive — 7 regras injetadas no system prompt: proíbe dados de treinamento e invenção de fatos, mas permite sintetizar, resumir e reorganizar informações presentes no contexto.
- Prompt de Contexto Reforçado — instruções no prompt de query forçando uso exclusivo do contexto. Temperature forçada para 0.0.
- Guardrail de Compliance — segunda passagem LLM que verifica se a resposta é fundamentada no contexto do KB. Permite síntese legítima (paráfrase, combinação de trechos); remove apenas afirmações factuais inventadas. Se nada sobrevive, retorna "Esta informação não está disponível na base de conhecimento."
Aplicado em: query_simple, query_with_agent, evaluate_document, interpret_image.
Múltiplos documentos (PDF, PPTX, DOC/DOCX, TXT, MD) anexados via interface ou API para análise contextual cruzada com o KB.
Avaliação estruturada do documento contra o KB com Compliance Score (0-100), seções alinhadas/divergentes/não cobertas/ausentes, e recomendação. Respeita system prompt e guardrails do namespace.
Ctrl+V no chat cola imagem da área de transferência. Botão Interpretar envia para o modelo de visão (GPT-4o), que cruza o conteúdo visual com o KB do namespace selecionado, respeitando system prompt e guardrails.
Over-retrieve top_k × 4 → LLM scoring (0-10) → reorder top_k. Zero dependências extras.
Score do Pinecone exibido no rodapé de cada fonte com barra colorida (verde ≥85%, amarelo ≥70%, vermelho <70%). Threshold mínimo de 0.60 — chunks abaixo são descartados automaticamente antes de chegar ao LLM.
Todos os system prompts internos do sistema (DEFAULT_SYSTEM_PROMPT, guardrails de entrada/saída, guardrail KB-only, reranker, metrics judge, prompts de avaliação e interpretação, tool descriptions) são em português brasileiro. Detecção de bloqueio aceita tanto BLOCKED: quanto BLOQUEADO: para compatibilidade bidirecional.
Configuração por namespace: guardrail de entrada, toggle KB-only, system prompt customizado, guardrail de saída. Persistido em prompt_store.json. Badge "KB ONLY" visível na lista de configs quando ativo.
Na aba Bases, cada namespace expande para exibir a árvore de documentos e URLs ingeridos, agrupados por tipo (PDF, Word, PPTX, URL, etc.) com ícones coloridos e contagem de chunks. Cada fonte individual pode ser excluída sem afetar as demais — o sistema localiza e remove todos os vetores associados àquela fonte específica no Pinecone.
Recall@K, MRR, Faithfulness, Answer Relevance, Citation Coverage — exibidas após cada resposta.
Observability opcional para todas as interações LLM. Cada trace inclui o user_id do usuário autenticado (via sessão web), permitindo filtrar por usuário no dashboard Langfuse. Tag user:<login> adicionada automaticamente.
SHA256 + salt. Header X-API-Key obrigatório para acesso programático. Frontend web isento. Se não configurado, acesso livre.
Login com cookie httpOnly (superag_session), TTL 24h, gestão de usuários com tipos admin/user. Middleware SessionAuthMiddleware intercepta requisições sem sessão válida e redireciona para /login. Primeiro acesso auto-cria admin. Senhas SHA-256 + salt em SQLite (users.db).
CRUD completo de usuários via interface (tab "Usuarios", admin only) ou API (/auth/users). Criação com login, senha, tipo e descrição de perfil. Edição de tipo, senha e perfil. Exclusão com proteções (não excluir a si mesmo, não remover último admin). Avatar com inicial do login.
Instruções especializadas em Markdown injetadas nos prompts de avaliação (evaluate) e interpretação (interpret) por namespace. Um skill por tipo por namespace, extensível com tipos customizados. Persistido em SQLite (skills.db). Interface com formulário de edição, seletor de tipo e lista de skills configurados. Sem skill configurado, comportamento inalterado.
Login do usuário autenticado propagado como user_id nativo do Langfuse, tag user:<login> e campo nos metadados. Permite filtrar traces por usuário no dashboard Langfuse.
Todos os endpoints /api/* requerem header X-API-Key (acesso programático) ou cookie de sessão válido (acesso via browser) quando autenticação está habilitada. Endpoints /auth/* são públicos.
| Método | Endpoint | Descrição |
|---|---|---|
| GET | /api/health |
Health check |
| POST | /api/chat |
Consulta com Deep Agent ou Simple RAG |
| POST | /api/evaluate |
Avaliação de documentos contra o KB |
| POST | /api/interpret |
Interpretação de imagem com KB |
| POST | /api/discover/url |
Descoberta de sub-páginas a partir de uma URL |
| POST | /api/ingest/file |
Upload e indexação de arquivo |
| POST | /api/ingest/url |
Ingestão via URL |
| GET | /api/namespaces |
Lista namespaces e contagem de vetores |
| DELETE | /api/namespaces/{ns} |
Remove namespace inteiro |
| GET | /api/namespaces/{ns}/sources |
Lista fontes (docs/URLs) de um namespace |
| DELETE | /api/namespaces/{ns}/sources |
Remove uma fonte específica de um namespace |
| GET | /api/prompts |
Lista configs de prompts |
| GET | /api/prompts/{ns} |
Retorna config de um namespace |
| PUT | /api/prompts/{ns} |
Salva/atualiza config (inclui kb_only) |
| DELETE | /api/prompts/{ns} |
Remove config de prompts |
| Método | Endpoint | Acesso | Descrição |
|---|---|---|---|
| GET | /auth/check |
público | Verifica sessão ativa + se há usuários |
| POST | /auth/login |
público | Autenticação (retorna cookie sessão) |
| POST | /auth/logout |
autenticado | Invalida sessão |
| GET | /auth/me |
autenticado | Dados do usuário logado |
| GET | /auth/users |
admin | Lista todos os usuários |
| POST | /auth/users |
admin | Cria novo usuário |
| PUT | /auth/users/{login} |
admin | Edita usuário (tipo, senha, perfil) |
| DELETE | /auth/users/{login} |
admin | Remove usuário |
| PUT | /auth/password |
autenticado | Altera própria senha |
| Método | Endpoint | Descrição |
|---|---|---|
| GET | /api/skills |
Lista todos os skills (todos namespaces) |
| GET | /api/skills/{ns} |
Lista skills de um namespace |
| GET | /api/skills/{ns}/{type} |
Retorna skill específico |
| PUT | /api/skills/{ns}/{type} |
Cria/atualiza skill (upsert) |
| DELETE | /api/skills/{ns}/{type} |
Remove skill |
POST /api/chat (multipart/form-data)
| Parâmetro | Tipo | Default | Descrição |
|---|---|---|---|
| query | string | (required) | Pergunta do usuário |
| namespace | string | "default" | Namespace da KB |
| top_k | int | 5 | Número de documentos a recuperar |
| use_agent | string | "true" | "true" para Deep Agent, "false" para Simple RAG |
| use_rerank | string | "false" | "true" para ativar reranking |
| temperature | float | 0.7 | Temperature do LLM (0.0–1.0) |
| files | files | [] | Documentos anexados (opcional, múltiplos) |
POST /api/evaluate (multipart/form-data)
| Parâmetro | Tipo | Default | Descrição |
|---|---|---|---|
| namespace | string | "default" | Namespace da KB |
| top_k | int | 10 | Número de documentos para comparação |
| use_rerank | string | "false" | "true" para ativar reranking |
| temperature | float | 0.0 | Temperature do LLM |
| files | files | (required) | Documentos para avaliação |
POST /api/interpret (multipart/form-data)
| Parâmetro | Tipo | Default | Descrição |
|---|---|---|---|
| image | file | (required) | Arquivo de imagem |
| query | string | auto | Instrução para interpretação |
| namespace | string | "default" | Namespace da KB |
| top_k | int | 5 | Número de documentos de contexto |
| use_rerank | string | "false" | "true" para ativar reranking |
| temperature | float | 0.3 | Temperature do LLM |
PUT /api/prompts/{ns} (JSON)
| Parâmetro | Tipo | Default | Descrição |
|---|---|---|---|
| input_guardrail | string | "" | Regras para validar input do usuário |
| kb_only | bool | false | Restringir respostas exclusivamente ao KB |
| system_prompt | string | "" | System prompt customizado |
| output_guardrail | string | "" | Regras para validar output do LLM |
DELETE /api/namespaces/{ns}/sources (JSON body)
| Parâmetro | Tipo | Descrição |
|---|---|---|
| source_name | string | Nome exato da fonte a excluir (filename ou URL) |
POST /api/discover/url (JSON body)
| Parâmetro | Tipo | Default | Descrição |
|---|---|---|---|
| url | string | (required) | URL raiz para iniciar a descoberta |
| max_depth | int | 2 | Profundidade máxima do crawl (1–3) |
Retorna {url, links: [{url, path, depth, title}], total}.
POST /auth/login (JSON)
| Parâmetro | Tipo | Descrição |
|---|---|---|
| login | string | Login do usuário |
| password | string | Senha em texto plano (transmitida, nunca armazenada) |
Retorna cookie superag_session (httpOnly, 24h TTL) + {status, user}.
POST /auth/users (JSON, admin only)
| Parâmetro | Tipo | Default | Descrição |
|---|---|---|---|
| login | string | (required) | Login único do novo usuário |
| password | string | (required) | Senha inicial |
| user_type | string | "user" | "admin" ou "user" |
| profile_description | string | "" | Descrição do perfil do usuário |
PUT /auth/users/{login} (JSON, admin only)
| Parâmetro | Tipo | Descrição |
|---|---|---|
| password | string | Nova senha (omitir para manter a atual) |
| user_type | string | "admin" ou "user" |
| profile_description | string | Nova descrição do perfil |
PUT /api/skills/{ns}/{type} (JSON)
| Parâmetro | Tipo | Descrição |
|---|---|---|
| skill_name | string | Nome descritivo do skill |
| skill_content | string | Conteúdo Markdown com as instruções do skill |
import requests
import json
from pathlib import Path
BASE = "https://seu-servidor.onrender.com" # ou http://localhost:8000
API_KEY = "sua-chave-api-aqui"
HEADERS = {"X-API-Key": API_KEY}
NS = "financeiro"r = requests.get(f"{BASE}/api/health", headers=HEADERS)
r.json(){"status": "ok", "pinecone": true}config = {
"input_guardrail": "Apenas perguntas sobre indicadores financeiros e resultados trimestrais.",
"kb_only": True,
"system_prompt": (
"Você é um analista financeiro sênior.\n"
"Responda com precisão, cite fontes por número, use linguagem formal.\n"
"Nunca recomende ações de investimento."
),
"output_guardrail": (
"Não revelar valores absolutos de receita.\n"
"Garantir resposta em português brasileiro."
)
}
r = requests.put(f"{BASE}/api/prompts/{NS}", json=config, headers=HEADERS)
print(r.json())# Arquivo local com chunk_size e chunk_overlap customizados
with open("relatorio_q3.pdf", "rb") as f:
r = requests.post(
f"{BASE}/api/ingest/file",
files={"file": f},
data={
"source_type": "pdf",
"namespace": NS,
"chunk_size": 512,
"chunk_overlap": 128,
},
headers=HEADERS,
)
print(r.json())
# {"status": "indexed", "chunks_indexed": 47, "namespace": "financeiro", "source_name": "relatorio_q3.pdf"}
# Ingerir URL
r = requests.post(
f"{BASE}/api/ingest/url",
data={
"url": "https://empresa.com.br/ri/resultados-q3",
"namespace": NS,
"chunk_size": 512,
"chunk_overlap": 128,
},
headers=HEADERS,
)
print(r.json())arquivos = Path("./documentos").glob("*.pdf")
for arq in arquivos:
with open(arq, "rb") as f:
r = requests.post(
f"{BASE}/api/ingest/file",
files={"file": f},
data={"source_type": "pdf", "namespace": NS, "chunk_size": 512, "chunk_overlap": 128},
headers=HEADERS,
)
status = r.json()
print(f"{arq.name}: {status.get('chunks_indexed', 'ERRO')} chunks")r = requests.post(
f"{BASE}/api/chat",
data={
"query": "Qual foi o crescimento percentual do Q3?",
"namespace": NS,
"top_k": 5,
"use_agent": "false",
"use_rerank": "false",
"temperature": 0.3,
},
headers=HEADERS,
)
data = r.json()
print(data["answer"])
print(f"\nFontes: {len(data['sources'])}")
for s in data["sources"]:
score = s.get("similarity_score", "?")
print(f" - {s['metadata'].get('source_name', '?')} (cosine: {score})")r = requests.post(
f"{BASE}/api/chat",
data={
"query": "Compare a margem EBITDA do Q3 com Q2 e explique as variações",
"namespace": NS,
"top_k": 8,
"use_agent": "true",
"use_rerank": "true",
"temperature": 0.5,
},
headers=HEADERS,
)
data = r.json()
print(data["answer"])
# Métricas RAG
if data.get("metrics"):
m = data["metrics"]
print(f"\nRecall@K: {m.get('recall_at_k', 0):.2f}")
print(f"MRR: {m.get('mrr', 0):.2f}")
print(f"Faithfulness: {m.get('faithfulness', 0):.2f}")
print(f"Relevance: {m.get('answer_relevance', 0):.2f}")
print(f"Coverage: {m.get('citation_coverage', 0):.2f}")with open("proposta_parceria.pdf", "rb") as f:
r = requests.post(
f"{BASE}/api/chat",
files={"files": f},
data={
"query": "Compare esta proposta com os dados do Q3 e identifique riscos",
"namespace": NS,
"top_k": 5,
"use_agent": "true",
"use_rerank": "true",
"temperature": 0.4,
},
headers=HEADERS,
)
print(r.json()["answer"])with open("relatorio_externo.pdf", "rb") as f:
r = requests.post(
f"{BASE}/api/evaluate",
files={"files": f},
data={"namespace": NS, "top_k": 10, "use_rerank": "true", "temperature": 0.0},
headers=HEADERS,
)
data = r.json()
print(data["answer"])
# Saída estruturada: Compliance Score, Aligned, Divergent, Not Covered, Missing, Recommendationwith open("grafico_vendas.png", "rb") as f:
r = requests.post(
f"{BASE}/api/interpret",
files={"image": ("grafico_vendas.png", f, "image/png")},
data={
"query": "Analise este gráfico comparando com os dados do KB",
"namespace": NS,
"top_k": 5,
"use_rerank": "false",
"temperature": 0.2,
},
headers=HEADERS,
)
print(r.json()["answer"])r = requests.get(f"{BASE}/api/namespaces/{NS}/sources", headers=HEADERS)
data = r.json()
for s in data["sources"]:
print(f" [{s['source_type']}] {s['source_name']} — {s['chunks']} chunks")# Excluir um documento
r = requests.delete(
f"{BASE}/api/namespaces/{NS}/sources",
json={"source_name": "relatorio_q2.pdf"},
headers=HEADERS,
)
print(r.json())
# {"namespace": "financeiro", "source_name": "relatorio_q2.pdf", "deleted_vectors": 35}
# Excluir uma URL
r = requests.delete(
f"{BASE}/api/namespaces/{NS}/sources",
json={"source_name": "https://empresa.com.br/ri/resultados-q2"},
headers=HEADERS,
)
print(r.json())# Listar
r = requests.get(f"{BASE}/api/namespaces", headers=HEADERS)
for ns in r.json()["namespaces"]:
print(f" {ns['namespace']}: {ns['vector_count']} vetores")
# Deletar namespace inteiro
r = requests.delete(f"{BASE}/api/namespaces/teste", headers=HEADERS)
print(r.json())# Listar todas as configs
r = requests.get(f"{BASE}/api/prompts", headers=HEADERS)
for cfg in r.json()["configs"]:
kb = "✓" if cfg.get("kb_only") else "✗"
p = "✓" if cfg.get("system_prompt") else "✗"
i = "✓" if cfg.get("input_guardrail") else "✗"
o = "✓" if cfg.get("output_guardrail") else "✗"
print(f" {cfg['namespace']}: kb_only:{kb} prompt:{p} in:{i} out:{o}")
# Ler config de um namespace
r = requests.get(f"{BASE}/api/prompts/{NS}", headers=HEADERS)
print(json.dumps(r.json(), indent=2, ensure_ascii=False))
# Ativar KB-only em um namespace existente
r = requests.put(f"{BASE}/api/prompts/{NS}", json={"kb_only": True}, headers=HEADERS)
# Desativar KB-only
r = requests.put(f"{BASE}/api/prompts/{NS}", json={"kb_only": False}, headers=HEADERS)
# Deletar config
r = requests.delete(f"{BASE}/api/prompts/teste", headers=HEADERS)# 1. Ingerir vários documentos
for arq in Path("./docs").glob("*.*"):
ext = arq.suffix.lstrip(".")
type_map = {"pdf": "pdf", "docx": "docx", "pptx": "pptx", "txt": "txt", "csv": "csv"}
if ext not in type_map:
continue
with open(arq, "rb") as f:
r = requests.post(
f"{BASE}/api/ingest/file",
files={"file": f},
data={"source_type": type_map[ext], "namespace": "projeto-x"},
headers=HEADERS,
)
print(f" {arq.name}: {r.json().get('chunks_indexed', '?')} chunks")
# 2. Configurar prompt + KB-only
requests.put(f"{BASE}/api/prompts/projeto-x", json={
"kb_only": True,
"system_prompt": "Responda exclusivamente sobre o Projeto X. Use linguagem técnica.",
"input_guardrail": "Bloquear perguntas não relacionadas ao Projeto X.",
"output_guardrail": "Responder em português. Citar fontes.",
}, headers=HEADERS)
# 3. Consultar
r = requests.post(f"{BASE}/api/chat", data={
"query": "Quais são os riscos identificados na fase 2?",
"namespace": "projeto-x",
"use_agent": "true",
"temperature": 0.0,
}, headers=HEADERS)
print(r.json()["answer"])
# 4. Listar fontes
r = requests.get(f"{BASE}/api/namespaces/projeto-x/sources", headers=HEADERS)
for s in r.json()["sources"]:
print(f" {s['source_name']} ({s['chunks']} chunks)")
# 5. Remover documento obsoleto
requests.delete(f"{BASE}/api/namespaces/projeto-x/sources",
json={"source_name": "rascunho_v1.pdf"}, headers=HEADERS)
# 6. Verificar atualização
r = requests.get(f"{BASE}/api/namespaces/projeto-x/sources", headers=HEADERS)
print(f"Fontes restantes: {len(r.json()['sources'])}")# 1. Descobrir sub-páginas de um site
r = requests.post(f"{BASE}/api/discover/url", json={
"url": "https://falagaiotto.com.br",
"max_depth": 2,
}, headers=HEADERS)
data = r.json()
print(f"Total de páginas encontradas: {data['total']}")
for link in data["links"]:
print(f" [d{link['depth']}] {link['path']} — {link['title'][:50]}")
# 2. Filtrar apenas artigos (excluir páginas de navegação, tags, categorias)
artigos = [l for l in data["links"] if "/20" in l["path"] or "/blog/" in l["path"]]
print(f"\nArtigos selecionados: {len(artigos)}")
# 3. Ingerir apenas os artigos selecionados
for art in artigos:
r = requests.post(f"{BASE}/api/ingest/url", data={
"url": art["url"],
"namespace": "blog",
"chunk_size": 512,
"chunk_overlap": 128,
}, headers=HEADERS)
if r.ok:
print(f" ✓ {art['url']}: {r.json()['chunks_indexed']} chunks")
else:
print(f" ✗ {art['url']}: {r.json().get('detail', 'erro')}")
# 4. Verificar fontes ingeridas
r = requests.get(f"{BASE}/api/namespaces/blog/sources", headers=HEADERS)
for s in r.json()["sources"]:
print(f" [{s['source_type']}] {s['source_name'][:60]} — {s['chunks']} chunks")# Criar skill de avaliação financeira
r = requests.put(
f"{BASE}/api/skills/financeiro/evaluate",
json={
"skill_name": "avaliador-financeiro",
"skill_content": (
"# Avaliação Financeira Especializada\n\n"
"Ao avaliar documentos neste namespace:\n\n"
"## Indicadores Obrigatórios\n"
"- EBITDA e margem EBITDA\n"
"- Margem líquida, ROE e ROIC\n"
"- Endividamento (Dívida Líquida / EBITDA)\n\n"
"## Conformidade Regulatória\n"
"- Verificar aderência IFRS/CPC\n"
"- Sinalizar operações com partes relacionadas\n\n"
"## Formato Adicional\n"
"Incluir seção extra: **7. Análise Financeira Quantitativa**"
),
},
headers=HEADERS,
)
print(r.json())
# {"skill": {"id": 1, "namespace": "financeiro", "skill_type": "evaluate", ...}}
# Criar skill de interpretação de gráficos
r = requests.put(
f"{BASE}/api/skills/financeiro/interpret",
json={
"skill_name": "interpretador-graficos",
"skill_content": (
"# Interpretação de Gráficos Financeiros\n\n"
"1. Identificar tipo de gráfico\n"
"2. Extrair todos os valores numéricos\n"
"3. Comparar com indicadores do KB\n"
"4. Alertar quando valores divergirem mais de 10%"
),
},
headers=HEADERS,
)
print(r.json())
# Listar skills de um namespace
r = requests.get(f"{BASE}/api/skills/financeiro", headers=HEADERS)
for s in r.json()["skills"]:
print(f" [{s['skill_type']}] {s['skill_name']} — {len(s['skill_content'])} chars")
# Excluir um skill
r = requests.delete(f"{BASE}/api/skills/financeiro/interpret", headers=HEADERS)
print(r.json())import requests
BASE = "http://localhost:8000"
# Login (retorna cookie de sessão)
session = requests.Session()
r = session.post(f"{BASE}/auth/login", json={
"login": "admin",
"password": "minha-senha-segura",
})
print(r.json()) # {"status": "ok", "user": {"login": "admin", "user_type": "admin", ...}}
# Verificar sessão
r = session.get(f"{BASE}/auth/check")
print(r.json()) # {"authenticated": true, "user": {...}}
# Criar novo usuário (requer admin)
r = session.post(f"{BASE}/auth/users", json={
"login": "analista.jr",
"password": "senha-inicial",
"user_type": "user",
"profile_description": "Analista júnior, equipe de RI",
})
print(r.json())
# Listar usuários
r = session.get(f"{BASE}/auth/users")
for u in r.json()["users"]:
print(f" {u['login']} ({u['user_type']}) — {u['profile_description']}")
# Alterar tipo de usuário
r = session.put(f"{BASE}/auth/users/analista.jr", json={
"user_type": "admin",
"profile_description": "Promovido a admin",
})
print(r.json())
# Logout
r = session.post(f"{BASE}/auth/logout")
print(r.json()) # {"status": "ok"}Bot completo que expõe todas as funcionalidades do SupeRAG via comandos Telegram.
pip install python-telegram-bot requests"""SupeRAG Telegram Bot — acessa todas as features via API."""
import logging
import requests
from io import BytesIO
from telegram import Update
from telegram.ext import (
Application, CommandHandler, MessageHandler,
ContextTypes, filters,
)
# ── Config ──
TELEGRAM_TOKEN = "SEU_TOKEN_TELEGRAM"
SUPERAG_BASE = "https://seu-servidor.onrender.com"
SUPERAG_API_KEY = "sua-chave-api-aqui"
DEFAULT_NS = "default"
HEADERS = {"X-API-Key": SUPERAG_API_KEY}
logging.basicConfig(level=logging.INFO)
log = logging.getLogger(__name__)
# ── Helpers ──
def api_post(endpoint, data=None, files=None):
r = requests.post(
f"{SUPERAG_BASE}/api/{endpoint}",
data=data, files=files, headers=HEADERS, timeout=120,
)
r.raise_for_status()
return r.json()
def api_get(endpoint):
r = requests.get(f"{SUPERAG_BASE}/api/{endpoint}", headers=HEADERS, timeout=30)
r.raise_for_status()
return r.json()
def api_put(endpoint, json_data):
r = requests.put(
f"{SUPERAG_BASE}/api/{endpoint}",
json=json_data, headers=HEADERS, timeout=30,
)
r.raise_for_status()
return r.json()
def api_delete(endpoint, json_data=None):
r = requests.delete(
f"{SUPERAG_BASE}/api/{endpoint}",
json=json_data, headers=HEADERS, timeout=30,
)
r.raise_for_status()
return r.json()
def api_post_json(endpoint, json_data):
r = requests.post(
f"{SUPERAG_BASE}/api/{endpoint}",
json=json_data, headers=HEADERS, timeout=120,
)
r.raise_for_status()
return r.json()
def get_ns(context) -> str:
return context.user_data.get("namespace", DEFAULT_NS)
# ── Comandos ──
async def cmd_start(update: Update, context: ContextTypes.DEFAULT_TYPE):
await update.message.reply_text(
"🤖 *SupeRAG Bot*\n\n"
"Comandos disponíveis:\n"
"/ns — selecionar namespace\n"
"/ask — consulta Simple RAG\n"
"/agent — consulta Deep Agent\n"
"/ingest_url — ingerir URL\n"
"/discover [profundidade] — explorar sub-páginas\n"
"/ingest_discovered [n1,n2,...] — ingerir URLs descobertas\n"
"/evaluate — avaliar doc (envie arquivo depois)\n"
"/interpret — interpretar imagem (envie foto depois)\n"
"/sources — listar fontes do namespace\n"
"/del_source — excluir fonte do namespace\n"
"/namespaces — listar namespaces\n"
"/prompts — listar configs\n"
"/set_prompt — configurar prompt (veja ajuda)\n"
"/kb_only on|off — ativar/desativar modo KB-only\n"
"/metrics — mostrar métricas da última resposta\n\n"
"📎 Envie um arquivo para ingerir automaticamente\n"
"📸 Envie uma foto para interpretar com o KB",
parse_mode="Markdown",
)
async def cmd_ns(update: Update, context: ContextTypes.DEFAULT_TYPE):
if not context.args:
ns = get_ns(context)
await update.message.reply_text(f"Namespace atual: `{ns}`", parse_mode="Markdown")
return
context.user_data["namespace"] = context.args[0]
await update.message.reply_text(f"Namespace alterado para: `{context.args[0]}`", parse_mode="Markdown")
async def cmd_ask(update: Update, context: ContextTypes.DEFAULT_TYPE):
query = " ".join(context.args) if context.args else ""
if not query:
await update.message.reply_text("Uso: /ask ")
return
msg = await update.message.reply_text("🔍 Consultando KB...")
try:
data = api_post("chat", data={
"query": query,
"namespace": get_ns(context),
"top_k": 5,
"use_agent": "false",
"use_rerank": "false",
"temperature": 0.3,
})
context.user_data["last_metrics"] = data.get("metrics")
answer = data["answer"][:4000]
sources = "\n".join(
f" • {s['metadata'].get('source_name','?')} ({s.get('similarity_score','?')})"
for s in data.get("sources", [])[:5]
)
await msg.edit_text(f"{answer}\n\n📚 *Fontes:*\n{sources}", parse_mode="Markdown")
except Exception as e:
await msg.edit_text(f"❌ Erro: {e}")
async def cmd_agent(update: Update, context: ContextTypes.DEFAULT_TYPE):
query = " ".join(context.args) if context.args else ""
if not query:
await update.message.reply_text("Uso: /agent ")
return
msg = await update.message.reply_text("🧠 Deep Agent processando...")
try:
data = api_post("chat", data={
"query": query,
"namespace": get_ns(context),
"top_k": 5,
"use_agent": "true",
"use_rerank": "true",
"temperature": 0.5,
})
context.user_data["last_metrics"] = data.get("metrics")
await msg.edit_text(data["answer"][:4000])
except Exception as e:
await msg.edit_text(f"❌ Erro: {e}")
async def cmd_ingest_url(update: Update, context: ContextTypes.DEFAULT_TYPE):
url = context.args[0] if context.args else ""
if not url:
await update.message.reply_text("Uso: /ingest_url ")
return
msg = await update.message.reply_text("📥 Ingerindo URL...")
try:
data = api_post("ingest/url", data={
"url": url,
"namespace": get_ns(context),
"chunk_size": 512,
"chunk_overlap": 128,
})
await msg.edit_text(f"✅ {data['chunks_indexed']} chunks indexados em `{data['namespace']}`", parse_mode="Markdown")
except Exception as e:
await msg.edit_text(f"❌ Erro: {e}")
async def cmd_discover(update: Update, context: ContextTypes.DEFAULT_TYPE):
url = context.args[0] if context.args else ""
if not url:
await update.message.reply_text("Uso: /discover [profundidade]\nEx: /discover https://site.com 2")
return
depth = int(context.args[1]) if len(context.args) > 1 else 2
depth = min(max(depth, 1), 3)
msg = await update.message.reply_text("🔍 Explorando links...")
try:
data = api_post_json("discover/url", {"url": url, "max_depth": depth})
links = data.get("links", [])
if not links:
await msg.edit_text("Nenhum link encontrado.")
return
# Store for later use with /ingest_discovered
context.user_data["discovered_links"] = links
context.user_data["discover_ns"] = get_ns(context)
lines = []
for i, l in enumerate(links[:50]):
title = f" — {l['title'][:40]}" if l.get("title") else ""
lines.append(f"`{i+1}.` [d{l['depth']}] {l['path']}{title}")
text = f"🌐 *{len(links)} páginas encontradas:*\n" + "\n".join(lines)
if len(links) > 50:
text += f"\n\n... e mais {len(links) - 50} páginas."
text += f"\n\nUse /ingest_discovered para ingerir todas, ou /ingest_discovered 1,3,5 para selecionar por número."
await msg.edit_text(text, parse_mode="Markdown")
except Exception as e:
await msg.edit_text(f"❌ Erro: {e}")
async def cmd_ingest_discovered(update: Update, context: ContextTypes.DEFAULT_TYPE):
links = context.user_data.get("discovered_links")
if not links:
await update.message.reply_text("Nenhuma descoberta ativa. Use /discover primeiro.")
return
# Parse selection
if context.args:
try:
indices = [int(x.strip()) - 1 for x in " ".join(context.args).replace(",", " ").split()]
selected = [links[i] for i in indices if 0 <= i < len(links)]
except ValueError:
await update.message.reply_text("Uso: /ingest_discovered 1,3,5 (números das páginas)")
return
else:
selected = links
ns = context.user_data.get("discover_ns", get_ns(context))
msg = await update.message.reply_text(f"📥 Ingerindo {len(selected)} URLs em `{ns}`...", parse_mode="Markdown")
ok, fail = 0, 0
for l in selected:
try:
api_post("ingest/url", data={
"url": l["url"], "namespace": ns,
"chunk_size": 512, "chunk_overlap": 128,
})
ok += 1
except Exception:
fail += 1
context.user_data.pop("discovered_links", None)
result = f"✅ {ok} URLs indexadas em `{ns}`"
if fail:
result += f", {fail} erros"
await msg.edit_text(result, parse_mode="Markdown")
async def cmd_evaluate(update: Update, context: ContextTypes.DEFAULT_TYPE):
context.user_data["pending_action"] = "evaluate"
await update.message.reply_text("📄 Envie o arquivo para avaliação.")
async def cmd_interpret(update: Update, context: ContextTypes.DEFAULT_TYPE):
query = " ".join(context.args) if context.args else "Interprete esta imagem com base na base de conhecimento"
context.user_data["pending_action"] = "interpret"
context.user_data["interpret_query"] = query
await update.message.reply_text("📸 Envie a imagem para interpretação.")
async def cmd_sources(update: Update, context: ContextTypes.DEFAULT_TYPE):
ns = get_ns(context)
try:
data = api_get(f"namespaces/{ns}/sources")
sources = data.get("sources", [])
if not sources:
await update.message.reply_text(f"Nenhuma fonte em `{ns}`", parse_mode="Markdown")
return
lines = [f" • `[{s['source_type']}]` {s['source_name']} — {s['chunks']} chunks"
for s in sources]
await update.message.reply_text(
f"📂 *Fontes em {ns}:*\n" + "\n".join(lines), parse_mode="Markdown"
)
except Exception as e:
await update.message.reply_text(f"❌ Erro: {e}")
async def cmd_del_source(update: Update, context: ContextTypes.DEFAULT_TYPE):
source_name = " ".join(context.args) if context.args else ""
if not source_name:
await update.message.reply_text("Uso: /del_source ")
return
ns = get_ns(context)
msg = await update.message.reply_text(f"🗑️ Excluindo `{source_name}`...", parse_mode="Markdown")
try:
data = api_delete(f"namespaces/{ns}/sources", {"source_name": source_name})
await msg.edit_text(
f"✅ Excluído: {data['deleted_vectors']} vetores de `{source_name}` em `{ns}`",
parse_mode="Markdown",
)
except Exception as e:
await msg.edit_text(f"❌ Erro: {e}")
async def cmd_namespaces(update: Update, context: ContextTypes.DEFAULT_TYPE):
try:
data = api_get("namespaces")
lines = [f" • `{ns['namespace']}`: {ns['vector_count']} vetores"
for ns in data["namespaces"]]
await update.message.reply_text("📂 *Namespaces:*\n" + "\n".join(lines), parse_mode="Markdown")
except Exception as e:
await update.message.reply_text(f"❌ Erro: {e}")
async def cmd_prompts(update: Update, context: ContextTypes.DEFAULT_TYPE):
try:
data = api_get("prompts")
lines = []
for c in data["configs"]:
kb = "✓" if c.get("kb_only") else "✗"
p = "✓" if c.get("system_prompt") else "✗"
i = "✓" if c.get("input_guardrail") else "✗"
o = "✓" if c.get("output_guardrail") else "✗"
lines.append(f" • `{c['namespace']}` kb:{kb} prompt:{p} in:{i} out:{o}")
await update.message.reply_text("⚙️ *Configs:*\n" + "\n".join(lines), parse_mode="Markdown")
except Exception as e:
await update.message.reply_text(f"❌ Erro: {e}")
async def cmd_set_prompt(update: Update, context: ContextTypes.DEFAULT_TYPE):
"""Uso: /set_prompt campo valor
Campos: system_prompt, input_guardrail, output_guardrail
"""
text = update.message.text.replace("/set_prompt", "").strip()
parts = text.split(" ", 1)
if len(parts) < 2 or parts[0] not in ("system_prompt", "input_guardrail", "output_guardrail"):
await update.message.reply_text(
"Uso: /set_prompt \n"
"Campos: system_prompt, input_guardrail, output_guardrail"
)
return
try:
api_put(f"prompts/{get_ns(context)}", {parts[0]: parts[1]})
await update.message.reply_text(f"✅ `{parts[0]}` atualizado para namespace `{get_ns(context)}`", parse_mode="Markdown")
except Exception as e:
await update.message.reply_text(f"❌ Erro: {e}")
async def cmd_kb_only(update: Update, context: ContextTypes.DEFAULT_TYPE):
arg = (context.args[0] if context.args else "").lower()
if arg not in ("on", "off"):
await update.message.reply_text("Uso: /kb_only on|off")
return
ns = get_ns(context)
val = arg == "on"
try:
api_put(f"prompts/{ns}", {"kb_only": val})
status = "ativado" if val else "desativado"
await update.message.reply_text(f"✅ KB-only {status} para `{ns}`", parse_mode="Markdown")
except Exception as e:
await update.message.reply_text(f"❌ Erro: {e}")
async def cmd_metrics(update: Update, context: ContextTypes.DEFAULT_TYPE):
m = context.user_data.get("last_metrics")
if not m:
await update.message.reply_text("Nenhuma métrica disponível. Faça uma consulta primeiro.")
return
text = (
f"📊 *Métricas RAG:*\n"
f" Recall@K: {m.get('recall_at_k', 0):.2f}\n"
f" MRR: {m.get('mrr', 0):.2f}\n"
f" Faithfulness: {m.get('faithfulness', 0):.2f}\n"
f" Relevance: {m.get('answer_relevance', 0):.2f}\n"
f" Coverage: {m.get('citation_coverage', 0):.2f}"
)
await update.message.reply_text(text, parse_mode="Markdown")
# ── Handlers de mídia ──
async def handle_document(update: Update, context: ContextTypes.DEFAULT_TYPE):
"""Auto-ingest de documentos OU avaliação se /evaluate pendente."""
doc = update.message.document
file = await doc.get_file()
buf = BytesIO()
await file.download_to_memory(buf)
buf.seek(0)
ext = (doc.file_name or "").rsplit(".", 1)[-1].lower()
type_map = {"pdf": "pdf", "pptx": "pptx", "docx": "docx", "doc": "docx", "txt": "txt", "md": "markdown", "csv": "csv", "html": "html"}
source_type = type_map.get(ext)
if not source_type:
await update.message.reply_text(f"Formato .{ext} não suportado.")
return
action = context.user_data.pop("pending_action", None)
if action == "evaluate":
msg = await update.message.reply_text("📋 Avaliando documento...")
try:
data = api_post("evaluate", data={
"namespace": get_ns(context),
"top_k": 10,
"use_rerank": "true",
"temperature": 0.0,
}, files={"files": (doc.file_name, buf)})
await msg.edit_text(data["answer"][:4000])
except Exception as e:
await msg.edit_text(f"❌ Erro: {e}")
else:
msg = await update.message.reply_text("📥 Indexando documento...")
try:
data = api_post("ingest/file", data={
"source_type": source_type,
"namespace": get_ns(context),
"chunk_size": 512,
"chunk_overlap": 128,
}, files={"file": (doc.file_name, buf)})
await msg.edit_text(
f"✅ `{doc.file_name}`: {data['chunks_indexed']} chunks em `{data['namespace']}`",
parse_mode="Markdown",
)
except Exception as e:
await msg.edit_text(f"❌ Erro: {e}")
async def handle_photo(update: Update, context: ContextTypes.DEFAULT_TYPE):
"""Interpretar imagem enviada."""
photo = update.message.photo[-1] # maior resolução
file = await photo.get_file()
buf = BytesIO()
await file.download_to_memory(buf)
buf.seek(0)
query = context.user_data.pop("interpret_query", "Interprete esta imagem com base na base de conhecimento")
context.user_data.pop("pending_action", None)
msg = await update.message.reply_text("👁️ Interpretando imagem...")
try:
data = api_post("interpret", data={
"query": query,
"namespace": get_ns(context),
"top_k": 5,
"use_rerank": "false",
"temperature": 0.3,
}, files={"image": ("photo.jpg", buf, "image/jpeg")})
await msg.edit_text(data["answer"][:4000])
except Exception as e:
await msg.edit_text(f"❌ Erro: {e}")
async def handle_text(update: Update, context: ContextTypes.DEFAULT_TYPE):
"""Mensagem de texto sem comando → consulta Simple RAG."""
query = update.message.text.strip()
if not query:
return
msg = await update.message.reply_text("🔍 Consultando...")
try:
data = api_post("chat", data={
"query": query,
"namespace": get_ns(context),
"top_k": 5,
"use_agent": "false",
"use_rerank": "false",
"temperature": 0.5,
})
context.user_data["last_metrics"] = data.get("metrics")
await msg.edit_text(data["answer"][:4000])
except Exception as e:
await msg.edit_text(f"❌ Erro: {e}")
# ── Main ──
def main():
app = Application.builder().token(TELEGRAM_TOKEN).build()
app.add_handler(CommandHandler("start", cmd_start))
app.add_handler(CommandHandler("ns", cmd_ns))
app.add_handler(CommandHandler("ask", cmd_ask))
app.add_handler(CommandHandler("agent", cmd_agent))
app.add_handler(CommandHandler("ingest_url", cmd_ingest_url))
app.add_handler(CommandHandler("discover", cmd_discover))
app.add_handler(CommandHandler("ingest_discovered", cmd_ingest_discovered))
app.add_handler(CommandHandler("evaluate", cmd_evaluate))
app.add_handler(CommandHandler("interpret", cmd_interpret))
app.add_handler(CommandHandler("sources", cmd_sources))
app.add_handler(CommandHandler("del_source", cmd_del_source))
app.add_handler(CommandHandler("namespaces", cmd_namespaces))
app.add_handler(CommandHandler("prompts", cmd_prompts))
app.add_handler(CommandHandler("set_prompt", cmd_set_prompt))
app.add_handler(CommandHandler("kb_only", cmd_kb_only))
app.add_handler(CommandHandler("metrics", cmd_metrics))
app.add_handler(MessageHandler(filters.Document.ALL, handle_document))
app.add_handler(MessageHandler(filters.PHOTO, handle_photo))
app.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, handle_text))
log.info("Bot started")
app.run_polling()
if __name__ == "__main__":
main()| Comando | Descrição |
|---|---|
/start |
Menu de ajuda |
/ns <nome> |
Selecionar namespace |
/ask <pergunta> |
Simple RAG |
/agent <pergunta> |
Deep Agent + Reranking |
/ingest_url <url> |
Ingerir URL no namespace |
/discover <url> [depth] |
Explorar sub-páginas de um site |
/ingest_discovered [n1,n2,...] |
Ingerir URLs descobertas (todas ou por número) |
/evaluate + arquivo |
Avaliar documento contra KB |
/interpret <instrução> + foto |
Interpretar imagem com KB |
/sources |
Listar fontes do namespace |
/del_source <nome> |
Excluir fonte do namespace |
/namespaces |
Listar namespaces |
/prompts |
Listar configs de prompts |
/set_prompt <campo> <valor> |
Configurar prompt/guardrail |
/kb_only on|off |
Ativar/desativar modo KB-only |
/metrics |
Métricas da última resposta |
| Enviar arquivo | Auto-ingest no namespace |
| Enviar foto | Interpretar com KB |
| Texto livre | Consulta Simple RAG |
{
"answer": "Resposta gerada pelo agente...",
"sources": [
{
"content": "Preview do chunk...",
"metadata": { "source_name": "doc.pdf", "similarity_score": 0.87 },
"similarity_score": 0.87,
"rerank_score": 8.5,
"below_threshold": false
}
],
"namespace": "default",
"metrics": {
"recall_at_k": 0.8,
"mrr": 1.0,
"faithfulness": 0.95,
"answer_relevance": 0.88,
"citation_coverage": 0.6
},
"blocked": false
}