mirror of
https://github.com/hoshikawa2/first_contas.git
synced 2026-07-09 18:24:20 +00:00
549 lines
18 KiB
Python
549 lines
18 KiB
Python
from __future__ import annotations
|
|
|
|
import logging
|
|
|
|
from datetime import (
|
|
datetime,
|
|
timezone,
|
|
)
|
|
from typing import Any
|
|
from agente_contas_tim.integrations import agent_framework_bridge
|
|
from agente_contas_tim.constants.ic_tags_enum import (
|
|
CVNTag,
|
|
MPITag,
|
|
VEBTag,
|
|
)
|
|
from agente_contas_tim.integrations.agent_event_metadata import build_billing_id_payload
|
|
from agente_contas_tim.integrations.rct_policy import RCTOperation, rct_tags_for_attempt
|
|
from agente_contas_tim.observability import get_session_id
|
|
from agente_contas_tim.workflows.actions.registry import (
|
|
WorkflowRuntimeContext,
|
|
workflow_action,
|
|
)
|
|
from agente_contas_tim.workflows.runtime_types import ActionResult
|
|
from agente_contas_tim.workflows.actions.common.helpers import (
|
|
_emit_ic,
|
|
_first_text_from_params_or_state,
|
|
_result_failed_or_missing_data,
|
|
_runtime_llm_callbacks,
|
|
_runtime_llm_metadata,
|
|
_to_dict,
|
|
)
|
|
from agente_contas_tim.workflows.actions.invoice_explanation.helpers import (
|
|
_extract_invoice_explanation_text,
|
|
_extract_rag_context,
|
|
_normalizar_invoice_explanation_mensagem,
|
|
)
|
|
|
|
logger = logging.getLogger("agente_contas_tim.workflows.actions.tim_actions")
|
|
|
|
|
|
_INVOICE_EXPLANATION_TRAILER = "Com essa explicação, sanei sua dúvida?"
|
|
|
|
|
|
def _metadata_value(metadata: dict[str, Any], *keys: str) -> Any:
|
|
for key in keys:
|
|
value = metadata.get(key)
|
|
if value is None:
|
|
continue
|
|
if isinstance(value, str) and not value.strip():
|
|
continue
|
|
return value
|
|
return None
|
|
|
|
|
|
def _int_or_none(value: Any) -> int | None:
|
|
if value is None or isinstance(value, bool):
|
|
return None
|
|
if isinstance(value, int):
|
|
return value
|
|
raw = str(value).strip()
|
|
if not raw:
|
|
return None
|
|
try:
|
|
return int(raw)
|
|
except ValueError:
|
|
return None
|
|
|
|
|
|
def _rct_api_metadata_from_result(
|
|
result: Any,
|
|
input_state: dict[str, Any],
|
|
) -> dict[str, Any]:
|
|
result_metadata = getattr(result, "metadata", None)
|
|
if not isinstance(result_metadata, dict):
|
|
return {}
|
|
|
|
metadata: dict[str, Any] = {}
|
|
api_url = _metadata_value(result_metadata, "api_url", "apiUrl")
|
|
if api_url is not None:
|
|
metadata["apiUrl"] = str(api_url)
|
|
|
|
status_code = _int_or_none(
|
|
_metadata_value(
|
|
result_metadata,
|
|
"api_status_code",
|
|
"apiStatusCode",
|
|
"status_code",
|
|
)
|
|
)
|
|
if status_code is not None:
|
|
metadata["apiStatusCode"] = status_code
|
|
|
|
response_payload = _metadata_value(
|
|
result_metadata,
|
|
"api_response_payload",
|
|
"apiResponsePayload",
|
|
"error_body",
|
|
)
|
|
if response_payload is not None:
|
|
metadata["apiResponsePayload"] = str(response_payload)
|
|
|
|
latency_ms = _int_or_none(
|
|
_metadata_value(result_metadata, "latency_ms", "latencyMs")
|
|
)
|
|
if latency_ms is not None:
|
|
metadata["latencyMs"] = max(0, latency_ms)
|
|
|
|
agent_specific_data = build_billing_id_payload(
|
|
{
|
|
"invoice_id": str(
|
|
input_state.get("invoice_id")
|
|
or input_state.get("current_invoice_number")
|
|
or ""
|
|
).strip()
|
|
}
|
|
)
|
|
if agent_specific_data != "{}":
|
|
metadata["agentSpecificData"] = agent_specific_data
|
|
return metadata
|
|
|
|
|
|
@workflow_action("invoice_explanation")
|
|
def invoice_explanation(
|
|
state: dict[str, Any],
|
|
params: dict[str, Any],
|
|
runtime: WorkflowRuntimeContext,
|
|
) -> ActionResult:
|
|
"""Retorna explicacao fixa da fatura."""
|
|
msisdn = _first_text_from_params_or_state(
|
|
state,
|
|
params,
|
|
"msisdn",
|
|
"Msisdn",
|
|
"MSISDN",
|
|
)
|
|
if not msisdn:
|
|
return ActionResult.fail("msisdn obrigatório para invoice_explanation")
|
|
|
|
result = runtime.factory.create_invoice_explanation(
|
|
msisdn=msisdn,
|
|
customer_id=_first_text_from_params_or_state(
|
|
state,
|
|
params,
|
|
"customer_id",
|
|
"customerId",
|
|
),
|
|
current_invoice_number=_first_text_from_params_or_state(
|
|
state,
|
|
params,
|
|
"current_invoice_number",
|
|
"currentInvoiceNumber",
|
|
"invoice_id",
|
|
"invoiceId",
|
|
),
|
|
past_invoice_number=_first_text_from_params_or_state(
|
|
state,
|
|
params,
|
|
"past_invoice_number",
|
|
"pastInvoiceNumber",
|
|
),
|
|
current_invoice_due_date=_first_text_from_params_or_state(
|
|
state,
|
|
params,
|
|
"current_invoice_due_date",
|
|
"currentInvoiceDueDate",
|
|
),
|
|
past_invoice_due_date=_first_text_from_params_or_state(
|
|
state,
|
|
params,
|
|
"past_invoice_due_date",
|
|
"pastInvoiceDueDate",
|
|
),
|
|
channel=_first_text_from_params_or_state(
|
|
state,
|
|
params,
|
|
"channel",
|
|
"Channel",
|
|
),
|
|
).execute()
|
|
if _result_failed_or_missing_data(result, state=state):
|
|
return ActionResult.fail(
|
|
result.error or "Falha ao montar explicacao da fatura",
|
|
**result.metadata,
|
|
)
|
|
return ActionResult.ok(_to_dict(result.data), **result.metadata)
|
|
|
|
|
|
@workflow_action("preparar_invoice_explanation")
|
|
def preparar_invoice_explanation(
|
|
state: dict[str, Any],
|
|
params: dict[str, Any],
|
|
runtime: WorkflowRuntimeContext,
|
|
) -> ActionResult:
|
|
"""Busca a explicacao bruta da fatura via InvoiceExplanationCommand.
|
|
|
|
Reaproveita a explicacao gerada na primeira passagem ao re-pausar
|
|
(caminho OUTRO), ou a explicacao_base recebida do prefetch/cache,
|
|
evitando reexecutar o command. Nao aplica vocalizacao nem
|
|
enriquecimento — esses passos ficam em `formatar_invoice_explanation`.
|
|
"""
|
|
input_state = state.get("input", {}) if isinstance(state, dict) else {}
|
|
if not isinstance(input_state, dict):
|
|
input_state = {}
|
|
|
|
tentativa_anterior = int(params.get("tentativa_anterior") or 0)
|
|
if tentativa_anterior == 0:
|
|
_emit_ic(CVNTag.INICIO_FLUXO, input_state)
|
|
|
|
vars_state = state.get("vars", {}) if isinstance(state, dict) else {}
|
|
cached = vars_state.get("preparar", {}) if isinstance(vars_state, dict) else {}
|
|
explicacao_base = ""
|
|
if isinstance(cached, dict):
|
|
previous = cached.get("explicacao_base")
|
|
if isinstance(previous, str) and previous.strip():
|
|
explicacao_base = previous.strip()
|
|
|
|
metadata: dict[str, Any] = {}
|
|
if not explicacao_base:
|
|
explicacao_base = _first_text_from_params_or_state(
|
|
state,
|
|
params,
|
|
"explicacao_base",
|
|
"invoice_explanation_base",
|
|
"invoiceExplanationBase",
|
|
)
|
|
|
|
if not explicacao_base:
|
|
msisdn = _first_text_from_params_or_state(
|
|
state,
|
|
params,
|
|
"msisdn",
|
|
"Msisdn",
|
|
"MSISDN",
|
|
)
|
|
if not msisdn:
|
|
# Pre-validacao falhou antes de chamar o servico → loop de retry
|
|
_emit_ic(CVNTag.PRE_VALIDACAO_FAIL, input_state)
|
|
return ActionResult.ok(
|
|
{
|
|
"success": False,
|
|
"service_failed": False,
|
|
"tentativa": tentativa_anterior + 1,
|
|
}
|
|
)
|
|
|
|
# Pre-validacao ok → intencao validada antes de chamar o servico
|
|
_emit_ic(CVNTag.PRE_VALIDACAO_OK, input_state)
|
|
|
|
result = runtime.factory.create_invoice_explanation(
|
|
msisdn=msisdn,
|
|
customer_id=_first_text_from_params_or_state(
|
|
state,
|
|
params,
|
|
"customer_id",
|
|
"customerId",
|
|
),
|
|
current_invoice_number=_first_text_from_params_or_state(
|
|
state,
|
|
params,
|
|
"current_invoice_number",
|
|
"currentInvoiceNumber",
|
|
"invoice_id",
|
|
"invoiceId",
|
|
),
|
|
past_invoice_number=_first_text_from_params_or_state(
|
|
state,
|
|
params,
|
|
"past_invoice_number",
|
|
"pastInvoiceNumber",
|
|
),
|
|
current_invoice_due_date=_first_text_from_params_or_state(
|
|
state,
|
|
params,
|
|
"current_invoice_due_date",
|
|
"currentInvoiceDueDate",
|
|
),
|
|
past_invoice_due_date=_first_text_from_params_or_state(
|
|
state,
|
|
params,
|
|
"past_invoice_due_date",
|
|
"pastInvoiceDueDate",
|
|
),
|
|
channel=_first_text_from_params_or_state(
|
|
state,
|
|
params,
|
|
"channel",
|
|
"Channel",
|
|
),
|
|
).execute()
|
|
if _result_failed_or_missing_data(result, state=state):
|
|
# Falha no servico → vai direto pro FIM, sem retry
|
|
_emit_ic(CVNTag.SERVICO_FAIL, input_state)
|
|
rct_metadata = _rct_api_metadata_from_result(result, input_state)
|
|
for tag in rct_tags_for_attempt(
|
|
RCTOperation.BASE_CONHECIMENTO,
|
|
tentativa_anterior + 1,
|
|
success=False,
|
|
):
|
|
_emit_ic(tag, input_state, extra_metadata=rct_metadata)
|
|
return ActionResult.ok(
|
|
{
|
|
"success": False,
|
|
"service_failed": True,
|
|
}
|
|
)
|
|
explicacao_base = _extract_invoice_explanation_text(_to_dict(result.data))
|
|
if not explicacao_base:
|
|
_emit_ic(CVNTag.SERVICO_FAIL, input_state)
|
|
rct_metadata = _rct_api_metadata_from_result(result, input_state)
|
|
for tag in rct_tags_for_attempt(
|
|
RCTOperation.BASE_CONHECIMENTO,
|
|
tentativa_anterior + 1,
|
|
success=False,
|
|
):
|
|
_emit_ic(tag, input_state, extra_metadata=rct_metadata)
|
|
return ActionResult.ok(
|
|
{
|
|
"success": False,
|
|
"service_failed": True,
|
|
}
|
|
)
|
|
_emit_ic(CVNTag.SERVICO_OK, input_state)
|
|
for tag in rct_tags_for_attempt(
|
|
RCTOperation.BASE_CONHECIMENTO,
|
|
tentativa_anterior + 1,
|
|
success=True,
|
|
):
|
|
_emit_ic(tag, input_state)
|
|
metadata = dict(result.metadata)
|
|
else:
|
|
# Cache hit — intencao ja validada sem nova chamada ao servico
|
|
_emit_ic(CVNTag.PRE_VALIDACAO_OK, input_state)
|
|
_emit_ic(CVNTag.SERVICO_OK, input_state)
|
|
for tag in rct_tags_for_attempt(
|
|
RCTOperation.BASE_CONHECIMENTO,
|
|
tentativa_anterior + 1,
|
|
success=True,
|
|
):
|
|
_emit_ic(tag, input_state)
|
|
|
|
return ActionResult.ok(
|
|
{
|
|
"success": True,
|
|
"explicacao_base": explicacao_base,
|
|
"tentativa": 0,
|
|
},
|
|
**metadata,
|
|
)
|
|
|
|
|
|
@workflow_action("checar_tentativa_cvn")
|
|
def checar_tentativa_cvn(
|
|
state: dict[str, Any],
|
|
params: dict[str, Any],
|
|
runtime: WorkflowRuntimeContext,
|
|
) -> ActionResult:
|
|
"""Verifica contador de tentativas do fluxo CVN e emite CVN.004 ou CVN.005."""
|
|
input_state = state.get("input", {}) if isinstance(state, dict) else {}
|
|
if not isinstance(input_state, dict):
|
|
input_state = {}
|
|
tentativa = int(params.get("tentativa") or 1)
|
|
if tentativa > 2:
|
|
_emit_ic(CVNTag.LIMITE_TENTATIVAS, input_state)
|
|
else:
|
|
_emit_ic(CVNTag.DENTRO_LIMITE, input_state)
|
|
return ActionResult.ok({"tentativa": tentativa})
|
|
|
|
|
|
@workflow_action("formatar_invoice_explanation")
|
|
def formatar_invoice_explanation(
|
|
state: dict[str, Any],
|
|
params: dict[str, Any],
|
|
runtime: WorkflowRuntimeContext,
|
|
) -> ActionResult:
|
|
"""Reescreve a explicacao bruta para voz, enriquecendo com juros/multas
|
|
a partir do invoice_detail, via capability LLM dedicada.
|
|
|
|
Recebe `explicacao_base` (obrigatorio) e `invoice_detail` (opcional)
|
|
nos params. Quando `trailer_override` for informado, instrui o LLM a
|
|
usar essa frase como fechamento (em vez da pergunta padrao). Util
|
|
para o caminho de reforco (OUTRO).
|
|
|
|
Em caso de falha do LLM, faz fallback para concatenacao simples
|
|
`explicacao_base + trailer` para nao bloquear o fluxo.
|
|
"""
|
|
explicacao_base = str(
|
|
params.get("explicacao_base", "") or ""
|
|
).strip()
|
|
if not explicacao_base:
|
|
return ActionResult.fail(
|
|
"explicacao_base obrigatoria para formatar_invoice_explanation"
|
|
)
|
|
|
|
invoice_detail = str(
|
|
params.get("invoice_detail", "") or ""
|
|
).strip()
|
|
trailer_override = str(
|
|
params.get("trailer_override", "") or ""
|
|
).strip()
|
|
fallback_trailer = (
|
|
trailer_override or _INVOICE_EXPLANATION_TRAILER
|
|
)
|
|
|
|
mensagem = ""
|
|
if runtime.llm_gateway is not None:
|
|
try:
|
|
llm_result = runtime.llm_gateway.execute(
|
|
capability_id="fluxo_invoice_explanation_reescrita",
|
|
variables={
|
|
"explicacao_base": explicacao_base,
|
|
"invoice_detail": invoice_detail or "(vazio)",
|
|
"trailer_override": trailer_override,
|
|
},
|
|
user_text=explicacao_base,
|
|
callbacks=_runtime_llm_callbacks(runtime),
|
|
tags=["workflow_action"],
|
|
metadata=_runtime_llm_metadata(runtime),
|
|
)
|
|
mensagem = str(
|
|
getattr(llm_result, "content", "") or ""
|
|
).strip()
|
|
mensagem = _normalizar_invoice_explanation_mensagem(
|
|
mensagem,
|
|
trailer_override=trailer_override,
|
|
)
|
|
except Exception:
|
|
logger.exception(
|
|
"formatar_invoice_explanation: falha ao invocar "
|
|
"capability fluxo_invoice_explanation_reescrita"
|
|
)
|
|
|
|
if not mensagem:
|
|
mensagem = f"{explicacao_base}\n\n{fallback_trailer}"
|
|
|
|
return ActionResult.ok(
|
|
{
|
|
"mensagem": mensagem,
|
|
"await_user_input": True,
|
|
}
|
|
)
|
|
|
|
|
|
@workflow_action("registrar_atendimento_invoice_explanation")
|
|
def registrar_atendimento_invoice_explanation(
|
|
state: dict[str, Any],
|
|
params: dict[str, Any],
|
|
runtime: WorkflowRuntimeContext,
|
|
) -> ActionResult:
|
|
"""Stub de registro do atendimento de invoice_explanation.
|
|
|
|
Implementacao real ainda nao definida; deixa apenas um warning para
|
|
indicar que o caminho foi exercitado. A resposta ao cliente fica a
|
|
cargo do orchestrator (LLM) no proximo round.
|
|
"""
|
|
resposta = params.get("resposta_usuario")
|
|
ic = str(params.get("ic", "")).strip()
|
|
resposta_norm = str(resposta or "").strip().upper()
|
|
ic_extra = ""
|
|
cvn_ic = ""
|
|
if resposta_norm == "SIM":
|
|
ic_extra = MPITag.EXPLICACAO_SIM
|
|
cvn_ic = CVNTag.CLIENTE_CONCORDOU
|
|
elif resposta_norm == "NAO":
|
|
ic_extra = MPITag.EXPLICACAO_NAO
|
|
cvn_ic = CVNTag.CLIENTE_NAO_CONCORDOU
|
|
logger.warning(
|
|
"registrar_atendimento_invoice_explanation: registro ainda nao "
|
|
"implementado (resposta_usuario=%s)",
|
|
resposta,
|
|
)
|
|
if cvn_ic:
|
|
input_state = state.get("input", {}) if isinstance(state, dict) else {}
|
|
if not isinstance(input_state, dict):
|
|
input_state = {}
|
|
_emit_ic(cvn_ic, input_state)
|
|
if ic_extra:
|
|
try:
|
|
input_state = (
|
|
state.get("input", {}) if isinstance(state, dict) else {}
|
|
)
|
|
if not isinstance(input_state, dict):
|
|
input_state = {}
|
|
rag_info = _extract_rag_context(state)
|
|
message_id = _first_text_from_params_or_state(
|
|
state,
|
|
params,
|
|
"message_id",
|
|
"messageId",
|
|
)
|
|
channel_id = str(
|
|
input_state.get("channel_id")
|
|
or input_state.get("channelId")
|
|
or "URA"
|
|
).strip() or "URA"
|
|
event_date = int(datetime.now(timezone.utc).timestamp() * 1000)
|
|
customer_message = str(
|
|
input_state.get("customer_message")
|
|
or input_state.get("customerMessage")
|
|
or resposta
|
|
or ""
|
|
).strip()
|
|
metadata = {
|
|
"sessionId": get_session_id(),
|
|
"tag": ic_extra,
|
|
"eventDate": event_date,
|
|
"uraCallId": str(
|
|
input_state.get("ura_call_id", "")
|
|
).strip(),
|
|
"ani": str(input_state.get("ani", "")).strip(),
|
|
"gsm": str(input_state.get("msisdn", "")).strip(),
|
|
"agentId": "contas",
|
|
"channelId": channel_id,
|
|
"ragRetrievedDocuments": str(
|
|
rag_info.get("ragRetrievedDocuments", "")
|
|
),
|
|
"ragSelectedDocuments": str(
|
|
rag_info.get("ragSelectedDocuments", "")
|
|
),
|
|
"noMatchRag": bool(rag_info.get("noMatchRag", True)),
|
|
"llmResponse": str(input_state.get("llm_response", "") or ""),
|
|
"customerMessage": customer_message,
|
|
}
|
|
if message_id:
|
|
metadata["messageId"] = message_id
|
|
agent_framework_bridge.event(ic_extra, metadata=metadata)
|
|
if ic == VEBTag.EXPLICACAO_ACEITA:
|
|
veb_metadata = dict(metadata)
|
|
veb_metadata["tag"] = ic
|
|
agent_framework_bridge.event(ic, metadata=veb_metadata)
|
|
except Exception:
|
|
logger.debug(
|
|
"registrar_atendimento_invoice_explanation: falha ao emitir %s",
|
|
ic_extra,
|
|
exc_info=True,
|
|
)
|
|
payload: dict[str, Any] = {}
|
|
if ic:
|
|
payload["ic"] = ic
|
|
return ActionResult.ok(payload)
|
|
|
|
|
|
__all__ = [
|
|
'invoice_explanation',
|
|
'preparar_invoice_explanation',
|
|
'checar_tentativa_cvn',
|
|
'formatar_invoice_explanation',
|
|
'registrar_atendimento_invoice_explanation',
|
|
]
|