from __future__ import annotations from typing import Any from app.agents.prompting import apply_agent_profile_prompt from app.identity_extraction import extract_identity_from_text from app.agents.runtime import AgentRuntimeMixin class BackofficeAgent(AgentRuntimeMixin): """Agente Backoffice/ANATEL usando somente contratos nativos do framework. Este agente não contém grafo próprio, node legado ou orquestração paralela ao framework. A execução segue o padrão uniforme: 1. LangGraph do framework faz guardrails de entrada, roteamento e checkpoint. 2. EnterpriseRouter injeta intent, route e mcp_tools a partir de YAML. 3. AgentRuntimeMixin executa MCP Tool Router, RAG, cache, IC/NOC/GRL. 4. LLM recebe contexto já normalizado e gera a resposta. 5. Workflow do framework aplica OutputSupervisor, output guardrails, judges e persistência. Regras específicas do domínio ficam em prompt_policy/routing/tools/mcp mapping, não em um fluxo particular dentro do agente. """ name = "backoffice_agent" _domain_graph_cache: dict[str, Any] = {} DEFAULT_TOOLS_BY_INTENT: dict[str, list[str]] = { "backoffice_anatel_triage": [ "consultar_reclamacao", "consultar_cliente_backoffice", "consultar_siebel_caso", "consultar_imdb_cliente", "consultar_speech_analytics", "consultar_tais_kb", "consultar_abrt", "consultar_portabilidade", ], "backoffice_customer_lookup": [ "consultar_cliente_backoffice", "consultar_imdb_cliente", "consultar_portabilidade", ], "backoffice_action_register": [ "consultar_reclamacao", "consultar_siebel_caso", "registrar_acao_backoffice", "registrar_acao_siebel", ], "backoffice_response_emulator": [ "consultar_reclamacao", "buscar_templates_emulador", "gerar_rascunho_emulador", ], } def __init__(self, llm, telemetry=None, tool_router=None, rag_service=None, cache=None, settings=None, observer=None): self.llm = llm self.telemetry = telemetry self.tool_router = tool_router self.rag_service = rag_service self.cache = cache self.settings = settings self.observer = observer async def run(self, state: dict[str, Any]) -> dict[str, Any]: await self._emit_ic( "AGA.001", state, { "status": "Entrada recebida pelo agente nativo do framework", "framework_native": True, "business_component": "backoffice_anatel", }, component="agent.backoffice.native.start", ) normalized_state = self._normalize_state_tools(state) await self._emit_ic( "AGA.018", normalized_state, { "status": "Contexto canônico validado pelo framework", "missing_fields": self._missing_required_context(normalized_state), "framework_native": True, }, component="agent.backoffice.native.context", ) if self._has_original_ticket_context(normalized_state): await self._emit_ic( "BACKOFFICE_TICKET_CONTEXT_DETECTED", normalized_state, { "status": "Contexto de ticket detectado; execução operacional deve entrar como canal backoffice_rest via ChannelGateway/dispatcher", "framework_native": True, "reason": "o agente conversacional não compila nem executa grafos de domínio", }, component="agent.backoffice.native.ticket_context", ) rag_context, rag_metadata = await self._retrieve_rag_context(normalized_state) if rag_metadata.get("enabled"): await self._emit_ic( "AGA.012", normalized_state, { "status": "RAG/TAIS KB consultado pelo serviço nativo do framework", "document_count": rag_metadata.get("document_count", 0), "top_document_ids": rag_metadata.get("top_document_ids", []), }, component="agent.backoffice.native.rag", ) mcp_results = await self._execute_tools_by_intent(normalized_state) await self._emit_ic( "AGA.014", normalized_state, { "status": "Ferramentas MCP selecionadas pelo roteador do framework", "intent": normalized_state.get("intent"), "tool_count": len(mcp_results), "tools": [r.get("tool_name") or r.get("tool") for r in mcp_results], }, component="agent.backoffice.native.tools", ) answer = await self._generate_answer(normalized_state, rag_context, rag_metadata, mcp_results) await self._emit_ic( "AGA.043", normalized_state, { "status": "Resposta produzida pelo agente nativo e entregue ao workflow do framework", "answer_chars": len(answer or ""), "framework_native": True, }, component="agent.backoffice.native.completed", ) await self._emit_noc( "001", normalized_state, {"message": "Backoffice native agent completed", "type": "INFO"}, component="agent.backoffice.native.completed", ) siebel = self._extract_siebel_result(mcp_results) return { "answer": answer, "next_state": self._next_state(normalized_state, mcp_results), "mcp_results": mcp_results, "rag_metadata": rag_metadata, "crmProtocol": siebel.get("crmProtocol"), "fieldsToUpdate": siebel.get("fieldsToUpdate"), "transitions": siebel.get("transitions"), "metadata": { **(normalized_state.get("metadata") or {}), "siebel_response": siebel.get("siebel_response"), }, "framework_native": { "agent": self.name, "intent": normalized_state.get("intent"), "route": normalized_state.get("route"), "tools": normalized_state.get("mcp_tools") or [], }, } def _normalize_state_tools(self, state: dict[str, Any]) -> dict[str, Any]: intent = state.get("intent") or "backoffice_anatel_triage" configured_tools = list(state.get("mcp_tools") or []) default_tools = self.DEFAULT_TOOLS_BY_INTENT.get(intent, self.DEFAULT_TOOLS_BY_INTENT["backoffice_anatel_triage"]) tools = configured_tools or default_tools # Garante ordem estável e remove duplicidade sem perder a configuração do roteador. seen: set[str] = set() deduped = [] for tool in tools: if tool and tool not in seen: seen.add(tool) deduped.append(tool) return { **state, "route": state.get("route") or self.name, "active_agent": self.name, "intent": intent, "mcp_tools": deduped, } def _missing_required_context(self, state: dict[str, Any]) -> list[str]: ctx = state.get("context") or {} bc = ctx.get("business_context") or state.get("business_context") or {} interaction = (bc.get("interaction_key") if isinstance(bc, dict) else None) or ctx.get("interaction_key") if not interaction: interaction = ( ctx.get("protocol_id") or ctx.get("protocolo") or ctx.get("complaint_id") or ctx.get("complaintProtocol") or ctx.get("transactionId") or ctx.get("transaction_id") or ctx.get("ticket_id") ) return [] if interaction else ["interaction_key"] async def _execute_tools_by_intent(self, state: dict[str, Any]) -> list[dict[str, Any]]: intent = state.get("intent") or "backoffice_anatel_triage" tools = list(state.get("mcp_tools") or []) results: list[dict[str, Any]] = [] for tool in tools: arguments = self._tool_arguments(tool, intent, state) if tool.startswith("registrar_") and not arguments.get("action_text"): # O framework não deve registrar ação operacional sem texto explícito. results.append({ "ok": False, "tool_name": tool, "skipped": True, "reason": "action_text ausente; registro não executado por segurança operacional", }) await self._emit_ic( "AGA.008", state, {"status": "Registro operacional não executado sem action_text", "tool_name": tool}, component="agent.backoffice.native.tool.skip", ) continue result = await self._call_mcp_tool(tool, arguments, state) results.append(result) if tool in {"consultar_speech_analytics"}: await self._emit_ic("AGA.010", state, {"status": "Speech Analytics consultado", "tool_ok": result.get("ok")}, component="agent.backoffice.native.speech") elif tool in {"consultar_imdb_cliente", "consultar_cliente_backoffice"}: await self._emit_ic("AGA.011", state, {"status": "Cliente/IMDB consultado", "tool_ok": result.get("ok")}, component="agent.backoffice.native.customer") elif tool in {"consultar_tais_kb", "buscar_templates_emulador"}: await self._emit_ic("AGA.020", state, {"status": "Templates/TAIS consultados", "tool_ok": result.get("ok")}, component="agent.backoffice.native.templates") elif tool in {"registrar_acao_backoffice", "registrar_acao_siebel"}: await self._emit_ic("AGA.006", state, {"status": "Ação operacional solicitada", "tool_ok": result.get("ok")}, component="agent.backoffice.native.action") return results def _tool_arguments(self, tool: str, intent: str, state: dict[str, Any]) -> dict[str, Any]: ctx = state.get("context") or {} # Para query/prompt usamos o texto sanitizado; para extração de CPF/CNPJ # precisamos do texto original, pois o guardrail MSK mascara o documento. text = state.get("sanitized_input") or state.get("user_text") or "" original_text = ( ctx.get("message") or ctx.get("text") or ctx.get("query") or state.get("user_text") or text or "" ) bc = ctx.get("business_context") or state.get("business_context") or {} explicit = ctx.get("tool_arguments") or state.get("tool_arguments") or {} def pick(*names: str) -> Any: for name in names: cur: Any = None if name in explicit: cur = explicit.get(name) elif isinstance(bc, dict) and name in bc: cur = bc.get(name) elif name in ctx: cur = ctx.get(name) elif name in state: cur = state.get(name) if cur not in (None, "", {}, []): return cur return None protocol_id = pick( "protocol_id", "protocolo", "interaction_key", "complaint_id", "complaintProtocol", "transactionId", "transaction_id", "ticket_id", ) extracted_identity = extract_identity_from_text(str(original_text)) # Identificador explicitamente digitado na mensagem atual prevalece sobre # defaults da sessão/front (ex.: user_id/msisdn 11999999999). customer_key = extracted_identity.get("customer_key") or pick("customer_key", "cpf", "cnpj", "document", "msisdn", "customer_id", "cpf_hash", "document_hash", "user_id") if not protocol_id: protocol_id = extracted_identity.get("protocol_id") or extracted_identity.get("interaction_key") contract_key = pick("contract_key", "contract_id", "account_id", "asset_id") session_key = pick("session_key", "conversation_key", "session_id") # Extra args são passados ao MCPToolRouter. O mapper do framework ainda # aplica mcp_parameter_mapping.yaml, mas estes aliases tornam a chamada # resiliente para tools que exigem protocol_id diretamente. common = { "query": text, "operator_instructions": text, "selected_actions": explicit.get("selected_actions") or [], } if protocol_id: common["protocol_id"] = str(protocol_id) common["interaction_key"] = str(protocol_id) if customer_key: common["customer_key"] = str(customer_key) # Mantém aliases documentais para tools TIM que aceitam CPF/CNPJ/document. for identity_key in ("cpf", "cnpj", "document", "document_type"): if extracted_identity.get(identity_key): common[identity_key] = str(extracted_identity[identity_key]) if contract_key: common["contract_key"] = str(contract_key) if session_key: common["session_key"] = str(session_key) common["operator_session"] = str(session_key) if tool.startswith("registrar_"): action_text = explicit.get("action_text") if not action_text and intent == "backoffice_action_register": action_text = text common["action_text"] = action_text return common async def _generate_answer( self, state: dict[str, Any], rag_context: str, rag_metadata: dict[str, Any], mcp_results: list[dict[str, Any]], ) -> str: missing = self._missing_required_context(state) if missing: return ( "[BackofficeAgent] Para seguir no padrão do framework, preciso do identificador canônico " f"{', '.join(missing)}. Informe o protocolo/reclamação/chamado ou configure o mapping em identity.yaml." ) system_prompt = apply_agent_profile_prompt(state, self._system_prompt()) messages = [ {"role": "system", "content": system_prompt}, { "role": "user", "content": ( "Mensagem do usuário:\n" f"{state.get('sanitized_input') or state.get('user_text') or ''}\n\n" "Intent/rota escolhidos pelo framework:\n" f"intent={state.get('intent')} route={state.get('route')}\n\n" "Resultados MCP normalizados pelo framework:\n" f"{mcp_results}\n\n" "Contexto RAG nativo do framework:\n" f"{rag_context or '[sem contexto RAG]'}\n\n" "Metadados RAG:\n" f"{rag_metadata}" ), }, ] try: generated = await self._invoke_llm_cached(state, "BackofficeNativeAgent", messages) return f"[BackofficeAgent] {generated}" except Exception as exc: await self._emit_noc( "005", state, {"message": f"LLM failed in native backoffice agent: {type(exc).__name__}: {exc}", "type": "FAILURE"}, component="agent.backoffice.native.llm", ) return self._fallback_answer(state, mcp_results) def _has_original_ticket_context(self, state: dict[str, Any]) -> bool: ctx = state.get("context") or {} rc = ctx.get("request_context") or ctx return bool( rc.get("transactionId") and isinstance(rc.get("complaint") or {}, dict) and isinstance(rc.get("customer") or {}, dict) ) async def _run_original_checklist_workflow(self, state: dict[str, Any]) -> dict[str, Any]: """Deprecated compatibility stub. A migração framework-native não permite que o agente conversacional compile/execute ``src.agent.graphs`` diretamente. Os fluxos checklist e response emulator são executados pelo dispatcher de workflows após normalização pelo canal backoffice_rest/ChannelGateway. """ return { "executed": False, "error": { "type": "DeprecatedDirectGraphExecution", "message": "Use BackofficeRestChannelAdapter + BackofficeWorkflowDispatcher", }, } def _system_prompt(self) -> str: return """ Você é o agente Backoffice/ANATEL executando exclusivamente pelo framework corporativo. Use apenas estes insumos: - BusinessContext canônico do framework. - Ferramentas MCP selecionadas pelo EnterpriseRouter. - Contexto RAG retornado pelo RagService do framework. - Memória/checkpoint já carregados pelo workflow. Regras obrigatórias: 1. Não invente protocolo, cliente, contrato, status, SLA, parecer ou ação Siebel. 2. Se uma ferramenta MCP retornar erro ou ausência de dados, diga exatamente que a evidência não foi encontrada. 3. Não confirme registro de ação sem retorno ok/registered da tool correspondente. 4. Se faltar protocolo/chamado/reclamação, peça somente esse identificador. 5. Responda de forma operacional, curta e rastreável. 6. A resposta será validada por OutputSupervisor, guardrails de saída e judges do framework. """.strip() def _fallback_answer(self, state: dict[str, Any], mcp_results: list[dict[str, Any]]) -> str: ok_tools = [r.get("tool_name") or r.get("tool") for r in mcp_results if r.get("ok")] failed_tools = [r.get("tool_name") or r.get("tool") for r in mcp_results if not r.get("ok")] return ( "[BackofficeAgent] Fluxo nativo executado pelo framework. " f"Intent: {state.get('intent')}. " f"Tools com sucesso: {ok_tools or 'nenhuma'}. " f"Tools pendentes/erro: {failed_tools or 'nenhuma'}. " "A resposta final não foi enriquecida pelo LLM porque houve falha controlada nessa etapa." ) def _next_state(self, state: dict[str, Any], mcp_results: list[dict[str, Any]]) -> str: if self._missing_required_context(state): return "WAITING_BACKOFFICE_IDENTIFIER" if any(r.get("skipped") for r in mcp_results): return "BACKOFFICE_WAITING_ACTION_TEXT" return "BACKOFFICE_ACTIVE" def _extract_siebel_result(self, mcp_results: list[dict[str, Any]]) -> dict[str, Any]: for item in mcp_results: tool = item.get("tool_name") or item.get("tool") if tool in {"registrar_acao_siebel", "abrir_chamado_siebel", "backOfficeSRopening"}: result = item.get("result") or item.get("data") or item return { "crmProtocol": result.get("crmProtocol"), "fieldsToUpdate": result.get("fieldsToUpdate"), "transitions": result.get("transitions") or [], "siebel_response": result, } return {}