292 lines
13 KiB
Python
292 lines
13 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 _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 _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',
|
|
},
|
|
},
|
|
)
|
|
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}',
|
|
},
|
|
},
|
|
)
|
|
return await super().evaluate(text, context)
|
|
|
|
class TimPrematureActionRail(_TimPromptRail):
|
|
code='TIM_REVPREC'; stage='output'; profile_name='grl'; prompt_builder=staticmethod(build_revprec_prompt)
|
|
|
|
@staticmethod
|
|
def _structured_insufficient_evidence_message(text: str, context: dict[str, Any]) -> bool:
|
|
"""Allow an epistemically conservative tool/workflow answer without LLM re-judgment.
|
|
|
|
The bypass is intentionally structural and exact-message based. An arbitrary
|
|
assistant sentence saying "não sei" is *not* enough: the current-turn tool
|
|
result must explicitly declare ``epistemic_status=insufficient_evidence`` and
|
|
expose the same ``mensagem`` that is being sent to the customer.
|
|
"""
|
|
expected = " ".join(str(text or "").split())
|
|
if not expected:
|
|
return False
|
|
|
|
roots = []
|
|
ctx = context or {}
|
|
for key in ('mcp_results', 'tool_result', 'evidence'):
|
|
value = ctx.get(key)
|
|
if value is not None:
|
|
roots.append(value)
|
|
|
|
def walk(value: Any) -> bool:
|
|
if isinstance(value, dict):
|
|
status = str(value.get('epistemic_status') or '').strip().lower()
|
|
message = " ".join(str(value.get('mensagem') or '').split())
|
|
if status == 'insufficient_evidence' and message and message == expected:
|
|
return True
|
|
return any(walk(item) for item in value.values())
|
|
if isinstance(value, (list, tuple)):
|
|
return any(walk(item) for item in value)
|
|
return False
|
|
|
|
return any(walk(root) for root in roots)
|
|
|
|
async def evaluate(self, text: str, context: dict[str, Any]) -> RailDecision:
|
|
if self._structured_insufficient_evidence_message(text, context):
|
|
return RailDecision(
|
|
code=self.code,
|
|
allowed=True,
|
|
reason='insufficient_evidence_non_assertive',
|
|
sanitized_text=text,
|
|
metadata={
|
|
'external': True,
|
|
'domain': 'TIM_CONTAS',
|
|
'mechanism': 'deterministic_epistemic_bypass',
|
|
'epistemic_status': 'insufficient_evidence',
|
|
},
|
|
)
|
|
|
|
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 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']
|
|
# Prompt original: 1 = violação / verbalização prematura; 0 = permitido.
|
|
allowed = not digits or digits[-1] != '1'
|
|
return RailDecision(code=self.code, allowed=allowed, reason='' if allowed else 'verbalização prematura segundo política TIM Contas', sanitized_text=text, metadata={'external':True,'domain':'TIM_CONTAS','raw':output[:100], 'terminal_action':'retry'})
|
|
|
|
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
|