Files
agent_contas/app/extensions/tim_guardrails.py
2026-09-01 11:24:03 -03:00

494 lines
22 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 _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}',
},
},
)
# 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',
}
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)
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',
},
)
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