Upload 2 files

processors/processor_base.py

@@ -1,268 +1,282 @@
-"""
-Classe Base Refatorada - Sem simulações, apenas LLM real
-"""
-from abc import ABC, abstractmethod
-from typing import Dict, Any, Optional
-from datetime import datetime
+"""Classe base para processadores com injeção automática de schemas."""
+
 import logging
-import json
+from abc import ABC, abstractmethod
+from typing import Dict, Optional, Any
+from pathlib import Path
+import sys
 
-from processors.config_loader import get_config_loader, SpecialistConfig
+# Importa schema_loader
+sys.path.insert(0, str(Path(__file__).parent.parent))
+from llm.schema_loader import get_schema_loader
 
 logger = logging.getLogger(__name__)
 
 
 class ProcessorBase(ABC):
-    """
-    Classe abstrata base para processadores
+    """Classe base para todos os processadores com schema injection."""
 
-    MUDANÇAS:
-    - Carrega configuração do YAML
-    - Não tem métodos auxiliares hardcoded
-    - Todos os dados vêm do LLM
-    """
-
-    def __init__(
-        self,
-        specialist_id: int,
-        llm_manager=None
-    ):
+    def __init__(self, 
+                 specialist_name: str,
+                 llm_client: Any,
+                 system_prompt: str,
+                 config: Optional[Dict] = None):
         """
+        Inicializa processador base.
+
         Args:
-            specialist_id: ID do especialista (1-9)
-            llm_manager: LLMManager para chamadas LLM
+            specialist_name: Nome do especialista (usado para carregar schema)
+            llm_client: Cliente LLM (GroqClient, OpenAIClient, etc)
+            system_prompt: Prompt base do sistema
+            config: Configurações adicionais
         """
-        self.specialist_id = specialist_id
-        self.llm_manager = llm_manager
-
-        # Carregar configuração do YAML
-        config_loader = get_config_loader()
-        self.config: Optional[SpecialistConfig] = config_loader.get_specialist(specialist_id)
-
-        if not self.config:
-            raise ValueError(f"Especialista {specialist_id} não encontrado na configuração")
-
-        self.specialist_name = self.config.name
-        self.execution_time = 0
-        self.confidence_score = 0
-        self.errors = []
-        self.warnings = []
-
-        logger.info(
-            f"✅ {self.specialist_name} inicializado "
-            f"(provider={self.config.llm_config.provider}, "
-            f"model={self.config.llm_config.model})"
-        )
+        self.specialist_name = specialist_name
+        self.llm_client = llm_client
+        self.config = config or {}
+
+        # 🔥 INJEÇÃO AUTOMÁTICA DE SCHEMA
+        self.schema_loader = get_schema_loader()
+        self.system_prompt = self._enhance_system_prompt(system_prompt)
 
-    async def process(self, acordao_data: Dict[str, Any]) -> Dict[str, Any]:
+        logger.info(f"✅ {self.__class__.__name__} inicializado (specialist={specialist_name})")
+
+    def _enhance_system_prompt(self, base_prompt: str) -> str:
         """
-        Processa acórdão usando LLM REAL
+        Injeta schema JSON no system prompt.
 
         Args:
-            acordao_data: Dados do acórdão
+            base_prompt: Prompt original
 
         Returns:
-            Resultado do processamento
+            Prompt com schema injetado
         """
-        if not self.llm_manager:
-            raise ValueError("❌ LLMManager não configurado")
-
-        start_time = datetime.now()
-
         try:
-            # 1. Formatar prompt do usuário com dados do acórdão
-            user_prompt = self.config.format_user_prompt(
-                tribunal=acordao_data.get('tribunal', 'N/A'),
-                numero_processo=acordao_data.get('numero_processo', 'N/A'),
-                ementa=acordao_data.get('ementa', ''),
-                integra=acordao_data.get('integra', ''),
-                classe_processual=acordao_data.get('classe_processual', 'N/A')
-            )
-
-            # 2. Chamar LLM Manager
-            response = await self.llm_manager.generate(
-                provider=self.config.llm_config.provider,
-                model=self.config.llm_config.model,
-                system_prompt=self.config.system_prompt,
-                user_prompt=user_prompt,
-                temperature=self.config.llm_config.temperature,
-                max_tokens=self.config.llm_config.max_tokens
+            enhanced = self.schema_loader.inject_schema_in_prompt(
+                system_prompt=base_prompt,
+                specialist_name=self.specialist_name,
+                format_style="json"  # Pode ser "json", "compact" ou "markdown"
             )
 
-            # 3. Extrair resultado (compatível com GroqResponse ou dict)
-            result_text = None
-
-            # Verificar se é GroqResponse (objeto com atributo content)
-            if hasattr(response, 'content'):
-                result_text = response.content
-                logger.debug(f"[{self.specialist_name}] Resposta GroqResponse extraída")
-
-            # Verificar se é dict com estrutura padrão
-            elif isinstance(response, dict):
-                if response.get('status') == 'success':
-                    # Pode estar em 'result', 'content' ou 'response'
-                    result_text = (
-                        response.get('result') or 
-                        response.get('content') or 
-                        response.get('response')
-                    )
-                    logger.debug(f"[{self.specialist_name}] Resposta dict extraída")
-                else:
-                    error_msg = response.get('error', 'Erro desconhecido')
-                    raise ValueError(f"Erro na chamada LLM: {error_msg}")
-
-            # Tipo inesperado
-            else:
-                raise ValueError(
-                    f"Tipo de resposta inesperado: {type(response).__name__}. "
-                    f"Esperado: GroqResponse ou dict"
-                )
-
-            if not result_text:
-                raise ValueError("Resposta vazia do LLM")
-
-            # Parse JSON
-            result = self._parse_json_response(result_text)
-
-            # Validar
-            if not self.validate(result):
-                self.add_warning("Resultado não passou na validação completa")
-
-            # Calcular confiança
-            self.confidence_score = self._calculate_confidence(result)
-
-            # 4. Pós-processar
-            self.execution_time = (datetime.now() - start_time).total_seconds()
-
-            return self.post_process(result)
+            logger.debug(f"✅ Schema injetado para {self.specialist_name}")
+            return enhanced
 
         except Exception as e:
-            self.add_error(f"Erro no processamento: {e}")
-            self.execution_time = (datetime.now() - start_time).total_seconds()
+            logger.warning(f"⚠️ Falha ao injetar schema: {e}, usando prompt original")
+            return base_prompt
 
-            return {
-                "specialist_id": self.specialist_id,
-                "specialist_name": self.specialist_name,
-                "status": "error",
-                "error": str(e),
-                "execution_time": self.execution_time,
-                "timestamp": datetime.now().isoformat()
-            }
+    @abstractmethod
+    async def process(self, data: Dict) -> Dict:
+        """
+        Processa dados com LLM.
+
+        Args:
+            data: Dados de entrada
+
+        Returns:
+            Resultado processado
+        """
+        pass
 
-    def _parse_json_response(self, text: str) -> Dict[str, Any]:
+    def validate_response(self, response: str) -> tuple:
         """
-        Parse resposta JSON do LLM (com limpeza se necessário)
+        Valida resposta contra schema.
 
         Args:
-            text: Texto da resposta
+            response: Resposta JSON do LLM
 
         Returns:
-            Dict parseado
+            (is_valid: bool, errors: list)
         """
-        try:
-            # Tentar parse direto
-            return json.loads(text)
-        except json.JSONDecodeError:
-            # Limpar markdown code blocks
-            text = text.strip()
-
-            if '```json' in text:
-                text = text.split('```json', 1)[1].split('```', 1)[0].strip()
-            elif '```' in text:
-                text = text.split('```', 1)[1].split('```', 1)[0].strip()
-
-            # Tentar parse novamente
-            try:
-                return json.loads(text)
-            except json.JSONDecodeError as e:
-                self.add_error(f"Erro ao parsear JSON: {e}")
-                logger.error(f"[{self.specialist_name}] Resposta inválida: {text[:200]}...")
-                raise ValueError(f"Resposta não é JSON válido: {e}")
-
-    def validate(self, result: Dict[str, Any]) -> bool:
+        return self.schema_loader.validate_response(response, self.specialist_name)
+
+    async def process_with_validation(self, data: Dict) -> Dict:
         """
-        Valida resultado contra schema
+        Processa dados e valida resposta contra schema.
 
         Args:
-            result: Resultado a validar
+            data: Dados de entrada
 
         Returns:
-            True se válido
+            Dict com {
+                "status": "success" | "error",
+                "result": resultado processado,
+                "validation_errors": lista de erros (se houver)
+            }
         """
         try:
-            # Validação básica: verificar campos required do schema
-            schema = self.config.schema
-            required_fields = schema.get('required', [])
+            # Processa
+            result = await self.process(data)
+
+            # Valida se resultado contém JSON
+            if 'content' in result:
+                is_valid, errors = self.validate_response(result['content'])
 
-            for field in required_fields:
-                if field not in result:
-                    self.add_warning(f"Campo obrigatório ausente: {field}")
-                    return False
+                if not is_valid:
+                    logger.warning(f"⚠️ Validação falhou para {self.specialist_name}: {errors}")
+                    return {
+                        "status": "validation_error",
+                        "result": result,
+                        "validation_errors": errors
+                    }
 
-            return True
+            return {
+                "status": "success",
+                "result": result,
+                "validation_errors": []
+            }
 
         except Exception as e:
-            self.add_error(f"Erro na validação: {e}")
-            return False
+            logger.error(f"❌ Erro no processamento de {self.specialist_name}: {e}")
+            return {
+                "status": "error",
+                "result": None,
+                "validation_errors": [str(e)]
+            }
+
+    def get_model_config(self) -> Dict:
+        """Retorna configuração do modelo LLM."""
+        return self.config.get('llm_config', {})
 
-    def _calculate_confidence(self, result: Dict[str, Any]) -> int:
+    def get_temperature(self) -> float:
+        """Retorna temperatura configurada."""
+        return self.get_model_config().get('temperature', 0.7)
+
+    def get_max_tokens(self) -> Optional[int]:
+        """Retorna max_tokens configurado."""
+        return self.get_model_config().get('max_tokens')
+
+    def get_model_name(self) -> str:
+        """Retorna nome do modelo."""
+        return self.get_model_config().get('model', 'llama-3.3-70b-versatile')
+
+
+# ============================================================================
+# EXEMPLO DE USO: Processador Decisão
+# ============================================================================
+
+class ProcessorDecisao(ProcessorBase):
+    """Processador especializado em análise de decisão."""
+
+    def __init__(self, llm_client: Any, config: Optional[Dict] = None):
+        system_prompt = """Você é um especialista em análise de decisões judiciais.
+Extraia com precisão: tipo de decisão, dispositivos, placar de votação, 
+prazos e efeitos da decisão."""
+
+        super().__init__(
+            specialist_name="decisao",
+            llm_client=llm_client,
+            system_prompt=system_prompt,
+            config=config
+        )
+
+    async def process(self, data: Dict) -> Dict:
         """
-        Calcula score de confiança baseado no resultado
+        Processa análise de decisão.
+
+        Args:
+            data: Dict com {
+                "acordao_texto": texto do acórdão,
+                "numero_processo": número do processo,
+                ...
+            }
 
         Returns:
-            Score 0-100
+            Dict com resultado do LLM
         """
-        # Lógica simples: quanto mais campos preenchidos, maior confiança
-        schema = self.config.schema
-        required_fields = schema.get('required', [])
+        # Monta user prompt
+        user_prompt = f"""Analise a seguinte decisão judicial:
 
-        if not required_fields:
-            return 85  # Padrão
+PROCESSO: {data.get('numero_processo', 'N/A')}
 
-        filled = sum(1 for field in required_fields if result.get(field))
-        confidence = int((filled / len(required_fields)) * 100)
+TEXTO:
+{data.get('acordao_texto', '')}
 
-        return min(confidence, 100)
+Extraia e retorne em JSON seguindo o schema fornecido."""
+
+        # Chama LLM com schema injetado automaticamente
+        response = self.llm_client.generate(
+            prompt=user_prompt,
+            system_prompt=self.system_prompt,
+            model=self.get_model_name(),
+            temperature=self.get_temperature(),
+            max_tokens=self.get_max_tokens(),
+            response_format={"type": "json_object"}
+        )
 
-    def post_process(self, result: Dict[str, Any]) -> Dict[str, Any]:
-        """Pós-processa resultado"""
         return {
-            "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(),
-            "config": {
-                "provider": self.config.llm_config.provider,
-                "model": self.config.llm_config.model,
-                "temperature": self.config.llm_config.temperature
-            }
+            "content": response.content,
+            "tokens": response.total_tokens,
+            "model": response.model
         }
 
-    def add_error(self, error_msg: str):
-        """Adiciona erro"""
-        self.errors.append(error_msg)
-        logger.error(f"[{self.specialist_name}] {error_msg}")
 
-    def add_warning(self, warning_msg: str):
-        """Adiciona aviso"""
-        self.warnings.append(warning_msg)
-        logger.warning(f"[{self.specialist_name}] {warning_msg}")
+# ============================================================================
+# EXEMPLO DE USO: Processador Fundamentação
+# ============================================================================
+
+class ProcessorFundamentacao(ProcessorBase):
+    """Processador especializado em análise de fundamentação jurídica."""
 
-    def get_schema(self) -> Dict[str, Any]:
-        """Retorna schema do especialista"""
-        return self.config.schema
+    def __init__(self, llm_client: Any, config: Optional[Dict] = None):
+        system_prompt = """Você é um especialista em fundamentação jurídica.
+Identifique e classifique: legislação citada, jurisprudência, doutrina e princípios."""
+
+        super().__init__(
+            specialist_name="fundamentacao",
+            llm_client=llm_client,
+            system_prompt=system_prompt,
+            config=config
+        )
+
+    async def process(self, data: Dict) -> Dict:
+        """Processa análise de fundamentação."""
+        user_prompt = f"""Analise a fundamentação jurídica:
+
+TEXTO:
+{data.get('acordao_texto', '')}
+
+Extraia legislação, jurisprudência e doutrina citadas em JSON."""
+
+        response = self.llm_client.generate(
+            prompt=user_prompt,
+            system_prompt=self.system_prompt,
+            model=self.get_model_name(),
+            temperature=self.get_temperature(),
+            max_tokens=self.get_max_tokens(),
+            response_format={"type": "json_object"}
+        )
 
-    @property
-    def info(self) -> Dict[str, Any]:
-        """Retorna informações sobre o processador"""
         return {
-            "specialist_id": self.specialist_id,
-            "specialist_name": self.specialist_name,
-            "provider": self.config.llm_config.provider,
-            "model": self.config.llm_config.model,
-            "schema": self.config.schema
+            "content": response.content,
+            "tokens": response.total_tokens,
+            "model": response.model
         }
+
+
+# ============================================================================
+# FACTORY: Criador de processadores
+# ============================================================================
+
+def create_processor(specialist_type: str, llm_client: Any, config: Dict) -> ProcessorBase:
+    """
+    Factory para criar processador baseado no tipo.
+
+    Args:
+        specialist_type: Tipo do especialista ("decisao", "fundamentacao", etc)
+        llm_client: Cliente LLM
+        config: Configuração
+
+    Returns:
+        Instância do processador apropriado
+    """
+    processors = {
+        "decisao": ProcessorDecisao,
+        "fundamentacao": ProcessorFundamentacao,
+        # Adicione outros processadores aqui
+    }
+
+    processor_class = processors.get(specialist_type)
+    if not processor_class:
+        raise ValueError(f"Processador desconhecido: {specialist_type}")
+
+    return processor_class(llm_client=llm_client, config=config)

processors/schema_loader.py

@@ -0,0 +1,212 @@
+"""Schema Loader - Carrega e injeta schemas JSON nos system prompts."""
+
+import os
+import json
+import logging
+from pathlib import Path
+from typing import Dict, Optional
+
+logger = logging.getLogger(__name__)
+
+
+class SchemaLoader:
+    """Carrega schemas JSON e injeta em system prompts."""
+
+    # Mapeamento especialista -> arquivo schema
+    SCHEMA_MAP = {
+        "metadados": "metadados_schema.json",
+        "segmentacao": "segmentacao_schema.json",
+        "transcricao": "transcricao_schema.json",
+        "contexto": "contexto_schema.json",
+        "fundamentacao": "fundamentacao_schema.json",
+        "decisao": "decisao_schema.json",
+        "arquivo": "arquivo_schema.json",
+        "relatorio": "relatorio_schema.json",
+        "auditoria": "auditoria_schema.json",
+    }
+
+    def __init__(self, schemas_dir: Optional[str] = None):
+        """
+        Inicializa SchemaLoader.
+
+        Args:
+            schemas_dir: Diretório dos schemas (padrão: prompts/schemas/)
+        """
+        if schemas_dir is None:
+            # Tenta encontrar o diretório automaticamente
+            base_dir = Path(__file__).parent.parent
+            schemas_dir = base_dir / "prompts" / "schemas"
+
+        self.schemas_dir = Path(schemas_dir)
+
+        if not self.schemas_dir.exists():
+            logger.warning(f"⚠️ Diretório de schemas não encontrado: {self.schemas_dir}")
+        else:
+            logger.info(f"✅ SchemaLoader inicializado: {self.schemas_dir}")
+
+        # Cache de schemas carregados
+        self._cache: Dict[str, Dict] = {}
+
+    def load_schema(self, specialist_name: str) -> Optional[Dict]:
+        """
+        Carrega schema JSON de um especialista.
+
+        Args:
+            specialist_name: Nome do especialista (ex: "decisao", "metadados")
+
+        Returns:
+            Dict com schema JSON ou None se não encontrado
+        """
+        # Verifica cache
+        if specialist_name in self._cache:
+            return self._cache[specialist_name]
+
+        # Busca arquivo schema
+        schema_filename = self.SCHEMA_MAP.get(specialist_name.lower())
+        if not schema_filename:
+            logger.warning(f"⚠️ Schema não mapeado para especialista: {specialist_name}")
+            return None
+
+        schema_path = self.schemas_dir / schema_filename
+
+        if not schema_path.exists():
+            logger.warning(f"⚠️ Arquivo schema não encontrado: {schema_path}")
+            return None
+
+        try:
+            with open(schema_path, 'r', encoding='utf-8') as f:
+                schema = json.load(f)
+
+            # Armazena no cache
+            self._cache[specialist_name] = schema
+
+            logger.debug(f"✅ Schema carregado: {specialist_name}")
+            return schema
+
+        except Exception as e:
+            logger.error(f"❌ Erro ao carregar schema {schema_path}: {e}")
+            return None
+
+    def inject_schema_in_prompt(self, 
+                                system_prompt: str, 
+                                specialist_name: str,
+                                format_style: str = "json") -> str:
+        """
+        Injeta schema JSON no system prompt.
+
+        Args:
+            system_prompt: Prompt original do sistema
+            specialist_name: Nome do especialista
+            format_style: Estilo de formatação ("json", "markdown", "compact")
+
+        Returns:
+            System prompt com schema injetado
+        """
+        schema = self.load_schema(specialist_name)
+
+        if not schema:
+            logger.warning(f"⚠️ Schema não disponível para {specialist_name}, retornando prompt original")
+            return system_prompt
+
+        # Formata schema conforme estilo
+        if format_style == "json":
+            schema_text = json.dumps(schema, indent=2, ensure_ascii=False)
+        elif format_style == "compact":
+            schema_text = json.dumps(schema, ensure_ascii=False)
+        elif format_style == "markdown":
+            schema_text = self._format_schema_markdown(schema)
+        else:
+            schema_text = json.dumps(schema, indent=2, ensure_ascii=False)
+
+        # Monta prompt final com schema
+        enhanced_prompt = f"""{system_prompt}
+
+# JSON SCHEMA OBRIGATÓRIO
+
+Você DEVE retornar sua resposta seguindo EXATAMENTE este JSON Schema:
+
+```json
+{schema_text}
+```
+
+IMPORTANTE:
+- Retorne APENAS JSON válido
+- Siga TODOS os campos required do schema
+- Respeite os tipos de dados (string, integer, array, object)
+- Valide enums e patterns quando especificados
+- NÃO adicione comentários ou texto fora do JSON
+- NÃO invente campos que não estão no schema
+"""
+
+        return enhanced_prompt
+
+    def _format_schema_markdown(self, schema: Dict) -> str:
+        """Formata schema em Markdown legível."""
+        lines = []
+
+        if 'title' in schema:
+            lines.append(f"## {schema['title']}")
+        if 'description' in schema:
+            lines.append(f"{schema['description']}
+")
+
+        if 'properties' in schema:
+            lines.append("### Campos:")
+            for field, props in schema['properties'].items():
+                field_type = props.get('type', 'any')
+                desc = props.get('description', '')
+                required = '(obrigatório)' if field in schema.get('required', []) else '(opcional)'
+                lines.append(f"- **{field}** ({field_type}) {required}: {desc}")
+
+        return '
+'.join(lines)
+
+    def get_available_specialists(self) -> list:
+        """Retorna lista de especialistas com schemas disponíveis."""
+        return list(self.SCHEMA_MAP.keys())
+
+    def validate_response(self, response: str, specialist_name: str) -> tuple:
+        """
+        Valida resposta JSON contra schema.
+
+        Args:
+            response: Resposta JSON (string)
+            specialist_name: Nome do especialista
+
+        Returns:
+            (is_valid: bool, errors: list)
+        """
+        try:
+            from jsonschema import validate, ValidationError
+        except ImportError:
+            logger.warning("jsonschema não instalado, validação desabilitada")
+            return True, []
+
+        schema = self.load_schema(specialist_name)
+        if not schema:
+            return True, ["Schema não encontrado"]
+
+        try:
+            data = json.loads(response)
+            validate(instance=data, schema=schema)
+            return True, []
+        except json.JSONDecodeError as e:
+            return False, [f"JSON inválido: {e}"]
+        except ValidationError as e:
+            return False, [f"Validação falhou: {e.message}"]
+        except Exception as e:
+            return False, [f"Erro na validação: {e}"]
+
+
+# ============================================================================
+# SINGLETON GLOBAL (Opcional)
+# ============================================================================
+
+_global_loader: Optional[SchemaLoader] = None
+
+def get_schema_loader() -> SchemaLoader:
+    """Retorna instância global do SchemaLoader (singleton)."""
+    global _global_loader
+    if _global_loader is None:
+        _global_loader = SchemaLoader()
+    return _global_loader