mirror of
https://github.com/hoshikawa2/first_contas.git
synced 2026-07-09 10:14:20 +00:00
126 lines
4.1 KiB
Python
126 lines
4.1 KiB
Python
from __future__ import annotations
|
|
|
|
import re
|
|
|
|
from typing import Any
|
|
|
|
_INVOICE_EXPLANATION_TRAILER = "Com essa explicação, sanei sua dúvida?"
|
|
_INVOICE_EXPLANATION_TRAILER_PATTERN = re.compile(
|
|
r"com essa explica[cç][aã]o,?\s+sanei sua d[uú]vida\??$",
|
|
re.IGNORECASE,
|
|
)
|
|
_INVOICE_EXPLANATION_PADRAO_SENTINEL = re.compile(
|
|
r"\s*\(use o padr[aã]o\)\s*$",
|
|
re.IGNORECASE,
|
|
)
|
|
|
|
|
|
def _normalizar_mensagem_voz(
|
|
mensagem: str,
|
|
*,
|
|
trailer_override: str,
|
|
trailer_default: str = "",
|
|
trailer_pattern: re.Pattern[str] | None = None,
|
|
) -> str:
|
|
"""Pos-processa a mensagem reescrita pelo LLM.
|
|
|
|
Remove o sentinel "(use o padrao)" caso o modelo o tenha emitido e,
|
|
quando `trailer_override` estiver vazio, garante o `trailer_default`
|
|
ao final (a menos que o texto ja termine com `trailer_pattern`).
|
|
"""
|
|
if not mensagem:
|
|
return mensagem
|
|
if trailer_override:
|
|
return mensagem
|
|
|
|
mensagem = _INVOICE_EXPLANATION_PADRAO_SENTINEL.sub("", mensagem).strip()
|
|
if not trailer_default:
|
|
return mensagem
|
|
if trailer_pattern is not None and trailer_pattern.search(mensagem):
|
|
return mensagem
|
|
return f"{mensagem} {trailer_default}".strip()
|
|
|
|
|
|
def _normalizar_invoice_explanation_mensagem(
|
|
mensagem: str,
|
|
*,
|
|
trailer_override: str,
|
|
) -> str:
|
|
return _normalizar_mensagem_voz(
|
|
mensagem,
|
|
trailer_override=trailer_override,
|
|
trailer_default=_INVOICE_EXPLANATION_TRAILER,
|
|
trailer_pattern=_INVOICE_EXPLANATION_TRAILER_PATTERN,
|
|
)
|
|
|
|
|
|
def _extract_rag_context(state: dict[str, Any]) -> dict[str, Any]:
|
|
"""Extrai campos de RAG do estado para inclusão em eventos MPI."""
|
|
vars_state = state.get("vars", {})
|
|
if not isinstance(vars_state, dict):
|
|
return {}
|
|
|
|
# Procura por qualquer nó que tenha retornado dados de RAG.
|
|
# Prioriza os nós conhecidos por realizar busca RAG.
|
|
for node_id in ("buscar_informacao", "buscar_informacao_rag", "preparar"):
|
|
node_result = vars_state.get(node_id)
|
|
if not isinstance(node_result, dict):
|
|
continue
|
|
|
|
# Se o nó já tem os campos formatados, usa eles
|
|
if "ragRetrievedDocuments" in node_result:
|
|
return {
|
|
"ragRetrievedDocuments": node_result.get("ragRetrievedDocuments", ""),
|
|
"ragSelectedDocuments": node_result.get("ragSelectedDocuments", ""),
|
|
"noMatchRag": node_result.get("noMatchRag", True),
|
|
}
|
|
|
|
# Caso contrário, tenta construir a partir da lista 'documents'
|
|
documents = node_result.get("documents")
|
|
if isinstance(documents, (list, tuple)):
|
|
retrieved_titles = []
|
|
selected_titles = []
|
|
for doc in documents:
|
|
if not isinstance(doc, dict):
|
|
continue
|
|
t = (
|
|
doc.get("title_proc")
|
|
or doc.get("title")
|
|
or doc.get("chunk_texto")
|
|
or ""
|
|
)
|
|
if t:
|
|
title = str(t)
|
|
retrieved_titles.append(title)
|
|
# Threshold de 0.6 para considerar como selecionado (distância menor é melhor)
|
|
distance = float(doc.get("distance", 1.0))
|
|
if distance <= 0.6:
|
|
selected_titles.append(title)
|
|
|
|
return {
|
|
"ragRetrievedDocuments": "|".join(retrieved_titles),
|
|
"ragSelectedDocuments": "|".join(selected_titles),
|
|
"noMatchRag": len(documents) == 0,
|
|
}
|
|
|
|
return {}
|
|
|
|
|
|
def _extract_invoice_explanation_text(data: Any) -> str:
|
|
if isinstance(data, dict):
|
|
for key in ("mensagem", "message", "explicacao", "texto"):
|
|
value = data.get(key)
|
|
if isinstance(value, str) and value.strip():
|
|
return value.strip()
|
|
if isinstance(data, str) and data.strip():
|
|
return data.strip()
|
|
return ""
|
|
|
|
|
|
__all__ = [
|
|
'_normalizar_mensagem_voz',
|
|
'_normalizar_invoice_explanation_mensagem',
|
|
'_extract_rag_context',
|
|
'_extract_invoice_explanation_text',
|
|
]
|