86 lines
4.3 KiB
Python
86 lines
4.3 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 _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)
|
|
|
|
class TimProactiveOfferRail(_TimPromptRail):
|
|
code='TIM_AOFERTA'; stage='output'; prompt_builder=staticmethod(build_aoferta_prompt)
|
|
|
|
class TimPrematureActionRail(_TimPromptRail):
|
|
code='TIM_REVPREC'; stage='output'; profile_name='grl'; prompt_builder=staticmethod(build_revprec_prompt)
|
|
|
|
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 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
|