File size: 4,933 Bytes
88b683e |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 |
"""
Nó para refinamento de respostas
"""
import logging
from typing import Dict, Any
from agents.tools import refine_response_with_llm
async def refine_response_node(state: Dict[str, Any]) -> Dict[str, Any]:
"""
Nó para refinar a resposta usando LLM adicional
Args:
state: Estado atual do agente
Returns:
Estado atualizado com resposta refinada
"""
if not state.get("advanced_mode", False) or state.get("error"):
# Pula refinamento se modo avançado desabilitado ou há erro
logging.info("[REFINE] Pulando refinamento - modo avançado desabilitado ou erro presente")
return state
logging.info("[REFINE] Iniciando refinamento da resposta")
try:
original_response = state.get("response", "")
user_input = state.get("user_input", "")
if not original_response or not user_input:
logging.warning("[REFINE] Resposta ou entrada do usuário não disponível")
return state
# Refina resposta com LLM adicional
refined_response = await refine_response_with_llm(
user_input,
original_response
)
# Atualiza estado com resposta refinada
state["response"] = refined_response
state["refined"] = True
logging.info("[REFINE] Resposta refinada com sucesso")
except Exception as e:
error_msg = f"Erro ao refinar resposta: {e}"
logging.error(f"[REFINE] {error_msg}")
# Mantém resposta original em caso de erro
state["refinement_error"] = error_msg
return state
async def check_refinement_quality_node(state: Dict[str, Any]) -> Dict[str, Any]:
"""
Nó para verificar qualidade do refinamento
Args:
state: Estado com resposta refinada
Returns:
Estado com avaliação da qualidade
"""
try:
original_response = state.get("sql_result", {}).get("output", "")
refined_response = state.get("response", "")
if not state.get("refined", False):
state["refinement_quality"] = "not_refined"
return state
# Métricas simples de qualidade
quality_metrics = {
"length_increase": len(refined_response) - len(original_response),
"has_insights": any(word in refined_response.lower() for word in [
"insight", "análise", "interpretação", "conclusão", "tendência"
]),
"has_statistics": any(word in refined_response.lower() for word in [
"média", "total", "percentual", "proporção", "estatística"
]),
"improved": len(refined_response) > len(original_response) * 1.1
}
# Determina qualidade geral
if quality_metrics["improved"] and (quality_metrics["has_insights"] or quality_metrics["has_statistics"]):
quality_score = "high"
elif quality_metrics["length_increase"] > 0:
quality_score = "medium"
else:
quality_score = "low"
state["refinement_quality"] = quality_score
state["quality_metrics"] = quality_metrics
logging.info(f"[REFINE] Qualidade avaliada: {quality_score}")
except Exception as e:
logging.error(f"[REFINE] Erro ao avaliar qualidade: {e}")
state["refinement_quality"] = "error"
return state
async def format_final_response_node(state: Dict[str, Any]) -> Dict[str, Any]:
"""
Nó para formatação final da resposta
Args:
state: Estado com resposta processada
Returns:
Estado com resposta formatada
"""
try:
response = state.get("response", "")
execution_time = state.get("execution_time", 0.0)
advanced_mode = state.get("advanced_mode", False)
refined = state.get("refined", False)
# Adiciona informações de contexto se necessário
if advanced_mode and refined:
quality = state.get("refinement_quality", "unknown")
if quality == "high":
response += "\n\n💡 *Resposta aprimorada com análise avançada*"
elif quality == "medium":
response += "\n\n🔍 *Resposta complementada*"
# Adiciona tempo de execução se significativo
if execution_time > 2.0:
response += f"\n\n⏱️ *Processado em {execution_time:.1f}s*"
# Formatação final
state["response"] = response.strip()
state["formatted"] = True
logging.info(f"[FORMAT] Resposta formatada - {len(response)} caracteres")
except Exception as e:
logging.error(f"[FORMAT] Erro na formatação: {e}")
# Mantém resposta original se houver erro na formatação
return state
|