Spaces:
Runtime error
Runtime error
File size: 4,837 Bytes
f0322a6 |
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 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 |
"""
Classe Abstrata Base para Processadores
Define interface padrão para todos os especialistas
"""
from abc import ABC, abstractmethod
from typing import Dict, Any, Optional, List
from datetime import datetime
import logging
import json
logger = logging.getLogger(__name__)
class ProcessorBase(ABC):
"""
Classe abstrata que define a interface padrão para todos os processadores
Cada especialista herda desta classe e implementa:
- process(): lógica específica do especialista
- validate(): validação de saída
- get_schema(): schema JSON esperado
"""
def __init__(self, specialist_id: int, specialist_name: str, llm_model=None):
"""
Inicializa processador base
Args:
specialist_id: ID do especialista (1-9)
specialist_name: Nome descritivo
llm_model: Modelo LLM para usar (Gemini, OpenAI, etc)
"""
self.specialist_id = specialist_id
self.specialist_name = specialist_name
self.llm_model = llm_model
self.execution_time = 0
self.confidence_score = 0
self.errors = []
self.warnings = []
@abstractmethod
def process(self, acordao_data: Dict[str, Any]) -> Dict[str, Any]:
"""
Processa dados do acórdão e retorna resultado específico do especialista
Args:
acordao_data: Dados estruturados do acórdão
Returns:
Dicionário com resultado da análise
"""
pass
@abstractmethod
def validate(self, result: Dict[str, Any]) -> bool:
"""
Valida resultado contra schema esperado
Args:
result: Resultado a validar
Returns:
True se válido, False caso contrário
"""
pass
@abstractmethod
def get_schema(self) -> Dict[str, Any]:
"""
Retorna schema JSON esperado para validação
Returns:
Schema JSON format
"""
pass
def pre_process(self, raw_data: str) -> Dict[str, Any]:
"""
Pré-processa dados brutos (limpeza, formatação)
Args:
raw_data: Dados brutos (texto do acórdão)
Returns:
Dados estruturados
"""
return {
"raw_text": raw_data.strip(),
"length": len(raw_data),
"processed_at": datetime.now().isoformat()
}
def post_process(self, result: Dict[str, Any]) -> Dict[str, Any]:
"""
Pós-processa resultado antes de retornar
Args:
result: Resultado bruto do processamento
Returns:
Resultado formatado e enriquecido
"""
enriched = {
"specialist_id": self.specialist_id,
"specialist_name": self.specialist_name,
"result": result,
"execution_time": self.execution_time,
"confidence_score": self.confidence_score,
"errors": self.errors,
"warnings": self.warnings,
"timestamp": datetime.now().isoformat()
}
return enriched
def add_error(self, error_msg: str):
"""Adiciona erro à lista"""
self.errors.append(error_msg)
logger.error(f"[{self.specialist_name}] {error_msg}")
def add_warning(self, warning_msg: str):
"""Adiciona aviso à lista"""
self.warnings.append(warning_msg)
logger.warning(f"[{self.specialist_name}] {warning_msg}")
def set_confidence(self, score: int):
"""Define score de confiança (0-100)"""
if 0 <= score <= 100:
self.confidence_score = score
else:
self.add_warning(f"Confidence score {score} fora do range 0-100")
def to_json(self, result: Dict[str, Any]) -> str:
"""Converte resultado para JSON string"""
try:
return json.dumps(result, ensure_ascii=False, indent=2)
except Exception as e:
self.add_error(f"Erro ao serializar JSON: {e}")
return "{}"
def from_json(self, json_str: str) -> Dict[str, Any]:
"""Parse JSON string para dicionário"""
try:
return json.loads(json_str)
except Exception as e:
self.add_error(f"Erro ao parsear JSON: {e}")
return {}
@property
def info(self) -> Dict[str, Any]:
"""Retorna informações sobre o processador"""
return {
"specialist_id": self.specialist_id,
"specialist_name": self.specialist_name,
"llm_model": self.llm_model.__class__.__name__ if self.llm_model else None,
"schema": self.get_schema()
}
|