720 lines
32 KiB
Python
720 lines
32 KiB
Python
from __future__ import annotations
|
|
|
|
import json, re
|
|
from typing import Any
|
|
|
|
from agent_framework.guardrails.base import Guardrail, RailDecision
|
|
from .tim_prompts.ausencia_oferta_proativa import build_aoferta_prompt
|
|
from .tim_prompts.out_of_scope import build_oos_prompt
|
|
from .tim_prompts.revprec import build_revprec_prompt
|
|
from .tim_prompts.fraseologia import build_fraseologia_prompt
|
|
|
|
|
|
def _llm(context: dict[str, Any]):
|
|
return context.get('guardrail_llm') or context.get('llm') or context.get('model')
|
|
|
|
|
|
def _context_text(context: dict[str, Any]) -> str:
|
|
try: return json.dumps(context or {}, ensure_ascii=False, default=str)[:16000]
|
|
except Exception: return str(context or {})[:16000]
|
|
|
|
|
|
def _compact_aoferta_context(context: dict[str, Any]) -> str:
|
|
"""High-signal context for the proactive-offer LLM.
|
|
|
|
The generic output context can contain very large MCP/workflow payloads. Sending
|
|
the first N characters of that payload may truncate the customer request and the
|
|
resolved transaction target, which makes the auditor treat legitimate fallback
|
|
guidance as a new unsolicited action. Keep only authoritative facts needed by
|
|
AOFERTA and cap nested operational results.
|
|
"""
|
|
ctx = context or {}
|
|
|
|
history = []
|
|
for item in list(ctx.get('conversation_history') or [])[-8:]:
|
|
if isinstance(item, dict):
|
|
role = str(item.get('role') or item.get('type') or '')
|
|
content = str(item.get('content') or '')
|
|
if content:
|
|
history.append({'role': role, 'content': content[:1200]})
|
|
|
|
pre = ctx.get('transaction_pre_validation') if isinstance(ctx.get('transaction_pre_validation'), dict) else {}
|
|
requested = pre.get('requested_arguments') if isinstance(pre.get('requested_arguments'), dict) else {}
|
|
resolved = pre.get('resolved_arguments') if isinstance(pre.get('resolved_arguments'), dict) else {}
|
|
|
|
def compact_result(value: Any, depth: int = 0) -> Any:
|
|
if depth > 6:
|
|
return None
|
|
if isinstance(value, list):
|
|
return [compact_result(v, depth + 1) for v in value[:8]]
|
|
if not isinstance(value, dict):
|
|
return value if isinstance(value, (str, int, float, bool)) or value is None else str(value)[:500]
|
|
keep = {
|
|
'tool_name', 'ok', 'error', 'status', 'transaction_status', 'workflow_status',
|
|
'subject', 'name', 'resolved_subject', 'resolved_subjects', 'success', 'reason',
|
|
'mensagem', 'message', 'terminal_status', 'effective_tool_name',
|
|
'requested_arguments', 'resolved_arguments', 'items', 'results', 'errors',
|
|
'cancelados', 'nao_cancelados', 'nao_encontrados', 'output', 'result',
|
|
}
|
|
out = {}
|
|
for k, v in value.items():
|
|
if k in keep:
|
|
cv = compact_result(v, depth + 1)
|
|
if cv not in (None, {}, []):
|
|
out[k] = cv
|
|
return out
|
|
|
|
payload = {
|
|
'current_user_message': str(ctx.get('current_user_message') or '')[:1200],
|
|
'current_intent': str(ctx.get('current_intent') or ctx.get('intent') or ''),
|
|
'current_route': str(ctx.get('current_route') or ctx.get('route') or ''),
|
|
'transaction_status': str(ctx.get('transaction_status') or ''),
|
|
'conversation_history': history,
|
|
'transaction_request': {
|
|
'requested_arguments': compact_result(requested),
|
|
'resolved_arguments': compact_result(resolved),
|
|
'effective_tool_name': pre.get('effective_tool_name'),
|
|
'confirmation_message': pre.get('confirmation_message'),
|
|
},
|
|
'current_execution': compact_result(ctx.get('mcp_results') or ctx.get('tool_result') or []),
|
|
}
|
|
return json.dumps(payload, ensure_ascii=False, default=str)[:16000]
|
|
|
|
|
|
|
|
|
|
def _authorized_human_handoff(context: dict[str, Any]) -> bool:
|
|
"""Return True only for structurally authorized human handoff on this turn."""
|
|
ctx = context or {}
|
|
route = str(ctx.get('current_route') or ctx.get('route') or '').strip().lower()
|
|
intent = str(ctx.get('current_intent') or ctx.get('intent') or '').strip().lower()
|
|
session_control = str(ctx.get('session_control') or '').strip().upper()
|
|
requested = ctx.get('human_handoff_requested') is True
|
|
handoff = ctx.get('handoff') is True
|
|
|
|
route_decision = ctx.get('route_decision') if isinstance(ctx.get('route_decision'), dict) else {}
|
|
route_meta = route_decision.get('metadata') if isinstance(route_decision.get('metadata'), dict) else {}
|
|
rd_route = str(route_decision.get('route') or route_decision.get('agent') or '').strip().lower()
|
|
rd_intent = str(route_decision.get('intent') or '').strip().lower()
|
|
rd_handoff = route_decision.get('handoff') is True
|
|
rd_session_control = str(route_meta.get('session_control') or '').strip().upper()
|
|
|
|
control_evidence = (
|
|
session_control == 'HUMAN_HANDOFF'
|
|
or requested
|
|
or handoff
|
|
or rd_handoff
|
|
or rd_session_control == 'HUMAN_HANDOFF'
|
|
)
|
|
route_evidence = (
|
|
route == 'human_handoff'
|
|
or intent == 'human_handoff'
|
|
or rd_route == 'human_handoff'
|
|
or rd_intent == 'human_handoff'
|
|
)
|
|
if control_evidence and route_evidence:
|
|
return True
|
|
|
|
# A resumed domain workflow may decide the handoff after the router has
|
|
# already been bypassed for workflow continuation. In that case the
|
|
# current route/intent legitimately remain the domain values, while the
|
|
# *current-turn workflow result* is the authoritative orchestration
|
|
# decision. Accept only a terminal, internally consistent handoff result;
|
|
# an isolated `handoff=true` or transfer-like sentence is never enough.
|
|
roots = []
|
|
for key in ('mcp_results', 'tool_result', 'workflow_result'):
|
|
value = ctx.get(key)
|
|
if value is not None:
|
|
roots.append(value)
|
|
|
|
def terminal_workflow_handoff(value: Any) -> bool:
|
|
if isinstance(value, dict):
|
|
workflow_control = str(value.get('session_control') or '').strip().upper()
|
|
workflow_terminal = str(value.get('terminal_status') or '').strip().lower()
|
|
workflow_requested = value.get('human_handoff_requested') is True
|
|
workflow_handoff = value.get('handoff') is True
|
|
workflow_session_ended = value.get('session_ended') is True
|
|
|
|
has_control = workflow_control == 'HUMAN_HANDOFF'
|
|
has_request = workflow_requested or workflow_handoff
|
|
has_terminal = workflow_terminal == 'human_handoff' or workflow_session_ended
|
|
if has_control and has_request and has_terminal:
|
|
return True
|
|
return any(terminal_workflow_handoff(item) for item in value.values())
|
|
if isinstance(value, (list, tuple)):
|
|
return any(terminal_workflow_handoff(item) for item in value)
|
|
return False
|
|
|
|
return any(terminal_workflow_handoff(root) for root in roots)
|
|
|
|
|
|
|
|
def _deterministic_tim_domain_response(text: str, context: dict[str, Any]) -> bool:
|
|
"""Conservative fast path for unmistakable TIM Contas domain output.
|
|
|
|
This extension may know the TIM Contas vocabulary; the generic framework does
|
|
not. Human-transfer language is intentionally excluded so handoff authorization
|
|
remains exclusively structural. Ambiguous domain prose still goes to TIM_OOS LLM.
|
|
"""
|
|
normalized = re.sub(r'\s+', ' ', str(text or '').strip().lower())
|
|
if not normalized:
|
|
return False
|
|
|
|
handoff_markers = (
|
|
'encaminhar', 'encaminhado', 'transferir', 'transferido',
|
|
'atendente', 'pessoa', 'humano', 'especialista',
|
|
)
|
|
if any(m in normalized for m in handoff_markers):
|
|
return False
|
|
|
|
domain_markers = (
|
|
'contestação', 'contestacao', 'fatura', 'cobrança', 'cobranca',
|
|
'plano', 'serviço', 'servico', 'vas', 'boleto', 'desconto',
|
|
)
|
|
completion_markers = (
|
|
'registrad', 'cancelad', 'concluíd', 'concluid', 'executad',
|
|
'processad', 'enviad', 'emitid', 'protocolo',
|
|
)
|
|
|
|
# Keep the bypass intentionally narrow: an unmistakable domain noun plus an
|
|
# operational/result marker. Ordinary explanations and ambiguous responses are
|
|
# still semantically audited by TIM_OOS.
|
|
return (
|
|
any(m in normalized for m in domain_markers)
|
|
and any(m in normalized for m in completion_markers)
|
|
)
|
|
|
|
def _parse_json(raw: Any) -> dict[str, Any]:
|
|
text = str(getattr(raw, 'content', raw) or '').strip()
|
|
m = re.search(r'\{[\s\S]*\}', text)
|
|
if m: text=m.group(0)
|
|
try: return json.loads(text)
|
|
except Exception: return {'allowed': False, 'reason': f'Resposta inválida do guardrail TIM: {text[:300]}'}
|
|
|
|
|
|
class _TimPromptRail(Guardrail):
|
|
prompt_builder = None
|
|
profile_name = 'guardrail'
|
|
|
|
async def evaluate(self, text: str, context: dict[str, Any]) -> RailDecision:
|
|
llm = _llm(context)
|
|
if llm is None:
|
|
return RailDecision(code=self.code, allowed=False, reason='LLM do framework indisponível para guardrail TIM', metadata={'external': True, 'fail_closed': True})
|
|
prompt = self.prompt_builder(text or '', _context_text(context))
|
|
raw = await llm.ainvoke(
|
|
[{'role':'system','content':'Responda apenas JSON válido, sem markdown.'}, {'role':'user','content':prompt}],
|
|
profile_name=self.profile_name,
|
|
component_name=f'guardrail.external.{self.code.lower()}',
|
|
generation_name=f'guardrail.external.{self.code.lower()}',
|
|
)
|
|
out = _parse_json(raw)
|
|
return RailDecision(code=self.code, allowed=bool(out.get('allowed', False)), reason=str(out.get('reason') or out.get('label') or ''), sanitized_text=text, metadata={'external': True, 'domain':'TIM_CONTAS', 'data':out})
|
|
|
|
|
|
class TimOutOfScopeRail(_TimPromptRail):
|
|
code='TIM_OOS'; stage='output'; prompt_builder=staticmethod(build_oos_prompt)
|
|
|
|
async def evaluate(self, text: str, context: dict[str, Any]) -> RailDecision:
|
|
# A handoff structurally selected by the router/workflow is an authorized
|
|
# orchestration response, not a domain answer to be judged as OOS by text.
|
|
if _authorized_human_handoff(context):
|
|
return RailDecision(
|
|
code=self.code,
|
|
allowed=True,
|
|
reason='handoff_humano_autorizado',
|
|
sanitized_text=text,
|
|
metadata={
|
|
'external': True,
|
|
'domain': 'TIM_CONTAS',
|
|
'mechanism': 'deterministic_handoff_bypass',
|
|
'data': {
|
|
'allowed': True,
|
|
'reason': 'handoff_humano_autorizado',
|
|
},
|
|
},
|
|
)
|
|
if _deterministic_tim_domain_response(text, context):
|
|
return RailDecision(
|
|
code=self.code,
|
|
allowed=True,
|
|
reason='resposta suportada pelo contexto do domínio TIM Contas',
|
|
sanitized_text=text,
|
|
metadata={
|
|
'external': True,
|
|
'domain': 'TIM_CONTAS',
|
|
'mechanism': 'deterministic_domain_bypass',
|
|
'data': {
|
|
'allowed': True,
|
|
'reason': 'resposta suportada pelo contexto do domínio TIM Contas',
|
|
},
|
|
},
|
|
)
|
|
return await super().evaluate(text, context)
|
|
|
|
class TimProactiveOfferRail(_TimPromptRail):
|
|
code='TIM_AOFERTA'; stage='output'; prompt_builder=staticmethod(build_aoferta_prompt)
|
|
|
|
# Mantém a mesma semântica do AOFERTA nativo do framework: mensagens de
|
|
# continuidade de uma transação já aberta não constituem nova oferta
|
|
# proativa. O bypass é estritamente dirigido pelo estado transacional, sem
|
|
# inferência textual e sem desabilitar outros guardrails de saída.
|
|
_TRANSACTION_CONTINUATION_STATUSES = {
|
|
'COLLECTING_PARAMETERS',
|
|
'AWAITING_CONFIRMATION',
|
|
}
|
|
|
|
@staticmethod
|
|
def _authorized_human_handoff(context: dict[str, Any]) -> bool:
|
|
return _authorized_human_handoff(context)
|
|
|
|
@classmethod
|
|
def _transaction_continuation_status(cls, context: dict[str, Any]) -> str | None:
|
|
ctx = context or {}
|
|
status = str(ctx.get('transaction_status') or '').strip().upper()
|
|
if status in cls._TRANSACTION_CONTINUATION_STATUSES:
|
|
return status
|
|
|
|
# Compatibilidade com o contexto de output do Contas e com callers que
|
|
# expõem o estado apenas no resultado da tool. É o mesmo fallback usado
|
|
# pelo ProactiveOfferRail nativo: mcp_results -> tool_result.
|
|
results = ctx.get('mcp_results') or ctx.get('tool_result') or []
|
|
if isinstance(results, dict):
|
|
results = [results]
|
|
for result in reversed(list(results)):
|
|
if not isinstance(result, dict):
|
|
continue
|
|
result_status = str(result.get('transaction_status') or '').strip().upper()
|
|
if result_status in cls._TRANSACTION_CONTINUATION_STATUSES:
|
|
return result_status
|
|
return None
|
|
|
|
async def evaluate(self, text: str, context: dict[str, Any]) -> RailDecision:
|
|
if self._authorized_human_handoff(context):
|
|
return RailDecision(
|
|
code=self.code,
|
|
allowed=True,
|
|
reason='handoff_humano_autorizado',
|
|
sanitized_text=text,
|
|
metadata={
|
|
'external': True,
|
|
'domain': 'TIM_CONTAS',
|
|
'mechanism': 'deterministic_handoff_bypass',
|
|
'data': {
|
|
'allowed': True,
|
|
'reason': 'handoff_humano_autorizado',
|
|
},
|
|
},
|
|
)
|
|
|
|
continuation_status = self._transaction_continuation_status(context)
|
|
if continuation_status:
|
|
return RailDecision(
|
|
code=self.code,
|
|
allowed=True,
|
|
reason=f'continuidade_transacional:{continuation_status}',
|
|
sanitized_text=text,
|
|
metadata={
|
|
'external': True,
|
|
'domain': 'TIM_CONTAS',
|
|
'mechanism': 'deterministic_transaction_bypass',
|
|
'transaction_status': continuation_status,
|
|
'data': {
|
|
'allowed': True,
|
|
'reason': f'continuidade_transacional:{continuation_status}',
|
|
},
|
|
},
|
|
)
|
|
|
|
# Terminal/partial-success responses still need semantic auditing, but with
|
|
# compact authoritative transaction evidence. Do not use a completed-state
|
|
# bypass: the LLM must still block genuinely new targets/actions.
|
|
llm = _llm(context)
|
|
if llm is None:
|
|
return RailDecision(code=self.code, allowed=False, reason='LLM do framework indisponível para guardrail TIM', metadata={'external': True, 'fail_closed': True})
|
|
compact_context = _compact_aoferta_context(context)
|
|
prompt = self.prompt_builder(text or '', compact_context)
|
|
raw = await llm.ainvoke(
|
|
[{'role':'system','content':'Responda apenas JSON válido, sem markdown.'}, {'role':'user','content':prompt}],
|
|
profile_name=self.profile_name,
|
|
component_name=f'guardrail.external.{self.code.lower()}',
|
|
generation_name=f'guardrail.external.{self.code.lower()}',
|
|
)
|
|
out = _parse_json(raw)
|
|
return RailDecision(
|
|
code=self.code,
|
|
allowed=bool(out.get('allowed', False)),
|
|
reason=str(out.get('reason') or out.get('label') or ''),
|
|
sanitized_text=text,
|
|
metadata={
|
|
'external': True,
|
|
'domain':'TIM_CONTAS',
|
|
'mechanism': 'llm_semantic_compact_transaction_context',
|
|
'data':out,
|
|
},
|
|
)
|
|
|
|
class TimPrematureActionRail(_TimPromptRail):
|
|
code='TIM_REVPREC'; stage='output'; profile_name='grl'; prompt_builder=staticmethod(build_revprec_prompt)
|
|
|
|
_MESSAGE_KEYS = {
|
|
'mensagem', 'message', 'response', 'response_text', 'final_answer',
|
|
'customer_message', 'customer_response', 'answer', 'text',
|
|
}
|
|
|
|
@classmethod
|
|
def _current_execution_evidence(cls, context: dict[str, Any]) -> list[dict[str, Any]]:
|
|
"""Return compact, current-turn execution evidence only.
|
|
|
|
REVPREC must judge whether an operational claim is supported by what actually
|
|
happened in this turn. Conversation history, LTM and previous workflow state
|
|
are deliberately excluded from this evidence set.
|
|
"""
|
|
ctx = context or {}
|
|
roots = ctx.get('mcp_results')
|
|
if roots is None:
|
|
roots = ctx.get('tool_result')
|
|
if roots is None:
|
|
return []
|
|
if isinstance(roots, dict):
|
|
roots = [roots]
|
|
if not isinstance(roots, (list, tuple)):
|
|
return []
|
|
|
|
def compact(value: Any, depth: int = 0) -> Any:
|
|
if depth > 7:
|
|
return None
|
|
if isinstance(value, (str, int, float, bool)) or value is None:
|
|
if isinstance(value, str):
|
|
return value[:2000]
|
|
return value
|
|
if isinstance(value, (list, tuple)):
|
|
out = [compact(v, depth + 1) for v in value[:20]]
|
|
return [v for v in out if v not in (None, {}, [])]
|
|
if not isinstance(value, dict):
|
|
return str(value)[:500]
|
|
keep = {
|
|
'tool_name', 'server_name', 'ok', 'error', 'status', 'workflow_name',
|
|
'execution_id', 'success', 'allowed', 'reason', 'subject', 'name',
|
|
'resolved_subject', 'resolved_subjects', 'value', 'amount', 'protocol',
|
|
'protocolo', 'mensagem', 'message', 'response', 'response_text',
|
|
'final_answer', 'customer_message', 'customer_response', 'answer', 'text',
|
|
'epistemic_status', 'output', 'result', 'items', 'results', 'errors',
|
|
'cancelados', 'nao_cancelados', 'nao_encontrados', 'terminal_status',
|
|
'contestation_registered', 'contested_invoice_amount', 'contested_invoice_amount_open',
|
|
'sms_sent', 'barcode', 'itemName', 'itemsResponse', 'contested_items', 'sr',
|
|
}
|
|
out = {}
|
|
for k, v in value.items():
|
|
if k in keep:
|
|
cv = compact(v, depth + 1)
|
|
if cv not in (None, {}, []):
|
|
out[k] = cv
|
|
return out
|
|
|
|
evidence = []
|
|
for item in roots:
|
|
if not isinstance(item, dict):
|
|
continue
|
|
# Only successful/current tool executions can prove that an action happened.
|
|
# Failed results are still included so a contradictory success claim can be blocked.
|
|
cv = compact(item)
|
|
if isinstance(cv, dict) and cv:
|
|
evidence.append(cv)
|
|
return evidence
|
|
|
|
@classmethod
|
|
def _authoritative_output_messages(cls, context: dict[str, Any]) -> list[str]:
|
|
"""Extract canonical customer-facing messages from current successful tool outputs.
|
|
|
|
This is generic: no tool/workflow names are known here. Only message-like fields
|
|
underneath the current result/output tree are considered; state/input/history are
|
|
intentionally ignored.
|
|
"""
|
|
ctx = context or {}
|
|
roots = ctx.get('mcp_results')
|
|
if roots is None:
|
|
roots = ctx.get('tool_result')
|
|
if isinstance(roots, dict):
|
|
roots = [roots]
|
|
if not isinstance(roots, (list, tuple)):
|
|
return []
|
|
|
|
found: list[str] = []
|
|
|
|
def walk(value: Any, *, in_output: bool = False) -> None:
|
|
if isinstance(value, dict):
|
|
for k, v in value.items():
|
|
key = str(k).lower()
|
|
child_output = in_output or key in {'output', 'result', 'results'}
|
|
if child_output and key in cls._MESSAGE_KEYS and isinstance(v, str) and v.strip():
|
|
found.append(' '.join(v.split()))
|
|
if key not in {'state', 'input', 'metadata', 'conversation_history', 'history'}:
|
|
walk(v, in_output=child_output)
|
|
elif isinstance(value, (list, tuple)):
|
|
for v in value:
|
|
walk(v, in_output=in_output)
|
|
|
|
for item in roots:
|
|
if not isinstance(item, dict):
|
|
continue
|
|
if item.get('ok') is False:
|
|
continue
|
|
walk(item.get('result', item), in_output=True)
|
|
return list(dict.fromkeys(found))
|
|
|
|
@classmethod
|
|
def _candidate_is_authoritative_output(cls, text: str, context: dict[str, Any]) -> bool:
|
|
candidate = ' '.join(str(text or '').split())
|
|
if not candidate:
|
|
return False
|
|
return candidate in cls._authoritative_output_messages(context)
|
|
|
|
@staticmethod
|
|
def _normalized_text(value: Any) -> str:
|
|
text = str(value or '').lower()
|
|
text = text.replace('r$', ' ')
|
|
text = re.sub(r'\s+', ' ', text)
|
|
return text.strip()
|
|
|
|
@classmethod
|
|
def _candidate_has_operational_completion_claim(cls, text: str) -> bool:
|
|
"""Detect completion/effect claims, not ordinary explanations.
|
|
|
|
REVPREC exists to stop claims that an operation already happened. A
|
|
billing explanation containing values/dates is not such a claim and must
|
|
not depend on a probabilistic binary LLM classification.
|
|
"""
|
|
normalized = cls._normalized_text(text)
|
|
patterns = (
|
|
r'\b(cancelad[oa]s?|cancelei|cancelamos|cancelou|cancelamento .*conclu)',
|
|
r'\b(contesta(?:ção|cao) .*?(?:registrad|abert|conclu|criad))',
|
|
r'\b(registrad[oa]s?|executei|executamos|executou|executad[oa]s?|conclu[ií]d[oa]s?|processad[oa]s?)\b',
|
|
r'\b(enviad[oa]s?|emitid[oa]s?|retirad[oa]s?|removid[oa]s?)\b',
|
|
r'\b(protocol(?:o)? .*?(?:abert|gerad|registrad))',
|
|
r'\b(?:foi|foram) (?:cancelad|contestad|registrad|enviad|emitid|retirad|removid)',
|
|
r'\b(?:was|were|has been) (?:cancelled|canceled|registered|sent|issued|removed)',
|
|
)
|
|
return any(re.search(pattern, normalized, re.I) for pattern in patterns)
|
|
|
|
@classmethod
|
|
def _current_execution_scalar_evidence(cls, context: dict[str, Any]) -> tuple[list[str], bool, bool]:
|
|
"""Flatten scalar facts from current successful result/output only.
|
|
|
|
Returns (facts, completed, positive_effect). Inputs, conversation state,
|
|
history and metadata are excluded so user claims can never prove an action.
|
|
"""
|
|
roots = (context or {}).get('mcp_results')
|
|
if roots is None:
|
|
roots = (context or {}).get('tool_result')
|
|
if isinstance(roots, dict):
|
|
roots = [roots]
|
|
if not isinstance(roots, (list, tuple)):
|
|
return [], False, False
|
|
|
|
facts: list[str] = []
|
|
completed = False
|
|
positive = False
|
|
positive_status = {'completed', 'opened', 'open', 'iniciada', 'iniciado', 'success', 'succeeded', 'closed', 'fechado'}
|
|
positive_keys = {
|
|
'success', 'executed', 'contestation_registered', 'sms_sent',
|
|
'cancelled', 'canceled', 'registered', 'created', 'updated',
|
|
}
|
|
skip = {'state', 'input', 'metadata', 'conversation_history', 'history', 'session', 'session_metadata'}
|
|
|
|
def walk(value: Any, path: tuple[str, ...] = ()) -> None:
|
|
nonlocal completed, positive
|
|
if isinstance(value, dict):
|
|
for k, v in value.items():
|
|
key = str(k).lower()
|
|
if key in skip:
|
|
continue
|
|
if key == 'status' and str(v).strip().lower() == 'completed':
|
|
completed = True
|
|
if key in positive_keys and v is True:
|
|
positive = True
|
|
if key == 'status' and str(v).strip().lower() in positive_status:
|
|
positive = True
|
|
# Keys such as sms_sent/barcode are themselves meaningful
|
|
# evidence and can support paraphrased customer-facing text.
|
|
if key not in {'output', 'result', 'results'}:
|
|
facts.append(key)
|
|
walk(v, path + (key,))
|
|
elif isinstance(value, (list, tuple)):
|
|
for item in value[:50]:
|
|
walk(item, path)
|
|
elif value not in (None, ''):
|
|
facts.append(str(value))
|
|
|
|
for item in roots:
|
|
if not isinstance(item, dict) or item.get('ok') is False:
|
|
continue
|
|
if str(((item.get('result') or {}) if isinstance(item.get('result'), dict) else {}).get('status') or '').upper() == 'COMPLETED':
|
|
completed = True
|
|
walk(item.get('result', item))
|
|
return facts, completed, positive
|
|
|
|
@classmethod
|
|
def _candidate_is_structurally_grounded_completion(cls, text: str, context: dict[str, Any]) -> bool:
|
|
"""Allow a paraphrased completion only when current execution proves it.
|
|
|
|
This deliberately does not know tool names. It requires a completed
|
|
current execution, a positive side-effect marker, grounding for every
|
|
material number mentioned by the candidate, and at least one meaningful
|
|
textual fact/entity shared with the execution result.
|
|
"""
|
|
if not cls._candidate_has_operational_completion_claim(text):
|
|
return False
|
|
facts, completed, positive = cls._current_execution_scalar_evidence(context)
|
|
if not (completed and positive and facts):
|
|
return False
|
|
|
|
candidate = cls._normalized_text(text)
|
|
evidence_text = cls._normalized_text(' '.join(facts))
|
|
|
|
def canon_number(raw: str) -> str:
|
|
raw = raw.strip().replace(' ', '')
|
|
if ',' in raw and '.' in raw:
|
|
if raw.rfind(',') > raw.rfind('.'):
|
|
raw = raw.replace('.', '').replace(',', '.')
|
|
else:
|
|
raw = raw.replace(',', '')
|
|
else:
|
|
raw = raw.replace(',', '.')
|
|
try:
|
|
num = float(raw)
|
|
return (f'{num:.6f}').rstrip('0').rstrip('.')
|
|
except ValueError:
|
|
return re.sub(r'\D', '', raw)
|
|
|
|
candidate_numbers = [canon_number(x) for x in re.findall(r'(?<!\w)\d[\d .,-]*\d|(?<!\w)\d', candidate)]
|
|
evidence_numbers = {canon_number(x) for x in re.findall(r'(?<!\w)\d[\d .,-]*\d|(?<!\w)\d', evidence_text)}
|
|
material_numbers = [n for n in candidate_numbers if len(re.sub(r'\D', '', n)) >= 2]
|
|
if not material_numbers:
|
|
return False
|
|
if any(n not in evidence_numbers for n in material_numbers):
|
|
return False
|
|
|
|
# Require an entity/message overlap beyond generic success vocabulary.
|
|
candidate_words = set(re.findall(r'[a-zà-ÿ0-9+_-]{4,}', candidate, re.I))
|
|
stop = {
|
|
'sucesso', 'registrada', 'registrado', 'concluido', 'concluida',
|
|
'cliente', 'valor', 'protocolo', 'cobranca', 'contestacao', 'servico',
|
|
'mensal', 'novo', 'boleto', 'recebeu', 'detalhes',
|
|
}
|
|
evidence_words = set(re.findall(r'[a-zà-ÿ0-9+_-]{4,}', evidence_text, re.I))
|
|
meaningful_overlap = (candidate_words - stop) & (evidence_words - stop)
|
|
return bool(meaningful_overlap)
|
|
|
|
async def evaluate(self, text: str, context: dict[str, Any]) -> RailDecision:
|
|
# Strongest possible proof: the exact customer-facing candidate was emitted by
|
|
# the successful current-turn tool/workflow itself. REVPREC is about premature
|
|
# operational claims, so re-asking an LLM whether this exact authoritative result
|
|
# "really happened" only adds nondeterminism. Other safety rails still run.
|
|
if self._candidate_is_authoritative_output(text, context):
|
|
return RailDecision(
|
|
code=self.code,
|
|
allowed=True,
|
|
reason='current_execution_authoritative_output',
|
|
sanitized_text=text,
|
|
metadata={
|
|
'external': True,
|
|
'domain': 'TIM_CONTAS',
|
|
'mechanism': 'deterministic_current_execution_evidence',
|
|
'terminal_action': 'retry',
|
|
},
|
|
)
|
|
|
|
# REVPREC is not a general factuality rail. Ordinary descriptions and
|
|
# explanations contain no claim that an operation has already completed,
|
|
# so they are deterministically outside this rail's blocking scope.
|
|
if not self._candidate_has_operational_completion_claim(text):
|
|
return RailDecision(
|
|
code=self.code,
|
|
allowed=True,
|
|
reason='no_operational_completion_claim',
|
|
sanitized_text=text,
|
|
metadata={
|
|
'external': True,
|
|
'domain': 'TIM_CONTAS',
|
|
'mechanism': 'deterministic_claim_scope',
|
|
'terminal_action': 'retry',
|
|
},
|
|
)
|
|
|
|
# A composed/paraphrased response can still be proven by structured
|
|
# current-turn execution evidence even when it is not byte-for-byte equal
|
|
# to a workflow message. This removes nondeterministic false positives
|
|
# while keeping unsupported or contradictory claims on the semantic path.
|
|
if self._candidate_is_structurally_grounded_completion(text, context):
|
|
return RailDecision(
|
|
code=self.code,
|
|
allowed=True,
|
|
reason='current_execution_structurally_grounded',
|
|
sanitized_text=text,
|
|
metadata={
|
|
'external': True,
|
|
'domain': 'TIM_CONTAS',
|
|
'mechanism': 'deterministic_structured_execution_evidence',
|
|
'terminal_action': 'retry',
|
|
},
|
|
)
|
|
|
|
evidence = self._current_execution_evidence(context)
|
|
llm = _llm(context)
|
|
if llm is None:
|
|
return RailDecision(
|
|
code=self.code,
|
|
allowed=False,
|
|
reason='LLM do framework indisponível para guardrail TIM',
|
|
metadata={'external': True, 'fail_closed': True},
|
|
)
|
|
|
|
# For composed/paraphrased answers, ask the semantic rail the correct question:
|
|
# whether the operational claim is unsupported or contradicted by current-turn
|
|
# evidence. This replaces the obsolete assumption that reaching REVPREC means
|
|
# no tool ran.
|
|
evidence_text = json.dumps(evidence, ensure_ascii=False, default=str)[:14000]
|
|
prompt = self.prompt_builder(text or '', evidence_text)
|
|
raw = await llm.ainvoke(
|
|
[
|
|
{'role':'system','content':'Responda apenas com o dígito 1 ou 0, sem texto adicional.'},
|
|
{'role':'user','content':prompt},
|
|
],
|
|
profile_name=self.profile_name,
|
|
component_name='guardrail.external.tim_revprec',
|
|
generation_name='guardrail.external.tim_revprec',
|
|
)
|
|
output = str(getattr(raw, 'content', raw) or '').strip()
|
|
digits = [ch for ch in output if ch in '01']
|
|
# 1 = unsupported/contradicted operational completion claim; 0 = allowed.
|
|
allowed = not digits or digits[-1] != '1'
|
|
return RailDecision(
|
|
code=self.code,
|
|
allowed=allowed,
|
|
reason='' if allowed else 'resultado operacional afirmado sem suporte na evidência atual',
|
|
sanitized_text=text,
|
|
metadata={
|
|
'external': True,
|
|
'domain':'TIM_CONTAS',
|
|
'raw':output[:100],
|
|
'terminal_action':'retry',
|
|
'mechanism':'llm_current_execution_evidence',
|
|
'current_execution_evidence_count': len(evidence),
|
|
},
|
|
)
|
|
|
|
class TimPhraseologyRail(_TimPromptRail):
|
|
code='TIM_FRASEOLOGIA'; stage='output'; profile_name='grl'; prompt_builder=staticmethod(build_fraseologia_prompt)
|
|
|
|
async def evaluate(self, text: str, context: dict[str, Any]) -> RailDecision:
|
|
result = await super().evaluate(text, context)
|
|
result.metadata = {
|
|
**dict(result.metadata or {}),
|
|
'remediation': {
|
|
'type': 'rewrite', 'max_attempts': 1, 'prompt_id': 'FALLBACK',
|
|
'profile_name': 'grl', 'component_name': 'guardrail.wording.rewrite',
|
|
'generation_name': 'guardrail.wording.rewrite',
|
|
},
|
|
}
|
|
return result
|