Funcional

This commit is contained in:
2026-04-28 20:31:06 -03:00
parent ad6f9d4c0c
commit 387a662a39
9 changed files with 347 additions and 250 deletions

View File

@@ -859,8 +859,7 @@ Já o código **app_nemo.py** utiliza o framework **Nemo Guardrails** para ilust
python -m src.app_nemo
![img_3.png](img_3.png)
![img_4.png](img_4.png)
## 12. Mapeamento da planilha para implementação

BIN
img_4.png Normal file

Binary file not shown.

After

Width:  |  Height:  |  Size: 268 KiB

View File

@@ -0,0 +1,43 @@
from nemoguardrails import LLMRails, RailsConfig
from src.actions import (
mask_pii_action,
detectar_toxicidade_action,
detectar_out_of_scope_action,
validar_alcada_action,
verbalizacao_prematura_action,
validar_groundedness_action,
supervisor_vas_avulso_action,
enforce_compliance_anatel_action,
calcular_tcr_action,
detectar_fallback_action,
registrar_violacao_action,
validar_consistencia_historica_action,
contabilizar_tokens_action,
calcular_eficiencia_nlu_action,
calcular_eficiencia_nlu_action,
detectar_loop_action,
medir_tamanho_mensagem_action,
calcular_precisao_revocacao_action,
avaliar_acuracia_semantica_action
)
def init(app: LLMRails):
app.register_action(mask_pii_action)
app.register_action(detectar_toxicidade_action)
app.register_action(detectar_out_of_scope_action)
app.register_action(validar_alcada_action)
app.register_action(verbalizacao_prematura_action)
app.register_action(validar_groundedness_action)
app.register_action(supervisor_vas_avulso_action)
app.register_action(enforce_compliance_anatel_action)
app.register_action(calcular_tcr_action)
app.register_action(detectar_fallback_action)
app.register_action(registrar_violacao_action)
app.register_action(validar_consistencia_historica_action)
app.register_action(contabilizar_tokens_action)
app.register_action(calcular_eficiencia_nlu_action)
app.register_action(calcular_eficiencia_nlu_action)
app.register_action(detectar_loop_action)
app.register_action(medir_tamanho_mensagem_action)
app.register_action(calcular_precisao_revocacao_action)
app.register_action(avaliar_acuracia_semantica_action)

View File

@@ -1,11 +1,27 @@
colang_version: "1.0"
models:
- type: main
engine: openai
model: gpt-5
api_key_env_var: OPENAI_API_KEY
parameters:
temperature: 0
base_url: http://127.0.0.1:8051/v1
max_tokens: 50 # 🔥 evita respostas longas do LLM
# 🔥 usado apenas se você chamar explicitamente no flow
- type: self_check_input
engine: openai
model: openai.gpt-oss-120b
api_key_env_var: OPENAI_API_KEY
parameters:
temperature: 0
base_url: http://127.0.0.1:8051/v1
rails:
input:
flows:
- main
- check_input_terms
output:
flows:
- check_output_terms

View File

@@ -1,9 +0,0 @@
define user express input
$input_text
define flow main
user express input
execute executar_pipeline_validacoes
bot say $nemo_response

View File

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

View File

@@ -1,150 +1,284 @@
# src/actions.py
import json
import uuid
from typing import Optional
from nemoguardrails.actions import action
from .deterministic_rails import (
mask_pii,
validar_alcada,
enforce_compliance_anatel,
calcular_tcr,
detectar_fallback,
registrar_violacao,
validar_consistencia_historica,
contabilizar_tokens,
calcular_eficiencia_nlu,
detectar_no_match_rag,
detectar_loop,
medir_tamanho_mensagem,
calcular_precisao_revocacao,
avaliar_acuracia_semantica,
)
from .llm_rails import (
detectar_toxicidade,
detectar_out_of_scope,
verbalizacao_prematura,
validar_groundedness,
supervisor_vas_avulso,
)
from .deterministic_rails import validar_alcada
# =========================
# HELPERS
# =========================
try:
from .judges import avaliar_qualidade_resposta
except Exception:
avaliar_qualidade_resposta = None
def get_payload(context: Optional[dict]) -> dict:
return (context or {}).get("payload", {})
# =========================
# ACTIONS
# =========================
@action(is_system_action=True)
async def mask_pii_action(context: Optional[dict] = None, **kwargs):
print("🔥 MSK")
payload = get_payload(context)
input_text = payload.get("input_text") or context.get("user_message", "")
result = mask_pii(input_text)
if context is not None:
context["text"] = getattr(result, "sanitized_text", input_text)
return result
PIPELINE_RESULTS = {}
# -------------------------
@action(is_system_action=True)
async def detectar_toxicidade_action(context: Optional[dict] = None, **kwargs):
print("🔥 TOX")
text = context.get("text") or context.get("user_message", "")
result = detectar_toxicidade(text)
return result
def extrair_payload(context: dict) -> dict:
try:
messages = context.get("messages", [])
content = messages[-1]["content"]
return json.loads(content)
except Exception:
return {}
# -------------------------
@action(is_system_action=True)
async def detectar_out_of_scope_action(context: Optional[dict] = None, **kwargs):
print("🔥 OOS")
text = context.get("text") or context.get("user_message", "")
result = detectar_out_of_scope(text)
return result
def add_trace(trace, label, result):
trace.append({
"rail": label,
"allowed": result.allowed,
"reason": result.reason,
"code": getattr(result, "code", label),
"mechanism": getattr(result, "mechanism", ""),
"data": getattr(result, "data", {}),
})
# -------------------------
@action(is_system_action=True)
async def validar_alcada_action(context: Optional[dict] = None, **kwargs):
print("🔥 ADJ")
payload = get_payload(context)
ctx = payload.get("context", {})
valor = ctx.get("ajuste_valor", 0)
result = validar_alcada(valor)
return result
def executar_pipeline_validacoes(context: dict):
print("🔥🔥🔥 ACTION FOI EXECUTADA")
payload = extrair_payload(context)
# -------------------------
request_id = payload.get("request_id") or str(uuid.uuid4())
input_text = payload.get("input_text", "")
ctx = payload.get("context", {}) or {}
@action(is_system_action=True)
async def verbalizacao_prematura_action(context: Optional[dict] = None, **kwargs):
print("🔥 REVPREC")
trace = []
failures = []
payload = get_payload(context)
ctx = payload.get("context", {})
# =========================
# INPUT RAILS - LLM
# =========================
resposta = ctx.get("resposta_llm", "")
r_tox = detectar_toxicidade(input_text)
add_trace(trace, "TOX", r_tox)
if not r_tox.allowed:
failures.append(("TOX", r_tox.reason))
result = verbalizacao_prematura(resposta, ctx)
r_oos = detectar_out_of_scope(input_text)
add_trace(trace, "OOS", r_oos)
if not r_oos.allowed:
failures.append(("OOS", r_oos.reason))
return result
# =========================
# BUSINESS RAIL - DETERMINISTIC
# =========================
valor = ctx.get("ajuste_valor")
r_adj = validar_alcada(valor)
add_trace(trace, "ADJ", r_adj)
if not r_adj.allowed:
failures.append(("ADJ", r_adj.reason))
# -------------------------
# =========================
# LLM RESPONSE
# =========================
@action(is_system_action=True)
async def validar_groundedness_action(context: Optional[dict] = None, **kwargs):
print("🔥 GND")
final_response = ctx.get("resposta_llm", "")
payload = get_payload(context)
ctx = payload.get("context", {})
trace.append({
"step": "LLM",
"allowed": True,
"input": input_text,
"output_preview": final_response[:200],
"mechanism": "provided_response_or_proxy",
})
resposta = ctx.get("resposta_llm", "")
# =========================
# OUTPUT RAILS - LLM
# =========================
result = validar_groundedness(resposta, ctx)
r_revprec = verbalizacao_prematura(final_response, ctx)
add_trace(trace, "REVPREC", r_revprec)
if not r_revprec.allowed:
failures.append(("REVPREC", r_revprec.reason))
return result
r_gnd = validar_groundedness(final_response, ctx)
add_trace(trace, "GND", r_gnd)
if not r_gnd.allowed:
failures.append(("GND", r_gnd.reason))
# -------------------------
# =========================
# OPTIONAL JUDGE / CMP
# =========================
@action(is_system_action=True)
async def supervisor_vas_avulso_action(context: Optional[dict] = None, **kwargs):
print("🔥 REVPREC_SUP")
if avaliar_qualidade_resposta is not None:
r_cmp = avaliar_qualidade_resposta(input_text, final_response)
add_trace(trace, "CMP", r_cmp)
if not r_cmp.allowed:
failures.append(("CMP", r_cmp.reason))
else:
trace.append({
"rail": "CMP",
"allowed": True,
"reason": "CMP não configurado",
"mechanism": "skipped",
"data": {},
})
payload = get_payload(context)
# =========================
# FINAL DECISION
# =========================
result = supervisor_vas_avulso(payload)
blocked = len(failures) > 0
return result
if blocked:
first_code, first_reason = failures[0]
nemo_response = f"BLOCKED:{first_code} - {first_reason}"
else:
nemo_response = final_response
@action(is_system_action=True)
async def enforce_compliance_anatel_action(context=None, **kwargs):
print("🔥 CMP")
text = context.get("text") or context.get("user_message", "")
payload = get_payload(context)
ctx = payload.get("context", {})
result = enforce_compliance_anatel(text, ctx)
return result
@action(is_system_action=True)
async def calcular_tcr_action(context=None, **kwargs):
print("🔥 TCR")
payload = get_payload(context)
status = payload.get("context", {}).get("status", "")
result = calcular_tcr(status)
return result
@action(is_system_action=True)
async def detectar_fallback_action(context=None, **kwargs):
print("🔥 FALLBACK")
text = context.get("text") or context.get("user_message", "")
result = detectar_fallback(text)
return result
@action(is_system_action=True)
async def registrar_violacao_action(context=None, **kwargs):
print("🔥 VIOL")
payload = get_payload(context)
agent_id = payload.get("agent_id", "unknown")
code = payload.get("violation_code", "UNKNOWN")
result = registrar_violacao(agent_id, code)
return result
@action(is_system_action=True)
async def validar_consistencia_historica_action(context=None, **kwargs):
print("🔥 HIST")
payload = get_payload(context)
ctx = payload.get("context", {})
result = validar_consistencia_historica(ctx)
return result
@action(is_system_action=True)
async def contabilizar_tokens_action(context=None, **kwargs):
print("🔥 PMPTK")
payload = get_payload(context)
prompt = payload.get("prompt_tokens", 0)
completion = payload.get("completion_tokens", 0)
result = contabilizar_tokens(prompt, completion)
return result
@action(is_system_action=True)
async def calcular_eficiencia_nlu_action(context=None, **kwargs):
print("🔥 EFIC")
payload = get_payload(context)
ctx = payload.get("context", {})
result = calcular_eficiencia_nlu(
ctx.get("chunks_retornados", 0),
ctx.get("chunks_utilizados", 0)
)
return result
@action(is_system_action=True)
async def detectar_no_match_rag_action(context=None, **kwargs):
print("🔥 NO-M")
payload = get_payload(context)
ctx = payload.get("context", {})
result = detectar_no_match_rag(
ctx.get("chunks", []),
ctx.get("resposta_llm", "")
)
return result
@action(is_system_action=True)
async def detectar_loop_action(context=None, **kwargs):
print("🔥 VLOOP")
payload = get_payload(context)
mensagens = payload.get("context", {}).get("mensagens", [])
result = detectar_loop(mensagens)
return result
@action(is_system_action=True)
async def medir_tamanho_mensagem_action(context=None, **kwargs):
print("🔥 MSIZE")
text = context.get("text") or context.get("user_message", "")
result = medir_tamanho_mensagem(text)
return result
@action(is_system_action=True)
async def calcular_precisao_revocacao_action(context=None, **kwargs):
print("🔥 REVPREC_METRIC")
payload = get_payload(context)
ctx = payload.get("context", {})
result = calcular_precisao_revocacao(
ctx.get("y_true", []),
ctx.get("y_pred", [])
)
return result
@action(is_system_action=True)
async def avaliar_acuracia_semantica_action(context=None, **kwargs):
print("🔥 SEMAC")
payload = get_payload(context)
ctx = payload.get("context", {})
result = avaliar_acuracia_semantica(
ctx.get("audio_transcrito", ""),
ctx.get("referencia_humana", "")
)
return result
result = {
"allowed": not blocked,
"label": "CONFORME" if not blocked else "PROBLEMA",
"response": final_response,
"reason": nemo_response if blocked else "",
"failures": failures,
"trace": trace,
}
PIPELINE_RESULTS[request_id] = result
return {
"nemo_response": nemo_response
}

View File

@@ -1,133 +1,47 @@
import json
import uuid
#https://docs.nvidia.com/nemo/guardrails/latest/configure-rails/actions/index.html
#https://docs.nvidia.com/nemo/guardrails/latest/configure-rails/actions/registering-actions.html
#https://docs.nvidia.com/nemo/guardrails/latest/observability/logging/index.html
from nemoguardrails import LLMRails, RailsConfig
from src.actions import executar_pipeline_validacoes, PIPELINE_RESULTS
config = RailsConfig.from_path("./config")
rails = LLMRails(config)
def extract_return_values(response):
results = []
def build_rails():
config = RailsConfig.from_path("./config")
rails = LLMRails(config)
log = response.log
rails.register_action(
executar_pipeline_validacoes,
"executar_pipeline_validacoes"
)
for rail in log.activated_rails:
for action in rail.executed_actions:
rv = action.return_value
if rv is not None:
results.append({
"action": action.action_name,
"allowed": getattr(rv, "allowed", None),
"reason": getattr(rv, "reason", None),
"sanitized_text": getattr(rv, "sanitized_text", None),
"code": getattr(rv, "code", None),
"mechanism": getattr(rv, "mechanism", None),
"data": getattr(rv, "data", None)
})
return rails
return results
MESSAGE = "Meu CPF é 169.323.728-86"
rails = build_rails()
def executar_atendimento(user_input: str, context: dict):
request_id = str(uuid.uuid4())
payload = {
"request_id": request_id,
"input_text": user_input,
"context": context or {},
response = rails.generate(
messages=[{"role": "user", "content": MESSAGE}],
options={
"output_vars": ["triggered_input_rail", "relevant_chunks"],
"log": {
"activated_rails": True,
"llm_calls": True
}
}
)
# 🔥 chama direto seu pipeline
executar_pipeline_validacoes({
"messages": [
{
"role": "user",
"content": json.dumps(payload)
}
]
})
result = PIPELINE_RESULTS.pop(request_id, None)
return result
if __name__ == "__main__":
def rodar_teste(nome, user_input, context):
print("\n" + "="*60)
print(f"🧪 TESTE: {nome}")
print("-"*60)
print("INPUT:", user_input)
print("CONTEXT:", context)
result = executar_atendimento(user_input, context)
print("\n📌 RESULTADO FINAL:")
print(result)
print("\n🔍 TRACE:")
for step in result.get("trace", []):
print(step)
print("="*60)
# =========================
# ✅ TESTE 1 - OK (CONFORME)
# =========================
rodar_teste(
"OK - Cancelamento válido",
"quero cancelar VAS",
{
"ajuste_valor": 20,
"resposta_llm": "Cancelamento realizado com sucesso."
}
)
# =========================
# ❌ TESTE 2 - TOXICIDADE
# =========================
rodar_teste(
"TOX - Linguagem ofensiva",
"quero cancelar VAS seu lixo",
{
"ajuste_valor": 20,
"resposta_llm": "Cancelamento realizado."
}
)
# =========================
# ❌ TESTE 3 - ALÇADA
# =========================
rodar_teste(
"ADJ - Valor acima da alçada",
"quero cancelar VAS",
{
"ajuste_valor": 2000,
"resposta_llm": "Cancelamento realizado."
}
)
# =========================
# ❌ TESTE 4 - GROUNDEDNESS
# =========================
rodar_teste(
"GND - Resposta não confiável",
"quero cancelar VAS",
{
"ajuste_valor": 20,
"resposta_llm": "O cancelamento ativa cashback infinito e crédito ilimitado."
}
)
# =========================
# ⚠️ TESTE 5 - MÚLTIPLOS PROBLEMAS
# =========================
rodar_teste(
"MULTI - TOX + ADJ + GND",
"quero cancelar VAS seu inútil",
{
"ajuste_valor": 5000,
"resposta_llm": "Você ganha benefícios ilimitados com esse cancelamento."
}
)
feedback = extract_return_values(response)
for f in feedback:
print(f)

View File

@@ -6,7 +6,7 @@ Texto:
{text}
Classifique como:
- TOXICO
- TOXICO: Apenas se encontrar palavras ofensivas ou condenatórias
- NORMAL
Responda JSON: