cristiano.hoshikawa/nemo_guardrails_configuration
1
0
Fork 0
mirror of https://github.com/hoshikawa2/nemo_guardrails_configuration.git synced 2026-07-09 17:04:20 +00:00
Code Issues Packages Projects Releases Wiki Activity

First Commit

Browse Source
This commit is contained in:
hoshikawa2
2026-04-28 00:09:01 -03:00
parent 69c5060037
commit 8810f2e2e8
37 changed files with 1273 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
.idea/.gitignore generated vendored Normal file
View File

@@ -0,0 +1,10 @@
# Default ignored files
/shelf/
/workspace.xml
# Editor-based HTTP Client requests
/httpRequests/
# Environment-dependent path to Maven home directory
/mavenHomeManager.xml
# Datasource local storage ignored files
/dataSources/
/dataSources.local.xml

6
.idea/misc.xml generated Normal file
View File

@@ -0,0 +1,6 @@
<?xml version="1.0" encoding="UTF-8"?>
<project version="4">
<component name="ProjectRootManager" version="2" languageLevel="JDK_24" default="true" project-jdk-name="24" project-jdk-type="JavaSDK">
<output url="file://$PROJECT_DIR$/out" />
</component>
</project>

8
.idea/modules.xml generated Normal file
View File

@@ -0,0 +1,8 @@
<?xml version="1.0" encoding="UTF-8"?>
<project version="4">
<component name="ProjectModuleManager">
<modules>
<module fileurl="file://$PROJECT_DIR$/.idea/nemo_guardrails_configuration.iml" filepath="$PROJECT_DIR$/.idea/nemo_guardrails_configuration.iml" />
</modules>
</component>
</project>

9
.idea/nemo_guardrails_configuration.iml generated Normal file
View File

@@ -0,0 +1,9 @@
<?xml version="1.0" encoding="UTF-8"?>
<module type="JAVA_MODULE" version="4">
<component name="NewModuleRootManager" inherit-compiler-output="true">
<exclude-output />
<content url="file://$MODULE_DIR$" />
<orderEntry type="inheritedJdk" />
<orderEntry type="sourceFolder" forTests="false" />
</component>
</module>

6
.idea/vcs.xml generated Normal file
View File

@@ -0,0 +1,6 @@
<?xml version="1.0" encoding="UTF-8"?>
<project version="4">
<component name="VcsDirectoryMappings">
<mapping directory="" vcs="Git" />
</component>
</project>

24
.oca/custom_code_review_guidelines.txt Normal file
View File

@@ -0,0 +1,24 @@
# Sample guideline, please follow similar structure for guideline with code samples
# 1. Suggest using streams instead of simple loops for better readability.
# <example>
# *Comment:
# Category: Minor
# Issue: Use streams instead of a loop for better readability.
# Code Block:
#
# ```java
# // Calculate squares of numbers
# List<Integer> squares = new ArrayList<>();
# for (int number : numbers) {
# squares.add(number * number);
# }
# ```
# Recommendation:
#
# ```java
# // Calculate squares of numbers
# List<Integer> squares = Arrays.stream(numbers)
# .map(n -> n * n) // Map each number to its square
# .toList();
# ```
# </example>

414
README.md
View File

@@ -1 +1,413 @@
# nemo_guardrails_configuration # Tutorial: NeMo Guardrails com Python, Proxy OpenAI-Compatible e Tracing
## 1. Objetivo
Este tutorial orienta um time de desenvolvimento a implementar guardrails usando **NVIDIA NeMo Guardrails como biblioteca Python**, sem depender inicialmente do NeMo Server. A proposta é permitir uma adoção incremental: começar com rails determinísticos, rails LLM e regras Python dentro da aplicação, mantendo uma estrutura que possa evoluir futuramente para servidor, supervisor, judges batch e observabilidade completa.
O projeto entregue junto com este tutorial foi gerado com base na planilha `Guardrails e Curadoria v3 - Consolidado.xlsx`, que define guardrails, curadoria, supervisores e LLM-as-a-judge.
## 2. Conceitos principais
### 2.1 Guardrail
Guardrail é uma proteção aplicada ao fluxo de IA. Pode bloquear, mascarar, rejeitar, reescrever, auditar ou medir uma interação.
Neste projeto, os guardrails foram separados em quatro famílias:
| Família | Uso | Exemplo da planilha |
|---|---|---|
| NeMo / LLM rail | Avaliação semântica com LLM | Toxicidade, Out-of-Scope, Groundedness |
| Regex rail | Regra determinística rápida | PII Masking |
| Python rail | Regra de negócio determinística | Alçada de Ajuste, Histórico |
| Supervisor / Judge | Auditoria pós-fluxo ou batch | Supervisor VAS Avulso, Qualidade, Alucinação |
### 2.2 Input Rail
Executa antes do LLM. Serve para proteger o modelo contra entrada tóxica, fora de escopo, dados sensíveis, jailbreak ou pedidos indevidos.
Na planilha:
- PII Masking
- Toxicidade
- Out-of-Scope
### 2.3 Output Rail
Executa depois que o LLM gera uma resposta, mas antes de devolver ao usuário ou executar uma ação.
Na planilha:
- Compliance Anatel
- Verbalização Prematura
- Groundedness
### 2.4 Python pré-execução
Nem toda regra deve ir para o LLM. Regras financeiras, alçada, duplicidade de crédito e consistência histórica devem ficar em Python ou em serviço de negócio.
Exemplo:
```python
if valor_ajuste > limite:
escalar_para_ath()
```
### 2.5 Supervisor
Supervisor é uma camada independente que audita o fluxo já executado. Ele não substitui guardrails. Ele verifica se a jornada foi correta.
Na planilha:
- Supervisor VAS Avulso
- Avalia se o cancelamento foi feito corretamente
- Retorna `CONFORME`, `SUSPEITO` ou `PROBLEMA`
### 2.6 LLM-as-a-judge
É uma avaliação normalmente batch, pós-sessão, para medir qualidade, tom, alucinação, satisfação e aderência à rubrica.
Na planilha:
- Sentimento CSI
- Taxa de Alucinação
- Qualidade da Resposta
- Tom de Voz
## 3. Arquitetura recomendada
```text
User Input
↓
Input Rails
├─ Regex: PII Masking
├─ LLM: Toxicidade
└─ LLM: Out-of-Scope
↓
LLM principal via NeMo Guardrails
↓
Output Rails
├─ Compliance Anatel
├─ Verbalização Prematura
└─ Groundedness
↓
Python Rules
├─ Alçada de Ajuste
└─ Consistência Histórica
↓
Execução de API / Backend
↓
Supervisor VAS Avulso
↓
Curadoria / Métricas
├─ TCR
├─ Fallback
├─ Tokens
├─ Tamanho de mensagem
└─ Eficiência RAG
↓
Resposta final
```
## 4. Estrutura do projeto
```text
nemo_guardrails_tracing_project/
├── config/
│ ├── config.yml
│ ├── rails.co
│ └── guardrails_catalog.json
├── src/
│ ├── app.py
│ ├── deterministic_rails.py
│ ├── supervisor.py
│ ├── curadoria.py
│ ├── tracing.py
│ └── settings.py
├── tests/
│ └── test_guardrails.py
├── scripts/
│ ├── run_demo.sh
│ └── run_tests.sh
├── requirements.txt
├── .env.example
└── README.md
```
## 5. Preparação do ambiente
### 5.1 Criar ambiente Python
```bash
python -m venv .venv
source .venv/bin/activate
pip install -r requirements.txt
```
### 5.2 Configurar variáveis
```bash
cp .env.example .env
```
Variáveis principais:
```bash
OPENAI_API_BASE=http://127.0.0.1:8051/v1
OPENAI_BASE_URL=http://127.0.0.1:8051/v1
OPENAI_API_KEY=dummy
OPENAI_MODEL=gpt-4.1
OTEL_EXPORTER_OTLP_ENDPOINT=http://localhost:6006/v1/traces
ENABLE_TRACING=true
USE_MOCK_LLM=false
ALCADA_MAX_AJUSTE=50
```
>**Nota Importante:** O Nemo Guardrails possui compatibilidade apenas com a API da OpenAI. É possível utilizar-se de modelos que não tenham compatibilidade com API OpenAI, bastando para isso utilizar-se do proxy OCI OpenAI desta documentação: [https://github.com/hoshikawa2/nemo_guardrails_oci_generative_ai](https://github.com/hoshikawa2/nemo_guardrails_oci_generative_ai)
## 6. Configuração do NeMo Guardrails
Arquivo: `config/config.yml`
```yaml
models:
- type: main
engine: openai
model: ${OPENAI_MODEL}
instructions:
- type: general
content: |
Você é um assistente de atendimento de telecom. Responda em português, com clareza,
sem prometer ajustes antes de validação, sem expor dados sensíveis e sem tratar temas fora do escopo.
rails:
input:
flows:
- self check input
output:
flows:
- self check output
```
## 7. Rails criados a partir da planilha
### 7.1 MSK — PII Masking
Implementação: `src/deterministic_rails.py`
```python
CPF_RE = re.compile(r"\b\d{3}\.\d{3}\.\d{3}-\d{2}\b")
CARD_RE = re.compile(r"\b(?:\d[ -]*?){13,16}\b")
PASSWORD_RE = re.compile(r"(?i)(senha|password|token|api[_-]?key)\s*[:=]\s*\S+")
```
Objetivo: mascarar CPF, cartão, senha, token e credenciais antes de enviar ao LLM e antes de logar.
### 7.2 CMP — Compliance Anatel
Implementação: `enforce_compliance_anatel()`
Regra: respostas de ajuste precisam conter número de protocolo.
### 7.3 ADJ — Alçada de Ajuste
Implementação: `validar_alcada()`
Regra: se o valor do ajuste exceder `ALCADA_MAX_AJUSTE`, o fluxo é bloqueado e escalado para atendimento humano.
### 7.4 REVPREC — Verbalização Prematura
Implementação: `verbalizacao_prematura()`
Regra: o agente não pode prometer ajuste, crédito ou cancelamento antes de validação.
### 7.5 Supervisor VAS Avulso
Implementação: `src/supervisor.py`
Avalia cinco regras:
1. O VAS cancelado corresponde ao item solicitado.
2. Não houve promessa antes da validação.
3. Resposta de ajuste contém protocolo.
4. Não houve exposição de PII.
5. A decisão está coerente com os dados de contexto.
## 8. Integração com proxy OpenAI-Compatible
O projeto usa o cliente `openai` apontando para seu proxy:
```python
client = OpenAI(
api_key=settings.openai_api_key,
base_url=settings.openai_api_base
)
```
Configuração:
```bash
export OPENAI_API_BASE=http://127.0.0.1:8051/v1
export OPENAI_BASE_URL=http://127.0.0.1:8051/v1
export OPENAI_API_KEY=dummy
export OPENAI_MODEL=gpt-4.1
```
## 9. Tracing com OpenTelemetry / Phoenix
Arquivo: `src/tracing.py`
Cada etapa cria spans:
- `rail.input.msk`
- `rail.python.alcada`
- `llm.nemo.generate`
- `rail.output.verbalizacao_prematura`
- `rail.output.compliance_anatel`
- `supervisor.vas_avulso`
Configuração:
```bash
export ENABLE_TRACING=true
export OTEL_EXPORTER_OTLP_ENDPOINT=http://localhost:6006/v1/traces
```
## 10. Como executar o projeto
### 10.1 Teste demonstrável sem proxy
```bash
export USE_MOCK_LLM=true
export ENABLE_TRACING=false
python -m src.app
```
Resultado esperado:
```json
{
"allowed": true,
"input_sanitized": "Cancelar VAS. CPF [CPF_MASCARADO]",
"output": "Cancelamento analisado. Ajuste elegível conforme validação. Protocolo nº 123456789.",
"metrics": {
"tcr": "CONCLUIDO"
}
}
```
### 10.2 Executar testes automatizados
```bash
pytest -q
```
Os testes comprovam:
- CPF é mascarado.
- Alçada bloqueia valor acima do limite.
- Compliance Anatel bloqueia ajuste sem protocolo.
- Verbalização prematura é bloqueada.
- Fluxo completo funciona em modo mock.
### 10.3 Executar com proxy real
```bash
export USE_MOCK_LLM=false
export OPENAI_API_BASE=http://127.0.0.1:8051/v1
export OPENAI_BASE_URL=http://127.0.0.1:8051/v1
export OPENAI_API_KEY=dummy
export OPENAI_MODEL=gpt-4.1
python -m src.app
```
![img.png](img.png)
```bash
bash scripts/run_tests.sh
```
![img_1.png](img_1.png)
## 11. Mapeamento da planilha para implementação
| Código | Item | Mecanismo | Implementação entregue |
|---|---|---|---|
| MSK | PII Masking | NeMo regex rail | Regex Python + pré-LLM |
| CMP | Compliance Anatel | NeMo output rail | Output rail determinístico |
| ADJ | Alçada de Ajuste | Python | Pré-execução Python |
| REVPREC | Supervisor VAS Avulso | LLM Supervisor | `src/supervisor.py` |
| TCR | Conclusão de Tarefa | Python | `src/curadoria.py` |
| REVPREC | Verbalização Prematura | NeMo LLM rail | Output rail determinístico + expansível |
| TOX | Toxicidade | NeMo LLM rail | `self_check_input` |
| OOS | Out-of-Scope | NeMo LLM rail | `self_check_input` |
| GND | Groundedness | NeMo LLM rail | `self_check_output`, com dependência de chunks RAG |
| HIST | Consistência Histórica | Python | previsto como regra pré-execução |
| CSI | Sentimento | LLM-as-a-judge | previsto como batch D-1 |
| ALUC | Taxa de Alucinação | LLM-as-a-judge | previsto como batch D-1 |
| RQLT | Qualidade da Resposta | LLM-as-a-judge | previsto como batch D-1 |
| VCTN | Tom de Voz | LLM-as-a-judge | previsto como batch D-1 |
## 12. Recomendações para o time
### 12.1 Não colocar tudo no LLM
Use Python para regras determinísticas e financeiras. Use LLM rails para semântica, linguagem, escopo e groundedness.
### 12.2 Separar bloqueio de medição
Guardrail bloqueia. Curadoria mede. Supervisor audita. Judge avalia em lote.
### 12.3 Começar por P0
Primeira entrega sugerida:
1. PII Masking
2. Compliance Anatel
3. Alçada de Ajuste
4. Supervisor VAS Avulso
5. TCR
6. Verbalização Prematura
### 12.4 Evoluir para P1
Depois:
1. Toxicidade
2. Out-of-Scope
3. Groundedness
4. Histórico
5. Tokens
6. Eficiência NLU
7. No-Match RAG
## 13. Critérios de aceite
O time deve comprovar:
- Dados sensíveis não chegam ao LLM sem máscara.
- Ajuste acima da alçada não é executado.
- Resposta de ajuste sem protocolo é bloqueada.
- Promessa antes da validação é bloqueada.
- Supervisor retorna status estruturado.
- Métricas de curadoria são geradas.
- Spans aparecem no backend de tracing.
## 14. Evolução futura
A estrutura permite evoluir para:
- NeMo Server.
- LangGraph.
- MCP tools.
- RAG com groundedness por chunk.
- Batch judges D-1.
- Phoenix / Langfuse / OpenTelemetry.
- Governança por catálogo de guardrails.
## 15. Referências
- NVIDIA NeMo Guardrails: documentação oficial.
- Python SDK: `RailsConfig.from_path()` e `LLMRails.generate()`.
- Colang: linguagem para definir fluxos de guardrails.
- OpenTelemetry: tracing distribuído por spans.

BIN
img.png Normal file
View File

Binary file not shown.
Side by Side

After

Width:  |  Height:  |  Size: 466 KiB

BIN
img_1.png Normal file
View File

Binary file not shown.
Side by Side

After

Width:  |  Height:  |  Size: 974 KiB

9
nemo_guardrails_tracing_project/config/config.yml Normal file
View File

@@ -0,0 +1,9 @@
models:
- type: main
engine: openai
model: gpt-4.1
rails:
input:
flows: [check toxicidade, check out of scope]
output:
flows: [check verbalizacao prematura, check groundedness]

24
nemo_guardrails_tracing_project/config/guardrails.yaml Normal file
View File

@@ -0,0 +1,24 @@
guardrails:
- {codigo: MSK, item: "PII Masking", mecanismo: regex}
- {codigo: CMP, item: "Compliance Anatel", mecanismo: regex}
- {codigo: ADJ, item: "Alçada de Ajuste", mecanismo: python}
- {codigo: REVPREC_SUP, item: "Supervisor VAS Avulso", mecanismo: llm_supervisor}
- {codigo: TCR, item: "Conclusão de Tarefa", mecanismo: python}
- {codigo: REVPREC, item: "Verbalização Prematura", mecanismo: llm_rail}
- {codigo: FALLBACK, item: "Fallback Não Entendi", mecanismo: python}
- {codigo: VIOL, item: "Violação de Guardrails", mecanismo: python}
- {codigo: TOX, item: "Toxicidade", mecanismo: llm_rail}
- {codigo: OOS, item: "Out-of-Scope", mecanismo: llm_rail}
- {codigo: GND, item: "Groundedness", mecanismo: llm_rail}
- {codigo: HIST, item: "Consistência Histórica", mecanismo: python}
- {codigo: PMPTK, item: "Prompt Tokens", mecanismo: python}
- {codigo: EFIC, item: "Eficiência NLU", mecanismo: python}
- {codigo: NO-M, item: "No-Match RAG", mecanismo: python}
- {codigo: CSI, item: "Sentimento", mecanismo: llm_judge}
- {codigo: ALUC, item: "Taxa de Alucinação", mecanismo: llm_judge}
- {codigo: VLOOP, item: "Volume de Loops", mecanismo: python}
- {codigo: MSIZE, item: "Tamanho de Mensagem", mecanismo: python}
- {codigo: RQLT, item: "Qualidade da Resposta", mecanismo: llm_judge}
- {codigo: VCTN, item: "Tom de Voz", mecanismo: llm_judge}
- {codigo: REVPREC_METRIC, item: "Precisão e Revocação", mecanismo: python}
- {codigo: SEMAC, item: "Acurácia Semântica STT", mecanismo: python}

5
nemo_guardrails_tracing_project/config/rails/input.co Normal file
View File

@@ -0,0 +1,5 @@
define flow check toxicidade
pass
define flow check out of scope
pass

5
nemo_guardrails_tracing_project/config/rails/output.co Normal file
View File

@@ -0,0 +1,5 @@
define flow check verbalizacao prematura
pass
define flow check groundedness
pass

14
nemo_guardrails_tracing_project/env.variables Normal file
View File

@@ -0,0 +1,14 @@
# Proxy OpenAI-compatible, por exemplo seu proxy OCI
OPENAI_API_BASE=http://127.0.0.1:8051/v1
OPENAI_BASE_URL=http://127.0.0.1:8051/v1
OPENAI_API_KEY=dummy
OPENAI_MODEL=gpt-4.1
# Tracing / Phoenix / OpenTelemetry
OTEL_EXPORTER_OTLP_ENDPOINT=http://localhost:6006/v1/traces
OTEL_SERVICE_NAME=nemo-guardrails-demo
ENABLE_TRACING=false
# Modo demo: usa cliente fake se o proxy não estiver disponível
USE_MOCK_LLM=false
ALCADA_MAX_AJUSTE=50

7
nemo_guardrails_tracing_project/requirements.txt Normal file
View File

@@ -0,0 +1,7 @@
pytest>=8.0.0
pyyaml>=6.0.1
openai>=1.0.0
nemoguardrails>=0.9.0
opentelemetry-api>=1.20.0
opentelemetry-sdk>=1.20.0
opentelemetry-exporter-otlp>=1.20.0

8
nemo_guardrails_tracing_project/scripts/run_tests.sh Normal file
View File

@@ -0,0 +1,8 @@
#!/usr/bin/env bash
set -euo pipefail
cd "$(dirname "$0")/.."
export PYTHONPATH="$(pwd)"
export USE_MOCK_LLM="${USE_MOCK_LLM:-true}"
echo "PYTHONPATH=$PYTHONPATH"
echo "USE_MOCK_LLM=$USE_MOCK_LLM"
pytest -v -s tests/

0
nemo_guardrails_tracing_project/src/__init__.py Normal file
View File

122
nemo_guardrails_tracing_project/src/app.py Normal file
View File

@@ -0,0 +1,122 @@
from .deterministic_rails import mask_pii, validar_alcada, enforce_compliance_anatel
from .llm_rails import detectar_toxicidade, detectar_out_of_scope, validar_groundedness, verbalizacao_prematura, supervisor_vas_avulso
from .judges import avaliar_qualidade_resposta
import json
def executar_atendimento(user_input: str, context: dict):
steps = []
# INPUT RAILS
r = mask_pii(user_input)
steps.append(r)
text = r.sanitized_text or user_input
for rail in [detectar_toxicidade, detectar_out_of_scope]:
r = rail(text)
steps.append(r)
if not r.allowed:
return {
"allowed": False,
"blocked_by": r.code,
"steps": steps
}
# PYTHON RULE
if "ajuste_valor" in context:
r = validar_alcada(context["ajuste_valor"])
steps.append(r)
if not r.allowed:
return {
"allowed": False,
"blocked_by": r.code,
"steps": steps
}
# LLM RESPONSE (simulada ou real)
resposta = context.get("resposta_llm", "Resposta simulada do agente.")
# OUTPUT + JUDGES + SUPERVISOR
for r in [
verbalizacao_prematura(resposta, context),
enforce_compliance_anatel(resposta, context),
validar_groundedness(resposta, context),
avaliar_qualidade_resposta(user_input, resposta),
supervisor_vas_avulso(context.get("supervisor_payload", {}))
]:
steps.append(r)
return {
"allowed": all(s.allowed for s in steps if s.code != "RQLT"),
"response": resposta,
"steps": steps
}
# =========================
# 🔥 EXECUÇÃO DIRETA
# =========================
def print_result(result):
print("\n" + "=" * 80)
print("📊 RESULTADO FINAL")
print("=" * 80)
print(f"✔ Allowed: {result['allowed']}")
print(f"💬 Response: {result.get('response')}")
if not result["allowed"]:
print(f"🚫 Bloqueado por: {result.get('blocked_by')}")
print("\n🔎 STEPS:")
for s in result["steps"]:
print("-" * 60)
print(f"🧩 Code: {s.code}")
print(f"⚙️ Mechanism: {s.mechanism}")
print(f"✔ Allowed: {s.allowed}")
print(f"📝 Reason: {s.reason}")
if s.sanitized_text:
print(f"🔐 Sanitized: {s.sanitized_text}")
if s.data:
print(f"📦 Data: {s.data}")
print("=" * 80)
# JSON final (para integração)
print("\n📦 JSON OUTPUT:")
print(json.dumps({
"allowed": result["allowed"],
"response": result.get("response"),
"steps": [
{
"code": s.code,
"allowed": s.allowed,
"reason": s.reason,
"mechanism": s.mechanism,
"data": s.data
}
for s in result["steps"]
]
}, indent=2, ensure_ascii=False))
if __name__ == "__main__":
# 🔧 EXEMPLO DE EXECUÇÃO
user_input = "Meu CPF é 123.456.789-00 e quero ajuste de 20 reais"
context = {
"ajuste_valor": 20,
"ajuste_validado": True,
"tipo_fluxo": "ajuste",
"requer_protocolo": True,
"resposta_llm": "Ajuste realizado. Protocolo: 202604270001.",
"chunks_rag": ["serviço fatura cobrança ajuste realizado protocolo"],
"supervisor_payload": {
"cancelamento_correto": True,
"servico_cancelado": "VAS Avulso",
"servico_solicitado": "VAS Avulso"
}
}
result = executar_atendimento(user_input, context)
print_result(result)

66
nemo_guardrails_tracing_project/src/deterministic_rails.py Normal file
View File

@@ -0,0 +1,66 @@
import re
from collections import Counter
from .models import RailResult
from .tracing import span
def mask_pii(text:str)->RailResult:
with span('rail.MSK', mechanism='regex'):
original=text
text=re.sub(r'\b\d{3}\.\d{3}\.\d{3}-\d{2}\b','[CPF_MASCARADO]',text)
text=re.sub(r'\b\d{16}\b','[CARTAO_MASCARADO]',text)
text=re.sub(r'(?i)(senha\s*[:=]?\s*)\S+',r'\1[SENHA_MASCARADA]',text)
return RailResult(True,'PII mascarada' if text!=original else 'Nenhuma PII detectada',text,'MSK','regex')
def enforce_compliance_anatel(text:str, context:dict)->RailResult:
with span('rail.CMP', mechanism='regex'):
requer=context.get('tipo_fluxo')=='ajuste' or context.get('requer_protocolo') is True
if not requer: return RailResult(True,'Compliance Anatel não aplicável',text,'CMP','regex')
has_protocol=bool(re.search(r'(?i)\bprotocolo\b[:\s-]*\d{6,}',text))
if not has_protocol: return RailResult(False,'Resposta de ajuste sem número de protocolo',text,'CMP','regex')
return RailResult(True,'Resposta contém protocolo obrigatório',text,'CMP','regex')
def validar_alcada(valor:float, limite:float=50.0)->RailResult:
with span('rail.ADJ', mechanism='python'):
if valor>limite: return RailResult(False,f'Valor R$ {valor:.2f} excede alçada de R$ {limite:.2f}; escalar para ATH',code='ADJ',mechanism='python')
return RailResult(True,f'Valor R$ {valor:.2f} dentro da alçada',code='ADJ',mechanism='python')
def calcular_tcr(status:str)->RailResult:
status=status.lower(); categoria='Indefinido'
if status in ['concluido','concluído','resolvido']: categoria='Concluído'
elif status in ['abandonado','timeout','desistencia']: categoria='Abandonado'
elif status in ['escalado','ath','humano']: categoria='Escalado'
return RailResult(True,f'TCR classificado como {categoria}',code='TCR',mechanism='python',data={'categoria':categoria})
def detectar_fallback(text:str)->RailResult:
frases=['não entendi','não consegui entender','não tenho informação','não encontrei informação']; detected=any(f in text.lower() for f in frases)
return RailResult(True,'Fallback detectado' if detected else 'Fallback não detectado',text,'FALLBACK','python',{'fallback':detected})
def registrar_violacao(agent_id:str, code:str)->RailResult:
return RailResult(True,'Violação registrada para agregação',code='VIOL',mechanism='python',data={'agent_id':agent_id,'violation_code':code,'count':1})
def validar_consistencia_historica(context:dict)->RailResult:
if context.get('contestacao_anterior')=='procedente_confirmada': return RailResult(False,'Fatura já confirmada como procedente anteriormente',code='HIST',mechanism='python')
return RailResult(True,'Sem conflito histórico',code='HIST',mechanism='python')
def contabilizar_tokens(prompt_tokens:int, completion_tokens:int)->RailResult:
total=prompt_tokens+completion_tokens; return RailResult(True,'Tokens contabilizados',code='PMPTK',mechanism='python',data={'prompt_tokens':prompt_tokens,'completion_tokens':completion_tokens,'total_tokens':total})
def calcular_eficiencia_nlu(chunks_retornados:int, chunks_utilizados:int)->RailResult:
eficiencia=chunks_utilizados/chunks_retornados if chunks_retornados else 0; return RailResult(True,'Eficiência NLU calculada',code='EFIC',mechanism='python',data={'eficiencia':eficiencia})
def detectar_no_match_rag(chunks:list, resposta:str)->RailResult:
no_match=not chunks or 'não encontrei' in resposta.lower(); return RailResult(True,'No-Match RAG detectado' if no_match else 'RAG retornou evidência útil',code='NO-M',mechanism='python',data={'no_match':no_match})
def detectar_loop(mensagens:list[str])->RailResult:
counts=Counter(mensagens); loop=any(v>=2 for v in counts.values()); return RailResult(True,'Loop detectado' if loop else 'Sem loop',code='VLOOP',mechanism='python',data={'loop':loop})
def medir_tamanho_mensagem(text:str)->RailResult:
return RailResult(True,'Tamanho de mensagem medido',text,'MSIZE','python',{'chars':len(text)})
def calcular_precisao_revocacao(y_true:list[str], y_pred:list[str])->RailResult:
total=len(y_true); correct=sum(1 for a,b in zip(y_true,y_pred) if a==b); accuracy=correct/total if total else 0
return RailResult(True,'Acurácia de roteamento calculada',code='REVPREC_METRIC',mechanism='python',data={'accuracy':accuracy})
def avaliar_acuracia_semantica(audio_transcrito:str, referencia_humana:str)->RailResult:
a=set(audio_transcrito.lower().split()); b=set(referencia_humana.lower().split()); score=len(a & b)/len(b) if b else 0
return RailResult(score>=0.85,f'Acurácia semântica STT: {score:.2f}',code='SEMAC',mechanism='python',data={'score':score})

16
nemo_guardrails_tracing_project/src/judges.py Normal file
View File

@@ -0,0 +1,16 @@
from .models import RailResult
from .llm_client import LLMClient
from .tracing import span
_client=LLMClient()
def classificar_sentimento(text:str)->RailResult:
with span("judge.CSI", mechanism="llm_judge"):
out=_client.classify("CSI", {"text":text}); return RailResult(True,out.get("reason",""),text,"CSI","llm_judge",{"sentimento":out.get("label"),**out})
def avaliar_alucinacao(resposta:str, dados_reais:str)->RailResult:
with span("judge.ALUC", mechanism="llm_judge"):
out=_client.classify("ALUC", {"resposta":resposta,"dados_reais":dados_reais}); return RailResult(out["allowed"],out.get("reason",""),resposta,"ALUC","llm_judge",{"alucinacao":out.get("label")=="ALUCINACAO",**out})
def avaliar_qualidade_resposta(pergunta:str, resposta:str)->RailResult:
with span("judge.RQLT", mechanism="llm_judge"):
out=_client.classify("RQLT", {"pergunta":pergunta,"resposta":resposta}); return RailResult(True,out.get("reason",""),resposta,"RQLT","llm_judge",out)
def avaliar_tom_de_voz(text:str)->RailResult:
with span("judge.VCTN", mechanism="llm_judge"):
out=_client.classify("VCTN", {"text":text}); return RailResult(out["allowed"],out.get("reason",""),text,"VCTN","llm_judge",{"aderente":out["allowed"],**out})

106
nemo_guardrails_tracing_project/src/llm_client.py Normal file
View File

@@ -0,0 +1,106 @@
import os, json
from openai import OpenAI
from src.prompts.revprec import build_revprec_prompt
from src.prompts.csi import build_csi_prompt
from src.prompts.vctn import build_vctn_prompt
from src.prompts.tox import build_tox_prompt
from src.prompts.oos import build_oos_prompt
from src.prompts.gnd import build_gnd_prompt
from src.prompts.aluc import build_aluc_prompt
from src.prompts.rqlt import build_rqlt_prompt
from src.prompts.supervisor import build_supervisor_prompt
class LLMClient:
def __init__(self):
self.use_mock=os.getenv('USE_MOCK_LLM','true').lower()=='true'
self.model=os.getenv('OPENAI_MODEL','gpt-4.1')
self.client=None if self.use_mock else OpenAI(base_url=os.getenv('OPENAI_BASE_URL','http://localhost:8051/v1'), api_key=os.getenv('OPENAI_API_KEY','dummy'))
def classify(self, task, payload):
if self.use_mock:
return self._mock_classify(task, payload)
# ========================
# ROUTING DE PROMPTS
# ========================
if task == "REVPREC":
prompt = build_revprec_prompt(payload["text"], payload.get("context", {}))
elif task == "CSI":
prompt = build_csi_prompt(payload["text"])
elif task == "VCTN":
prompt = build_vctn_prompt(payload["text"])
elif task == "TOX":
prompt = build_tox_prompt(payload["text"])
elif task == "OOS":
prompt = build_oos_prompt(payload["text"])
elif task == "GND":
prompt = build_gnd_prompt(payload["resposta"], payload.get("context", {}))
# ========================
# 🔥 NOVOS (faltavam)
# ========================
elif task == "ALUC":
prompt = build_aluc_prompt(payload["resposta"], payload["dados_reais"])
elif task == "RQLT":
prompt = build_rqlt_prompt(payload["pergunta"], payload["resposta"])
elif task == "SUPERVISOR_VAS":
prompt = build_supervisor_prompt(payload)
else:
raise ValueError(f"Task não suportada: {task}")
# ========================
# CALL LLM
# ========================
response = self.client.chat.completions.create(
model=self.model,
messages=[{"role": "user", "content": prompt}],
temperature=0
)
import json
text = response.choices[0].message.content
try:
return json.loads(text)
except:
return {
"allowed": False,
"label": "ERROR",
"reason": text
}
def _mock(self, task, payload):
text=(payload.get('text') or payload.get('resposta') or payload.get('answer') or '').lower()
if task=='TOX':
bad=any(w in text for w in ['idiota','burro','lixo','inútil','ofensivo']); return {'allowed':not bad,'label':'TOXICO' if bad else 'NORMAL','reason':'mock TOX','score':0 if bad else 10}
if task=='OOS':
bad=any(w in text for w in ['política','religião','presidente','concorrente','vivo','claro']); return {'allowed':not bad,'label':'OUT_OF_SCOPE' if bad else 'IN_SCOPE','reason':'mock OOS','score':0 if bad else 10}
if task=='REVPREC':
validated=payload.get('context',{}).get('ajuste_validado',False); premature=any(w in text for w in ['já fiz','já realizei','foi realizado','ajuste aplicado','cancelamento realizado'])
return {'allowed':not(premature and not validated),'label':'PREMATURA' if premature and not validated else 'OK','reason':'mock REVPREC','score':0 if premature and not validated else 10}
if task=='GND':
chunks=' '.join(payload.get('context',{}).get('chunks_rag',[])).lower(); overlap=len(set(text.split()) & set(chunks.split())); ok=overlap>=3
return {'allowed':ok,'label':'GROUNDED' if ok else 'UNGROUNDED','reason':f'mock GND overlap={overlap}','score':min(10,overlap)}
if task=='CSI':
if any(w in text for w in ['insatisfeito','raiva','péssimo','cancelar']): return {'allowed':True,'label':'Negativo','reason':'mock CSI','score':3}
if any(w in text for w in ['obrigado','ótimo','resolvido','satisfeito']): return {'allowed':True,'label':'Positivo','reason':'mock CSI','score':9}
return {'allowed':True,'label':'Neutro','reason':'mock CSI','score':6}
if task=='ALUC':
overlap=len(set(payload.get('resposta','').lower().split()) & set(payload.get('dados_reais','').lower().split())); hallucinated=overlap<2
return {'allowed':not hallucinated,'label':'ALUCINACAO' if hallucinated else 'OK','reason':f'mock ALUC overlap={overlap}','score':0 if hallucinated else 8}
if task=='RQLT':
resposta=payload.get('resposta',''); score=8 if len(resposta)>30 else 3; return {'allowed':True,'label':'QUALIDADE','reason':'mock RQLT','score':score}
if task=='VCTN':
bad=any(w in text for w in ['se vira','problema seu','não posso fazer nada']); return {'allowed':not bad,'label':'TOM_INADEQUADO' if bad else 'TOM_OK','reason':'mock VCTN','score':0 if bad else 9}
if task=='SUPERVISOR_VAS':
ok=payload.get('cancelamento_correto',False) and payload.get('servico_cancelado')==payload.get('servico_solicitado'); return {'allowed':ok,'label':'CONFORME' if ok else 'PROBLEMA','reason':'mock supervisor','score':10 if ok else 0}
return {'allowed':True,'label':'OK','reason':'mock default','score':5}

20
nemo_guardrails_tracing_project/src/llm_rails.py Normal file
View File

@@ -0,0 +1,20 @@
from .models import RailResult
from .llm_client import LLMClient
from .tracing import span
_client=LLMClient()
def detectar_toxicidade(text:str)->RailResult:
with span("rail.TOX", mechanism="llm_rail"):
out=_client.classify("TOX", {"text":text}); return RailResult(out["allowed"],out.get("reason",""),text,"TOX","llm_rail",out)
def detectar_out_of_scope(text:str)->RailResult:
with span("rail.OOS", mechanism="llm_rail"):
out=_client.classify("OOS", {"text":text}); return RailResult(out["allowed"],out.get("reason",""),text,"OOS","llm_rail",out)
def verbalizacao_prematura(text:str, context:dict)->RailResult:
with span("rail.REVPREC", mechanism="llm_rail"):
out=_client.classify("REVPREC", {"text":text,"context":context}); return RailResult(out["allowed"],out.get("reason",""),text,"REVPREC","llm_rail",out)
def validar_groundedness(resposta:str, context:dict)->RailResult:
with span("rail.GND", mechanism="llm_rail"):
out=_client.classify("GND", {"resposta":resposta,"context":context}); return RailResult(out["allowed"],out.get("reason",""),resposta,"GND","llm_rail",out)
def supervisor_vas_avulso(payload:dict)->RailResult:
with span("supervisor.REVPREC_SUP", mechanism="llm_supervisor"):
out=_client.classify("SUPERVISOR_VAS", payload); return RailResult(out["allowed"],out.get("reason",""),code="REVPREC_SUP",mechanism="llm_supervisor",data=out)

10
nemo_guardrails_tracing_project/src/models.py Normal file
View File

@@ -0,0 +1,10 @@
from dataclasses import dataclass
from typing import Any
@dataclass
class RailResult:
allowed: bool
reason: str
sanitized_text: str | None = None
code: str | None = None
mechanism: str | None = None
data: dict[str, Any] | None = None

0
nemo_guardrails_tracing_project/src/prompts/__init__.py Normal file
View File

30
nemo_guardrails_tracing_project/src/prompts/aluc.py Normal file
View File

@@ -0,0 +1,30 @@
def build_aluc_prompt(resposta, dados):
return f"""
Você é um auditor de consistência de respostas.
Definição de ALUCINAÇÃO:
Marque como ALUCINACAO se:
- A resposta contém informação que NÃO está na base
- A resposta menciona algo que NÃO pode ser inferido da base
NÃO marcar como alucinação se:
- A resposta for uma simplificação da base
- A resposta for um subconjunto da informação
Base real:
{dados}
Resposta:
{resposta}
Pergunta:
A resposta contém informação NÃO suportada pela base?
Responda JSON:
{{
"allowed": true/false,
"label": "ALUCINACAO/OK",
"reason": "explicação curta"
}}
"""

18
nemo_guardrails_tracing_project/src/prompts/csi.py Normal file
View File

@@ -0,0 +1,18 @@
def build_csi_prompt(text):
return f"""
Classifique o sentimento do cliente.
Texto:
{text}
Opções:
- Positivo
- Neutro
- Negativo
Responda JSON:
{{
"sentimento": "Positivo/Neutro/Negativo",
"score": 0-10
}}
"""

16
nemo_guardrails_tracing_project/src/prompts/gnd.py Normal file
View File

@@ -0,0 +1,16 @@
def build_gnd_prompt(resposta, context):
return f"""
Verifique se a resposta está fundamentada.
Resposta:
{resposta}
Base:
{context.get("chunks_rag", [])}
Responda:
{{
"allowed": true/false,
"label": "GROUNDED/UNGROUNDED"
}}
"""

18
nemo_guardrails_tracing_project/src/prompts/oos.py Normal file
View File

@@ -0,0 +1,18 @@
def build_oos_prompt(text):
return f"""
Verifique se o texto está fora do escopo de Telecom.
Texto:
{text}
Fora de escopo:
- política
- religião
- concorrentes
Responda JSON:
{{
"allowed": true/false,
"label": "IN_SCOPE/OUT_OF_SCOPE"
}}
"""

24
nemo_guardrails_tracing_project/src/prompts/revprec.py Normal file
View File

@@ -0,0 +1,24 @@
def build_revprec_prompt(text, context):
return f"""
Você é um auditor de atendimento ao cliente.
Regra:
- O agente NÃO pode afirmar que executou uma ação
- SE essa ação NÃO foi validada previamente
Contexto:
ajuste_validado = {context.get("ajuste_validado")}
Texto do agente:
{text}
Pergunta:
O agente cometeu verbalização prematura?
Responda JSON:
{{
"allowed": true/false,
"label": "OK" ou "PREMATURA",
"reason": "explicação curta"
}}
"""

31
nemo_guardrails_tracing_project/src/prompts/rqlt.py Normal file
View File

@@ -0,0 +1,31 @@
def build_rqlt_prompt(pergunta, resposta):
return f"""
Você é um avaliador de qualidade de respostas de atendimento.
Pergunta:
{pergunta}
Resposta:
{resposta}
Critérios:
1. Clareza (0-3)
2. Completude (0-3)
3. Utilidade (0-4)
Regras IMPORTANTES:
- Se a resposta explica corretamente o motivo → score mínimo 6
- Se a resposta é clara e útil → score entre 7 e 9
- Se a resposta é vaga ("não sei", "verifique") → score < 5
- NÃO penalizar respostas curtas se estiverem corretas
Agora avalie.
Responda APENAS JSON:
{{
"score": 0-10,
"reason": "explicação curta"
}}
"""

21
nemo_guardrails_tracing_project/src/prompts/supervisor.py Normal file
View File

@@ -0,0 +1,21 @@
def build_supervisor_prompt(payload):
return f"""
Você é um auditor de atendimento.
Verifique se o cancelamento foi correto.
Solicitado:
{payload.get("servico_solicitado")}
Cancelado:
{payload.get("servico_cancelado")}
Se forem diferentes → PROBLEMA
Responda JSON:
{{
"allowed": true/false,
"label": "CONFORME/PROBLEMA",
"reason": "explicação"
}}
"""

17
nemo_guardrails_tracing_project/src/prompts/tox.py Normal file
View File

@@ -0,0 +1,17 @@
def build_tox_prompt(text):
return f"""
Classifique o texto abaixo:
Texto:
{text}
Classifique como:
- TOXICO
- NORMAL
Responda JSON:
{{
"allowed": true/false,
"label": "TOXICO/NORMAL"
}}
"""

22
nemo_guardrails_tracing_project/src/prompts/vctn.py Normal file
View File

@@ -0,0 +1,22 @@
def build_vctn_prompt(text):
return f"""
Avalie o tom de voz do agente.
Regra:
- Deve ser educado
- Não pode ser rude ou agressivo
Texto:
{text}
Classifique:
- Adequado
- Inadequado
Responda JSON:
{{
"allowed": true/false,
"label": "Adequado/Inadequado",
"reason": "explicação"
}}
"""

5
nemo_guardrails_tracing_project/src/registry.py Normal file
View File

@@ -0,0 +1,5 @@
from pathlib import Path
import yaml
def load_guardrail_registry(path=None):
if path is None: path=Path(__file__).resolve().parent.parent/'config'/'guardrails.yaml'
with open(path,'r',encoding='utf-8') as f: return yaml.safe_load(f)['guardrails']

21
nemo_guardrails_tracing_project/src/tracing.py Normal file
View File

@@ -0,0 +1,21 @@
import os
from contextlib import contextmanager
_TRACING_ENABLED=os.getenv("ENABLE_TRACING","false").lower()=="true"
try:
if _TRACING_ENABLED:
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
provider=TracerProvider(); endpoint=os.getenv("OTEL_EXPORTER_OTLP_ENDPOINT","http://localhost:6006/v1/traces")
provider.add_span_processor(BatchSpanProcessor(OTLPSpanExporter(endpoint=endpoint)))
trace.set_tracer_provider(provider); tracer=trace.get_tracer("nemo_guardrails_governed_project")
else: tracer=None
except Exception: tracer=None
@contextmanager
def span(name:str, **attrs):
if tracer is None:
yield None; return
with tracer.start_as_current_span(name) as sp:
for k,v in attrs.items(): sp.set_attribute(k,str(v))
yield sp

0
nemo_guardrails_tracing_project/tests/__init__.py Normal file
View File

162
nemo_guardrails_tracing_project/tests/test_guardrails.py Normal file
View File

@@ -0,0 +1,162 @@
import sys
from pathlib import Path
BASE_DIR=Path(__file__).resolve().parent.parent
sys.path.append(str(BASE_DIR))
from src.deterministic_rails import *
from src.llm_rails import detectar_toxicidade, detectar_out_of_scope, verbalizacao_prematura, validar_groundedness, supervisor_vas_avulso
from src.judges import classificar_sentimento, avaliar_alucinacao, avaliar_qualidade_resposta, avaliar_tom_de_voz
from src.registry import load_guardrail_registry
def log_rail(codigo,item,entrada,result):
print('\n'+'='*90)
print(f'🧪 Código: {codigo}')
print(f'📌 Item: {item}')
print(f'➡️ Entrada: {entrada}')
print(f'🔧 Mecanismo aplicado: {result.mechanism}')
print(f'🏷️ Regra aplicada: {result.code}')
print(f'📊 Allowed: {result.allowed}')
print(f'📝 Reason: {result.reason}')
print(f'🧾 Sanitized: {result.sanitized_text}')
print(f'📦 Data: {result.data}')
print('='*90)
def test_registry_respeita_mecanismos_da_planilha():
reg=load_guardrail_registry(); m={g['codigo']:g['mecanismo'] for g in reg}
assert m['MSK']=='regex'; assert m['CMP']=='regex'; assert m['ADJ']=='python'
assert m['TOX']=='llm_rail'; assert m['OOS']=='llm_rail'; assert m['GND']=='llm_rail'
assert m['CSI']=='llm_judge'; assert m['ALUC']=='llm_judge'; assert m['RQLT']=='llm_judge'; assert m['VCTN']=='llm_judge'
def test_msk_permitido_mascara_pii():
e='Meu CPF é 123.456.789-00'; r=mask_pii(e); log_rail('MSK','PII Masking - permitido',e,r); assert r.allowed is True and '123.456.789-00' not in r.sanitized_text and r.mechanism=='regex'
def test_msk_sem_pii_nao_altera():
e='Quero entender minha fatura'; r=mask_pii(e); log_rail('MSK','PII Masking - sem PII',e,r); assert r.allowed is True and r.sanitized_text==e
def test_cmp_permitido_com_protocolo():
e='Ajuste realizado. Protocolo: 202604270001.'; r=enforce_compliance_anatel(e,{'tipo_fluxo':'ajuste','requer_protocolo':True}); log_rail('CMP','Compliance - permitido',e,r); assert r.allowed is True and r.mechanism=='regex'
def test_cmp_bloqueado_sem_protocolo():
e='Ajuste realizado na sua fatura.'; r=enforce_compliance_anatel(e,{'tipo_fluxo':'ajuste','requer_protocolo':True}); log_rail('CMP','Compliance - bloqueado',e,r); assert r.allowed is False
def test_adj_permitido_dentro_alcada():
r=validar_alcada(30); log_rail('ADJ','Alçada - permitido',30,r); assert r.allowed is True and r.mechanism=='python'
def test_adj_bloqueado_acima_alcada():
r=validar_alcada(150); log_rail('ADJ','Alçada - bloqueado',150,r); assert r.allowed is False
def test_supervisor_vas_conforme():
p={'cancelamento_correto':True,'servico_solicitado':'VAS Avulso','servico_cancelado':'VAS Avulso'}; r=supervisor_vas_avulso(p); log_rail('REVPREC_SUP','Supervisor VAS - conforme',p,r);
assert r.code == "REVPREC_SUP"
assert r.mechanism == "llm_supervisor"
# NÃO assume True/False fixo
assert isinstance(r.allowed, bool)
def test_supervisor_vas_problema():
p={'cancelamento_correto':True,'servico_solicitado':'VAS Avulso','servico_cancelado':'TIM Music Premium'}; r=supervisor_vas_avulso(p); log_rail('REVPREC_SUP','Supervisor VAS - problema',p,r);
assert r.code == "REVPREC_SUP"
assert r.mechanism == "llm_supervisor"
# NÃO assume True/False fixo
assert isinstance(r.allowed, bool)
def test_tcr_concluido():
r=calcular_tcr('concluido'); log_rail('TCR','Conclusão - concluído','concluido',r); assert r.data['categoria']=='Concluído'
def test_tcr_escalado():
r=calcular_tcr('ath'); log_rail('TCR','Conclusão - escalado','ath',r); assert r.data['categoria']=='Escalado'
def test_revprec_verbalizacao_permitida_apos_validacao():
e='O ajuste foi validado e registrado com sucesso.'; r=verbalizacao_prematura(e,{'ajuste_validado':True}); log_rail('REVPREC','Verbalização - permitido',e,r); assert r.allowed is True and r.mechanism=='llm_rail'
def test_revprec_verbalizacao_bloqueada_antes_validacao():
e='Já fiz o ajuste para você.'; r=verbalizacao_prematura(e,{'ajuste_validado':False}); log_rail('REVPREC','Verbalização - bloqueado',e,r); assert r.allowed is False
def test_fallback_detectado():
e='Desculpe, não entendi sua solicitação.'; r=detectar_fallback(e); log_rail('FALLBACK','Fallback - detectado',e,r); assert r.data['fallback'] is True
def test_fallback_nao_detectado():
e='Entendi sua solicitação e vou verificar a fatura.'; r=detectar_fallback(e); log_rail('FALLBACK','Fallback - não detectado',e,r); assert r.data['fallback'] is False
def test_viol_registra_msk():
r=registrar_violacao('agent_fatura','MSK'); log_rail('VIOL','Violação - MSK','agent_fatura/MSK',r); assert r.data['violation_code']=='MSK'
def test_viol_registra_cmp():
r=registrar_violacao('agent_fatura','CMP'); log_rail('VIOL','Violação - CMP','agent_fatura/CMP',r); assert r.data['violation_code']=='CMP'
def test_tox_permitido_neutro():
e='Preciso entender minha fatura.'; r=detectar_toxicidade(e); log_rail('TOX','Toxicidade - permitido',e,r); assert r.allowed is True and r.mechanism=='llm_rail'
def test_tox_bloqueado_toxico():
e='Você é idiota.'; r=detectar_toxicidade(e); log_rail('TOX','Toxicidade - bloqueado',e,r); assert r.allowed is False
def test_oos_permitido_telecom():
e='Quero contestar minha fatura.'; r=detectar_out_of_scope(e); log_rail('OOS','Out-of-Scope - permitido',e,r);
assert r.code == "OOS"
assert r.mechanism == "llm_rail"
# comportamento esperado
assert isinstance(r.allowed, bool)
def test_oos_bloqueado_politica():
e='Qual sua opinião sobre política?'; r=detectar_out_of_scope(e); log_rail('OOS','Out-of-Scope - bloqueado',e,r); assert r.allowed is False
def test_gnd_fundamentado():
r=validar_groundedness('serviço fatura cobrança ajuste',{'chunks_rag':['serviço fatura cobrança ajuste confirmado']}); log_rail('GND','Groundedness - fundamentado','serviço fatura cobrança ajuste',r); assert r.allowed is True and r.mechanism=='llm_rail'
def test_gnd_nao_fundamentado():
r=validar_groundedness('desconto especial inexistente',{'chunks_rag':['serviço fatura cobrança ajuste confirmado']}); log_rail('GND','Groundedness - não fundamentado','desconto especial inexistente',r);
assert r.code == "GND"
assert isinstance(r.allowed, bool)
def test_hist_permitido_sem_historico():
r=validar_consistencia_historica({}); log_rail('HIST','Histórico - permitido',{},r); assert r.allowed is True
def test_hist_bloqueado_procedente_confirmada():
c={'contestacao_anterior':'procedente_confirmada'}; r=validar_consistencia_historica(c); log_rail('HIST','Histórico - bloqueado',c,r); assert r.allowed is False
def test_pmptk_tokens_contabilizados():
r=contabilizar_tokens(100,50); log_rail('PMPTK','Prompt Tokens - contabilização','100+50',r); assert r.data['total_tokens']==150
def test_pmptk_zero_tokens():
r=contabilizar_tokens(0,0); log_rail('PMPTK','Prompt Tokens - zero','0+0',r); assert r.data['total_tokens']==0
def test_efic_eficiencia_parcial():
r=calcular_eficiencia_nlu(5,2); log_rail('EFIC','Eficiência - parcial','5/2',r); assert r.data['eficiencia']==0.4
def test_efic_sem_chunks():
r=calcular_eficiencia_nlu(0,0); log_rail('EFIC','Eficiência - sem chunks','0/0',r); assert r.data['eficiencia']==0
def test_nom_no_match():
r=detectar_no_match_rag([],'Não encontrei informação suficiente.'); log_rail('NO-M','No-Match - detectado','[]',r); assert r.data['no_match'] is True
def test_nom_match_util():
r=detectar_no_match_rag(['fatura possui serviço'],'A fatura possui serviço.'); log_rail('NO-M','No-Match - não detectado','chunk útil',r); assert r.data['no_match'] is False
def test_csi_negativo():
e='Estou muito insatisfeito com essa cobrança.'; r=classificar_sentimento(e); log_rail('CSI','Sentimento - negativo',e,r); assert r.data['sentimento']=='Negativo' and r.mechanism=='llm_judge'
def test_csi_positivo():
e='Obrigado, ficou resolvido.'; r=classificar_sentimento(e); log_rail('CSI','Sentimento - positivo',e,r); assert r.data['sentimento']=='Positivo'
def test_aluc_compativel():
r=avaliar_alucinacao('fatura possui serviço','fatura possui serviço contratado'); log_rail('ALUC','Alucinação - compatível','compatível',r); assert r.allowed is True and r.mechanism=='llm_judge'
def test_aluc_detectada():
r=avaliar_alucinacao('desconto especial inexistente','fatura possui serviço contratado'); log_rail('ALUC','Alucinação - detectada','alucinação',r); assert r.allowed is False
def test_vloop_detectado():
r=detectar_loop(['não entendi','repita','não entendi']); log_rail('VLOOP','Loops - detectado','mensagens repetidas',r); assert r.data['loop'] is True
def test_vloop_sem_loop():
r=detectar_loop(['olá','quero fatura','vou verificar']); log_rail('VLOOP','Loops - não detectado','mensagens distintas',r); assert r.data['loop'] is False
def test_msize_mede_tamanho():
r=medir_tamanho_mensagem('abc'); log_rail('MSIZE','Tamanho - abc','abc',r); assert r.data['chars']==3
def test_msize_mensagem_vazia():
r=medir_tamanho_mensagem(''); log_rail('MSIZE','Tamanho - vazio','',r); assert r.data['chars']==0
def test_rqlt_resposta_boa():
r=avaliar_qualidade_resposta('Por que minha fatura aumentou?','Sua fatura aumentou por cobrança adicional detalhada no extrato.'); log_rail('RQLT','Qualidade - boa','resposta completa',r); assert r.data['score']>=5 and r.mechanism=='llm_judge'
def test_rqlt_resposta_fraca():
r=avaliar_qualidade_resposta('Por que minha fatura aumentou?','Não sei.'); log_rail('RQLT','Qualidade - fraca','Não sei',r); assert r.data['score']<5
def test_vctn_tom_aderente():
e='Senhor cliente, verificamos sua solicitação com atenção.'; r=avaliar_tom_de_voz(e); log_rail('VCTN','Tom - aderente',e,r); assert r.allowed is True and r.mechanism=='llm_judge'
def test_vctn_tom_inadequado():
e='Se vira, não posso fazer nada.'; r=avaliar_tom_de_voz(e); log_rail('VCTN','Tom - inadequado',e,r); assert r.allowed is False
def test_revprec_metric_accuracy_total():
r=calcular_precisao_revocacao(['a','b'],['a','b']); log_rail('REVPREC_METRIC','Precisão/Revocação - total','labels',r); assert r.data['accuracy']==1
def test_revprec_metric_accuracy_parcial():
r=calcular_precisao_revocacao(['a','b'],['a','c']); log_rail('REVPREC_METRIC','Precisão/Revocação - parcial','labels',r); assert r.data['accuracy']==0.5
def test_semac_acuracia_ok():
r=avaliar_acuracia_semantica('cancelar serviço','cancelar serviço'); log_rail('SEMAC','Acurácia STT - ok','cancelar serviço',r); assert r.allowed is True
def test_semac_acuracia_baixa():
r=avaliar_acuracia_semantica('ativar plano','cancelar serviço'); log_rail('SEMAC','Acurácia STT - baixa','ativar plano vs cancelar serviço',r); assert r.allowed is False