File size: 3,264 Bytes
6b29104 |
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 |
"""
Sistema de logging para o AgentPDF.
Fornece logging estruturado e colorido para melhor debugging
e monitoramento do sistema.
"""
import logging
import sys
from datetime import datetime
from typing import Optional
class ColoredFormatter(logging.Formatter):
"""Formatter personalizado com cores para diferentes níveis de log."""
# Códigos de cores ANSI
COLORS = {
'DEBUG': '\033[36m', # Ciano
'INFO': '\033[32m', # Verde
'WARNING': '\033[33m', # Amarelo
'ERROR': '\033[31m', # Vermelho
'CRITICAL': '\033[35m', # Magenta
'RESET': '\033[0m' # Reset
}
def format(self, record):
# Adiciona cor baseada no nível
color = self.COLORS.get(record.levelname, self.COLORS['RESET'])
reset = self.COLORS['RESET']
# Formato personalizado
record.levelname = f"{color}{record.levelname}{reset}"
return super().format(record)
def setup_logger(name: str = "AgentPDF", level: str = "INFO") -> logging.Logger:
"""
Configura e retorna um logger personalizado.
Args:
name: Nome do logger
level: Nível de logging (DEBUG, INFO, WARNING, ERROR, CRITICAL)
Returns:
logging.Logger: Logger configurado
"""
logger = logging.getLogger(name)
# Evita duplicação de handlers
if logger.handlers:
return logger
# Configura nível
numeric_level = getattr(logging, level.upper(), logging.INFO)
logger.setLevel(numeric_level)
# Handler para console
console_handler = logging.StreamHandler(sys.stdout)
console_handler.setLevel(numeric_level)
# Formatter com cores
formatter = ColoredFormatter(
'%(asctime)s | %(levelname)s | %(name)s | %(message)s',
datefmt='%H:%M:%S'
)
console_handler.setFormatter(formatter)
logger.addHandler(console_handler)
return logger
def log_node_execution(node_name: str, status: str, details: Optional[str] = None):
"""
Log específico para execução de nós do LangGraph.
Args:
node_name: Nome do nó
status: Status da execução (START, SUCCESS, ERROR)
details: Detalhes adicionais
"""
logger = logging.getLogger("AgentPDF.Nodes")
emoji_map = {
"START": "🚀",
"SUCCESS": "✅",
"ERROR": "❌",
"PROCESSING": "⚙️"
}
emoji = emoji_map.get(status, "📝")
message = f"{emoji} {node_name} - {status}"
if details:
message += f" | {details}"
if status == "ERROR":
logger.error(message)
elif status == "START" or status == "PROCESSING":
logger.info(message)
else:
logger.info(message)
def log_graph_execution(action: str, details: Optional[str] = None):
"""
Log específico para execução do grafo principal.
Args:
action: Ação sendo executada
details: Detalhes adicionais
"""
logger = logging.getLogger("AgentPDF.Graph")
message = f"🔄 {action}"
if details:
message += f" | {details}"
logger.info(message)
# Logger principal do sistema
main_logger = setup_logger("AgentPDF", "INFO")
|