ParaAIV3.1

config/pipeline_config.yaml

@@ -0,0 +1,62 @@
+##PARA.AI/config/pipeline_config.yaml
+# =============================================================================
+# Pipeline Configuration V13.6 - Orquestração em Fases com Dependências
+# =============================================================================
+
+pipeline:
+  name: "Para.AI V13.6 Pipeline"
+  version: "v13.6"
+  description: "Pipeline de processamento de acórdãos TJPR em 6 fases"
+
+  # Fases sequenciais com dependências explícitas
+  phases:
+    - id: 1
+      name: "Segmentação"
+      description: "Segmenta inteiro teor em RELATÓRIO + FUNDAMENTAÇÃO + DISPOSITIVO"
+      parallel: false
+      specialists: [1]
+      depends_on: []
+
+    - id: 2
+      name: "Metadados"
+      description: "Extrai metadados estruturais do acórdão"
+      parallel: false
+      specialists: [2]
+      depends_on: [1]
+
+    - id: 3
+      name: "Classificação"
+      description: "Classifica ramo do direito e assuntos"
+      parallel: false
+      specialists: [3]
+      depends_on: [2]
+
+    - id: 4
+      name: "Tripartite"
+      description: "Extrai RELATÓRIO + FUNDAMENTAÇÃO + DECISÃO em paralelo"
+      parallel: true
+      specialists: [4, 5, 6]
+      depends_on: [1, 2, 3]
+
+    - id: 5
+      name: "Arquivista"
+      description: "Análise meta-cognitiva e avaliação de qualidade"
+      parallel: false
+      specialists: [7]
+      depends_on: [4]
+
+    - id: 6
+      name: "Validação"
+      description: "Validação jsonschema e cálculo de completude"
+      parallel: false
+      specialists: []
+      depends_on: [5]
+
+# Configuração Global
+global_config:
+  max_workers: 3
+  default_timeout: 120
+  batch_size: 50
+  enable_retry: true
+  max_retries: 3
+  retry_delay: 2

core/__init__.py

@@ -0,0 +1,18 @@
+##PARA.AI/core/__init__.py
+"""
+Core Module - Componentes principais da pipeline V13.6
+"""
+
+from .orchestrator import PipelineOrchestrator
+from .base_specialist import BaseSpecialist
+from .context_builder import ContextBuilder
+from .validator import SchemaValidator
+
+__all__ = [
+    'PipelineOrchestrator',
+    'BaseSpecialist',
+    'ContextBuilder',
+    'SchemaValidator'
+]
+
+__version__ = "13.6.0"

core/base_specialist.py

@@ -0,0 +1,309 @@
+##PARA.AI/core/base_specialist.py
+"""
+Base Specialist - Classe base refatorada para todos os especialistas V13.6
+MUDANÇAS DO V13.1:
+- Context injection obrigatório
+- Validação com jsonschema
+- Prompts ultra-simples (<1000 chars)
+- Retry logic automático
+"""
+import logging
+import json
+import yaml
+from typing import Dict, Any, Optional
+from datetime import datetime
+from abc import ABC, abstractmethod
+
+from api.utils.logger import setup_logger
+
+logger = setup_logger(__name__)
+
+
+class BaseSpecialist(ABC):
+    """
+    Classe base para todos os especialistas V13.6
+
+    REGRAS DE OURO:
+    1. NUNCA pedir mais de 10 campos por especialista
+    2. SEMPRE exibir JSON parcial no prompt (context injection)
+    3. SEMPRE validar output com jsonschema
+    4. SEMPRE logar input + output (debug)
+    5. NUNCA usar placeholders ou dados simulados
+    """
+
+    def __init__(
+        self, 
+        specialist_id: int,
+        config_path: str,
+        llm_manager,
+        schema_validator=None
+    ):
+        """
+        Args:
+            specialist_id: ID único do especialista (1-9)
+            config_path: Caminho para specialists_v13_6.yaml
+            llm_manager: Instância do LLMManager
+            schema_validator: Validador de schema (opcional)
+        """
+        self.id = specialist_id
+        self.llm_manager = llm_manager
+        self.schema_validator = schema_validator
+
+        # Carregar configuração do especialista
+        self.config = self._load_config(specialist_id, config_path)
+        self.name = self.config.get('name', f'Specialist_{specialist_id}')
+
+        # Carregar prompt template
+        self.prompt_template = self._load_prompt_template()
+
+        # Configuração LLM
+        self.llm_config = self.config.get('llm_config', {})
+
+        # Retry logic
+        self.max_retries = self.config.get('max_retries', 3)
+        self.retry_delay = self.config.get('retry_delay', 2)
+
+        logger.info(f"✅ {self.name} (ID {self.id}) inicializado")
+
+    def _load_config(self, specialist_id: int, config_path: str) -> Dict[str, Any]:
+        """Carrega configuração do especialista do YAML"""
+        try:
+            with open(config_path, 'r', encoding='utf-8') as f:
+                full_config = yaml.safe_load(f)
+
+            specialist_config = full_config.get('specialists', {}).get(specialist_id, {})
+
+            if not specialist_config:
+                raise ValueError(f"Configuração não encontrada para especialista {specialist_id}")
+
+            return specialist_config
+
+        except Exception as e:
+            logger.error(f"❌ Erro ao carregar config: {e}")
+            raise
+
+    def _load_prompt_template(self) -> str:
+        """Carrega template de prompt do arquivo .txt"""
+        prompt_file = self.config.get('prompt_file', '')
+
+        if not prompt_file:
+            logger.warning(f"⚠️ Prompt file não configurado para {self.name}")
+            return ""
+
+        try:
+            with open(prompt_file, 'r', encoding='utf-8') as f:
+                return f.read()
+        except Exception as e:
+            logger.error(f"❌ Erro ao carregar prompt: {e}")
+            return ""
+
+    async def process(
+        self, 
+        input_data: Dict[str, Any],
+        context: Optional[Dict[str, Any]] = None
+    ) -> Dict[str, Any]:
+        """
+        Processa input com context injection e retry logic
+
+        Args:
+            input_data: Dados brutos do acórdão
+            context: JSON parcial já preenchido por especialistas anteriores
+
+        Returns:
+            Resultado parcial (apenas os campos deste especialista)
+        """
+        logger.info(f"🤖 {self.name} processando...")
+
+        for attempt in range(self.max_retries):
+            try:
+                # 1. Construir prompt com context injection
+                prompt = self._build_prompt(input_data, context)
+
+                # 2. Log do prompt (debug)
+                logger.debug(f"  📝 Prompt ({len(prompt)} chars): {prompt[:200]}...")
+
+                # 3. Chamar LLM
+                response = await self._call_llm(prompt)
+
+                # 4. Parse JSON
+                result = self._parse_response(response)
+
+                # 5. Validar schema (se configurado)
+                if self.schema_validator:
+                    is_valid, errors = self._validate_output(result)
+                    if not is_valid:
+                        logger.warning(f"  ⚠️ Validação falhou: {errors[:2]}")
+                        if attempt < self.max_retries - 1:
+                            continue  # Retry
+
+                # 6. Log de sucesso
+                logger.info(f"  ✅ {self.name} completou: {len(result)} campos")
+
+                return result
+
+            except Exception as e:
+                logger.error(f"  ❌ Tentativa {attempt + 1}/{self.max_retries} falhou: {e}")
+
+                if attempt == self.max_retries - 1:
+                    # Última tentativa falhou - retornar estrutura vazia
+                    logger.error(f"  ❌ {self.name} FALHOU após {self.max_retries} tentativas")
+                    return self._get_empty_structure()
+
+        return self._get_empty_structure()
+
+    def _build_prompt(
+        self, 
+        input_data: Dict[str, Any],
+        context: Optional[Dict[str, Any]] = None
+    ) -> str:
+        """
+        Constrói prompt final com substituição de variáveis e context injection
+
+        NOVIDADE V13.6:
+        - Exibe JSON parcial já preenchido
+        - LLM entende que deve "completar", não "gerar do zero"
+        """
+        prompt = self.prompt_template
+
+        # Substituir variáveis básicas
+        prompt = prompt.replace('{ementa}', input_data.get('ementa', ''))
+        prompt = prompt.replace('{inteiro_teor}', input_data.get('inteiro_teor', '')[:10000])  # Limitar
+
+        # Substituir seções específicas (se disponíveis no context)
+        if context:
+            secoes = context.get('secoes_originais', {})
+            prompt = prompt.replace(
+                '{RELATORIO_texto}', 
+                secoes.get('RELATORIO_texto_completo', '')[:5000]
+            )
+            prompt = prompt.replace(
+                '{FUNDAMENTACAO_texto}', 
+                secoes.get('FUNDAMENTACAO_texto_completo', '')[:5000]
+            )
+            prompt = prompt.replace(
+                '{DISPOSITIVO_texto}', 
+                secoes.get('DISPOSITIVO_texto_completo', '')[:3000]
+            )
+
+            # Substituir metadados
+            if 'metadados' in context:
+                metadados_str = json.dumps(context['metadados'], indent=2, ensure_ascii=False)
+                prompt = prompt.replace('{metadados}', metadados_str[:1000])
+
+        # CONTEXT INJECTION: Exibir JSON parcial já preenchido
+        if context:
+            # Remover campos muito grandes para não explodir o prompt
+            context_compact = self._compact_context(context)
+
+            context_str = json.dumps(context_compact, indent=2, ensure_ascii=False)
+
+            prompt += f"""
+
+━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
+JSON JÁ PREENCHIDO (contexto dos especialistas anteriores):
+━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
+
+{context_str}
+
+━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
+IMPORTANTE:
+- NÃO repita os campos acima
+- Complete APENAS os campos NOVOS da sua responsabilidade
+- Retorne SOMENTE JSON válido
+━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
+"""
+
+        return prompt
+
+    def _compact_context(self, context: Dict[str, Any]) -> Dict[str, Any]:
+        """Remove campos muito grandes do contexto para não explodir o prompt"""
+        compact = {}
+
+        for key, value in context.items():
+            if key in ['secoes_originais']:
+                # Não incluir textos completos (já foram substituídos)
+                continue
+
+            if isinstance(value, str) and len(value) > 500:
+                # Truncar strings muito longas
+                compact[key] = value[:500] + "..."
+            else:
+                compact[key] = value
+
+        return compact
+
+    async def _call_llm(self, prompt: str) -> str:
+        """
+        Chama LLM via LLMManager
+
+        NOTA: Adaptar para API específica do llm_manager
+        """
+        try:
+            # Assumindo que llm_manager tem método generate()
+            response = await self.llm_manager.generate(
+                system_prompt=self.config.get('system_prompt', ''),
+                user_prompt=prompt,
+                temperature=self.llm_config.get('temperature', 0.3),
+                max_tokens=self.llm_config.get('max_tokens', 2000),
+                model=self.llm_config.get('model', 'groq/llama-3-70b')
+            )
+
+            return response
+
+        except Exception as e:
+            logger.error(f"❌ Erro ao chamar LLM: {e}")
+            raise
+
+    def _parse_response(self, response: str) -> Dict[str, Any]:
+        """Parse da resposta JSON do LLM"""
+        try:
+            # Tentar extrair JSON (pode vir com markdown)
+            response_clean = response.strip()
+
+            # Remover markdown code blocks se existir
+            if response_clean.startswith('```'):
+                lines = response_clean.split('
+')
+                response_clean = '
+'.join(lines[1:-1])  # Remove primeira e última linha
+
+            result = json.loads(response_clean)
+            return result
+
+        except json.JSONDecodeError as e:
+            logger.error(f"❌ Erro ao parsear JSON: {e}")
+            logger.error(f"Response: {response[:500]}")
+            raise
+
+    def _validate_output(self, result: Dict[str, Any]) -> tuple[bool, list]:
+        """Valida output contra schema parcial do especialista"""
+        if not self.schema_validator:
+            return True, []
+
+        schema_file = self.config.get('schema_file', '')
+        if not schema_file:
+            return True, []
+
+        try:
+            return self.schema_validator.validate_partial(result, schema_file)
+        except Exception as e:
+            logger.error(f"❌ Erro na validação: {e}")
+            return False, [str(e)]
+
+    @abstractmethod
+    def _get_empty_structure(self) -> Dict[str, Any]:
+        """
+        Retorna estrutura vazia em caso de falha total
+        Deve ser implementado por cada especialista
+        """
+        pass
+
+    def get_info(self) -> Dict[str, Any]:
+        """Retorna informações sobre o especialista"""
+        return {
+            'id': self.id,
+            'name': self.name,
+            'enabled': self.config.get('enabled', True),
+            'llm_model': self.llm_config.get('model', 'unknown'),
+            'max_retries': self.max_retries
+        }

core/context_builder.py

@@ -0,0 +1,79 @@
+##PARA.AI/core/context_builder.py
+"""
+Context Builder - Constrói contexto para context injection V13.6
+Monta JSON parcial que especialista N deve VER (mas não modificar)
+"""
+import logging
+from typing import Dict, Any, Optional
+
+from api.utils.logger import setup_logger
+
+logger = setup_logger(__name__)
+
+
+class ContextBuilder:
+    """
+    Constrói contexto para cada especialista
+
+    REGRA:
+    - Especialista N vê output de especialistas 1..N-1
+    - Mas vê apenas os campos RELEVANTES (não tudo)
+    - Evita explodir o prompt com dados desnecessários
+    """
+
+    def __init__(self):
+        # Mapear quais campos cada especialista precisa ver
+        self.context_rules = {
+            1: [],  # Segmentador: não precisa de contexto
+            2: ['secoes_originais'],  # Metadados: vê segmentação
+            3: ['metadados'],  # Classificador: vê metadados
+            4: ['metadados', 'classificacao_tematica', 'secoes_originais'],  # Relatório
+            5: ['metadados', 'classificacao_tematica', 'secoes_originais'],  # Fundamentação
+            6: ['metadados', 'secoes_originais'],  # Dispositivo
+            7: ['metadados', 'classificacao_tematica', 'RELATORIO', 'FUNDAMENTACAO', 'DECISAO']  # Arquivista
+        }
+
+    def build_context(
+        self, 
+        current_result: Dict[str, Any],
+        specialist_id: int
+    ) -> Optional[Dict[str, Any]]:
+        """
+        Constrói contexto para um especialista específico
+
+        Args:
+            current_result: Resultado acumulado até agora
+            specialist_id: ID do especialista que vai receber o contexto
+
+        Returns:
+            Dicionário com campos relevantes ou None (se não precisa de contexto)
+        """
+        if specialist_id not in self.context_rules:
+            logger.warning(f"⚠️ Especialista {specialist_id} sem regra de contexto")
+            return None
+
+        fields_needed = self.context_rules[specialist_id]
+
+        if not fields_needed:
+            # Especialista não precisa de contexto
+            return None
+
+        # Montar contexto com apenas os campos necessários
+        context = {}
+
+        for field in fields_needed:
+            if field in current_result:
+                context[field] = current_result[field]
+
+        logger.debug(f"📦 Contexto para Especialista {specialist_id}: {list(context.keys())}")
+
+        return context if context else None
+
+    def add_context_rule(self, specialist_id: int, fields: list):
+        """Adiciona/modifica regra de contexto para um especialista"""
+        self.context_rules[specialist_id] = fields
+        logger.info(f"✅ Regra de contexto atualizada para Especialista {specialist_id}")
+
+    def get_context_rules(self) -> Dict[int, list]:
+        """Retorna todas as regras de contexto"""
+        return self.context_rules

core/orchestrator.py

@@ -0,0 +1,285 @@
+##PARA.AI/core/orchestrator.py
+"""
+Pipeline Orchestrator - Orquestrador principal V13.6
+Substitui processor_manager.py com arquitetura em fases e dependências explícitas
+"""
+import logging
+import asyncio
+from typing import Dict, Any, List, Optional
+from datetime import datetime
+import json
+
+from core.context_builder import ContextBuilder
+from core.validator import SchemaValidator
+from api.utils.logger import setup_logger
+
+logger = setup_logger(__name__)
+
+
+class PipelineOrchestrator:
+    """
+    Orquestrador de pipeline em fases com dependências explícitas
+
+    MUDANÇAS DO V13.1:
+    - Fases sequenciais com dependências explícitas (DAG)
+    - Context injection: especialista N recebe output de N-1
+    - Validação jsonschema após cada fase
+    - Logs detalhados por fase/especialista
+    - Suporte a execução paralela controlada (FASE 4)
+    """
+
+    def __init__(self, config: Dict[str, Any], llm_manager, specialists: Dict[int, Any]):
+        """
+        Args:
+            config: Configuração da pipeline (de pipeline_config.yaml)
+            llm_manager: Instância do LLMManager
+            specialists: Dicionário {id: instância do especialista}
+        """
+        self.config = config
+        self.llm_manager = llm_manager
+        self.specialists = specialists
+        self.context_builder = ContextBuilder()
+        self.validator = SchemaValidator('schemas/protocolo_v13_6_schema.json')
+
+        # Organizar fases
+        self.phases = self._organize_phases()
+
+        logger.info(
+            f"✅ PipelineOrchestrator inicializado: "
+            f"{len(self.phases)} fases, {len(self.specialists)} especialistas"
+        )
+
+    def _organize_phases(self) -> List[Dict[str, Any]]:
+        """Organiza fases em ordem de execução respeitando dependências"""
+        phases = self.config['pipeline']['phases']
+        # Ordenar por id (já está em ordem correta no YAML)
+        return sorted(phases, key=lambda p: p['id'])
+
+    async def process_acordao(
+        self, 
+        acordao_bruto: Dict[str, Any],
+        fase_inicial: int = 1,
+        fase_final: Optional[int] = None
+    ) -> Dict[str, Any]:
+        """
+        Processa acórdão através de todas as fases da pipeline
+
+        Args:
+            acordao_bruto: Dados brutos do acórdão (ementa, inteiro_teor, etc)
+            fase_inicial: Fase inicial (default: 1)
+            fase_final: Fase final (default: None = todas)
+
+        Returns:
+            JSON completo conforme Protocolo V13.6
+        """
+        start_time = datetime.now()
+
+        # Inicializar resultado acumulado
+        resultado = {
+            "protocolo_versao": "v13.6",
+            "id_manifestacao": acordao_bruto.get('id', 0),
+            "metadados_processamento": {
+                "protocolo_origem": "v13.6",
+                "data_processamento": start_time.isoformat(),
+                "versao_preprocessador": "v13.6.0",
+                "campos_enriquecidos": [],
+                "tempo_processamento_segundos": None
+            },
+            "campos_futuros": {
+                "relatorio_transcript_exato": None,
+                "fundamentacao_transcript_exato": None,
+                "dispositivo_transcript_exato": None,
+                "embeddings_metadata": None,
+                "tags_embedding_baldes": None
+            }
+        }
+
+        logger.info(f"🚀 Iniciando pipeline para acórdão ID {resultado['id_manifestacao']}")
+
+        # Executar fases
+        fase_final = fase_final or len(self.phases)
+        fases_para_executar = [p for p in self.phases if fase_inicial <= p['id'] <= fase_final]
+
+        for phase in fases_para_executar:
+            phase_id = phase['id']
+            phase_name = phase['name']
+            is_parallel = phase.get('parallel', False)
+
+            logger.info(f"📍 FASE {phase_id}: {phase_name} (parallel={is_parallel})")
+
+            try:
+                if is_parallel and len(phase.get('specialists', [])) > 1:
+                    # Execução paralela
+                    resultado = await self._run_phase_parallel(
+                        phase=phase,
+                        current_result=resultado,
+                        input_data=acordao_bruto
+                    )
+                else:
+                    # Execução sequencial
+                    resultado = await self._run_phase_sequential(
+                        phase=phase,
+                        current_result=resultado,
+                        input_data=acordao_bruto
+                    )
+
+                logger.info(f"✅ FASE {phase_id} concluída")
+
+            except Exception as e:
+                logger.error(f"❌ Erro na FASE {phase_id} ({phase_name}): {e}")
+                resultado['metadados_processamento']['alertas_qualidade'] =                     resultado['metadados_processamento'].get('alertas_qualidade', []) +                     [f"Erro na fase {phase_id}: {str(e)}"]
+                # Continuar para próximas fases (não travar tudo)
+
+        # Validação final (FASE 6)
+        if fase_final >= 6:
+            is_valid, errors = self.validator.validate(resultado)
+            if not is_valid:
+                logger.warning(f"⚠️ Validação final: {len(errors)} erros encontrados")
+                resultado['metadados_processamento']['alertas_validacao'] = errors[:5]  # Max 5
+
+        # Calcular tempo total
+        end_time = datetime.now()
+        resultado['metadados_processamento']['tempo_processamento_segundos'] =             (end_time - start_time).total_seconds()
+
+        logger.info(
+            f"✅ Pipeline completa: {resultado['metadados_processamento']['tempo_processamento_segundos']:.2f}s"
+        )
+
+        return resultado
+
+    async def _run_phase_sequential(
+        self,
+        phase: Dict[str, Any],
+        current_result: Dict[str, Any],
+        input_data: Dict[str, Any]
+    ) -> Dict[str, Any]:
+        """Executa fase sequencialmente (um especialista por vez)"""
+        specialist_ids = phase.get('specialists', [])
+
+        for spec_id in specialist_ids:
+            specialist = self.specialists.get(spec_id)
+            if not specialist:
+                logger.warning(f"⚠️ Especialista {spec_id} não encontrado")
+                continue
+
+            # Context injection: passar JSON parcial já preenchido
+            context = self.context_builder.build_context(
+                current_result=current_result,
+                specialist_id=spec_id
+            )
+
+            logger.info(f"  🤖 Executando Especialista {spec_id}: {specialist.__class__.__name__}")
+
+            # Executar especialista
+            try:
+                partial_result = await specialist.process(
+                    input_data=input_data,
+                    context=context
+                )
+
+                # Mesclar resultado parcial no resultado acumulado
+                current_result = self._merge_results(current_result, partial_result)
+
+                # Registrar campo enriquecido
+                campos_novos = list(partial_result.keys())
+                current_result['metadados_processamento']['campos_enriquecidos'].extend(campos_novos)
+
+                logger.info(f"  ✅ Especialista {spec_id} completou: {len(campos_novos)} campos")
+
+            except Exception as e:
+                logger.error(f"  ❌ Erro no Especialista {spec_id}: {e}")
+                raise
+
+        return current_result
+
+    async def _run_phase_parallel(
+        self,
+        phase: Dict[str, Any],
+        current_result: Dict[str, Any],
+        input_data: Dict[str, Any]
+    ) -> Dict[str, Any]:
+        """Executa fase em paralelo (múltiplos especialistas simultaneamente)"""
+        specialist_ids = phase.get('specialists', [])
+
+        # Preparar tarefas paralelas
+        tasks = []
+        for spec_id in specialist_ids:
+            specialist = self.specialists.get(spec_id)
+            if not specialist:
+                logger.warning(f"⚠️ Especialista {spec_id} não encontrado")
+                continue
+
+            # Context injection
+            context = self.context_builder.build_context(
+                current_result=current_result,
+                specialist_id=spec_id
+            )
+
+            logger.info(f"  🤖 Agendando Especialista {spec_id} (paralelo)")
+
+            # Criar tarefa async
+            task = specialist.process(input_data=input_data, context=context)
+            tasks.append((spec_id, task))
+
+        # Executar em paralelo
+        results = await asyncio.gather(*[t[1] for t in tasks], return_exceptions=True)
+
+        # Processar resultados
+        for (spec_id, _), result in zip(tasks, results):
+            if isinstance(result, Exception):
+                logger.error(f"  ❌ Erro no Especialista {spec_id}: {result}")
+                continue
+
+            # Mesclar resultado
+            current_result = self._merge_results(current_result, result)
+            campos_novos = list(result.keys())
+            current_result['metadados_processamento']['campos_enriquecidos'].extend(campos_novos)
+
+            logger.info(f"  ✅ Especialista {spec_id} completou: {len(campos_novos)} campos")
+
+        return current_result
+
+    def _merge_results(self, current: Dict[str, Any], partial: Dict[str, Any]) -> Dict[str, Any]:
+        """
+        Mescla resultado parcial no resultado acumulado
+
+        Estratégia:
+        - Campos de primeiro nível: sobrescrever
+        - Campos aninhados: deep merge
+        """
+        for key, value in partial.items():
+            if key in ['metadados_processamento', 'campos_futuros']:
+                # Não sobrescrever metadados de controle
+                continue
+
+            if isinstance(value, dict) and key in current and isinstance(current[key], dict):
+                # Deep merge
+                current[key].update(value)
+            else:
+                # Sobrescrever
+                current[key] = value
+
+        return current
+
+    def get_phase_info(self, phase_id: int) -> Optional[Dict[str, Any]]:
+        """Retorna informações sobre uma fase específica"""
+        for phase in self.phases:
+            if phase['id'] == phase_id:
+                return phase
+        return None
+
+    def get_pipeline_status(self) -> Dict[str, Any]:
+        """Retorna status atual da pipeline"""
+        return {
+            'total_phases': len(self.phases),
+            'total_specialists': len(self.specialists),
+            'phases': [
+                {
+                    'id': p['id'],
+                    'name': p['name'],
+                    'parallel': p.get('parallel', False),
+                    'specialists_count': len(p.get('specialists', []))
+                }
+                for p in self.phases
+            ]
+        }

core/validator.py

@@ -0,0 +1,160 @@
+##PARA.AI/core/validator.py
+"""
+Schema Validator - Validação com jsonschema V13.6
+"""
+import logging
+import json
+from typing import Dict, Any, Tuple, List
+
+try:
+    import jsonschema
+    from jsonschema import validate, ValidationError
+    HAS_JSONSCHEMA = True
+except ImportError:
+    HAS_JSONSCHEMA = False
+    logging.warning("⚠️ jsonschema não instalado. Validação desabilitada.")
+
+from api.utils.logger import setup_logger
+
+logger = setup_logger(__name__)
+
+
+class SchemaValidator:
+    """
+    Validador de schema JSON usando jsonschema
+
+    MUDANÇA DO V13.1:
+    - V13.1: Método validate() existia mas não fazia nada
+    - V13.6: Validação real com jsonschema
+    """
+
+    def __init__(self, schema_path: str):
+        """
+        Args:
+            schema_path: Caminho para protocolo_v13_6_schema.json
+        """
+        self.schema_path = schema_path
+        self.schema = self._load_schema(schema_path)
+        self.enabled = HAS_JSONSCHEMA
+
+        if not self.enabled:
+            logger.warning("⚠️ SchemaValidator inicializado SEM jsonschema (validação desabilitada)")
+        else:
+            logger.info(f"✅ SchemaValidator inicializado: {schema_path}")
+
+    def _load_schema(self, schema_path: str) -> Dict[str, Any]:
+        """Carrega schema JSON"""
+        try:
+            with open(schema_path, 'r', encoding='utf-8') as f:
+                return json.load(f)
+        except FileNotFoundError:
+            logger.error(f"❌ Schema não encontrado: {schema_path}")
+            return {}
+        except json.JSONDecodeError as e:
+            logger.error(f"❌ Erro ao parsear schema: {e}")
+            return {}
+
+    def validate(self, data: Dict[str, Any]) -> Tuple[bool, List[str]]:
+        """
+        Valida dados completos contra schema V13.6
+
+        Args:
+            data: JSON completo para validar
+
+        Returns:
+            (is_valid, errors)
+        """
+        if not self.enabled:
+            return True, []
+
+        if not self.schema:
+            logger.warning("⚠️ Schema vazio, validação ignorada")
+            return True, []
+
+        try:
+            validate(instance=data, schema=self.schema)
+            logger.info("✅ Validação completa: SUCESSO")
+            return True, []
+
+        except ValidationError as e:
+            error_msg = f"{e.message} (campo: {'.'.join(map(str, e.path))})"
+            logger.warning(f"⚠️ Validação falhou: {error_msg}")
+            return False, [error_msg]
+
+        except Exception as e:
+            logger.error(f"❌ Erro na validação: {e}")
+            return False, [str(e)]
+
+    def validate_partial(
+        self, 
+        data: Dict[str, Any],
+        partial_schema_path: str
+    ) -> Tuple[bool, List[str]]:
+        """
+        Valida dados parciais contra schema de um especialista
+
+        Args:
+            data: JSON parcial (output de um especialista)
+            partial_schema_path: Caminho para schema parcial (ex: fase1_segmentacao.json)
+
+        Returns:
+            (is_valid, errors)
+        """
+        if not self.enabled:
+            return True, []
+
+        try:
+            with open(partial_schema_path, 'r', encoding='utf-8') as f:
+                partial_schema = json.load(f)
+
+            validate(instance=data, schema=partial_schema)
+            logger.debug(f"✅ Validação parcial ({partial_schema_path}): SUCESSO")
+            return True, []
+
+        except FileNotFoundError:
+            logger.warning(f"⚠️ Schema parcial não encontrado: {partial_schema_path}")
+            return True, []  # Ignorar se schema não existe
+
+        except ValidationError as e:
+            error_msg = f"{e.message} (campo: {'.'.join(map(str, e.path))})"
+            logger.warning(f"⚠️ Validação parcial falhou: {error_msg}")
+            return False, [error_msg]
+
+        except Exception as e:
+            logger.error(f"❌ Erro na validação parcial: {e}")
+            return False, [str(e)]
+
+    def validate_required_fields(self, data: Dict[str, Any]) -> Tuple[bool, List[str]]:
+        """
+        Valida apenas campos obrigatórios (verificação rápida)
+
+        Returns:
+            (is_valid, missing_fields)
+        """
+        if not self.schema:
+            return True, []
+
+        required_fields = self.schema.get('required', [])
+        missing = []
+
+        for field in required_fields:
+            if field not in data:
+                missing.append(field)
+
+        if missing:
+            logger.warning(f"⚠️ Campos obrigatórios faltando: {missing}")
+            return False, missing
+
+        return True, []
+
+    def get_schema_info(self) -> Dict[str, Any]:
+        """Retorna informações sobre o schema"""
+        if not self.schema:
+            return {}
+
+        return {
+            'title': self.schema.get('title', 'Unknown'),
+            'version': self.schema.get('description', '').split('DATA:')[-1].strip() if 'DATA:' in self.schema.get('description', '') else 'Unknown',
+            'required_fields': self.schema.get('required', []),
+            'total_properties': len(self.schema.get('properties', {}))
+        }

processors/processor_manager.py

@@ -1,33 +1,35 @@
+##PARA.AI/processors/processor_manager.py
 """
-Processor Manager - Gerencia os 9 especialistas com batch e paralelo
+Processor Manager - REFATORADO para usar PipelineOrchestrator V13.6
+MUDANÇA: Este arquivo agora é apenas um wrapper para manter compatibilidade
 """
 import logging
-import asyncio
+import yaml
 from typing import Dict, Any, List, Optional
-from datetime import datetime
 
+from core.orchestrator import PipelineOrchestrator
+from core.validator import SchemaValidator
+
+# Importar especialistas (mantidos do V13.1, serão refatorados na FASE 2)
 from processors.processor_metadados import ProcessorMetadados
 from processors.processor_segmentacao import ProcessorSegmentacao
-from processors.processor_transcricao import ProcessorTranscricao
-from processors.processor_decisao import ProcessorDecisao
+from processors.processor_relatorio import ProcessorRelatorio
 from processors.processor_fundamentacao import ProcessorFundamentacao
-from processors.processor_contexto import ProcessorContexto
+from processors.processor_decisao import ProcessorDecisao
 from processors.processor_arquivo import ProcessorArquivo
-from processors.processor_relatorio import ProcessorRelatorio
-from processors.processor_auditoria import ProcessorAuditoria
+from processors.processor_contexto import ProcessorContexto
 
 logger = logging.getLogger(__name__)
 
 
 class ProcessorManager:
     """
-    Gerenciador dos 9 especialistas
-
-    MUDANÇAS:
-    - Todos os especialistas usam LLM real
-    - Suporte a batch processing
-    - Execução paralela otimizada
-    - Sem simulações
+    Gerenciador de processadores - WRAPPER para PipelineOrchestrator
+    
+    MUDANÇA V13.6:
+    - Este arquivo agora delega para PipelineOrchestrator
+    - Mantém compatibilidade com código existente
+    - Especialistas serão migrados gradualmente para novo formato
     """
 
     def __init__(self, llm_manager, max_workers: int = 3):
@@ -39,22 +41,30 @@ class ProcessorManager:
         self.llm_manager = llm_manager
         self.max_workers = max_workers
 
-        # Inicializar os 9 especialistas com LLM Manager
-        self.processors = {
-            1: ProcessorMetadados(llm_manager),
-            2: ProcessorSegmentacao(llm_manager),
-            3: ProcessorTranscricao(llm_manager),
-            4: ProcessorDecisao(llm_manager),
-            5: ProcessorFundamentacao(llm_manager),
-            6: ProcessorContexto(llm_manager),
-            7: ProcessorArquivo(llm_manager),
-            8: ProcessorRelatorio(llm_manager),
-            9: ProcessorAuditoria(llm_manager)
+        # Carregar configuração da pipeline
+        with open('config/pipeline_config.yaml', 'r', encoding='utf-8') as f:
+            self.pipeline_config = yaml.safe_load(f)
+
+        # Inicializar especialistas (mapeamento temporário V13.1 → V13.6)
+        self.specialists = {
+            1: ProcessorSegmentacao(llm_manager),      # Segmentador
+            2: ProcessorMetadados(llm_manager),        # Metadados
+            3: ProcessorContexto(llm_manager),         # Classificador (renomear depois)
+            4: ProcessorRelatorio(llm_manager),        # Relatório
+            5: ProcessorFundamentacao(llm_manager),    # Fundamentação
+            6: ProcessorDecisao(llm_manager),          # Dispositivo
+            7: ProcessorArquivo(llm_manager),          # Arquivista
         }
 
+        # Inicializar orquestrador V13.6
+        self.orchestrator = PipelineOrchestrator(
+            config=self.pipeline_config,
+            llm_manager=llm_manager,
+            specialists=self.specialists
+        )
+
         logger.info(
-            f"✅ ProcessorManager: 9 especialistas inicializados "
-            f"(max_workers={max_workers})"
+            f"✅ ProcessorManager V13.6: Orquestração refatorada com {len(self.specialists)} especialistas"
         )
 
     async def process_acordao_sequential(
@@ -63,188 +73,52 @@ class ProcessorManager:
         specialist_ids: Optional[List[int]] = None
     ) -> Dict[str, Any]:
         """
-        Processa acórdão sequencialmente (um especialista por vez)
-
-        Args:
-            acordao_data: Dados do acórdão
-            specialist_ids: IDs dos especialistas a executar (None = todos)
-
-        Returns:
-            Resultado consolidado
+        Processa acórdão sequencialmente (compatibilidade V13.1)
+        
+        MUDANÇA: Delega para PipelineOrchestrator
         """
-        if specialist_ids is None:
-            specialist_ids = list(self.processors.keys())
-
-        results = []
-        start_time = datetime.now()
-
-        logger.info(
-            f"🔄 Processamento SEQUENCIAL: {len(specialist_ids)} especialistas"
-        )
-
-        for spec_id in specialist_ids:
-            processor = self.processors.get(spec_id)
-            if processor:
-                logger.info(f"  ⚙️ Executando especialista {spec_id}: {processor.specialist_name}")
-                try:
-                    result = await processor.process(acordao_data)
-                    results.append(result)
-                    logger.info(
-                        f"  ✅ {processor.specialist_name} concluído em "
-                        f"{result.get('execution_time', 0):.2f}s"
-                    )
-                except Exception as e:
-                    logger.error(f"  ❌ Erro no especialista {spec_id}: {e}")
-                    results.append({
-                        "specialist_id": spec_id,
-                        "status": "error",
-                        "error": str(e)
-                    })
-
-        execution_time = (datetime.now() - start_time).total_seconds()
-
-        # Contar sucessos e falhas
-        successful = sum(1 for r in results if r.get('status') != 'error')
-        failed = len(results) - successful
-
-        logger.info(
-            f"✅ Processamento concluído: {successful} sucessos, "
-            f"{failed} falhas em {execution_time:.2f}s"
+        logger.info("🔄 process_acordao_sequential() → delegando para PipelineOrchestrator")
+        
+        # Se specialist_ids fornecido, processar apenas fases específicas
+        if specialist_ids:
+            fase_inicial = min(specialist_ids)
+            fase_final = max(specialist_ids)
+        else:
+            fase_inicial = 1
+            fase_final = 6
+        
+        return await self.orchestrator.process_acordao(
+            acordao_bruto=acordao_data,
+            fase_inicial=fase_inicial,
+            fase_final=fase_final
         )
 
-        return {
-            "acordao_id": acordao_data.get('acordao_id', 'unknown'),
-            "status": "success" if failed == 0 else "partial",
-            "mode": "sequential",
-            "specialists_results": results,
-            "statistics": {
-                "total": len(results),
-                "successful": successful,
-                "failed": failed
-            },
-            "execution_time": execution_time,
-            "timestamp": datetime.now().isoformat()
-        }
-
     async def process_acordao_parallel(
         self,
         acordao_data: Dict[str, Any],
         specialist_ids: Optional[List[int]] = None
     ) -> Dict[str, Any]:
         """
-        Processa acórdão em paralelo (todos os especialistas ao mesmo tempo)
-
-        Args:
-            acordao_data: Dados do acórdão
-            specialist_ids: IDs dos especialistas a executar (None = todos)
-
-        Returns:
-            Resultado consolidado
+        Processa acórdão em paralelo (compatibilidade V13.1)
+        
+        MUDANÇA: Paralelo é controlado por fase no V13.6
         """
-        if specialist_ids is None:
-            specialist_ids = list(self.processors.keys())
-
-        start_time = datetime.now()
-
-        logger.info(
-            f"⚡ Processamento PARALELO: {len(specialist_ids)} especialistas"
-        )
-
-        # Criar tasks para execução paralela
-        tasks = []
-        for spec_id in specialist_ids:
-            processor = self.processors.get(spec_id)
-            if processor:
-                logger.info(f"  📤 Agendando especialista {spec_id}: {processor.specialist_name}")
-                tasks.append(processor.process(acordao_data))
-
-        # Executar todas as tasks em paralelo
-        results = await asyncio.gather(*tasks, return_exceptions=True)
-
-        # Processar resultados e exceções
-        processed_results = []
-        for i, result in enumerate(results):
-            if isinstance(result, Exception):
-                logger.error(f"  ❌ Exceção no especialista {specialist_ids[i]}: {result}")
-                processed_results.append({
-                    "specialist_id": specialist_ids[i],
-                    "status": "error",
-                    "error": str(result)
-                })
-            else:
-                processed_results.append(result)
-
-        execution_time = (datetime.now() - start_time).total_seconds()
-
-        # Contar sucessos e falhas
-        successful = sum(1 for r in processed_results if r.get('status') != 'error')
-        failed = len(processed_results) - successful
-
-        logger.info(
-            f"✅ Processamento paralelo concluído: {successful} sucessos, "
-            f"{failed} falhas em {execution_time:.2f}s"
+        logger.info("🔄 process_acordao_parallel() → delegando para PipelineOrchestrator")
+        
+        return await self.orchestrator.process_acordao(
+            acordao_bruto=acordao_data,
+            fase_inicial=1,
+            fase_final=6
         )
 
-        return {
-            "acordao_id": acordao_data.get('acordao_id', 'unknown'),
-            "status": "success" if failed == 0 else "partial",
-            "mode": "parallel",
-            "specialists_results": processed_results,
-            "statistics": {
-                "total": len(processed_results),
-                "successful": successful,
-                "failed": failed
-            },
-            "execution_time": execution_time,
-            "timestamp": datetime.now().isoformat()
-        }
-
-    async def process_acordao_batch(
-        self,
-        acordao_data: Dict[str, Any],
-        specialist_ids: Optional[List[int]] = None,
-        batch_id: Optional[str] = None
-    ) -> Dict[str, Any]:
-        """
-        Processa acórdão usando batch processing (via Groq Batch API)
-
-        Args:
-            acordao_data: Dados do acórdão
-            specialist_ids: IDs dos especialistas a executar (None = todos)
-            batch_id: ID do batch (gerado automaticamente se None)
-
-        Returns:
-            Resultado consolidado
-        """
-        if specialist_ids is None:
-            specialist_ids = list(self.processors.keys())
-
-        if batch_id is None:
-            batch_id = f"batch_{acordao_data.get('acordao_id', 'unknown')}_{int(datetime.now().timestamp())}"
-
-        start_time = datetime.now()
-
-        logger.info(
-            f"📦 Processamento BATCH: {len(specialist_ids)} especialistas "
-            f"(batch_id={batch_id})"
-        )
-
-        # TODO: Implementar batch processing
-        # Por enquanto, usar paralelo como fallback
-        logger.warning("⚠️ Batch processing ainda não implementado, usando paralelo")
-        return await self.process_acordao_parallel(acordao_data, specialist_ids)
-
     def get_processor(self, specialist_id: int):
-        """Obtém processador específico"""
-        return self.processors.get(specialist_id)
+        """Retorna especialista específico (compatibilidade V13.1)"""
+        return self.specialists.get(specialist_id)
 
     def get_all_processors(self) -> Dict[int, Any]:
-        """Retorna todos os processadores"""
-        return self.processors
+        """Retorna todos os especialistas (compatibilidade V13.1)"""
+        return self.specialists
 
-    def get_processors_info(self) -> List[Dict[str, Any]]:
-        """Retorna informações sobre todos os processadores"""
-        return [
-            processor.info
-            for processor in self.processors.values()
-        ]
+    def get_processors_info(self) -> Dict[str, Any]:
+        """Retorna informações sobre todos os especialistas"""
+        return self.orchestrator.get_pipeline_status()